bwstudio
595394fcb7
- 修改 admin/vite.config.js 添加 base: '/admin/' 解决静态资源路径问题 - 修复后台 index.html 资源引用从 /assets/ → /admin/assets/ - 更新优化器默认 API 地址为服务器地址 - 添加友链管理相关代码 - 修复多处分类管理页面
385 lines
14 KiB
Python
385 lines
14 KiB
Python
"""笑话优化核心逻辑：质量检测 → AI 润色 → 评价分类。"""
|
|
|
|
import hashlib
|
|
import json
|
|
import sys
|
|
import time
|
|
|
|
import httpx
|
|
from openai import OpenAI
|
|
|
|
from optimizer.prompts import (
|
|
QUALITY_CHECK_SYSTEM_PROMPT,
|
|
QUALITY_CHECK_USER_PROMPT,
|
|
POLISH_SYSTEM_PROMPT,
|
|
POLISH_USER_PROMPT,
|
|
EVALUATE_SYSTEM_PROMPT,
|
|
EVALUATE_USER_PROMPT,
|
|
)
|
|
|
|
|
|
class Optimizer:
|
|
def __init__(self, api_base: str, username: str, password: str):
|
|
self.api_base = api_base.rstrip("/")
|
|
self.username = username
|
|
self.password = password
|
|
self.token = None
|
|
self.ai_client = None
|
|
self.model_name = ""
|
|
self.types = []
|
|
self.crowds = []
|
|
# 统计
|
|
self.stats = {"checked": 0, "rejected": 0, "polished": 0, "evaluated": 0, "skipped": 0}
|
|
|
|
# === API 认证 ===
|
|
def _login(self) -> str:
|
|
resp = httpx.post(
|
|
f"{self.api_base}/api/auth/login",
|
|
json={"username": self.username, "password": self.password},
|
|
timeout=30,
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()["access_token"]
|
|
|
|
def _get(self, path: str) -> dict | list:
|
|
resp = httpx.get(
|
|
f"{self.api_base}{path}",
|
|
headers={"Authorization": f"Bearer {self.token}"},
|
|
timeout=30,
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
def _put(self, path: str, data: dict) -> dict:
|
|
resp = httpx.put(
|
|
f"{self.api_base}{path}",
|
|
json=data,
|
|
headers={"Authorization": f"Bearer {self.token}"},
|
|
timeout=30,
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
# === 初始化 ===
|
|
def setup(self):
|
|
print("[*] 正在登录...")
|
|
self.token = self._login()
|
|
print("[*] 登录成功")
|
|
|
|
ai_config = self._get("/api/admin/settings/active")
|
|
self.ai_client = OpenAI(
|
|
base_url=ai_config["api_base"],
|
|
api_key=ai_config["api_key"],
|
|
)
|
|
self.model_name = ai_config["model_name"]
|
|
print(f"[*] AI 模型: {self.model_name}")
|
|
|
|
self.types = self._get("/api/categories/types")
|
|
self.crowds = self._get("/api/categories/crowds")
|
|
print(f"[*] 分类: {len(self.types)} 种类型, {len(self.crowds)} 种人群")
|
|
|
|
# === 读取笑话 ===
|
|
def get_jokes(self, status: str | None = None, limit: int | None = None,
|
|
ids: list[int] | None = None) -> list[dict]:
|
|
"""从 API 分页读取笑话"""
|
|
if ids:
|
|
jokes = []
|
|
for jid in ids:
|
|
try:
|
|
j = self._get(f"/api/admin/jokes/{jid}")
|
|
jokes.append(j)
|
|
except Exception as e:
|
|
print(f" [!] 获取笑话 #{jid} 失败: {e}")
|
|
return jokes
|
|
|
|
page = 1
|
|
page_size = 100
|
|
all_jokes = []
|
|
|
|
while True:
|
|
try:
|
|
path = f"/api/admin/jokes?page={page}&page_size={page_size}"
|
|
if status:
|
|
path += f"&status={status}"
|
|
data = self._get(path)
|
|
items = data.get("items", [])
|
|
if not items:
|
|
break
|
|
all_jokes.extend(items)
|
|
print(f" [*] 已读取 {len(all_jokes)} 条...")
|
|
if limit and len(all_jokes) >= limit:
|
|
all_jokes = all_jokes[:limit]
|
|
break
|
|
page += 1
|
|
except Exception as e:
|
|
print(f" [!] 分页读取失败 (page={page}): {e}")
|
|
break
|
|
|
|
return all_jokes
|
|
|
|
# === Stage 1: 质量检测 ===
|
|
def quality_check(self, content: str) -> dict:
|
|
"""判断笑话是否有笑点，返回 {"has_punchline": bool, "reason": str}"""
|
|
def _call():
|
|
return self.ai_client.chat.completions.create(
|
|
model=self.model_name,
|
|
messages=[
|
|
{"role": "system", "content": QUALITY_CHECK_SYSTEM_PROMPT},
|
|
{"role": "user", "content": QUALITY_CHECK_USER_PROMPT.format(content=content[:2000])},
|
|
],
|
|
temperature=0.3,
|
|
max_tokens=200,
|
|
)
|
|
resp = self._safe_api_call(_call)
|
|
raw = resp.choices[0].message.content.strip()
|
|
return self._parse_json(raw, {"has_punchline": True, "reason": ""})
|
|
|
|
# === Stage 2: AI 润色 ===
|
|
def polish(self, content: str) -> str:
|
|
"""润色笑话内容"""
|
|
def _call():
|
|
return self.ai_client.chat.completions.create(
|
|
model=self.model_name,
|
|
messages=[
|
|
{"role": "system", "content": POLISH_SYSTEM_PROMPT},
|
|
{"role": "user", "content": POLISH_USER_PROMPT.format(content=content)},
|
|
],
|
|
temperature=0.8,
|
|
max_tokens=1024,
|
|
)
|
|
resp = self._safe_api_call(_call)
|
|
return resp.choices[0].message.content.strip()
|
|
|
|
# === Stage 3: 评价分类 ===
|
|
def evaluate(self, content: str) -> dict:
|
|
"""评价并分类，返回 {"types": [...], "crowds": [...], "score": int, "comment": str}"""
|
|
type_names = [t.get("name", "") for t in self.types]
|
|
crowd_names = [c.get("name", "") for c in self.crowds]
|
|
|
|
def _call():
|
|
return self.ai_client.chat.completions.create(
|
|
model=self.model_name,
|
|
messages=[
|
|
{"role": "system", "content": EVALUATE_SYSTEM_PROMPT},
|
|
{"role": "user", "content": EVALUATE_USER_PROMPT.format(
|
|
content=content[:2000],
|
|
known_types=", ".join(type_names),
|
|
known_crowds=", ".join(crowd_names),
|
|
)},
|
|
],
|
|
temperature=0.3,
|
|
max_tokens=300,
|
|
)
|
|
resp = self._safe_api_call(_call)
|
|
raw = resp.choices[0].message.content.strip()
|
|
result = self._parse_json(raw, {"types": [], "crowds": [], "score": 5, "comment": ""})
|
|
# Backward compatibility: if LLM returns old single format, convert to array
|
|
if isinstance(result.get("types"), str):
|
|
result["types"] = [result["types"]] if result["types"] else []
|
|
if isinstance(result.get("crowds"), str):
|
|
result["crowds"] = [result["crowds"]] if result["crowds"] else []
|
|
return result
|
|
|
|
# === 辅助方法 ===
|
|
def _safe_api_call(self, func, *args, max_retries: int = 3, **kwargs):
|
|
"""带重试的 API 调用，自动处理限流"""
|
|
import time as time_module
|
|
last_error = None
|
|
for attempt in range(max_retries):
|
|
try:
|
|
return func(*args, **kwargs)
|
|
except Exception as e:
|
|
last_error = e
|
|
error_str = str(e)
|
|
# 检查是否是限流错误
|
|
if "429" in error_str or "rate" in error_str.lower():
|
|
wait_time = (attempt + 1) * 30 # 30, 60, 90 秒
|
|
print(f" [!] API 限流，等待 {wait_time} 秒后重试 ({attempt+1}/{max_retries})...")
|
|
time_module.sleep(wait_time)
|
|
else:
|
|
# 其他错误，直接重试一次
|
|
if attempt < max_retries - 1:
|
|
time_module.sleep(5)
|
|
raise last_error
|
|
|
|
def _parse_json(self, raw: str, default: dict) -> dict:
|
|
"""安全解析 LLM 返回的 JSON"""
|
|
try:
|
|
data = json.loads(raw)
|
|
if isinstance(data, dict):
|
|
return data
|
|
return default
|
|
except json.JSONDecodeError:
|
|
if "```json" in raw:
|
|
raw = raw.split("```json")[1].split("```")[0]
|
|
elif "```" in raw:
|
|
raw = raw.split("```")[1].split("```")[0]
|
|
try:
|
|
return json.loads(raw.strip())
|
|
except Exception:
|
|
return default
|
|
|
|
def _get_type_id(self, type_name: str) -> int | None:
|
|
for t in self.types:
|
|
if t.get("name") == type_name:
|
|
return t.get("id")
|
|
return None
|
|
|
|
def _get_crowd_id(self, crowd_name: str) -> int | None:
|
|
for c in self.crowds:
|
|
if c.get("name") == crowd_name:
|
|
return c.get("id")
|
|
return None
|
|
|
|
# === 单条笑话处理（3 阶段） ===
|
|
def process_joke(self, joke: dict) -> bool:
|
|
"""处理单条笑话：质量检测 → 润色 → 评价分类，返回是否成功"""
|
|
joke_id = joke.get("id")
|
|
title = joke.get("title", "")
|
|
content = joke.get("content", "")
|
|
|
|
if not content:
|
|
print(f" [!] #{joke_id} 内容为空，跳过")
|
|
self.stats["skipped"] += 1
|
|
return False
|
|
|
|
print(f"\n {'='*40}")
|
|
print(f" 处理 #{joke_id}: {title[:30]}")
|
|
print(f" {'='*40}")
|
|
|
|
# Stage 1: 质量检测
|
|
print(f" [1/3] 质量检测...")
|
|
try:
|
|
check = self.quality_check(content)
|
|
if not check.get("has_punchline", True):
|
|
reason = check.get("reason", "无笑点")
|
|
print(f" [!] 无笑点: {reason}")
|
|
# 标记为 rejected
|
|
self._put(f"/api/admin/jokes/{joke_id}", {"status": "rejected"})
|
|
self.stats["rejected"] += 1
|
|
self.stats["checked"] += 1
|
|
return True # 处理完成（已拒绝）
|
|
print(f" [OK] 有笑点: {check.get('reason', '')}")
|
|
except Exception as e:
|
|
print(f" [!] 质量检测失败: {e}，跳过本条")
|
|
self.stats["skipped"] += 1
|
|
return False
|
|
|
|
self.stats["checked"] += 1
|
|
|
|
# Stage 2: AI 润色
|
|
print(f" [2/3] AI 润色...")
|
|
try:
|
|
polished = self.polish(content)
|
|
if polished and polished != content:
|
|
print(f" [OK] 润色完成 ({len(content)} -> {len(polished)} 字)")
|
|
else:
|
|
print(f" [*] 润色后无变化")
|
|
except Exception as e:
|
|
print(f" [!] 润色失败: {e}")
|
|
polished = content # 润色失败时使用原文
|
|
|
|
# Stage 3: 评价分类
|
|
print(f" [3/3] 评价分类...")
|
|
try:
|
|
eval_result = self.evaluate(polished)
|
|
type_names = eval_result.get("types", [])
|
|
crowd_names = eval_result.get("crowds", [])
|
|
score = eval_result.get("score", 5)
|
|
print(f" [OK] 类型={type_names}, 人群={crowd_names}, 评分={score}/10")
|
|
except Exception as e:
|
|
print(f" [!] 评价分类失败: {e}")
|
|
type_names = []
|
|
crowd_names = []
|
|
score = None
|
|
|
|
# 根据评分计算 AI 等级
|
|
ai_level = None
|
|
if score is not None:
|
|
if score >= 8:
|
|
ai_level = "excellent"
|
|
elif score >= 6:
|
|
ai_level = "good"
|
|
elif score >= 4:
|
|
ai_level = "ordinary"
|
|
else:
|
|
ai_level = "poor"
|
|
|
|
# 保存更新
|
|
try:
|
|
update = {
|
|
"polished_content": polished,
|
|
"status": "approved" if (score or 5) >= 4 else "pending",
|
|
}
|
|
if score is not None:
|
|
update["ai_score"] = float(score)
|
|
if ai_level:
|
|
update["ai_level"] = ai_level
|
|
if type_names:
|
|
update["type_ids"] = []
|
|
for n in type_names:
|
|
tid = self._get_type_id(n)
|
|
if tid:
|
|
update["type_ids"].append(tid)
|
|
if not update["type_ids"]:
|
|
del update["type_ids"]
|
|
if crowd_names:
|
|
update["crowd_ids"] = []
|
|
for n in crowd_names:
|
|
cid = self._get_crowd_id(n)
|
|
if cid:
|
|
update["crowd_ids"].append(cid)
|
|
if not update["crowd_ids"]:
|
|
del update["crowd_ids"]
|
|
|
|
self._put(f"/api/admin/jokes/{joke_id}", update)
|
|
self.stats["polished"] += 1
|
|
self.stats["evaluated"] += 1
|
|
print(f" [OK] 更新成功")
|
|
return True
|
|
except Exception as e:
|
|
print(f" [!] 更新失败: {e}")
|
|
return False
|
|
|
|
# === 主循环 ===
|
|
def run(self, status: str | None = None, limit: int | None = None,
|
|
ids: list[int] | None = None):
|
|
"""主入口：读取笑话并逐个处理"""
|
|
print(f"\n>> 笑话优化模式启动")
|
|
print(f" 筛选状态: {status or '全部'}")
|
|
if limit:
|
|
print(f" 处理数量: {limit}")
|
|
if ids:
|
|
print(f" 指定 ID: {ids}")
|
|
|
|
self.setup()
|
|
|
|
jokes = self.get_jokes(status, limit, ids)
|
|
print(f"\n[*] 共读取 {len(jokes)} 条笑话，开始处理")
|
|
|
|
for idx, joke in enumerate(jokes):
|
|
print(f"\n --- 进度 {idx+1}/{len(jokes)} ---")
|
|
try:
|
|
self.process_joke(joke)
|
|
except KeyboardInterrupt:
|
|
print("\n用户中断")
|
|
break
|
|
except Exception as e:
|
|
print(f" [!] 处理异常: {e}")
|
|
self.stats["skipped"] += 1
|
|
|
|
# 每条间稍等，避免 API 限流（每次请求间隔 2-3 秒）
|
|
if idx < len(jokes) - 1:
|
|
import time
|
|
time.sleep(2)
|
|
|
|
# 输出统计
|
|
print(f"\n{'='*40}")
|
|
print(f" 处理完成")
|
|
print(f" {'='*40}")
|
|
print(f" 检查: {self.stats['checked']} 条")
|
|
print(f" 拒绝(无笑点): {self.stats['rejected']} 条")
|
|
print(f" 润色: {self.stats['polished']} 条")
|
|
print(f" 评价分类: {self.stats['evaluated']} 条")
|
|
print(f" 跳过(失败): {self.stats['skipped']} 条")
|
|
print(f"{'='*40}") |