import os import time import logging from datetime import datetime, timedelta, date from zoneinfo import ZoneInfo import threading from flask import Flask, request, Response, abort, jsonify from icalendar import Calendar, Todo import xml.etree.ElementTree as ET import requests # ----------------------------------------------------------------------------- # Configuration via ENV # ----------------------------------------------------------------------------- BEAVER_URL = os.getenv("BEAVERHABITS_URL", "http://beaverhabits.lan") USERNAME = os.getenv("BEAVERHABITS_USERNAME", "") PASSWORD = os.getenv("BEAVERHABITS_PASSWORD", "") # How long to reuse a token (secs) TOKEN_TTL = int(os.getenv("BEAVERHABITS_TOKEN_TTL", "3600")) TZ = ZoneInfo(os.getenv("TZ", "EST")) # ----------------------------------------------------------------------------- # Logging # ----------------------------------------------------------------------------- logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s", ) log = logging.getLogger("caldav-proxy") # ----------------------------------------------------------------------------- # Flask app & XML namespaces # ----------------------------------------------------------------------------- app = Flask(__name__) NS = {"D": "DAV:", "C": "urn:ietf:params:xml:ns:caldav"} for p, uri in NS.items(): ET.register_namespace(p, uri) def to_bytes(elem: ET.Element) -> bytes: """Serialize an ElementTree Element to UTF-8 XML with declaration.""" return ET.tostring(elem, encoding="utf-8", xml_declaration=True) # ----------------------------------------------------------------------------- # Beaver Habits token cache with TTL # ----------------------------------------------------------------------------- _token_lock = threading.Lock() _token: str | None = None _token_ts: float = 0.0 _session = requests.Session() def get_token() -> str | None: """ Return a valid access token, or None if we're in no-auth mode. """ # 1) If either USERNAME or PASSWORD is missing, we are in "no-auth" mode. if not USERNAME or not PASSWORD: return None # 2) Otherwise, normal token+TTL caching: global _token, _token_ts now = time.time() with _token_lock: if _token and now - _token_ts < TOKEN_TTL: return _token # fetch a new token r = _session.post( f"{BEAVER_URL}/auth/login", headers={ "Accept": "application/json", "Content-Type": "application/x-www-form-urlencoded", }, data={ "grant_type": "password", "username": USERNAME, "password": PASSWORD, }, timeout=5, ) r.raise_for_status() js = r.json() _token = js["access_token"] _token_ts = now return _token def beaver_headers() -> dict[str,str]: """ Base headers for Beaver Habits API calls. In no-auth mode, we only send Accept: application/json. Otherwise we include the Bearer token. """ headers = { "Accept": "application/json", } token = get_token() if token: headers["Authorization"] = f"Bearer {token}" return headers # ----------------------------------------------------------------------------- # Utility: Datetime functions # ----------------------------------------------------------------------------- def local_now() -> datetime: """ Return “now” as an aware datetime in our configured TZ. """ return datetime.now(TZ) def next_midnight_local(now: datetime) -> datetime: """ Given an aware datetime ‘now’, return tomorrow at 00:00 in the same TZ. """ tomorrow = now.date() + timedelta(days=1) # Year, month, day at midnight return datetime(tomorrow.year, tomorrow.month, tomorrow.day, tzinfo=TZ) # ----------------------------------------------------------------------------- # Utility: build iCalendar VTODO # ----------------------------------------------------------------------------- def make_vtodo(uid: str, summary: str, completed: bool) -> bytes: """ Construct a VCALENDAR/VTODO where: - DTSTAMP = local_now() - DUE = next_midnight_local(local_now()) """ now = local_now() due = next_midnight_local(now) cal = Calendar() cal.add("prodid", "-//CalDAV⇆BeaverHabits Proxy//EN") cal.add("version", "2.0") todo = Todo() todo.add("uid", uid) todo.add("summary", summary) todo.add("dtstamp", now) # todo.add("due", due) # causes problems todo.add("status", "COMPLETED" if completed else "NEEDS-ACTION") cal.add_component(todo) return cal.to_ical() # ----------------------------------------------------------------------------- # Helper: fetch habits + today's completion from Beaver # ----------------------------------------------------------------------------- def fetch_tasks() -> dict[str, dict]: """ Fetch habit list + today's completion from Beaver Habits. Returns a dict: { habit_id: { "SUMMARY": , "STATUS": } } DATES are handled later by make_vtodo(). """ try: r = _session.get(f"{BEAVER_URL}/api/v1/habits", headers=beaver_headers(), timeout=5) r.raise_for_status() habits = r.json() except Exception as e: log.error("Failed to list habits: %s", e) return {} # Use the same date format Beaver expects today_str = local_now().strftime("%d-%m-%Y") out: dict[str, dict] = {} for h in habits: hid = h["id"] name = h["name"] try: params = { "date_fmt": "%d-%m-%Y", "date_start": today_str, "date_end": today_str, "sort": "asc", } rc = _session.get( f"{BEAVER_URL}/api/v1/habits/{hid}/completions", headers=beaver_headers(), params=params, timeout=5 ) rc.raise_for_status() done_list = rc.json() completed = today_str in done_list except Exception: log.warning("Failed to fetch completions for habit %s", hid) completed = False out[hid] = { "SUMMARY": name, "STATUS": "COMPLETED" if completed else "NEEDS-ACTION", } return out # ----------------------------------------------------------------------------- # Standard PROPFIND / discovery endpoints # ----------------------------------------------------------------------------- @app.route("/", methods=["PROPFIND"]) def propfind_root(): ms = ET.Element(f"{{{NS['D']}}}multistatus") # current-user-principal r1 = ET.SubElement(ms, f"{{{NS['D']}}}response") ET.SubElement(r1, f"{{{NS['D']}}}href").text = "/" ps1 = ET.SubElement(r1, f"{{{NS['D']}}}propstat") p1 = ET.SubElement(ps1, f"{{{NS['D']}}}prop") cup = ET.SubElement(p1, f"{{{NS['D']}}}current-user-principal") ET.SubElement(cup, f"{{{NS['D']}}}href").text = "/" ET.SubElement(ps1, f"{{{NS['D']}}}status").text = "HTTP/1.1 200 OK" # calendar-home-set r2 = ET.SubElement(ms, f"{{{NS['D']}}}response") ET.SubElement(r2, f"{{{NS['D']}}}href").text = "/" ps2 = ET.SubElement(r2, f"{{{NS['D']}}}propstat") p2 = ET.SubElement(ps2, f"{{{NS['D']}}}prop") chs = ET.SubElement(p2, f"{{{NS['C']}}}calendar-home-set") ET.SubElement(chs, f"{{{NS['D']}}}href").text = "/caldav/" ET.SubElement(ps2, f"{{{NS['D']}}}status").text = "HTTP/1.1 200 OK" return Response(to_bytes(ms), status=207, mimetype="application/xml") @app.route("/caldav/", methods=["PROPFIND"]) def propfind_caldav(): ms = ET.Element(f"{{{NS['D']}}}multistatus") # home collection r1 = ET.SubElement(ms, f"{{{NS['D']}}}response") ET.SubElement(r1, f"{{{NS['D']}}}href").text = "/caldav/" ps1 = ET.SubElement(r1, f"{{{NS['D']}}}propstat") p1 = ET.SubElement(ps1, f"{{{NS['D']}}}prop") rt1 = ET.SubElement(p1, f"{{{NS['D']}}}resourcetype") ET.SubElement(rt1, f"{{{NS['D']}}}collection") ET.SubElement(ps1, f"{{{NS['D']}}}status").text = "HTTP/1.1 200 OK" # our single tasks calendar href="/caldav/tasks/" r2 = ET.SubElement(ms, f"{{{NS['D']}}}response") ET.SubElement(r2, f"{{{NS['D']}}}href").text = href ps2 = ET.SubElement(r2, f"{{{NS['D']}}}propstat") p2 = ET.SubElement(ps2, f"{{{NS['D']}}}prop") rt2 = ET.SubElement(p2, f"{{{NS['D']}}}resourcetype") ET.SubElement(rt2, f"{{{NS['D']}}}collection") ET.SubElement(rt2, f"{{{NS['C']}}}calendar") sccs=ET.SubElement(p2, f"{{{NS['C']}}}supported-calendar-component-set") ET.SubElement(sccs, f"{{{NS['C']}}}comp", name="VTODO") ET.SubElement(p2, f"{{{NS['D']}}}displayname").text="Tasks" ET.SubElement(ps2,f"{{{NS['D']}}}status").text="HTTP/1.1 200 OK" return Response(to_bytes(ms), status=207, mimetype="application/xml") # ----------------------------------------------------------------------------- # PROPFIND + REPORT + GET/PUT on /caldav/tasks/ # ----------------------------------------------------------------------------- @app.route("/caldav/tasks/", methods=["PROPFIND", "REPORT"]) def tasks_collection(): """ PROPFIND: list resource URLs + etags REPORT: return full calendar-data for each VTODO """ tasks = fetch_tasks() multistatus = ET.Element(f"{{{NS['D']}}}multistatus") # Collection itself (must advertise calendar + VTODO support) r0 = ET.SubElement(multistatus, f"{{{NS['D']}}}response") ET.SubElement(r0, f"{{{NS['D']}}}href").text = "/caldav/tasks/" ps0 = ET.SubElement(r0, f"{{{NS['D']}}}propstat") p0 = ET.SubElement(ps0, f"{{{NS['D']}}}prop") rt0 = ET.SubElement(p0, f"{{{NS['D']}}}resourcetype") ET.SubElement(rt0, f"{{{NS['D']}}}collection") ET.SubElement(rt0, f"{{{NS['C']}}}calendar") sccs0 = ET.SubElement(p0, f"{{{NS['C']}}}supported-calendar-component-set") ET.SubElement(sccs0, f"{{{NS['C']}}}comp", name="VTODO") ET.SubElement(ps0, f"{{{NS['D']}}}status").text="HTTP/1.1 200 OK" for uid, props in tasks.items(): href = f"/caldav/tasks/{uid}.ics" etag = f"\"{uid}-{props['STATUS']}\"" body = make_vtodo(uid, props["SUMMARY"], props["STATUS"]=="COMPLETED") # PROPFIND section if request.method == "PROPFIND": r = ET.SubElement(multistatus, f"{{{NS['D']}}}response") ET.SubElement(r, f"{{{NS['D']}}}href").text = href ps = ET.SubElement(r, f"{{{NS['D']}}}propstat") p = ET.SubElement(ps, f"{{{NS['D']}}}prop") ET.SubElement(p, f"{{{NS['D']}}}getcontentlength").text = str(len(body)) ET.SubElement(p, f"{{{NS['D']}}}getcontenttype").text = "text/calendar; charset=utf-8" ET.SubElement(p, f"{{{NS['D']}}}getetag").text = etag ET.SubElement(ps,f"{{{NS['D']}}}status").text = "HTTP/1.1 200 OK" # REPORT section elif request.method == "REPORT": r = ET.SubElement(multistatus, f"{{{NS['D']}}}response") ET.SubElement(r, f"{{{NS['D']}}}href").text = href ps = ET.SubElement(r, f"{{{NS['D']}}}propstat") p = ET.SubElement(ps, f"{{{NS['D']}}}prop") cd = ET.SubElement(p, f"{{{NS['C']}}}calendar-data") cd.text = body.decode("utf-8") ET.SubElement(p, f"{{{NS['D']}}}getetag").text = etag ET.SubElement(ps,f"{{{NS['D']}}}status").text="HTTP/1.1 200 OK" return Response(to_bytes(multistatus), status=207, mimetype="application/xml") @app.route("/caldav/tasks/.ics", methods=["GET", "PUT"]) def single_task(uid): """ GET: return VTODO PUT: update completion for today in Beaver Habits """ # verify habit exists tasks_map = fetch_tasks() if uid not in tasks_map: abort(404) summary = tasks_map[uid]["SUMMARY"] completed = (tasks_map[uid]["STATUS"] == "COMPLETED") today = date.today() if request.method == "GET": return Response(make_vtodo(uid, summary, completed), mimetype="text/calendar") # PUT: parse STATUS and call Beaver Habits cal = Calendar.from_ical(request.data) new_status = None for comp in cal.walk(): if comp.name == "VTODO": new_status = str(comp.get("STATUS")) break if new_status not in ("NEEDS-ACTION", "COMPLETED"): abort(400, "Unsupported STATUS") done = (new_status == "COMPLETED") try: _session.post( f"{BEAVER_URL}/api/v1/habits/{uid}/completions", headers={**beaver_headers(), "Content-Type": "application/json"}, json={ "date_fmt": "%d-%m-%Y", "date": today.strftime("%d-%m-%Y"), "done": done }, timeout=5 ).raise_for_status() except Exception as e: log.error("Failed to set completion: %s", e) abort(500) return ("", 204) # ----------------------------------------------------------------------------- # Health‐check # ----------------------------------------------------------------------------- @app.route("/healthz", methods=["GET"]) def healthz(): return jsonify({"status": "ok"}), 200 # ----------------------------------------------------------------------------- # Start via WSGI # ----------------------------------------------------------------------------- if __name__ == "__main__": # Development only app.run(host="0.0.0.0", port=8000, debug=True)