df1856d793
- 产品编辑入口统一走 ProductFormFull(卖点/风格/人群/品牌词全字段); 修复开任务页 <form> 套 <form> 致"编辑产品"报错、改不了、跳回首个产品 - dashboard 入口卡片对齐实际路由: 系统管理(/config) 与 工作配置(/settings) 分开; settings ?tab=products 直达改用挂载后读 URL, 消除 hydration mismatch - 新增用户管理(users API/admin service/改密页) + alembic 022/023/024 - 上线部署: Dockerfile / docker-compose.prod+https / nginx https / .env.example - A8 三套正交叙事(痛点/场景/成分背书) + beige 调色去AI化 + 飞轮 text_import 高权重信号 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
353 lines
13 KiB
Python
353 lines
13 KiB
Python
"""
|
||
app/api/v1/benchmarks.py — 标杆笔记 + 违禁词路由（管理员）
|
||
主通道：截图+手填亮点（不做原始抓取）。
|
||
"""
|
||
|
||
import logging
|
||
import os
|
||
import uuid
|
||
from typing import Annotated
|
||
|
||
from fastapi import APIRouter, Depends, File, Form, UploadFile
|
||
from pydantic import BaseModel
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.config import get_settings
|
||
from app.core.database import get_db
|
||
from app.core.response import ok, paginate, raise_business, raise_not_found
|
||
from app.middleware.workspace_guard import CurrentUser, require_admin, require_write_permission
|
||
from app.models.product import BannedWord, BenchmarkNote, Product
|
||
|
||
logger = logging.getLogger(__name__)
|
||
router = APIRouter(tags=["products"])
|
||
|
||
|
||
class BenchmarkCreate(BaseModel):
|
||
screenshot_url: str | None = None
|
||
highlights: str | None = None # 手填亮点（主通道）
|
||
link_url: str | None = None # 可选，不做自动抓取
|
||
|
||
|
||
class BannedWordCreate(BaseModel):
|
||
word: str
|
||
level: str # auto_fix | soft_warn | hard_block
|
||
replacement: str | None = None
|
||
updatable: bool = True
|
||
|
||
|
||
def _fmt_benchmark(b: BenchmarkNote) -> dict:
|
||
import json
|
||
features = None
|
||
if b.features_json:
|
||
try:
|
||
features = json.loads(b.features_json)
|
||
except Exception:
|
||
features = None
|
||
return {
|
||
"id": b.id, "product_id": b.product_id,
|
||
"screenshot_url": b.screenshot_url,
|
||
"highlights": b.highlights, "link_url": b.link_url,
|
||
"features": features, "analyze_status": b.analyze_status,
|
||
"analysis_source": features.get("source") if isinstance(features, dict) else None,
|
||
"analysis_warning": features.get("warning") if isinstance(features, dict) else None,
|
||
"created_at": b.created_at.isoformat(),
|
||
}
|
||
|
||
|
||
def _fmt_banned(bw: BannedWord) -> dict:
|
||
return {
|
||
"id": bw.id, "word": bw.word, "level": bw.level,
|
||
"replacement": bw.replacement, "updatable": bw.updatable,
|
||
"workspace_id": bw.workspace_id,
|
||
}
|
||
|
||
|
||
_ALLOWED_IMAGE_CONTENT_TYPES = {"image/jpeg", "image/png", "image/webp"}
|
||
_MAX_IMAGE_SIZE_BYTES = 10 * 1024 * 1024
|
||
|
||
|
||
def _check_image_magic(data: bytes) -> bool:
|
||
if len(data) < 12:
|
||
return False
|
||
if data[:3] == b"\xff\xd8\xff":
|
||
return True
|
||
if data[:8] == b"\x89PNG\r\n\x1a\n":
|
||
return True
|
||
if data[:4] == b"RIFF" and data[8:12] == b"WEBP":
|
||
return True
|
||
return False
|
||
|
||
|
||
def _clean_link_url(link_url: str | None) -> str | None:
|
||
value = (link_url or "").strip()
|
||
if not value:
|
||
return None
|
||
if not value.startswith(("http://", "https://")):
|
||
raise_business("原始链接必须以 http:// 或 https:// 开头")
|
||
return value
|
||
|
||
|
||
def _is_benchmark_upload_path(path: str) -> bool:
|
||
upload_root = os.path.abspath(get_settings().UPLOAD_ABS_ROOT)
|
||
benchmark_root = os.path.join(upload_root, "benchmarks")
|
||
candidate = os.path.abspath(path)
|
||
try:
|
||
return os.path.commonpath([benchmark_root, candidate]) == benchmark_root
|
||
except ValueError:
|
||
return False
|
||
|
||
|
||
async def _save_benchmark_screenshot(
|
||
file: UploadFile,
|
||
workspace_id: int,
|
||
product_id: int,
|
||
) -> str:
|
||
if file.content_type not in _ALLOWED_IMAGE_CONTENT_TYPES:
|
||
raise_business(f"不支持的截图类型 {file.content_type}，仅支持 JPEG/PNG/WebP")
|
||
|
||
data = await file.read()
|
||
if len(data) > _MAX_IMAGE_SIZE_BYTES:
|
||
raise_business("截图超过 10 MB 限制")
|
||
if not _check_image_magic(data):
|
||
raise_business("截图文件内容与扩展名不符（非真实 JPEG/PNG/WebP 图片）")
|
||
|
||
settings = get_settings()
|
||
ext = os.path.splitext(file.filename or "benchmark.jpg")[1] or ".jpg"
|
||
abs_dir = os.path.join(
|
||
settings.UPLOAD_ABS_ROOT,
|
||
"benchmarks",
|
||
str(workspace_id),
|
||
str(product_id),
|
||
)
|
||
os.makedirs(abs_dir, exist_ok=True)
|
||
save_path = os.path.join(abs_dir, f"{uuid.uuid4().hex}{ext}")
|
||
with open(save_path, "wb") as f:
|
||
f.write(data)
|
||
return save_path
|
||
|
||
|
||
# ── 标杆笔记 ────────────────────────────────────────────────
|
||
@router.get("/products/{product_id}/benchmarks")
|
||
def list_benchmarks(
|
||
product_id: int,
|
||
current_user: Annotated[CurrentUser, Depends(require_write_permission)] = None,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
items = (
|
||
db.query(BenchmarkNote)
|
||
.filter(
|
||
BenchmarkNote.product_id == product_id,
|
||
BenchmarkNote.workspace_id == current_user.workspace_id,
|
||
)
|
||
.all()
|
||
)
|
||
return ok({"items": [_fmt_benchmark(b) for b in items]})
|
||
|
||
|
||
@router.post("/products/{product_id}/benchmarks")
|
||
def create_benchmark(
|
||
product_id: int, body: BenchmarkCreate,
|
||
current_user: Annotated[CurrentUser, Depends(require_admin)] = None,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
product = db.query(Product).filter(
|
||
Product.id == product_id,
|
||
Product.workspace_id == current_user.workspace_id,
|
||
).first()
|
||
if not product:
|
||
raise_not_found("产品不存在")
|
||
# 兼容旧前端：旧字段 screenshot_url 实际可能传的是外部原帖链接。
|
||
legacy_link = (body.screenshot_url or "").strip()
|
||
highlights = (body.highlights or "").strip()
|
||
link_url = _clean_link_url(body.link_url or legacy_link)
|
||
if not highlights and not link_url:
|
||
raise_business("请至少填写原始链接或亮点")
|
||
b = BenchmarkNote(
|
||
workspace_id=current_user.workspace_id,
|
||
product_id=product_id,
|
||
screenshot_url=None,
|
||
highlights=highlights,
|
||
link_url=link_url,
|
||
)
|
||
db.add(b); db.commit(); db.refresh(b)
|
||
return ok(_fmt_benchmark(b))
|
||
|
||
|
||
@router.post("/products/{product_id}/benchmarks/upload")
|
||
async def upload_benchmark(
|
||
product_id: int,
|
||
screenshot: UploadFile | None = File(None),
|
||
highlights: str | None = Form(None),
|
||
link_url: str | None = Form(None),
|
||
current_user: Annotated[CurrentUser, Depends(require_admin)] = None,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
product = db.query(Product).filter(
|
||
Product.id == product_id,
|
||
Product.workspace_id == current_user.workspace_id,
|
||
).first()
|
||
if not product:
|
||
raise_not_found("产品不存在")
|
||
|
||
clean_highlights = (highlights or "").strip()
|
||
clean_link_url = _clean_link_url(link_url)
|
||
if not screenshot and not clean_highlights and not clean_link_url:
|
||
raise_business("请至少上传截图、填写原始链接或填写亮点")
|
||
|
||
screenshot_path = None
|
||
if screenshot:
|
||
screenshot_path = await _save_benchmark_screenshot(
|
||
screenshot,
|
||
current_user.workspace_id,
|
||
product_id,
|
||
)
|
||
|
||
b = BenchmarkNote(
|
||
workspace_id=current_user.workspace_id,
|
||
product_id=product_id,
|
||
screenshot_url=screenshot_path,
|
||
highlights=clean_highlights,
|
||
link_url=clean_link_url,
|
||
)
|
||
db.add(b); db.commit(); db.refresh(b)
|
||
return ok(_fmt_benchmark(b))
|
||
|
||
|
||
@router.post("/benchmarks/{benchmark_id}/analyze")
|
||
async def analyze_benchmark_note(
|
||
benchmark_id: int,
|
||
current_user: Annotated[CurrentUser, Depends(require_admin)] = None,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
"""
|
||
第2环：对已存标杆笔记跑 8 维拆解。
|
||
输入源=screenshot_url 截图(vision) + highlights 手填亮点(参考)。
|
||
结果写入 features_json，analyze_status 流转 analyzing→done/failed。
|
||
key 链路遵基石B：plain_key 只活局部，立即清零。
|
||
"""
|
||
import json
|
||
from app.core.response import raise_business
|
||
from app.models.workspace import UserApiKey
|
||
from app.utils.fernet_utils import decrypt_key
|
||
from app.services.ai_engine.gemini_factory import build_ai_clients
|
||
from app.services.ai_engine.benchmark_analyzer import analyze_benchmark
|
||
|
||
b = db.query(BenchmarkNote).filter(
|
||
BenchmarkNote.id == benchmark_id,
|
||
BenchmarkNote.workspace_id == current_user.workspace_id,
|
||
).first()
|
||
if not b:
|
||
raise_not_found("标杆笔记不存在")
|
||
if not b.screenshot_url and not b.highlights and not b.link_url:
|
||
raise_business("该标杆无截图、原始链接和手填亮点，无法分析")
|
||
|
||
# 读截图字节：新数据存本地上传路径；旧数据可能误存外部 URL，不能直接 open。
|
||
screenshot = None
|
||
if b.screenshot_url:
|
||
if b.screenshot_url.startswith(("http://", "https://")):
|
||
logger.info("标杆截图字段为外部 URL，跳过本地读取 id=%s", benchmark_id)
|
||
elif _is_benchmark_upload_path(b.screenshot_url) and os.path.exists(b.screenshot_url):
|
||
try:
|
||
with open(b.screenshot_url, "rb") as f:
|
||
screenshot = f.read()
|
||
except Exception as e:
|
||
logger.warning("标杆截图读取失败 id=%s: %s", benchmark_id, e)
|
||
elif os.path.exists(b.screenshot_url):
|
||
logger.warning("标杆截图路径不在上传目录，跳过读取 id=%s path=%s", benchmark_id, b.screenshot_url)
|
||
else:
|
||
logger.warning("标杆截图路径不存在 id=%s path=%s", benchmark_id, b.screenshot_url)
|
||
|
||
api_key_row = db.query(UserApiKey).filter(
|
||
UserApiKey.user_id == current_user.user_id,
|
||
UserApiKey.workspace_id == current_user.workspace_id,
|
||
UserApiKey.provider.in_(["openai", "apiports"]),
|
||
).first()
|
||
if not api_key_row:
|
||
raise_business("未配置 API Key，请先在设置中录入")
|
||
plain_key = decrypt_key(api_key_row.encrypted_key)
|
||
|
||
# codeproxy 备用 key（可选，用户录入则用，没录回落 env）
|
||
alt_row = db.query(UserApiKey).filter(
|
||
UserApiKey.user_id == current_user.user_id,
|
||
UserApiKey.workspace_id == current_user.workspace_id,
|
||
UserApiKey.provider == "codeproxy",
|
||
).first()
|
||
alt_key = decrypt_key(alt_row.encrypted_key) if alt_row else None
|
||
clients = build_ai_clients(plain_key, alt_key=alt_key)
|
||
plain_key = None
|
||
alt_key = None
|
||
|
||
b.analyze_status = "analyzing"; db.commit()
|
||
try:
|
||
analysis_text_parts = []
|
||
if b.highlights:
|
||
analysis_text_parts.append(b.highlights)
|
||
if b.link_url:
|
||
analysis_text_parts.append(f"原始链接：{b.link_url}")
|
||
result = await analyze_benchmark(clients, screenshot, "\n".join(analysis_text_parts))
|
||
finally:
|
||
await clients.aclose()
|
||
|
||
b.features_json = json.dumps(result, ensure_ascii=False)
|
||
b.analyze_status = "done" if result.get("source") != "fallback" else "failed"
|
||
db.commit(); db.refresh(b)
|
||
logger.info("benchmark analyzed id=%s status=%s source=%s",
|
||
benchmark_id, b.analyze_status, result.get("source"))
|
||
return ok({**_fmt_benchmark(b), "features": result, "analyze_status": b.analyze_status})
|
||
@router.get("/banned-words")
|
||
def list_banned_words(
|
||
page: int = 1, page_size: int = 50,
|
||
current_user: Annotated[CurrentUser, Depends(require_write_permission)] = None,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
q = db.query(BannedWord).filter(BannedWord.workspace_id == current_user.workspace_id)
|
||
total = q.count()
|
||
items = q.offset((page - 1) * page_size).limit(page_size).all()
|
||
return ok(paginate([_fmt_banned(bw) for bw in items], total, page, page_size))
|
||
|
||
|
||
@router.post("/banned-words")
|
||
def create_banned_word(
|
||
body: BannedWordCreate,
|
||
current_user: Annotated[CurrentUser, Depends(require_admin)] = None,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
bw = BannedWord(
|
||
workspace_id=current_user.workspace_id,
|
||
word=body.word, level=body.level,
|
||
replacement=body.replacement, updatable=body.updatable,
|
||
)
|
||
db.add(bw); db.commit(); db.refresh(bw)
|
||
return ok(_fmt_banned(bw))
|
||
|
||
|
||
@router.put("/banned-words/{word_id}")
|
||
def update_banned_word(
|
||
word_id: int, body: BannedWordCreate,
|
||
current_user: Annotated[CurrentUser, Depends(require_admin)] = None,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
bw = db.query(BannedWord).filter(BannedWord.id == word_id, BannedWord.workspace_id == current_user.workspace_id).first()
|
||
if not bw:
|
||
raise_not_found("违禁词不存在")
|
||
if not bw.updatable:
|
||
from app.core.response import raise_forbidden
|
||
raise_forbidden("该违禁词不可修改")
|
||
bw.word = body.word; bw.level = body.level
|
||
bw.replacement = body.replacement; bw.updatable = body.updatable
|
||
db.commit(); db.refresh(bw)
|
||
return ok(_fmt_banned(bw))
|
||
|
||
|
||
@router.delete("/banned-words/{word_id}")
|
||
def delete_banned_word(
|
||
word_id: int,
|
||
current_user: Annotated[CurrentUser, Depends(require_admin)] = None,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
bw = db.query(BannedWord).filter(BannedWord.id == word_id, BannedWord.workspace_id == current_user.workspace_id).first()
|
||
if not bw:
|
||
raise_not_found("违禁词不存在")
|
||
db.delete(bw); db.commit()
|
||
return ok({"deleted": word_id})
|