Spaces:
Sleeping
Sleeping
File size: 2,075 Bytes
990895d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 | """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")
|