Yuzhiran Dev
5caf7bc52e
- Add org_id auto-population to articles.py, publishing.py, metrics.py - Add org_id filtering to search_rankings.py geo endpoints - Add org_id to calendar.py create/update/delete endpoints - Add org_id to tasks.py create endpoints - Create factory.html content pipeline page (3-tab: 待创作/进行中/待审查) - Create insights.html analytics page (4-tab: 概览/搜索排名/GEO/平台对比)
185 lines
7.8 KiB
Python
185 lines
7.8 KiB
Python
import re
|
|
from fastapi import APIRouter, HTTPException, Query, Depends, Body
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import or_
|
|
from pathlib import Path
|
|
import os
|
|
from datetime import datetime, date
|
|
|
|
from ..database import get_db
|
|
from ..models import User, Article, Topic
|
|
from .auth import get_current_user, org_filter
|
|
|
|
router = APIRouter(prefix="/api/articles", tags=["articles"])
|
|
|
|
@router.get("/drafts")
|
|
def list_drafts(topic_id: str = None, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)):
|
|
"""从 articles 表列出草稿"""
|
|
query = db.query(Article).join(Topic, Article.topic_id == Topic.id)
|
|
of = org_filter(current_user, Topic)
|
|
if of is not True:
|
|
query = query.filter(of)
|
|
if topic_id:
|
|
query = query.filter(Article.topic_id == topic_id)
|
|
articles = query.order_by(Article.created_at.desc()).all()
|
|
result = {}
|
|
for a in articles:
|
|
result.setdefault(a.platform, []).append({"id": a.id, "topic_id": a.topic_id, "status": a.status})
|
|
return {"articles": result}
|
|
|
|
@router.get("/list")
|
|
def list_articles(
|
|
platform: str = None,
|
|
status: str = None,
|
|
search: str = None,
|
|
topic_id: str = None,
|
|
current_user: User = Depends(get_current_user),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""列出所有文章（含选题标题）"""
|
|
query = db.query(Article, Topic.title.label("topic_title")).join(Topic, Article.topic_id == Topic.id)
|
|
of = org_filter(current_user, Topic)
|
|
if of is not True:
|
|
query = query.filter(of)
|
|
if platform:
|
|
query = query.filter(Article.platform == platform)
|
|
if status:
|
|
query = query.filter(Article.status == status)
|
|
if topic_id:
|
|
query = query.filter(Article.topic_id == topic_id)
|
|
if search:
|
|
query = query.filter(
|
|
or_(Article.id.ilike(f"%{search}%"), Topic.title.ilike(f"%{search}%"))
|
|
)
|
|
rows = query.order_by(Article.created_at.desc()).all()
|
|
return {
|
|
"articles": [
|
|
{
|
|
"id": r.Article.id,
|
|
"topic_id": r.Article.topic_id,
|
|
"topic_title": r.topic_title,
|
|
"platform": r.Article.platform,
|
|
"status": r.Article.status,
|
|
"compliance_score": r.Article.compliance_score,
|
|
"word_count": r.Article.word_count,
|
|
"created_at": r.Article.created_at.isoformat() if r.Article.created_at else None,
|
|
"images": r.Article.images or {},
|
|
}
|
|
for r in rows
|
|
]
|
|
}
|
|
|
|
@router.get("/detail/{article_id}")
|
|
def get_article_detail(article_id: str, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)):
|
|
"""获取文章详情"""
|
|
article = db.query(Article).filter(Article.id == article_id).first()
|
|
if not article:
|
|
raise HTTPException(status_code=404, detail="Article not found")
|
|
topic = db.query(Topic).filter(Topic.id == article.topic_id).first()
|
|
if topic and current_user.role != "admin" and topic.org_id != current_user.org_id:
|
|
raise HTTPException(status_code=404, detail="Article not found")
|
|
return {
|
|
"id": article.id,
|
|
"topic_id": article.topic_id,
|
|
"topic_title": topic.title if topic else None,
|
|
"platform": article.platform,
|
|
"file_path": article.file_path,
|
|
"title": article.title,
|
|
"content": article.content,
|
|
"status": article.status,
|
|
"compliance_score": article.compliance_score,
|
|
"word_count": article.word_count,
|
|
"outline": article.outline,
|
|
"html_content": article.html_content,
|
|
"images": article.images or {},
|
|
"created_at": article.created_at.isoformat() if article.created_at else None,
|
|
}
|
|
|
|
@router.delete("/{article_id}")
|
|
def delete_article(article_id: str, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)):
|
|
"""删除文章"""
|
|
article = db.query(Article).filter(Article.id == article_id).first()
|
|
if not article:
|
|
raise HTTPException(status_code=404, detail="Article not found")
|
|
topic = db.query(Topic).filter(Topic.id == article.topic_id).first()
|
|
if topic and current_user.role != "admin" and topic.org_id != current_user.org_id:
|
|
raise HTTPException(status_code=404, detail="Article not found")
|
|
file_path = Path(article.file_path) if article.file_path else None
|
|
db.delete(article)
|
|
db.commit()
|
|
if file_path and file_path.exists():
|
|
try:
|
|
file_path.unlink()
|
|
except OSError:
|
|
pass
|
|
return {"detail": "deleted"}
|
|
|
|
@router.get("/{topic_id}/images")
|
|
def get_article_images(topic_id: str, platform: str = None, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)):
|
|
"""获取某选题在各平台的配图路径"""
|
|
topic = db.query(Topic).filter(Topic.id == topic_id).first()
|
|
if topic and current_user.role != "admin" and topic.org_id != current_user.org_id:
|
|
raise HTTPException(status_code=404, detail="Topic not found")
|
|
query = db.query(Article).filter(Article.topic_id == topic_id)
|
|
if platform:
|
|
query = query.filter(Article.platform == platform)
|
|
articles = query.all()
|
|
return {
|
|
topic_id: {
|
|
a.platform: (a.images or {})
|
|
for a in articles
|
|
}
|
|
}
|
|
|
|
@router.get("/{topic_id}/preview")
|
|
def preview_article(topic_id: str, platform: str = "zhihu", current_user: User = Depends(get_current_user), db: Session = Depends(get_db)):
|
|
"""从 articles 表预览某选题的 HTML 内容"""
|
|
topic = db.query(Topic).filter(Topic.id == topic_id).first()
|
|
if topic and current_user.role != "admin" and topic.org_id != current_user.org_id:
|
|
raise HTTPException(status_code=404, detail="Topic not found")
|
|
article_id = f"{platform}_{topic_id}"
|
|
article = db.query(Article).filter(Article.id == article_id).first()
|
|
if not article or not article.html_content:
|
|
raise HTTPException(status_code=404, detail=f"Article not found for {topic_id} on {platform}")
|
|
return {"topic_id": topic_id, "platform": platform, "html": article.html_content, "images": article.images or {}}
|
|
|
|
@router.put("/{topic_id}/content")
|
|
def update_article_content(
|
|
topic_id: str,
|
|
platform: str = Body(...),
|
|
html_content: str = Body(...),
|
|
current_user: User = Depends(get_current_user),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""保存用户编辑后的文章 HTML 内容"""
|
|
topic = db.query(Topic).filter(Topic.id == topic_id).first()
|
|
if topic and current_user.role != "admin" and topic.org_id != current_user.org_id:
|
|
raise HTTPException(status_code=404, detail="Topic not found")
|
|
article_id = f"{platform}_{topic_id}"
|
|
article = db.query(Article).filter(Article.id == article_id).first()
|
|
if not article:
|
|
article = Article(
|
|
id=article_id, topic_id=topic_id, platform=platform,
|
|
file_path=f"db:{article_id}", status="draft",
|
|
org_id=topic.org_id
|
|
)
|
|
db.add(article)
|
|
article.html_content = html_content
|
|
db.commit()
|
|
return {"ok": True, "id": article_id}
|
|
|
|
@router.get("/optimization-report")
|
|
def get_optimization_report(publish_date: str = None, current_user: User = Depends(get_current_user)):
|
|
"""获取合规优化报告"""
|
|
if not publish_date:
|
|
publish_date = date.today().isoformat()
|
|
if not re.match(r'^\d{4}-\d{2}-\d{2}$', publish_date):
|
|
raise HTTPException(status_code=400, detail="Invalid date format (expected YYYY-MM-DD)")
|
|
PROJECT_ROOT = Path('/root/openclaw-workspace/projects/yu-zhi-ran')
|
|
report_path = PROJECT_ROOT / "automation" / "data" / "drafts" / publish_date / "optimization_report.json"
|
|
if not report_path.exists():
|
|
raise HTTPException(status_code=404, detail="No optimization report for this date")
|
|
report = report_path.read_text(encoding='utf-8')
|
|
import json
|
|
return json.loads(report)
|