"""Change History: reverse-chronological feed of recently modified items.""" from fastapi import APIRouter, Request, Depends from fastapi.templating import Jinja2Templates from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import text from typing import Optional from datetime import datetime, timezone from core.database import get_db from core.sidebar import get_sidebar_data router = APIRouter(prefix="/history", tags=["history"]) templates = Jinja2Templates(directory="templates") # Entity configs: (table, label_column, type_label, url_prefix) HISTORY_ENTITIES = [ ("domains", "name", "Domain", "/domains"), ("areas", "name", "Area", "/areas"), ("projects", "name", "Project", "/projects"), ("tasks", "title", "Task", "/tasks"), ("notes", "title", "Note", "/notes"), ("contacts", "first_name", "Contact", "/contacts"), ("meetings", "title", "Meeting", "/meetings"), ("decisions", "title", "Decision", "/decisions"), ("lists", "name", "List", "/lists"), ("appointments", "title", "Appointment", "/appointments"), ("links", "label", "Link", "/links"), ("files", "original_filename", "File", "/files"), ("capture", "content", "Capture", "/capture"), ] @router.get("/") async def history_view( request: Request, entity_type: Optional[str] = None, db: AsyncSession = Depends(get_db), ): sidebar = await get_sidebar_data(db) all_items = [] for table, label_col, type_label, url_prefix in HISTORY_ENTITIES: if entity_type and entity_type != table: continue try: result = await db.execute(text(f""" SELECT id, {label_col} as label, updated_at, created_at FROM {table} WHERE is_deleted = false ORDER BY updated_at DESC LIMIT 20 """)) for r in result: row = dict(r._mapping) # Determine action action = "created" if row["updated_at"] and row["created_at"]: diff = abs((row["updated_at"] - row["created_at"]).total_seconds()) if diff > 1: action = "modified" all_items.append({ "type": table, "type_label": type_label, "id": str(row["id"]), "label": str(row["label"] or "Untitled")[:80], "url": f"{url_prefix}/{row['id']}", "updated_at": row["updated_at"], "action": action, }) except Exception: continue # Sort by updated_at descending, take top 50 all_items.sort(key=lambda x: x["updated_at"] or datetime.min.replace(tzinfo=timezone.utc), reverse=True) all_items = all_items[:50] # Build entity type options for filter type_options = [{"value": t[0], "label": t[2]} for t in HISTORY_ENTITIES] return templates.TemplateResponse("history.html", { "request": request, "sidebar": sidebar, "items": all_items, "type_options": type_options, "current_type": entity_type or "", "page_title": "Change History", "active_nav": "history", })