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 市场调研简报
525 lines
17 KiB
Python
525 lines
17 KiB
Python
import os
|
|
import sys
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from pydantic import BaseModel
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import desc
|
|
from typing import Optional
|
|
from datetime import datetime, timezone
|
|
|
|
from ..database import get_db
|
|
from ..models import ExternalProduct, PromotionCampaign, CampaignKeyword, SEOAudit, KeywordRanking, OptimizationTask, User
|
|
from .auth import get_current_user, get_current_admin, org_filter
|
|
|
|
router = APIRouter(prefix="/api/external", tags=["external_promotion"])
|
|
|
|
PROJECT_ROOT = Path(__file__).parent.parent.parent.parent.parent
|
|
SEO_AUDITOR_PATH = PROJECT_ROOT / "scripts" / "seo_auditor.py"
|
|
|
|
|
|
class AuditRunRequest(BaseModel):
|
|
product_id: Optional[str] = None
|
|
url: Optional[str] = None
|
|
|
|
|
|
def _org_filtered_query(db, model, current_user):
|
|
query = db.query(model)
|
|
of = org_filter(current_user, model)
|
|
if of is not True:
|
|
query = query.filter(of)
|
|
return query
|
|
|
|
|
|
# ===== External Products =====
|
|
|
|
@router.get("/products")
|
|
def list_products(
|
|
type: Optional[str] = None,
|
|
limit: int = Query(50, ge=1, le=200),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
query = _org_filtered_query(db, ExternalProduct, current_user)
|
|
if type:
|
|
query = query.filter(ExternalProduct.type == type)
|
|
products = query.order_by(desc(ExternalProduct.created_at)).limit(limit).all()
|
|
return {"ok": True, "data": [p.to_dict() for p in products]}
|
|
|
|
|
|
@router.get("/products/{product_id}")
|
|
def get_product(
|
|
product_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
product = _org_filtered_query(db, ExternalProduct, current_user).filter(
|
|
ExternalProduct.id == product_id
|
|
).first()
|
|
if not product:
|
|
raise HTTPException(status_code=404, detail="产品不存在")
|
|
return {"ok": True, "data": product.to_dict()}
|
|
|
|
|
|
@router.post("/products")
|
|
def create_product(
|
|
body: dict,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
product = ExternalProduct(
|
|
name=body.get("name"),
|
|
type=body.get("type", "website"),
|
|
url=body.get("url"),
|
|
domain=body.get("domain"),
|
|
app_id=body.get("app_id"),
|
|
account_id=body.get("account_id"),
|
|
description=body.get("description"),
|
|
org_id=current_user.org_id,
|
|
)
|
|
db.add(product)
|
|
db.commit()
|
|
db.refresh(product)
|
|
return {"ok": True, "data": product.to_dict()}
|
|
|
|
|
|
@router.put("/products/{product_id}")
|
|
def update_product(
|
|
product_id: int,
|
|
body: dict,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
product = _org_filtered_query(db, ExternalProduct, current_user).filter(
|
|
ExternalProduct.id == product_id
|
|
).first()
|
|
if not product:
|
|
raise HTTPException(status_code=404, detail="产品不存在")
|
|
for field in ("name", "type", "url", "domain", "app_id", "account_id", "description"):
|
|
if field in body:
|
|
setattr(product, field, body[field])
|
|
db.commit()
|
|
db.refresh(product)
|
|
return {"ok": True, "data": product.to_dict()}
|
|
|
|
|
|
@router.delete("/products/{product_id}")
|
|
def delete_product(
|
|
product_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
product = _org_filtered_query(db, ExternalProduct, current_user).filter(
|
|
ExternalProduct.id == product_id
|
|
).first()
|
|
if not product:
|
|
raise HTTPException(status_code=404, detail="产品不存在")
|
|
db.delete(product)
|
|
db.commit()
|
|
return {"ok": True, "message": "已删除"}
|
|
|
|
|
|
# ===== Promotion Campaigns =====
|
|
|
|
@router.get("/campaigns")
|
|
def list_campaigns(
|
|
product_id: Optional[int] = None,
|
|
status: Optional[str] = None,
|
|
limit: int = Query(50, ge=1, le=200),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
query = _org_filtered_query(db, PromotionCampaign, current_user)
|
|
if product_id:
|
|
query = query.filter(PromotionCampaign.product_id == product_id)
|
|
if status:
|
|
query = query.filter(PromotionCampaign.status == status)
|
|
campaigns = query.order_by(desc(PromotionCampaign.created_at)).limit(limit).all()
|
|
result = []
|
|
for c in campaigns:
|
|
d = c.to_dict()
|
|
product = db.query(ExternalProduct).filter(ExternalProduct.id == c.product_id).first()
|
|
d["product_name"] = product.name if product else None
|
|
result.append(d)
|
|
return {"ok": True, "data": result}
|
|
|
|
|
|
@router.get("/campaigns/{campaign_id}")
|
|
def get_campaign(
|
|
campaign_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
campaign = _org_filtered_query(db, PromotionCampaign, current_user).filter(
|
|
PromotionCampaign.id == campaign_id
|
|
).first()
|
|
if not campaign:
|
|
raise HTTPException(status_code=404, detail="活动不存在")
|
|
d = campaign.to_dict()
|
|
product = db.query(ExternalProduct).filter(ExternalProduct.id == campaign.product_id).first()
|
|
d["product_name"] = product.name if product else None
|
|
return {"ok": True, "data": d}
|
|
|
|
|
|
@router.post("/campaigns")
|
|
def create_campaign(
|
|
body: dict,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
product = db.query(ExternalProduct).filter(ExternalProduct.id == body.get("product_id")).first()
|
|
if not product:
|
|
raise HTTPException(status_code=404, detail="产品不存在")
|
|
campaign = PromotionCampaign(
|
|
name=body.get("name"),
|
|
product_id=body.get("product_id"),
|
|
status=body.get("status", "active"),
|
|
keywords=body.get("keywords", []),
|
|
target_engines=body.get("target_engines", []),
|
|
notes=body.get("notes"),
|
|
org_id=current_user.org_id,
|
|
)
|
|
db.add(campaign)
|
|
db.commit()
|
|
db.refresh(campaign)
|
|
d = campaign.to_dict()
|
|
d["product_name"] = product.name
|
|
return {"ok": True, "data": d}
|
|
|
|
|
|
@router.put("/campaigns/{campaign_id}")
|
|
def update_campaign(
|
|
campaign_id: int,
|
|
body: dict,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
campaign = _org_filtered_query(db, PromotionCampaign, current_user).filter(
|
|
PromotionCampaign.id == campaign_id
|
|
).first()
|
|
if not campaign:
|
|
raise HTTPException(status_code=404, detail="活动不存在")
|
|
for field in ("name", "status", "keywords", "target_engines", "notes"):
|
|
if field in body:
|
|
setattr(campaign, field, body[field])
|
|
db.commit()
|
|
db.refresh(campaign)
|
|
d = campaign.to_dict()
|
|
product = db.query(ExternalProduct).filter(ExternalProduct.id == campaign.product_id).first()
|
|
d["product_name"] = product.name if product else None
|
|
return {"ok": True, "data": d}
|
|
|
|
|
|
@router.delete("/campaigns/{campaign_id}")
|
|
def delete_campaign(
|
|
campaign_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
campaign = _org_filtered_query(db, PromotionCampaign, current_user).filter(
|
|
PromotionCampaign.id == campaign_id
|
|
).first()
|
|
if not campaign:
|
|
raise HTTPException(status_code=404, detail="活动不存在")
|
|
db.delete(campaign)
|
|
db.commit()
|
|
return {"ok": True, "message": "已删除"}
|
|
|
|
|
|
# ===== Campaign Keywords =====
|
|
|
|
@router.get("/keywords")
|
|
def list_keywords(
|
|
campaign_id: Optional[int] = None,
|
|
limit: int = Query(100, ge=1, le=500),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
query = _org_filtered_query(db, CampaignKeyword, current_user)
|
|
if campaign_id:
|
|
query = query.filter(CampaignKeyword.campaign_id == campaign_id)
|
|
keywords = query.order_by(desc(CampaignKeyword.created_at)).limit(limit).all()
|
|
return {"ok": True, "data": [k.to_dict() for k in keywords]}
|
|
|
|
|
|
@router.post("/keywords")
|
|
def create_keyword(
|
|
body: dict,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
campaign = db.query(PromotionCampaign).filter(PromotionCampaign.id == body.get("campaign_id")).first()
|
|
if not campaign:
|
|
raise HTTPException(status_code=404, detail="活动不存在")
|
|
kw = CampaignKeyword(
|
|
campaign_id=body.get("campaign_id"),
|
|
keyword=body.get("keyword"),
|
|
search_volume=body.get("search_volume"),
|
|
difficulty=body.get("difficulty"),
|
|
current_rank=body.get("current_rank"),
|
|
target_rank=body.get("target_rank"),
|
|
org_id=current_user.org_id,
|
|
)
|
|
db.add(kw)
|
|
db.commit()
|
|
db.refresh(kw)
|
|
return {"ok": True, "data": kw.to_dict()}
|
|
|
|
|
|
@router.put("/keywords/{keyword_id}")
|
|
def update_keyword(
|
|
keyword_id: int,
|
|
body: dict,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
kw = _org_filtered_query(db, CampaignKeyword, current_user).filter(
|
|
CampaignKeyword.id == keyword_id
|
|
).first()
|
|
if not kw:
|
|
raise HTTPException(status_code=404, detail="关键词不存在")
|
|
for field in ("search_volume", "difficulty", "current_rank", "target_rank"):
|
|
if field in body:
|
|
setattr(kw, field, body[field])
|
|
if body.get("update_last_checked"):
|
|
kw.last_checked = datetime.now(timezone.utc)
|
|
db.commit()
|
|
db.refresh(kw)
|
|
return {"ok": True, "data": kw.to_dict()}
|
|
|
|
|
|
@router.delete("/keywords/{keyword_id}")
|
|
def delete_keyword(
|
|
keyword_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
kw = _org_filtered_query(db, CampaignKeyword, current_user).filter(
|
|
CampaignKeyword.id == keyword_id
|
|
).first()
|
|
if not kw:
|
|
raise HTTPException(status_code=404, detail="关键词不存在")
|
|
db.delete(kw)
|
|
db.commit()
|
|
return {"ok": True, "message": "已删除"}
|
|
|
|
|
|
# ===== SEO Audits =====
|
|
|
|
@router.get("/audits")
|
|
def list_audits(
|
|
product_id: Optional[int] = None,
|
|
limit: int = Query(50, ge=1, le=200),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
query = _org_filtered_query(db, SEOAudit, current_user)
|
|
if product_id:
|
|
query = query.filter(SEOAudit.product_id == product_id)
|
|
audits = query.order_by(desc(SEOAudit.checked_at)).limit(limit).all()
|
|
result = []
|
|
for a in audits:
|
|
d = a.to_dict()
|
|
product = db.query(ExternalProduct).filter(ExternalProduct.id == a.product_id).first()
|
|
d["product_name"] = product.name if product else None
|
|
result.append(d)
|
|
return {"ok": True, "data": result}
|
|
|
|
|
|
@router.get("/audits/{audit_id}")
|
|
def get_audit(
|
|
audit_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
audit = _org_filtered_query(db, SEOAudit, current_user).filter(
|
|
SEOAudit.id == audit_id
|
|
).first()
|
|
if not audit:
|
|
raise HTTPException(status_code=404, detail="审计记录不存在")
|
|
d = audit.to_dict()
|
|
d["raw_data"] = audit.raw_data or {}
|
|
tasks = db.query(OptimizationTask).filter(OptimizationTask.audit_id == audit_id).all()
|
|
d["optimization_tasks"] = [t.to_dict() for t in tasks]
|
|
return {"ok": True, "data": d}
|
|
|
|
|
|
# ===== Keyword Rankings =====
|
|
|
|
@router.get("/rankings")
|
|
def list_keyword_rankings(
|
|
campaign_id: Optional[int] = None,
|
|
keyword_id: Optional[int] = None,
|
|
product_id: Optional[int] = None,
|
|
search_engine: Optional[str] = None,
|
|
limit: int = Query(100, ge=1, le=500),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
query = _org_filtered_query(db, KeywordRanking, current_user)
|
|
if campaign_id:
|
|
query = query.filter(KeywordRanking.campaign_id == campaign_id)
|
|
if keyword_id:
|
|
query = query.filter(KeywordRanking.keyword_id == keyword_id)
|
|
if product_id:
|
|
query = query.filter(KeywordRanking.product_id == product_id)
|
|
if search_engine:
|
|
query = query.filter(KeywordRanking.search_engine == search_engine)
|
|
rankings = query.order_by(desc(KeywordRanking.checked_at)).limit(limit).all()
|
|
return {"ok": True, "data": [r.to_dict() for r in rankings]}
|
|
|
|
|
|
@router.get("/rankings/overview")
|
|
def rankings_overview(
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
base = _org_filtered_query(db, KeywordRanking, current_user)
|
|
total = base.count()
|
|
with_rank = base.filter(KeywordRanking.rank.isnot(None)).count()
|
|
engine_breakdown = {}
|
|
for row in base.with_entities(KeywordRanking.search_engine,
|
|
KeywordRanking.rank).all():
|
|
eng = row.search_engine
|
|
if eng not in engine_breakdown:
|
|
engine_breakdown[eng] = {"total": 0, "with_rank": 0}
|
|
engine_breakdown[eng]["total"] += 1
|
|
if row.rank is not None:
|
|
engine_breakdown[eng]["with_rank"] += 1
|
|
return {
|
|
"ok": True,
|
|
"data": {
|
|
"total": total,
|
|
"with_rank": with_rank,
|
|
"engine_breakdown": engine_breakdown,
|
|
}
|
|
}
|
|
|
|
|
|
# ===== Optimization Tasks =====
|
|
|
|
@router.get("/optimization-tasks")
|
|
def list_optimization_tasks(
|
|
product_id: Optional[int] = None,
|
|
audit_id: Optional[int] = None,
|
|
status: Optional[str] = None,
|
|
severity: Optional[str] = None,
|
|
limit: int = Query(100, ge=1, le=500),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
query = _org_filtered_query(db, OptimizationTask, current_user)
|
|
if product_id:
|
|
query = query.filter(OptimizationTask.product_id == product_id)
|
|
if audit_id:
|
|
query = query.filter(OptimizationTask.audit_id == audit_id)
|
|
if status:
|
|
query = query.filter(OptimizationTask.status == status)
|
|
if severity:
|
|
query = query.filter(OptimizationTask.severity == severity)
|
|
tasks = query.order_by(desc(OptimizationTask.created_at)).limit(limit).all()
|
|
return {"ok": True, "data": [t.to_dict() for t in tasks]}
|
|
|
|
|
|
@router.patch("/optimization-tasks/{task_id}")
|
|
def update_optimization_task(
|
|
task_id: int,
|
|
body: dict,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
task = _org_filtered_query(db, OptimizationTask, current_user).filter(
|
|
OptimizationTask.id == task_id
|
|
).first()
|
|
if not task:
|
|
raise HTTPException(status_code=404, detail="优化任务不存在")
|
|
if "status" in body:
|
|
task.status = body["status"]
|
|
if "severity" in body:
|
|
task.severity = body["severity"]
|
|
if "recommendation" in body:
|
|
task.recommendation = body["recommendation"]
|
|
db.commit()
|
|
db.refresh(task)
|
|
return {"ok": True, "data": task.to_dict()}
|
|
|
|
|
|
@router.get("/overview")
|
|
def external_overview(
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
products = _org_filtered_query(db, ExternalProduct, current_user).count()
|
|
campaigns = _org_filtered_query(db, PromotionCampaign, current_user).count()
|
|
active_campaigns = _org_filtered_query(db, PromotionCampaign, current_user).filter(
|
|
PromotionCampaign.status == "active"
|
|
).count()
|
|
audits = _org_filtered_query(db, SEOAudit, current_user).count()
|
|
keywords = _org_filtered_query(db, CampaignKeyword, current_user).count()
|
|
open_tasks = _org_filtered_query(db, OptimizationTask, current_user).filter(
|
|
OptimizationTask.status == "open"
|
|
).count()
|
|
|
|
type_breakdown = {}
|
|
for row in _org_filtered_query(db, ExternalProduct, current_user).with_entities(
|
|
ExternalProduct.type
|
|
).all():
|
|
t = row.type
|
|
type_breakdown[t] = type_breakdown.get(t, 0) + 1
|
|
|
|
return {
|
|
"ok": True,
|
|
"data": {
|
|
"products": products,
|
|
"campaigns": campaigns,
|
|
"active_campaigns": active_campaigns,
|
|
"audits": audits,
|
|
"keywords": keywords,
|
|
"open_tasks": open_tasks,
|
|
"type_breakdown": type_breakdown,
|
|
}
|
|
}
|
|
|
|
|
|
# ===== Run SEO Audit (delegate to scripts/seo_auditor.py) =====
|
|
|
|
@router.post("/audit/run")
|
|
def run_audit(
|
|
body: AuditRunRequest,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_admin),
|
|
):
|
|
cmd = [sys.executable, str(SEO_AUDITOR_PATH)]
|
|
if body.product_id:
|
|
cmd += ["--product-id", str(body.product_id)]
|
|
elif body.url:
|
|
cmd += ["--url", body.url]
|
|
else:
|
|
raise HTTPException(status_code=400, detail="product_id 或 url 至少提供一个")
|
|
|
|
try:
|
|
result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
|
|
except subprocess.TimeoutExpired:
|
|
raise HTTPException(status_code=408, detail="SEO 审计超时")
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"审计执行失败: {str(e)}")
|
|
|
|
if result.returncode != 0:
|
|
raise HTTPException(
|
|
status_code=502,
|
|
detail={
|
|
"message": "SEO 审计脚本返回错误",
|
|
"stderr": result.stderr,
|
|
"stdout": result.stdout,
|
|
},
|
|
)
|
|
|
|
return {
|
|
"ok": True,
|
|
"stdout": result.stdout,
|
|
"stderr": result.stderr,
|
|
"returncode": result.returncode,
|
|
}
|