91 lines
3.2 KiB
Python
91 lines
3.2 KiB
Python
"""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",
|
|
})
|