devplacepy/trash.py at 72e11db185e389306be50d6e6db733efa23d5e7a

 # retoor <retoor@molodetz.nl>
 import logging
 from fastapi import APIRouter, Request
 from fastapi.responses import HTMLResponse
 from devplacepy.database import (
     get_table,
     list_deleted,
     count_deleted,
     restore_event,
     purge_event,
     get_users_by_uids,
     resolve_object_url,
 )
 from devplacepy.attachments import _unlink_attachment_files
 from devplacepy.project_files import _unlink_blob
 from devplacepy.utils import require_admin, not_found
 from devplacepy.seo import base_seo_context, site_url, website_schema
 from devplacepy.responses import respond, action_result
 from devplacepy.schemas import AdminTrashOut
 from devplacepy.services.audit import record as audit
 logger = logging.getLogger(__name__)
 router = APIRouter()
 TRASH_TABLES = [
     {"key": "posts", "label": "Posts", "icon": "\U0001f4dd", "type": "post"},
     {"key": "comments", "label": "Comments", "icon": "\U0001f4ac", "type": "comment"},
     {"key": "gists", "label": "Gists", "icon": "\U0001f4cb", "type": "gist"},
     {"key": "projects", "label": "Projects", "icon": "\U0001f680", "type": "project"},
     {"key": "news", "label": "News", "icon": "\U0001f4f0", "type": "news"},
     {"key": "awards", "label": "Awards", "icon": "\U0001f3c6", "type": "award"},
     {"key": "quizzes", "label": "Quizzes", "icon": "\U0001f9e9", "type": "quiz"},
     {"key": "project_files", "label": "Project files", "icon": "\U0001f4c1", "type": None},
     {"key": "attachments", "label": "Attachments", "icon": "\U0001f4ce", "type": None},
 ]
 _TRASH_KEYS = {entry["key"] for entry in TRASH_TABLES}
 _TRASH_TYPE = {entry["key"]: entry["type"] for entry in TRASH_TABLES}
 def _trash_label(table: str, row: dict) -> str:
     for field in ("title", "name", "path", "original_filename", "question", "content"):
         value = row.get(field)
         if value:
             return str(value)[:80]
     return row.get("slug") or row.get("uid", "")
 def _trash_target_url(table: str, row: dict) -> str | None:
     target_type = _TRASH_TYPE.get(table)
     if target_type:
         return resolve_object_url(target_type, row["uid"])
     return None
 @router.get("/trash", response_class=HTMLResponse)
 async def admin_trash(request: Request, table: str = "posts", page: int = 1):
     admin = require_admin(request)
     if table not in _TRASH_KEYS:
         table = "posts"
     rows, pagination = list_deleted(table, page)
     actor_ids = [r.get("deleted_by") for r in rows if r.get("deleted_by")]
     actors = get_users_by_uids([a for a in actor_ids if a]) if actor_ids else {}
     items = []
     for row in rows:
         actor = actors.get(row.get("deleted_by"))
         items.append(
             {
                 "uid": row["uid"],
                 "table": table,
                 "label": _trash_label(table, row),
                 "deleted_at": row.get("deleted_at"),
                 "deleted_by": actor["username"] if actor else row.get("deleted_by"),
                 "target_url": _trash_target_url(table, row),
             }
         )
     tabs = [
         {**entry, "count": count_deleted(entry["key"]), "active": entry["key"] == table}
         for entry in TRASH_TABLES
     ]
     base = site_url(request)
     seo_ctx = base_seo_context(
         request,
         title="Trash - Admin",
         description="Restore or permanently remove soft-deleted content.",
         breadcrumbs=[
             {"name": "Home", "url": "/feed"},
             {"name": "Admin", "url": "/admin"},
             {"name": "Trash", "url": "/admin/trash"},
         ],
         schemas=[website_schema(base)],
     )
     return respond(
         request,
         "admin_trash.html",
         {
             **seo_ctx,
             "request": request,
             "user": admin,
             "items": items,
             "table": table,
             "tables": tabs,
             "pagination": pagination,
             "admin_section": "trash",
         },
         model=AdminTrashOut,
     )
 @router.post("/trash/{table}/{uid}/restore")
 async def admin_trash_restore(request: Request, table: str, uid: str):
     admin = require_admin(request)
     if table not in _TRASH_KEYS:
         raise not_found("Unknown trash table")
     row = get_table(table).find_one(uid=uid)
     if row and row.get("deleted_at"):
         restored = restore_event(row["deleted_at"])
         if table == "awards":
             from devplacepy.database.awards import recompute_user_award_stats
             recompute_user_award_stats(row.get("receiver_uid", ""))
         logger.info(
             f"Admin {admin['username']} restored {table} {uid} ({restored} rows)"
         )
         audit.record(
             request,
             "admin.trash.restore",
             user=admin,
             target_type=table,
             target_uid=uid,
             target_label=_trash_label(table, row),
             metadata={"table": table, "rows": restored},
             summary=f"{admin['username']} restored {table} {uid}",
         )
     return action_result(request, f"/admin/trash?table={table}")
 @router.post("/trash/{table}/{uid}/purge")
 async def admin_trash_purge(request: Request, table: str, uid: str):
     admin = require_admin(request)
     if table not in _TRASH_KEYS:
         raise not_found("Unknown trash table")
     row = get_table(table).find_one(uid=uid)
     if row and row.get("deleted_at"):
         purged = purge_event(row["deleted_at"])
         for purged_table, purged_rows in purged:
             if purged_table == "attachments":
                 for attachment in purged_rows:
                     _unlink_attachment_files(attachment)
             elif purged_table == "project_files":
                 for node in purged_rows:
                     if node.get("is_binary"):
                         _unlink_blob(node)
         logger.info(f"Admin {admin['username']} purged {table} {uid}")
         audit.record(
             request,
             "admin.trash.purge",
             user=admin,
             target_type=table,
             target_uid=uid,
             target_label=_trash_label(table, row),
             metadata={"table": table, "tables": [t for t, _ in purged]},
             summary=f"{admin['username']} permanently purged {table} {uid}",
         )
     return action_result(request, f"/admin/trash?table={table}")