feat(api): external promotion CRUD + audit endpoints

Add 20 API endpoints under /api/external/* covering products, campaigns, keywords, audits, rankings, optimization tasks with org_id filtering.

Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
This commit is contained in:
yuzhiran
2026-06-21 16:05:55 +08:00
parent 12e49f2811
commit 8f5a9e7ede
2 changed files with 471 additions and 1 deletions
@@ -0,0 +1,469 @@
from fastapi import APIRouter, Depends, HTTPException, Query
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, org_filter
router = APIRouter(prefix="/api/external", tags=["external_promotion"])
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,
}
}
+2 -1
View File
@@ -9,7 +9,7 @@ from pathlib import Path
from .database import engine, get_db, init_db from .database import engine, get_db, init_db
from .models import Base from .models import Base
from .api import topics, system, articles, publishing, auth, admin, audit, optimizer_logs, cases, task_logs, task_configs, prompt_configs, llm_configs, system_configs, topic_config, calendar, metrics, assets, tasks, platform_config, collector_mgmt, assistant, config_items, role_configs, menu_configs, search_providers, search_rankings from .api import topics, system, articles, publishing, auth, admin, audit, optimizer_logs, cases, task_logs, task_configs, prompt_configs, llm_configs, system_configs, topic_config, calendar, metrics, assets, tasks, platform_config, collector_mgmt, assistant, config_items, role_configs, menu_configs, search_providers, search_rankings, external_promotion
from .initial_data import import_initial_data from .initial_data import import_initial_data
from .core.scheduler import scheduler from .core.scheduler import scheduler
@@ -105,6 +105,7 @@ app.include_router(menu_configs.router)
app.include_router(menu_configs.public_router) app.include_router(menu_configs.public_router)
app.include_router(search_providers.router) app.include_router(search_providers.router)
app.include_router(search_rankings.router) app.include_router(search_rankings.router)
app.include_router(external_promotion.router)
# 挂载自动生成的图片(必须先于前端根挂载) # 挂载自动生成的图片(必须先于前端根挂载)
PROJECT_ROOT_DIR = Path(__file__).parent.parent.parent.parent PROJECT_ROOT_DIR = Path(__file__).parent.parent.parent.parent