import os import time import logging import datetime 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", "") 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")) # ----------------------------------------------------------------------------- # 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) # ----------------------------------------------------------------------------- # 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: global _token, _token_ts with _token_lock: now = time.time() if _token and (now - _token_ts) < TOKEN_TTL: return _token # fetch a new token log.info("Requesting new Beaver Habits 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]: return {"Accept": "application/json", "Authorization": f"Bearer {get_token()}"} # ----------------------------------------------------------------------------- # Utility: build iCalendar VTODO # ----------------------------------------------------------------------------- def make_vtodo(uid: str, summary: str, completed: bool) -> bytes: cal = Calendar() cal.add("prodid", "-//CalDAV⇆BeaverHabits Proxy//EN") cal.add("version", "2.0") todo = Todo() todo.add("uid", uid) todo.add("summary", summary) now = datetime.datetime.now(datetime.timezone.utc) todo.add("dtstamp", now) todo.add("due", now + datetime.timedelta(days=1)) todo.add("status", "COMPLETED" if completed else "NEEDS-ACTION") cal.add_component(todo) return cal.to_ical() def to_bytes(elem: ET.Element) -> bytes: return ET.tostring(elem, encoding="utf-8", xml_declaration=True) # ----------------------------------------------------------------------------- # Helper: fetch habits + today's completion from Beaver # ----------------------------------------------------------------------------- def fetch_tasks() -> dict[str, dict]: """Return a dict of {habit_id: {SUMMARY, STATUS, DTSTAMP, DUE}}.""" 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 {} today = datetime.date.today().strftime("%d-%m-%Y") out: dict[str, dict] = {} for h in habits: hid = h["id"] name = h["name"] # check completion for today try: params = { "date_fmt": "%d-%m-%Y", "date_start": today, "date_end": today, "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 in done_list except Exception: log.warning("Failed to get completions for %s", hid) completed = False out[hid] = { "SUMMARY": name, "STATUS": "COMPLETED" if completed else "NEEDS-ACTION", "DTSTAMP": datetime.datetime.now(datetime.timezone.utc), "DUE": datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(days=1), } 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 = datetime.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)