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("TIMEZONE", "UTC")) # ----------------------------------------------------------------------------- # 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: 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: 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)