250 lines
9.0 KiB
Python
Executable File
250 lines
9.0 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""odysseus-calendar — Unix-style CLI for the calendar feature.
|
|
|
|
Reads the same SQLite the web UI uses (`data/app.db`). Output is JSON on
|
|
stdout, errors on stderr, non-zero exit on failure. Composable:
|
|
|
|
odysseus-calendar list --start 2026-05-01 --end 2026-05-31 \\
|
|
| jq -r '.[] | .dtstart + "\\t" + .summary'
|
|
|
|
odysseus-calendar calendars | jq -r '.[].name'
|
|
|
|
Subcommands:
|
|
list List events in a date range (optionally per-calendar)
|
|
show Show one event by UID
|
|
calendars List configured calendars
|
|
create Create a new event
|
|
delete Delete an event by UID
|
|
"""
|
|
|
|
|
|
from __future__ import annotations
|
|
import sys
|
|
import os, sys
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "_lib"))
|
|
from cli import quiet_logs, emit, fail, common_parser, run, REPO_ROOT as _REPO_ROOT
|
|
quiet_logs()
|
|
|
|
|
|
import argparse
|
|
import json
|
|
import logging
|
|
import os
|
|
import sys
|
|
import uuid
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
def quiet_logs() -> None:
|
|
level_name = os.environ.get("LOG_LEVEL", "WARNING").upper()
|
|
level = getattr(logging, level_name, logging.WARNING)
|
|
root = logging.getLogger()
|
|
root.setLevel(level)
|
|
for handler in root.handlers:
|
|
handler.setLevel(level)
|
|
|
|
|
|
quiet_logs()
|
|
|
|
try:
|
|
from core.database import SessionLocal, CalendarCal, CalendarEvent
|
|
quiet_logs()
|
|
except ModuleNotFoundError as e:
|
|
sys.stderr.write(
|
|
f"error: {e}\nhint: run from repo root with venv active.\n"
|
|
)
|
|
sys.exit(2)
|
|
|
|
|
|
def fail(msg: str, code: int = 1) -> None:
|
|
sys.stderr.write(f"error: {msg}\n")
|
|
sys.exit(code)
|
|
|
|
|
|
def _parse_dt(s: str) -> datetime:
|
|
"""Accept either `YYYY-MM-DD` (treated as midnight local) or full
|
|
ISO 8601 with optional `Z`/offset."""
|
|
if len(s) == 10:
|
|
return datetime.fromisoformat(s + "T00:00:00")
|
|
return datetime.fromisoformat(s.replace("Z", "+00:00"))
|
|
|
|
|
|
def _serialize_event(ev: "CalendarEvent") -> dict:
|
|
return {
|
|
"uid": ev.uid,
|
|
"calendar_id": ev.calendar_id,
|
|
"calendar_name": ev.calendar.name if ev.calendar else "",
|
|
"summary": ev.summary,
|
|
"description": ev.description or "",
|
|
"location": ev.location or "",
|
|
"dtstart": ev.dtstart.isoformat() + ("Z" if ev.is_utc else "") if ev.dtstart else "",
|
|
"dtend": ev.dtend.isoformat() + ("Z" if ev.is_utc else "") if ev.dtend else "",
|
|
"all_day": bool(ev.all_day),
|
|
"is_utc": bool(ev.is_utc),
|
|
"rrule": ev.rrule or "",
|
|
"color": ev.color or "",
|
|
"status": ev.status or "",
|
|
"importance": ev.importance or "",
|
|
"event_type": ev.event_type or "",
|
|
}
|
|
|
|
|
|
# ─── list ────────────────────────────────────────────────────────────
|
|
|
|
def cmd_list(args) -> None:
|
|
"""List events in a date range. Defaults to next 30 days from today."""
|
|
start = _parse_dt(args.start) if args.start else datetime.now()
|
|
end = _parse_dt(args.end) if args.end else (start + timedelta(days=30))
|
|
db = SessionLocal()
|
|
try:
|
|
q = db.query(CalendarEvent).filter(
|
|
CalendarEvent.dtstart >= start,
|
|
CalendarEvent.dtstart < end,
|
|
)
|
|
if args.calendar:
|
|
cal = db.query(CalendarCal).filter(CalendarCal.name == args.calendar).first()
|
|
if not cal:
|
|
fail(f"no calendar named {args.calendar!r}")
|
|
q = q.filter(CalendarEvent.calendar_id == cal.id)
|
|
q = q.order_by(CalendarEvent.dtstart.asc()).limit(args.limit)
|
|
emit([_serialize_event(e) for e in q.all()], args)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
# ─── show ────────────────────────────────────────────────────────────
|
|
|
|
def cmd_show(args) -> None:
|
|
db = SessionLocal()
|
|
try:
|
|
ev = db.get(CalendarEvent, args.uid)
|
|
if not ev:
|
|
fail(f"no event with uid {args.uid!r}")
|
|
emit(_serialize_event(ev), args)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
# ─── calendars ───────────────────────────────────────────────────────
|
|
|
|
def cmd_calendars(args) -> None:
|
|
db = SessionLocal()
|
|
try:
|
|
cals = db.query(CalendarCal).order_by(CalendarCal.name.asc()).all()
|
|
emit([
|
|
{
|
|
"id": c.id,
|
|
"name": c.name,
|
|
"color": c.color or "",
|
|
"source": c.source or "local",
|
|
"event_count": len(c.events),
|
|
} for c in cals
|
|
], args)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
# ─── create ──────────────────────────────────────────────────────────
|
|
|
|
def cmd_create(args) -> None:
|
|
"""Create an event. --start/--end accept YYYY-MM-DD or ISO 8601.
|
|
--calendar selects by name; defaults to the first available."""
|
|
dtstart = _parse_dt(args.start)
|
|
dtend = _parse_dt(args.end) if args.end else (dtstart + timedelta(hours=1))
|
|
db = SessionLocal()
|
|
try:
|
|
cal_q = db.query(CalendarCal)
|
|
if args.calendar:
|
|
cal = cal_q.filter(CalendarCal.name == args.calendar).first()
|
|
if not cal:
|
|
fail(f"no calendar named {args.calendar!r}")
|
|
else:
|
|
cal = cal_q.order_by(CalendarCal.created_at.asc()).first()
|
|
if not cal:
|
|
fail("no calendars exist; create one in the web UI first")
|
|
ev = CalendarEvent(
|
|
uid=str(uuid.uuid4()),
|
|
calendar_id=cal.id,
|
|
summary=args.title,
|
|
description=args.description or "",
|
|
location=args.location or "",
|
|
dtstart=dtstart,
|
|
dtend=dtend,
|
|
all_day=bool(args.all_day),
|
|
is_utc=False,
|
|
importance=args.importance,
|
|
event_type=args.event_type or None,
|
|
)
|
|
db.add(ev)
|
|
db.commit()
|
|
db.refresh(ev)
|
|
emit(_serialize_event(ev), args)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
# ─── delete ──────────────────────────────────────────────────────────
|
|
|
|
def cmd_delete(args) -> None:
|
|
db = SessionLocal()
|
|
try:
|
|
ev = db.get(CalendarEvent, args.uid)
|
|
if not ev:
|
|
fail(f"no event with uid {args.uid!r}")
|
|
snapshot = _serialize_event(ev)
|
|
db.delete(ev)
|
|
db.commit()
|
|
emit({"ok": True, "deleted": snapshot}, args)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
# ─── argparse ────────────────────────────────────────────────────────
|
|
|
|
def _build_parser() -> argparse.ArgumentParser:
|
|
common = argparse.ArgumentParser(add_help=False)
|
|
common.add_argument("--pretty", action="store_true", help="Pretty-print JSON")
|
|
|
|
p = argparse.ArgumentParser(
|
|
prog="odysseus-calendar",
|
|
description="Shell-friendly wrapper around the Odysseus calendar.",
|
|
parents=[common],
|
|
)
|
|
sub = p.add_subparsers(dest="cmd", required=True)
|
|
|
|
pl = sub.add_parser("list", help="list events in a date range", parents=[common])
|
|
pl.add_argument("--start", help="YYYY-MM-DD or ISO datetime (default: today)")
|
|
pl.add_argument("--end", help="YYYY-MM-DD or ISO datetime (default: start + 30 days)")
|
|
pl.add_argument("--calendar", help="filter by calendar name")
|
|
pl.add_argument("--limit", type=int, default=100)
|
|
pl.set_defaults(func=cmd_list)
|
|
|
|
psh = sub.add_parser("show", help="show one event by UID", parents=[common])
|
|
psh.add_argument("uid")
|
|
psh.set_defaults(func=cmd_show)
|
|
|
|
pc = sub.add_parser("calendars", help="list configured calendars", parents=[common])
|
|
pc.set_defaults(func=cmd_calendars)
|
|
|
|
pcr = sub.add_parser("create", help="create an event", parents=[common])
|
|
pcr.add_argument("--title", required=True)
|
|
pcr.add_argument("--start", required=True, help="YYYY-MM-DD or ISO datetime")
|
|
pcr.add_argument("--end", help="YYYY-MM-DD or ISO datetime (default: start + 1h)")
|
|
pcr.add_argument("--calendar", help="calendar name (default: first available)")
|
|
pcr.add_argument("--description", default="")
|
|
pcr.add_argument("--location", default="")
|
|
pcr.add_argument("--all-day", action="store_true")
|
|
pcr.add_argument("--importance", choices=["low", "normal", "high", "critical"], default="normal")
|
|
pcr.add_argument("--event-type", choices=["work", "personal", "health", "travel", "meal", "social", "admin", "other"])
|
|
pcr.set_defaults(func=cmd_create)
|
|
|
|
pd = sub.add_parser("delete", help="delete an event by UID", parents=[common])
|
|
pd.add_argument("uid")
|
|
pd.set_defaults(func=cmd_delete)
|
|
|
|
return p
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(run(_build_parser()))
|