yuzhiran
7e953afbd2
- 新增创作工作台 studio.html：合并选题/内容工厂/文章管理为单一 tab 入口（iframe embed 模式） - 新增 AI味检测模块（ai_slop API + 页面，合规软硬问题分级） - 新增白标品牌配置（branding API + 页面 + deploy 私有化交付包） - 发布闭环：publishing 放宽至 editor + records/mark-published 接口 - 移动端响应式补全（admin/calendar/ai-slop 表格卡片兜底） - 修复菜单幂等播种缺陷（按 path 对齐，避免功能页孤立） - 新增短视频脚本 shortvideo.py 与 2026 市场调研简报
207 lines
6.9 KiB
Python
207 lines
6.9 KiB
Python
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"<title>\s*([^<]+?)\s*</title>", html, re.IGNORECASE)
|
||
if not m:
|
||
m = re.search(r"<h1[^>]*>\s*([^<]+?)\s*</h1>", html, re.IGNORECASE)
|
||
return m.group(1).strip() if m else ""
|
||
|
||
|
||
def _extract_content(html: str) -> str:
|
||
import re
|
||
text = re.sub(r"<style.*?</style>", "", html, flags=re.DOTALL | re.IGNORECASE)
|
||
text = re.sub(r"<script.*?</script>", "", 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,
|
||
)
|