import sys from pathlib import Path from typing import List, Optional from fastapi import APIRouter, HTTPException, Depends, Body from pydantic import BaseModel from ..database import get_db from ..models import User, Topic, Article from .auth import get_current_user, org_filter router = APIRouter(prefix="/api/ai-slop", tags=["ai-slop"]) PROJECT_ROOT = Path(__file__).resolve().parents[4] SCRIPTS_DIR = PROJECT_ROOT / "scripts" if not SCRIPTS_DIR.exists(): SCRIPTS_DIR = PROJECT_ROOT.parent / "scripts" sys.path.insert(0, str(SCRIPTS_DIR)) check_article = None polish_with_llm = None clean_html_content = None strip_ai_preface = None strip_thinking_html = None try: from compliance_checker import check_article except Exception: check_article = None try: from compliance_optimizer import polish_with_llm except Exception: polish_with_llm = None try: from content_cleaner import clean_html_content, strip_ai_preface, strip_thinking_html except Exception: clean_html_content = None strip_ai_preface = None strip_thinking_html = None HARD_ISSUE_TYPES = ("敏感词", "法律法规", "平台规则", "品牌规范", "资源合规") class IssueItem(BaseModel): type: str category: str = "" detail: str = "" suggestion: str = "" severity: str = "medium" class PlatformReport(BaseModel): platform: str score: int passed: bool issues: List[IssueItem] html_preview: str = "" class ReportResponse(BaseModel): topic_id: str platforms: List[PlatformReport] class PurifyRequest(BaseModel): topic_id: str platform: str class PurifyResponse(BaseModel): ok: bool platform: str score_before: Optional[int] = None score_after: Optional[int] = None issues_before: List[IssueItem] = [] issues_after: List[IssueItem] = [] preview: str = "" message: str = "" def _normalize_issues(issues: list) -> List[IssueItem]: result = [] for i in issues or []: itype = i.get("type", "") severity = "high" if itype in HARD_ISSUE_TYPES else "medium" detail = i.get("detail") or i.get("suggestion") or i.get("word") or i.get("tag") or i.get("pattern") or "" result.append(IssueItem( type=itype, category=i.get("category", ""), detail=detail, suggestion=i.get("suggestion", ""), severity=severity, )) return result def _extract_title(html: str) -> str: import re m = re.search(r"\s*([^<]+?)\s*", html, re.IGNORECASE) if not m: m = re.search(r"]*>\s*([^<]+?)\s*", html, re.IGNORECASE) return m.group(1).strip() if m else "" def _extract_content(html: str) -> str: import re text = re.sub(r"", "", html, flags=re.DOTALL | re.IGNORECASE) text = re.sub(r"", "", text, flags=re.DOTALL | re.IGNORECASE) text = re.sub(r"<[^>]+>", "", text) return re.sub(r"\s+", " ", text).strip() def _load_topic_data(db, topic_id: str) -> dict: topic = db.query(Topic).filter(Topic.id == topic_id).first() if not topic: return {} return { "topic": { "title": getattr(topic, "title", "") or "", "field": getattr(topic, "field", "") or "", "core_concept": getattr(topic, "core_concept", "") or "", } } @router.get("/report", response_model=ReportResponse) def get_report(topic_id: str, current_user: User = Depends(get_current_user), db=Depends(get_db)): if check_article is None: raise HTTPException(status_code=503, detail="合规检测模块不可用") topic = db.query(Topic).filter(Topic.id == topic_id).first() if not topic: raise HTTPException(status_code=404, detail="选题不存在") of = org_filter(current_user, Topic) if of is not True and topic.org_id != current_user.org_id: raise HTTPException(status_code=404, detail="选题不存在") topic_data = _load_topic_data(db, topic_id) from db_helper import get_articles_by_topic articles = get_articles_by_topic(topic_id) platforms = [] for art in articles: platform = art.get("platform") html = art.get("html_content") or "" if not html: continue res = check_article(html, platform, topic_data=topic_data) platforms.append(PlatformReport( platform=platform, score=res.get("score", 0), passed=res.get("passed", False), issues=_normalize_issues(res.get("issues", [])), html_preview=html[:600], )) return ReportResponse(topic_id=topic_id, platforms=platforms) @router.post("/purify", response_model=PurifyResponse) def purify(req: PurifyRequest, current_user: User = Depends(get_current_user), db=Depends(get_db)): if check_article is None: raise HTTPException(status_code=503, detail="合规检测模块不可用") topic = db.query(Topic).filter(Topic.id == req.topic_id).first() if not topic: raise HTTPException(status_code=404, detail="选题不存在") of = org_filter(current_user, Topic) if of is not True and topic.org_id != current_user.org_id: raise HTTPException(status_code=404, detail="选题不存在") from db_helper import get_articles_by_topic, save_article articles = get_articles_by_topic(req.topic_id) target = next((a for a in articles if a.get("platform") == req.platform), None) if not target or not target.get("html_content"): raise HTTPException(status_code=404, detail=f"未找到 {req.platform} 平台的文章") html = target["html_content"] topic_data = _load_topic_data(db, req.topic_id) before = check_article(html, req.platform, topic_data=topic_data) issues_before = _normalize_issues(before.get("issues", [])) raw_issues = before.get("issues", []) polished_html, log_msg = (html, None) if polish_with_llm is not None: polished_html, log_msg = polish_with_llm(html, req.platform, remaining_issues=raw_issues) cleaned = polished_html if clean_html_content is not None: cleaned = clean_html_content(cleaned) if strip_ai_preface is not None: cleaned = strip_ai_preface(cleaned) if strip_thinking_html is not None: cleaned = strip_thinking_html(cleaned) title = _extract_title(cleaned) content = _extract_content(cleaned) save_article(req.topic_id, req.platform, cleaned, title=title, content=content, db=db) after = check_article(cleaned, req.platform, topic_data=topic_data) issues_after = _normalize_issues(after.get("issues", [])) message = "净化完成" + (f"（{log_msg}）" if log_msg else "（仅执行清洗，未调用 LLM）") return PurifyResponse( ok=True, platform=req.platform, score_before=before.get("score"), score_after=after.get("score"), issues_before=issues_before, issues_after=issues_after, preview=cleaned[:600], message=message, )