import calendar import csv import io import logging import os import secrets import shutil import time from datetime import date, datetime, timedelta from pathlib import Path from typing import Annotated from urllib.parse import urlencode from uuid import uuid4 import pdfplumber from fastapi import Depends, FastAPI, File, Form, HTTPException, Request, UploadFile from fastapi.responses import HTMLResponse, RedirectResponse, Response from fastapi.templating import Jinja2Templates from pypdf import PdfReader from sqlalchemy import select from sqlalchemy.exc import OperationalError from sqlalchemy.orm import Session from starlette.middleware.sessions import SessionMiddleware from .database import Base, engine, get_db from .models import CallImport, RawEvent, TimeEntry from .schemas import TimeTrackSync, ZedSync from .services import TZ, derive_zed_suggestions, group_commits_by_day, group_entries_by_day, overlaps_accepted, overlaps_entries, parse_timestamp APP_TIMEZONE = os.environ.get("APP_TIMEZONE", "Europe/Vienna") ADMIN_PASSWORD = os.environ.get("ADMIN_PASSWORD", "") SYNC_API_TOKEN = os.environ.get("SYNC_API_TOKEN", "") UPLOAD_DIR = Path(os.environ.get("UPLOAD_DIR", "./uploads")) app = FastAPI(title="Time Track Server", version="0.1.0") app.add_middleware(SessionMiddleware, secret_key=os.environ.get("SESSION_SECRET", "development-only-change-me"), https_only=False) templates = Jinja2Templates(directory=str(Path(__file__).parent / "templates")) @app.on_event("startup") def startup() -> None: if not ADMIN_PASSWORD or not SYNC_API_TOKEN: raise RuntimeError("ADMIN_PASSWORD and SYNC_API_TOKEN must be configured") # Postgres may briefly restart or lag behind its health check at deploy time. for attempt in range(1, 11): try: Base.metadata.create_all(bind=engine) break except OperationalError as exc: if attempt == 10: raise logging.getLogger("uvicorn.error").warning( "Database not ready (attempt %s/10): %s", attempt, exc.orig ) time.sleep(2) UPLOAD_DIR.mkdir(parents=True, exist_ok=True) def require_api(request: Request) -> None: authorization = request.headers.get("Authorization", "") if not authorization.startswith("Bearer ") or not secrets.compare_digest(authorization[7:], SYNC_API_TOKEN): raise HTTPException(status_code=401, detail="Valid bearer token required") class LoginRequired(Exception): """A browser endpoint was accessed without an authenticated session.""" @app.exception_handler(LoginRequired) def login_required_handler(request: Request, exc: LoginRequired): # Browsers expect to be sent to the login page, not shown a bare 401. return RedirectResponse("/login", status_code=303) def require_web(request: Request) -> None: if not request.session.get("authenticated"): raise LoginRequired() def insert_raw_event(db: Session, source: str, device_id: str, external_id: str, project_slug: str | None, occurred_at: datetime | None, payload: dict) -> bool: existing = db.scalar(select(RawEvent.id).where( RawEvent.source == source, RawEvent.device_id == device_id, RawEvent.external_id == external_id, )) if existing is not None: return False db.add(RawEvent(source=source, device_id=device_id, external_id=external_id, project_slug=project_slug, occurred_at=occurred_at, payload=payload)) return True @app.get("/healthz") def healthz() -> dict: return {"status": "ok", "timezone": APP_TIMEZONE} @app.get("/favicon.ico", include_in_schema=False) def favicon(): return Response(status_code=204) @app.get("/login", response_class=HTMLResponse) def login_page(request: Request): return templates.TemplateResponse(request, "login.html", {"error": None}) @app.post("/login", response_class=HTMLResponse) def login(request: Request, password: Annotated[str, Form()]): if not secrets.compare_digest(password, ADMIN_PASSWORD): return templates.TemplateResponse(request, "login.html", {"error": "Invalid password"}, status_code=401) request.session["authenticated"] = True return RedirectResponse("/", status_code=303) @app.post("/logout") def logout(request: Request): request.session.clear() return RedirectResponse("/login", status_code=303) @app.post("/api/v1/sync/time-track", dependencies=[Depends(require_api)]) def sync_time_track(payload: TimeTrackSync, db: Session = Depends(get_db)) -> dict: inserted = 0 for item in payload.entries: start_at, end_at = parse_timestamp(item.start), parse_timestamp(item.end) if end_at < start_at: raise HTTPException(422, "Entry end must not precede start") if insert_raw_event(db, "time_track", payload.device_id, item.id, None, start_at, item.model_dump()): db.add(TimeEntry(source="time_track", device_id=payload.device_id, external_id=item.id, kind="tracked", status="accepted", start_at=start_at, end_at=end_at, duration_seconds=item.duration, task=item.task)) inserted += 1 db.commit() return {"received": len(payload.entries), "inserted": inserted} @app.post("/api/v1/sync/zed-heartbeats", dependencies=[Depends(require_api)]) def sync_zed_heartbeats(payload: ZedSync, db: Session = Depends(get_db)) -> dict: inserted = 0 for item in payload.records: occurred_at = parse_timestamp(item.timestamp) record = item.model_dump(by_alias=True) if insert_raw_event(db, "zed_heartbeat", payload.device_id, item.source_id, payload.project_slug, occurred_at, record): inserted += 1 db.flush() blocks = derive_zed_suggestions(db, payload.device_id, payload.project_slug) db.commit() return {"received": len(payload.records), "inserted": inserted, "suggestion_blocks": blocks} @app.post("/api/v1/sync/zed-commits", dependencies=[Depends(require_api)]) def sync_zed_commits(payload: ZedSync, db: Session = Depends(get_db)) -> dict: inserted = 0 for item in payload.records: occurred_at = parse_timestamp(item.timestamp) if insert_raw_event(db, "zed_commit", payload.device_id, item.source_id, payload.project_slug, occurred_at, item.model_dump(by_alias=True)): inserted += 1 db.commit() return {"received": len(payload.records), "inserted": inserted} @app.get("/", response_class=HTMLResponse) def index(request: Request, start: str | None = None, end: str | None = None, db: Session = Depends(get_db)): require_web(request) start_date = datetime.fromisoformat(start).replace(tzinfo=TZ) if start else None end_date = datetime.fromisoformat(end).replace(tzinfo=TZ).replace(hour=23, minute=59, second=59) if end else None entries = list(db.scalars(select(TimeEntry))) entries = [ entry for entry in entries if (start_date is None or entry.start_at >= start_date) and (end_date is None or entry.start_at <= end_date) ] commits = list(db.scalars(select(RawEvent).where(RawEvent.source == "zed_commit").order_by(RawEvent.occurred_at))) commits = [ commit for commit in commits if commit.occurred_at is not None and (start_date is None or commit.occurred_at >= start_date) and (end_date is None or commit.occurred_at <= end_date) ] entry_days = group_entries_by_day(entries) commit_days = group_commits_by_day(commits) days = [] for day_date in sorted(set(entry_days) | set(commit_days), reverse=True): day_entries = entry_days.get(day_date, []) days.append({ "date": day_date, "entries": day_entries, "accepted": sum(e.duration_seconds for e in day_entries if e.status == "accepted"), "suggested": sum(e.duration_seconds for e in day_entries if e.status == "suggested"), "commits": commit_days.get(day_date, []), }) return templates.TemplateResponse(request, "index.html", { "days": days, "accepted": sum(day["accepted"] for day in days), "suggested": sum(day["suggested"] for day in days), "start": start or "", "end": end or "", "timezone": APP_TIMEZONE, "display_tz": TZ, "format_seconds": lambda seconds: f"{seconds // 3600}h {(seconds % 3600) // 60:02d}m", "overlaps_accepted": lambda entry: overlaps_accepted(db, entry), }) def index_redirect(start: str | None, end: str | None) -> RedirectResponse: params = {name: value for name, value in (("start", start), ("end", end)) if value} suffix = "?" + urlencode(params) if params else "" return RedirectResponse("/" + suffix, status_code=303) @app.post("/entries/manual") def create_manual(request: Request, start_at: Annotated[str, Form()], end_at: Annotated[str, Form()], task: Annotated[str, Form()] = "", project_slug: Annotated[str, Form()] = "", db: Session = Depends(get_db)): require_web(request) start, end = parse_timestamp(start_at), parse_timestamp(end_at) if end <= start: raise HTTPException(422, "End must be after start") db.add(TimeEntry(source="manual", device_id="server", external_id=str(uuid4()), kind="manual", status="accepted", start_at=start, end_at=end, duration_seconds=int((end - start).total_seconds()), task=task, project_slug=project_slug or None)) db.commit() return RedirectResponse("/", status_code=303) @app.post("/entries/{entry_id}/edit") def edit_entry(entry_id: int, request: Request, start_at: Annotated[str, Form()], end_at: Annotated[str, Form()], task: Annotated[str, Form()] = "", project_slug: Annotated[str, Form()] = "", notes: Annotated[str, Form()] = "", db: Session = Depends(get_db)): require_web(request) entry = db.get(TimeEntry, entry_id) if not entry: raise HTTPException(404, "Entry not found") start, end = parse_timestamp(start_at), parse_timestamp(end_at) if end <= start: raise HTTPException(422, "End must be after start") entry.start_at = start entry.end_at = end entry.duration_seconds = int((end - start).total_seconds()) entry.task = task entry.project_slug = project_slug or None entry.notes = notes # Editing an inference explicitly makes it a durable manual timesheet row. if entry.kind == "zed_inferred": entry.kind = "manual" db.commit() return RedirectResponse("/", status_code=303) @app.post("/days/{day}/accept") def accept_day(day: str, request: Request, start: Annotated[str, Form()] = "", end: Annotated[str, Form()] = "", db: Session = Depends(get_db)): require_web(request) try: target = date.fromisoformat(day) except ValueError as exc: raise HTTPException(422, "Invalid day") from exc entries = list(db.scalars(select(TimeEntry))) accepted_rows = [entry for entry in entries if entry.status == "accepted"] for entry in entries: if entry.status != "suggested" or entry.start_at.astimezone(TZ).date() != target: continue # Inferred time overlapping accepted time is evidence only: accepting # it would double-count. It stays suggested (shown as overlapping) and # can be rejected individually. if overlaps_entries(entry, accepted_rows): continue entry.status = "accepted" if entry.kind == "zed_inferred": entry.kind = "manual" accepted_rows.append(entry) db.commit() return index_redirect(start or None, end or None) def format_duration(seconds: int) -> str: return f"{seconds // 3600}h {(seconds % 3600) // 60:02d}m" def _next_month(first_day: date) -> date: if first_day.month == 12: return date(first_day.year + 1, 1, 1) return date(first_day.year, first_day.month + 1, 1) def _month_table_row(day_date: date, entry_days: dict, commit_days: dict) -> tuple[str, int, int, int]: day_entries = entry_days.get(day_date, []) day_commits = commit_days.get(day_date, []) accepted = sum(e.duration_seconds for e in day_entries if e.status == "accepted") suggested = sum(e.duration_seconds for e in day_entries if e.status == "suggested") row = f"| {day_date.isoformat()} | {day_date.strftime('%a')} | {format_duration(accepted)} | {format_duration(suggested)} | {len(day_commits)} |" return row, accepted, suggested, len(day_commits) def _monthly_markdown_response(first_day: date, last_day: date, entry_days: dict, commit_days: dict) -> Response: lines = [ f"# Timesheet: {calendar.month_name[first_day.month]} {first_day.year} ({first_day.isoformat()} → {last_day.isoformat()})", "", f"Generated: {datetime.now(TZ).strftime('%Y-%m-%d %H:%M')} ({APP_TIMEZONE})", "", "## Summary by day", "", "| Date | Day | Accepted | Suggested (not counted) | Commits |", "|---|---|---:|---:|---:|", ] accepted_total = suggested_total = commit_total = days_with_accepted = 0 day_cursor = first_day while day_cursor <= last_day: row, accepted, suggested, commits = _month_table_row(day_cursor, entry_days, commit_days) lines.append(row) accepted_total += accepted suggested_total += suggested commit_total += commits if accepted: days_with_accepted += 1 day_cursor += timedelta(days=1) lines += [ "", "## Totals", "", f"- **Accepted:** {format_duration(accepted_total)}", f"- **Suggested (not counted):** {format_duration(suggested_total)}", f"- **Commits:** {commit_total}", f"- **Days with accepted time:** {days_with_accepted}", ] for day_date in sorted(entry_days): day_entries = entry_days[day_date] day_commits = commit_days.get(day_date, []) lines.append("") lines.append(f"### {day_date.strftime('%a %d %b %Y')}") for heading, statuses in ( ("Accepted time", ("accepted",)), ("Suggested time (not counted)", ("suggested",)), ): selected = [e for e in day_entries if e.status in statuses] if not selected: continue lines += ["", heading, "", "| Start | End | Source | Task | Project | Duration |", "|---|---|---|---|---|---:|"] for entry in selected: lines.append( f"| {entry.start_at.astimezone(TZ).strftime('%H:%M')} " f"| {entry.end_at.astimezone(TZ).strftime('%H:%M')} " f"| {entry.source} | {entry.task or '—'} | {entry.project_slug or ''} " f"| {format_duration(entry.duration_seconds)} |" ) if day_commits: lines += ["", "Commits", "", "| Time | Project | Author | Commit |", "|---|---|---|---|"] for commit in day_commits: sha = (commit.payload.get("sha", "") or "")[:7] lines.append( f"| {commit.occurred_at.astimezone(TZ).strftime('%H:%M')} | {commit.project_slug} " f"| {commit.payload.get('author', '')} | `{sha}` {commit.payload.get('subject', '')} |" ) return Response( "\n".join(lines) + "\n", media_type="text/markdown; charset=utf-8", headers={"Content-Disposition": f'attachment; filename="timesheet-{first_day.isoformat()[:7]}.md"'}, ) def _monthly_csv_response(first_day: date, entries: list[TimeEntry]) -> Response: buffer = io.StringIO() writer = csv.writer(buffer) writer.writerow(["date", "weekday", "start", "end", "source", "status", "task", "project", "duration_minutes"]) for entry in sorted(entries, key=lambda item: item.start_at): local_start = entry.start_at.astimezone(TZ) writer.writerow([ local_start.date().isoformat(), local_start.strftime("%a"), local_start.strftime("%Y-%m-%dT%H:%M"), entry.end_at.astimezone(TZ).strftime("%Y-%m-%dT%H:%M"), entry.source, entry.status, entry.task, entry.project_slug or "", entry.duration_seconds // 60, ]) return Response( buffer.getvalue(), media_type="text/csv; charset=utf-8", headers={"Content-Disposition": f'attachment; filename="timesheet-{first_day.isoformat()[:7]}.csv"'}, ) @app.get("/export/monthly") def export_monthly(request: Request, month: str, format: str = "md", db: Session = Depends(get_db)): require_web(request) try: first_day = date.fromisoformat(f"{month}-01") except ValueError as exc: raise HTTPException(422, "month must be formatted YYYY-MM") from exc last_day = _next_month(first_day) - timedelta(days=1) start_dt = datetime.combine(first_day, datetime.min.time(), tzinfo=TZ) end_dt = datetime.combine(_next_month(first_day), datetime.min.time(), tzinfo=TZ) entries = [ entry for entry in db.scalars(select(TimeEntry)) if entry.status != "rejected" and start_dt <= entry.start_at < end_dt ] commits = [ commit for commit in db.scalars(select(RawEvent).where(RawEvent.source == "zed_commit")) if commit.occurred_at is not None and start_dt <= commit.occurred_at < end_dt ] entry_days = group_entries_by_day(entries) commit_days = group_commits_by_day(commits) if format == "csv": return _monthly_csv_response(first_day, entries) return _monthly_markdown_response(first_day, last_day, entry_days, commit_days) @app.post("/entries/{entry_id}/status") def set_entry_status(entry_id: int, request: Request, status: Annotated[str, Form()], start: Annotated[str, Form()] = "", end: Annotated[str, Form()] = "", db: Session = Depends(get_db)): require_web(request) if status not in {"accepted", "rejected", "suggested"}: raise HTTPException(422, "Invalid status") entry = db.get(TimeEntry, entry_id) if not entry: raise HTTPException(404, "Entry not found") if status == "accepted" and overlaps_accepted(db, entry): raise HTTPException(422, "Entry overlaps accepted time and cannot be accepted") entry.status = status # An accepted inference becomes a durable user-approved manual entry. if status == "accepted" and entry.kind == "zed_inferred": entry.kind = "manual" db.commit() return index_redirect(start or None, end or None) @app.post("/call-imports", response_class=HTMLResponse) def upload_call_pdf(request: Request, file: Annotated[UploadFile, File()], db: Session = Depends(get_db)): require_web(request) if not file.filename or not file.filename.lower().endswith(".pdf"): raise HTTPException(422, "Only PDF call-detail logs are accepted") destination = UPLOAD_DIR / f"{uuid4()}.pdf" with destination.open("wb") as output: shutil.copyfileobj(file.file, output) try: with pdfplumber.open(destination) as pdf: text = "\n".join(page.extract_text() or "" for page in pdf.pages) status = "extracted" if text.strip() else "needs_ocr" except Exception: try: reader = PdfReader(destination) text = "\n".join(page.extract_text() or "" for page in reader.pages) status = "extracted" if text.strip() else "needs_ocr" except Exception as exc: text, status = f"Extraction failed: {exc}", "failed" call_import = CallImport(filename=file.filename, stored_path=str(destination), extraction_status=status, extracted_text=text) db.add(call_import) db.commit() return RedirectResponse("/", status_code=303)