habitica-xp/habitica_sync.py

#!/usr/bin/env python

import calendar
import os
import requests
import yaml

from datetime import datetime, date, time, timedelta
from dateutil.relativedelta import relativedelta, TH, FR, SA, SU, MO, TU, WE
from dateutil.rrule import rrulestr
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 = date.today()

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

    if not rec_type:
        return False

    elif 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 = (
        #     date.fromisoformat(start_str)
        #     if start_str
        #     else date(2000, 1, 1)  # fixed default epoch
        # )
        # delta = (check_date - anchor).days
        delta = (check_date - start_str).days
        return delta >= 0 and delta % interval == 0

    elif rec_type.startswith("flex"):
        event_parser = RecurringEvent()
        rule_date = event_parser.parse(rec.get("day"))
        rule_str = event_parser.get_RFC_rrule()
        if rule_str:
            rule = rrulestr(rule_str)
            # Define the start and end bounds for today's date
            today_start = datetime.combine(date.today(), time.min)
            today_end = today_start + timedelta(days=1)

            # Check if today's window contains a rule occurrence
            return len(rule.between(today_start, today_end, inc=True)) > 0

    return False


# --- Habitica API Wrappers ---
def get_existing_tasks():
    """Fetch current tasks from Habitica."""
    url = f"{BASE_URL}/tasks/user"
    try:
        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 = 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__":
    if not USER_ID and not API_TOKEN:
        print("Need to set the USER_ID and API_TOKEN variables.")
    else:
        sync_tasks()