initial commit
This commit is contained in:
commit
cc1dc2c4d8
4 changed files with 522 additions and 0 deletions
306
app.py
Normal file
306
app.py
Normal file
|
|
@ -0,0 +1,306 @@
|
|||
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)
|
||||
Loading…
Add table
Add a link
Reference in a new issue