"""Full-DB canonical export + media manifest. The export returns the SAME canonical shape scripts/export_v1.py produces and scripts/diff_harness.py consumes:: {"meta": {"exported_at", "source", "garden_id", "table_counts"}, "tables": {: [ {row}, ... sorted by id ]}} so the harness can run against the HTTP API too (its right side is ``SELECT * FROM ORDER BY id`` — identical rows, identical order, identical column set: every v2 column is included; the harness's ``canon_row(drop_v2_only=True)`` + ``V2_ONLY_COLUMNS`` allowlist strips the v2 additions during comparison). Row ordering matches the harness exactly: ``id`` ascending, except ``area_stations`` (composite key) which orders by ``(area_id, station_id)``. Garden scoping (deviation D1): tables with a ``garden_id`` column are filtered in SQL; child tables without one are filtered against their scoped parents' id sets (varieties→species, area_stations→areas, field_note_targets→field_notes, chat_messages/tool_suggestions→chat_sessions, overview_versions→its six polymorphic parents). Unscopeable rows — varieties with a NULL species link, tool_suggestions with a NULL session — are INCLUDED (single-garden reality; excluding them would silently drop data from the migration oracle). Values are JSON-ready: datetimes → ISO-8601 strings, dates → YYYY-MM-DD, booleans stay booleans — all forms scripts/canonical.py parses natively. The media manifest mirrors scripts/media_manifest.py: sha256 + size of every file under DATA_DIR's ``notes/`` and ``artifacts/`` subtrees, keyed by forward-slash relative path, sorted. """ from __future__ import annotations import hashlib import os from datetime import date, datetime, timezone from sqlalchemy import select from sqlalchemy.orm import Session from ..models.assistant import ChatMessage, ChatSession, ToolSuggestion from ..models.catalog import Species, Variety from ..models.garden import Area, AreaStation, Supply, WateringStation, Year from ..models.history import ActivityLog, Artifact, OverviewVersion from ..models.notes import FieldNote, FieldNoteTarget from ..models.planting import Plant, Planting # The 17 carried v1 tables — MUST stay in sync with scripts/export_v1.py's # V1_TABLES (the harness's table universe). V1_EXPORT_TABLES = ( "activity_log", "area_stations", "areas", "artifacts", "chat_messages", "chat_sessions", "field_note_targets", "field_notes", "overview_versions", "plantings", "plants", "species", "supplies", "tool_suggestions", "varieties", "watering_stations", "years", ) # Overview parents: entity_type -> the exported table its ids come from. _OVERVIEW_PARENT_TABLES = { "area": "areas", "planting": "plantings", "plant": "plants", "species": "species", "variety": "varieties", "year": "years", } def _jsonable(value): """DB value -> JSON-ready form the harness's canonicalizer parses.""" if isinstance(value, datetime): return value.isoformat() if isinstance(value, date): return value.isoformat() return value def _rows(db: Session, model, *where, order=None) -> list[dict]: """All rows of a model as plain dicts, deterministic order.""" if order is None: order = (model.__table__.c.id,) stmt = select(model.__table__).order_by(*order) if where: stmt = stmt.where(*where) return [ {k: _jsonable(v) for k, v in m.items()} for m in db.execute(stmt).mappings() ] def export_garden(db: Session, garden_id: int) -> dict: """Canonical full-DB export of one garden's carried tables.""" t: dict[str, list[dict]] = {} # --- directly scoped tables (garden_id column) --- t["species"] = _rows(db, Species, Species.garden_id == garden_id) t["years"] = _rows(db, Year, Year.garden_id == garden_id) t["areas"] = _rows(db, Area, Area.garden_id == garden_id) t["watering_stations"] = _rows( db, WateringStation, WateringStation.garden_id == garden_id ) t["plantings"] = _rows(db, Planting, Planting.garden_id == garden_id) t["plants"] = _rows(db, Plant, Plant.garden_id == garden_id) t["supplies"] = _rows(db, Supply, Supply.garden_id == garden_id) t["field_notes"] = _rows(db, FieldNote, FieldNote.garden_id == garden_id) t["artifacts"] = _rows(db, Artifact, Artifact.garden_id == garden_id) t["chat_sessions"] = _rows( db, ChatSession, ChatSession.garden_id == garden_id ) t["activity_log"] = _rows( db, ActivityLog, ActivityLog.garden_id == garden_id ) # --- child tables, scoped through their parents' id sets --- species_ids = {r["id"] for r in t["species"]} area_ids = {r["id"] for r in t["areas"]} note_ids = {r["id"] for r in t["field_notes"]} session_ids = {r["id"] for r in t["chat_sessions"]} # varieties: NULL species link is unscopeable -> included (see docstring). t["varieties"] = [ r for r in _rows(db, Variety) if r["species_id"] is None or r["species_id"] in species_ids ] t["area_stations"] = [ r for r in _rows( db, AreaStation, order=( AreaStation.__table__.c.area_id, AreaStation.__table__.c.station_id, ), ) if r["area_id"] in area_ids ] t["field_note_targets"] = [ r for r in _rows(db, FieldNoteTarget) if r["field_note_id"] in note_ids ] t["chat_messages"] = [ r for r in _rows(db, ChatMessage) if r["session_id"] in session_ids ] # tool_suggestions: NULL session (SET NULL) is unscopeable -> included. t["tool_suggestions"] = [ r for r in _rows(db, ToolSuggestion) if r["session_id"] is None or r["session_id"] in session_ids ] # overview_versions: polymorphic child of six scoped parents. parent_ids = { etype: {r["id"] for r in t[table]} for etype, table in _OVERVIEW_PARENT_TABLES.items() } t["overview_versions"] = [ r for r in _rows(db, OverviewVersion) if r["entity_id"] in parent_ids.get(r["entity_type"], set()) ] tables = {name: t[name] for name in V1_EXPORT_TABLES} return { "meta": { "exported_at": datetime.now(timezone.utc).isoformat( timespec="seconds" ), "source": "garden2 /api/export", "garden_id": garden_id, "table_counts": {name: len(rows) for name, rows in tables.items()}, }, "tables": tables, } # --- media manifest (mirrors scripts/media_manifest.py) --- MEDIA_SUBDIRS = ("notes", "artifacts") def _sha256_file(path: str) -> str: h = hashlib.sha256() with open(path, "rb") as f: for chunk in iter(lambda: f.read(1024 * 1024), b""): h.update(chunk) return h.hexdigest() def build_media_manifest(data_root: str) -> dict: """sha256 manifest of the media tree under ``data_root`` (notes/ + artifacts/ subtrees only — DB files and snapshots are not media). Same output shape as scripts/media_manifest.py:: {"meta": {...}, "files": {"notes/2026-04/x.webm": {"size": N, "sha256": "..."}, ...}} """ files: dict[str, dict] = {} for sub in MEDIA_SUBDIRS: base = os.path.join(data_root, sub) if not os.path.isdir(base): continue for dirpath, _dirnames, filenames in os.walk(base): for name in sorted(filenames): abs_path = os.path.join(dirpath, name) rel = os.path.relpath(abs_path, data_root).replace(os.sep, "/") files[rel] = { "size": os.path.getsize(abs_path), "sha256": _sha256_file(abs_path), } return { "meta": { "generated_at": datetime.now(timezone.utc).isoformat( timespec="seconds" ), "root": os.path.abspath(data_root), "file_count": len(files), "total_bytes": sum(f["size"] for f in files.values()), }, "files": dict(sorted(files.items())), }