"""Authenticated export of entries and daily rows as JSONL or CSV. Exports stream whole files from DATA_ROOT without mounting that directory. """ import csv import io from typing import Any from fastapi import APIRouter, Depends, Request from fastapi.responses import PlainTextResponse, Response from app.deps import require_login from app.fsutil import read_jsonl router = APIRouter( prefix="/api/export", tags=["export"], dependencies=[Depends(require_login)], ) CSV_COLUMNS = [ "id", "ts", "created_at", "updated_at", "activity", "happened", "emotions", "intensity", "remedy", "result", "tags", "notes", ] def _jsonl_response(path) -> Response: if not path.exists(): content = "" else: content = path.read_text(encoding="utf-8") return Response(content=content, media_type="application/x-ndjson") @router.get("/entries.jsonl") def export_entries_jsonl(request: Request) -> Response: """Download entries as newline-delimited JSON.""" return _jsonl_response(request.app.state.paths.entries) @router.get("/daily.jsonl") def export_daily_jsonl(request: Request) -> Response: """Download daily rows as newline-delimited JSON.""" return _jsonl_response(request.app.state.paths.daily) @router.get("/entries.csv") def export_entries_csv(request: Request) -> PlainTextResponse: """Download entries as CSV with fixed headers.""" records: list[dict[str, Any]] = read_jsonl(request.app.state.paths.entries) buffer = io.StringIO() writer = csv.DictWriter(buffer, fieldnames=CSV_COLUMNS, extrasaction="ignore") writer.writeheader() for record in records: row = {key: record.get(key, "") for key in CSV_COLUMNS} emotions = record.get("emotions") or [] tags = record.get("tags") or [] row["emotions"] = "|".join(str(item) for item in emotions) row["tags"] = "|".join(str(item) for item in tags) writer.writerow(row) return PlainTextResponse(content=buffer.getvalue(), media_type="text/csv")