habitica-xp/habitica_sync.py

#!/usr/bin/env python

import calendar
import datetime
import os
import requests
import yaml

from dateutil.relativedelta import relativedelta, TH, FR, SA, SU, MO, TU, WE
from recurrent.event_parser import RecurringEvent

# --- Configuration ---
USER_ID = os.getenv("HABITICA_USER_ID")
API_TOKEN = os.getenv("HABITICA_API_TOKEN")
YAML_FILE_PATH = "my_tasks.yaml"

BASE_URL = "https://habitica.com/api/v3"
HEADERS = {
    "x-api-user": USER_ID,
    "x-api-key": API_TOKEN,
    "x-client": f"{USER_ID}-PythonTaskSync",
    "Content-Type": "application/json",
}

WEEKDAYS = {
    "Monday": MO,
    "Tuesday": TU,
    "Wednesday": WE,
    "Thursday": TH,
    "Friday": FR,
    "Saturday": SA,
    "Sunday": SU,
}

def priorty(s):
    if not s:
        return 1.5
    elif isinstance(s, str):
        s = s.lower()
        if s.startswith("t"):
            return 0.1
        elif s.startswith("e"):
            return 1
        elif s.startswith("h"):
            return 2
        else:
            return 1.5
    else:
        return s

# --- Recurrence Engine ---
def is_task_due_today(task_def, check_date=None):
    """Evaluates if a YAML task definition is due on check_date (defaults to today)."""
    if check_date is None:
        check_date = datetime.date.today()

    rec = task_def.get("recurrence", {})
    rec_type = rec.get("type")
    print(f"Analyzing {rec}")

    if rec_type =="weekly":
        day_name = check_date.strftime("%A")
        return day_name in rec.get("days", [])

    elif rec_type =="day_of_month":
        target_day = rec.get("day")
        if target_day > 0:
            return check_date.day == target_day
        elif target_day == -1:
            last_day = calendar.monthrange(check_date.year, check_date.month)[1]
            return check_date.day == last_day

    elif rec_type =="relative_day":
        weekday_str = rec.get("weekday")
        ordinal = rec.get("ordinal", 1)
        first_of_month = check_date.replace(day=1)
        target_weekday = WEEKDAYS[weekday_str]
        calculated_date = first_of_month + relativedelta(day=1, weekday=target_weekday(ordinal))
        return check_date == calculated_date

    elif rec_type == "interval":
        interval = rec.get("interval", 2)
        start_str = rec.get("start")
        anchor = (
            datetime.date.fromisoformat(start_str)
            if start_str
            else datetime.date(2000, 1, 1)  # fixed default epoch
        )
        delta = (check_date - anchor).days
        return delta >= 0 and delta % interval == 0

    elif rec_type.startswith("flex"):
        r = RecurringEvent()
        target_date = r.parse(rec.get("day"))
        return check_date == target_date.date()

    return False


# --- Habitica API Wrappers ---
def get_existing_tasks():
    """Fetch current tasks from Habitica."""
    url = f"{BASE_URL}/tasks/user"
    try:
        print(f"headers: {HEADERS}")
        response = requests.get(url, headers=HEADERS)
        response.raise_for_status()
        res_json = response.json()
        if res_json.get("success"):
            return res_json.get("data", [])
    except Exception as err:
        print(f"Failed to fetch tasks from Habitica: {err}")
    return []


def create_habitica_task(task_def):
    """Create a task in Habitica via POST /api/v3/tasks/user."""
    url = f"{BASE_URL}/tasks/user"

    payload = {
        "text": task_def.get("text"),
        "type": task_def.get("type", "todo"),
        "priority": priorty(task_def.get("priority")),
    }
    if task_def.get("notes"):
        payload["notes"] = task_def.get("notes")
    if task_def.get("checklist"):
        payload["checklist"] = [{"text": t, "completed": False} for t in task_def.get("checklist")]

    print(payload)
    try:
        response = requests.post(url, headers=HEADERS, json=payload)
        response.raise_for_status()
        res_json = response.json()
        if res_json.get("success"):
            print(f" -> SUCCESSFULLY CREATED: '{payload['text']}'")
            return res_json.get("data")
    except Exception as err:
        print(f" -> ERROR creating '{payload['text']}': {err}")


# --- Main Logic ---
def sync_tasks():
    today = datetime.date.today()
    print(f"=== Syncing Habitica Tasks for {today} ===")

    # 1. Load YAML rules
    try:
        with open(YAML_FILE_PATH, "r", encoding="utf-8") as f:
            yaml_data = yaml.safe_load(f) or {}
            defined_tasks = yaml_data.get("tasks", [])
    except Exception as err:
        print(f"Could not read {YAML_FILE_PATH}: {err}")
        return
    print(f"Read YAML file: {YAML_FILE_PATH}")

    # 2. Filter tasks due today
    due_tasks = [task for task in defined_tasks if is_task_due_today(task, today)]
    print(f"Found {len(due_tasks)} task(s) due today in YAML.")

    if not due_tasks:
        print("No tasks due today. Exiting.")
        return

    # 3. Fetch existing Habitica tasks to avoid duplicates
    existing_tasks = get_existing_tasks()
    existing_titles = {t.get("text", "").strip().lower() for t in existing_tasks}

    # 4. Create missing tasks
    for task in due_tasks:
        title = task.get("text", "").strip()

        if title.lower() in existing_titles:
            print(f" -> SKIPPED (Already exists): '{title}'")
        else:
            create_habitica_task(task)


if __name__ == "__main__":
    sync_tasks()