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},
)