364 lines
No EOL
13 KiB
Python
364 lines
No EOL
13 KiB
Python
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": <name>, "STATUS": <flag> } }
|
||
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/<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 = 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) |