306 lines
No EOL
12 KiB
Python
306 lines
No EOL
12 KiB
Python
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/<uid>.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) |