import logging import html import re from typing import Any, List, Optional from fastapi import APIRouter, Query, Request from fastapi.responses import HTMLResponse from fastapi.templating import Jinja2Templates from app.modules.manual.backend.cache import manual_cache from app.core.database import execute_query, execute_query_single logger = logging.getLogger(__name__) router = APIRouter() templates = Jinja2Templates(directory="app") def _normalize_tag_list(value: Any) -> List[str]: if isinstance(value, list): return [str(v).strip() for v in value if str(v).strip()] return [] def _normalize_manual_text(value: Any) -> str: text = str(value or "") text = html.unescape(text) text = re.sub(r"(?i)", "\n", text) text = text.replace("\\n", "\n") return text @router.get("/manual", response_class=HTMLResponse) async def manual_index( request: Request, module: Optional[str] = Query(default=None), difficulty: Optional[str] = Query(default=None), tag: Optional[str] = Query(default=None), search: Optional[str] = Query(default=None), ): cache_key = f"views:list:{module}:{difficulty}:{tag}:{search}" cached = manual_cache.get(cache_key) if cached: rows, modules, unique_tags = cached else: filters = ["deleted_at IS NULL"] params = [] if module: filters.append("module = %s") params.append(module) if difficulty: filters.append("difficulty = %s") params.append(difficulty) if tag: filters.append("tags @> %s::jsonb") params.append(f'["{tag}"]') if search: filters.append("(title ILIKE %s OR content ILIKE %s)") params.extend([f"%{search}%", f"%{search}%"]) where_clause = " AND ".join(filters) rows = execute_query( f"SELECT id, slug, title, content, module, tags, difficulty, use_count, updated_at " f"FROM manual_articles WHERE {where_clause} ORDER BY updated_at DESC", tuple(params) ) or [] modules = execute_query( "SELECT DISTINCT module FROM manual_articles WHERE deleted_at IS NULL ORDER BY module ASC" ) or [] all_tags: List[str] = [] for row in rows: if "tags" in row and row["tags"]: try: import json if isinstance(row["tags"], str): t = json.loads(row["tags"]) if isinstance(t, list): all_tags.extend(t) elif isinstance(row["tags"], list): all_tags.extend(row["tags"]) except Exception: pass unique_tags = sorted(list(set(all_tags))) manual_cache.set(cache_key, (rows, modules, unique_tags)) return templates.TemplateResponse( "modules/manual/templates/list.html", { "request": request, "articles": rows, "available_modules": modules, "available_tags": unique_tags, "filters": { "module": module or "", "difficulty": difficulty or "", "tag": tag or "", "search": search or "", }, }, ) @router.get("/manual/admin", response_class=HTMLResponse) async def manual_admin(request: Request): articles = execute_query( """ SELECT id, title, slug, module, difficulty, use_count, updated_at FROM manual_articles WHERE deleted_at IS NULL ORDER BY updated_at DESC LIMIT 200 """ ) or [] return templates.TemplateResponse( "modules/manual/templates/admin.html", {"request": request, "articles": articles}, ) @router.get("/manual/{slug}", response_class=HTMLResponse) async def manual_detail(request: Request, slug: str): article = execute_query_single( """ SELECT id, title, slug, content, summary, module, tags, difficulty, use_count, created_at, updated_at FROM manual_articles WHERE slug = %s AND deleted_at IS NULL """, (slug,), ) if not article: return templates.TemplateResponse( "modules/manual/templates/detail.html", {"request": request, "article": None, "steps": [], "related": []}, status_code=404, ) execute_query( "UPDATE manual_articles SET use_count = COALESCE(use_count, 0) + 1 WHERE id = %s", (article["id"],), ) article["use_count"] = int(article.get("use_count") or 0) + 1 article["content_normalized"] = _normalize_manual_text(article.get("content")) article["summary_normalized"] = _normalize_manual_text(article.get("summary")) steps = execute_query( """ SELECT step_number, title, content, image_url, video_url FROM manual_steps WHERE manual_id = %s ORDER BY step_number ASC """, (article["id"],), ) or [] for step in steps: step["content_normalized"] = _normalize_manual_text(step.get("content")) related = execute_query( """ SELECT rm.slug, rm.title, rm.summary, rm.module FROM manual_relations mr JOIN manual_articles rm ON rm.id = mr.related_manual_id WHERE mr.manual_id = %s AND rm.deleted_at IS NULL ORDER BY rm.use_count DESC, rm.updated_at DESC LIMIT 12 """, (article["id"],), ) or [] return templates.TemplateResponse( "modules/manual/templates/detail.html", {"request": request, "article": article, "steps": steps, "related": related}, )