Yuzhiran Dev
415 lines
14 KiB
Python
415 lines
14 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, Query
|
||
from sqlalchemy.orm import Session
|
||
from sqlalchemy import func, desc
|
||
from typing import List, Optional
|
||
from datetime import datetime, timedelta, date
|
||
import logging
|
||
import re
|
||
import requests as http_requests
|
||
from pydantic import BaseModel
|
||
|
||
from ..database import get_db
|
||
from ..models import ContentMetrics, Topic, ContentCalendar
|
||
from ..schemas import (
|
||
ContentMetricsCreate, ContentMetricsUpdate, ContentMetricsResponse,
|
||
MetricsDashboard
|
||
)
|
||
from .auth import get_current_user, org_filter
|
||
|
||
router = APIRouter(prefix="/api/metrics", tags=["metrics"])
|
||
|
||
|
||
@router.get("/dashboard", response_model=MetricsDashboard)
|
||
def get_dashboard(
|
||
days: int = Query(30, ge=1, le=365),
|
||
db: Session = Depends(get_db),
|
||
current_user=Depends(get_current_user)
|
||
):
|
||
since = datetime.now() - timedelta(days=days)
|
||
|
||
topic_base = db.query(Topic)
|
||
of = org_filter(current_user, Topic)
|
||
if of is not True:
|
||
topic_base = topic_base.filter(of)
|
||
|
||
total_topics = topic_base.count()
|
||
|
||
raw_status = topic_base.with_entities(Topic.status, func.count()).group_by(Topic.status).all()
|
||
topics_by_status = {}
|
||
for s, cnt in raw_status:
|
||
topics_by_status[s] = cnt
|
||
|
||
metrics_q = db.query(ContentMetrics).join(Topic, ContentMetrics.topic_id == Topic.id)
|
||
if of is not True:
|
||
metrics_q = metrics_q.filter(of)
|
||
total_published = metrics_q.filter(ContentMetrics.views > 0).count()
|
||
|
||
all_metrics = metrics_q.filter(
|
||
ContentMetrics.created_at >= since
|
||
).all()
|
||
|
||
total_views = sum(m.views for m in all_metrics)
|
||
total_likes = sum(m.likes for m in all_metrics)
|
||
|
||
engagement_rates = [m.engagement_rate for m in all_metrics if m.views > 0]
|
||
avg_engagement = sum(engagement_rates) / len(engagement_rates) if engagement_rates else 0
|
||
|
||
top_q = db.query(
|
||
ContentMetrics.topic_id,
|
||
func.sum(ContentMetrics.views).label("total_views"),
|
||
func.sum(ContentMetrics.likes).label("total_likes")
|
||
).join(Topic, ContentMetrics.topic_id == Topic.id)
|
||
if of is not True:
|
||
top_q = top_q.filter(of)
|
||
top_topics_data = top_q.filter(
|
||
ContentMetrics.created_at >= since
|
||
).group_by(ContentMetrics.topic_id).order_by(desc("total_views")).limit(10).all()
|
||
|
||
top_topics = []
|
||
for row in top_topics_data:
|
||
topic = db.query(Topic).filter(Topic.id == row.topic_id).first()
|
||
top_topics.append({
|
||
"topic_id": row.topic_id,
|
||
"title": topic.title if topic else row.topic_id,
|
||
"total_views": row.total_views or 0,
|
||
"total_likes": row.total_likes or 0,
|
||
})
|
||
|
||
recent_q = db.query(ContentMetrics).join(Topic, ContentMetrics.topic_id == Topic.id)
|
||
if of is not True:
|
||
recent_q = recent_q.filter(of)
|
||
recent_metrics = recent_q.order_by(
|
||
ContentMetrics.created_at.desc()
|
||
).limit(10).all()
|
||
|
||
return MetricsDashboard(
|
||
total_topics=total_topics,
|
||
topics_by_status=topics_by_status,
|
||
total_published=total_published,
|
||
total_views=total_views,
|
||
total_likes=total_likes,
|
||
avg_engagement_rate=round(avg_engagement, 2),
|
||
top_topics=top_topics,
|
||
recent_metrics=[ContentMetricsResponse.model_validate(m) for m in recent_metrics]
|
||
)
|
||
|
||
|
||
@router.get("/trend")
|
||
def get_trend(
|
||
days: int = Query(30, ge=1, le=365),
|
||
group_by: str = Query("day", enum=["day", "week", "month"]),
|
||
db: Session = Depends(get_db),
|
||
current_user=Depends(get_current_user)
|
||
):
|
||
since = datetime.now() - timedelta(days=days)
|
||
|
||
if group_by == "day":
|
||
date_format = func.date(ContentMetrics.created_at)
|
||
else:
|
||
date_format = func.date_trunc(group_by, ContentMetrics.created_at)
|
||
|
||
q = db.query(
|
||
date_format.label("period"),
|
||
func.sum(ContentMetrics.views).label("views"),
|
||
func.sum(ContentMetrics.likes).label("likes"),
|
||
func.sum(ContentMetrics.comments).label("comments"),
|
||
func.count(ContentMetrics.id).label("count")
|
||
).join(Topic, ContentMetrics.topic_id == Topic.id)
|
||
of_m = org_filter(current_user, Topic)
|
||
if of_m is not True:
|
||
q = q.filter(of_m)
|
||
rows = q.filter(
|
||
ContentMetrics.created_at >= since
|
||
).group_by(date_format).order_by(date_format).all()
|
||
|
||
return [
|
||
{
|
||
"period": str(row.period),
|
||
"views": row.views or 0,
|
||
"likes": row.likes or 0,
|
||
"comments": row.comments or 0,
|
||
"count": row.count or 0
|
||
}
|
||
for row in rows
|
||
]
|
||
|
||
|
||
@router.get("/entries", response_model=List[ContentMetricsResponse])
|
||
def list_metrics(
|
||
topic_id: Optional[str] = None,
|
||
platform: Optional[str] = None,
|
||
limit: int = Query(50, le=200),
|
||
db: Session = Depends(get_db),
|
||
current_user=Depends(get_current_user)
|
||
):
|
||
query = db.query(ContentMetrics).join(Topic, ContentMetrics.topic_id == Topic.id)
|
||
of_m = org_filter(current_user, Topic)
|
||
if of_m is not True:
|
||
query = query.filter(of_m)
|
||
if topic_id:
|
||
query = query.filter(ContentMetrics.topic_id == topic_id)
|
||
if platform:
|
||
query = query.filter(ContentMetrics.platform == platform)
|
||
return query.order_by(ContentMetrics.created_at.desc()).limit(limit).all()
|
||
|
||
|
||
@router.post("/entries", response_model=ContentMetricsResponse)
|
||
def create_metric(
|
||
data: ContentMetricsCreate,
|
||
db: Session = Depends(get_db),
|
||
current_user=Depends(get_current_user)
|
||
):
|
||
topic = db.query(Topic).filter(Topic.id == data.topic_id).first()
|
||
if not topic:
|
||
raise HTTPException(status_code=404, detail="选题不存在")
|
||
if current_user.role != "admin" and topic.org_id != current_user.org_id:
|
||
raise HTTPException(status_code=404, detail="选题不存在")
|
||
|
||
existing = db.query(ContentMetrics).filter(
|
||
ContentMetrics.topic_id == data.topic_id,
|
||
ContentMetrics.platform == data.platform
|
||
).first()
|
||
|
||
if existing:
|
||
for k, v in data.model_dump(exclude_unset=True).items():
|
||
if k != "topic_id" and k != "platform":
|
||
setattr(existing, k, v)
|
||
existing.last_fetched = datetime.now()
|
||
db.commit()
|
||
db.refresh(existing)
|
||
return existing
|
||
|
||
metric = ContentMetrics(**data.model_dump(), last_fetched=datetime.now())
|
||
db.add(metric)
|
||
db.commit()
|
||
db.refresh(metric)
|
||
return metric
|
||
|
||
|
||
def _check_metric_org(metric, current_user, db):
|
||
"""Check if metric's topic belongs to user's org"""
|
||
topic = db.query(Topic).filter(Topic.id == metric.topic_id).first()
|
||
if topic and current_user.role != "admin" and topic.org_id != current_user.org_id:
|
||
raise HTTPException(status_code=404, detail="数据记录不存在")
|
||
|
||
|
||
@router.put("/entries/{metric_id}", response_model=ContentMetricsResponse)
|
||
def update_metric(
|
||
metric_id: int,
|
||
data: ContentMetricsUpdate,
|
||
db: Session = Depends(get_db),
|
||
current_user=Depends(get_current_user)
|
||
):
|
||
metric = db.query(ContentMetrics).filter(ContentMetrics.id == metric_id).first()
|
||
if not metric:
|
||
raise HTTPException(status_code=404, detail="数据记录不存在")
|
||
_check_metric_org(metric, current_user, db)
|
||
|
||
for k, v in data.model_dump(exclude_unset=True).items():
|
||
setattr(metric, k, v)
|
||
metric.last_fetched = datetime.now()
|
||
db.commit()
|
||
db.refresh(metric)
|
||
return metric
|
||
|
||
|
||
@router.delete("/entries/{metric_id}")
|
||
def delete_metric(
|
||
metric_id: int,
|
||
db: Session = Depends(get_db),
|
||
current_user=Depends(get_current_user)
|
||
):
|
||
metric = db.query(ContentMetrics).filter(ContentMetrics.id == metric_id).first()
|
||
if not metric:
|
||
raise HTTPException(status_code=404, detail="数据记录不存在")
|
||
_check_metric_org(metric, current_user, db)
|
||
db.delete(metric)
|
||
db.commit()
|
||
return {"ok": True}
|
||
|
||
|
||
@router.get("/topics/{topic_id}", response_model=List[ContentMetricsResponse])
|
||
def get_topic_metrics(
|
||
topic_id: str,
|
||
db: Session = Depends(get_db),
|
||
current_user=Depends(get_current_user)
|
||
):
|
||
topic = db.query(Topic).filter(Topic.id == topic_id).first()
|
||
if not topic:
|
||
raise HTTPException(status_code=404, detail="选题不存在")
|
||
if current_user.role != "admin" and topic.org_id != current_user.org_id:
|
||
raise HTTPException(status_code=404, detail="选题不存在")
|
||
return db.query(ContentMetrics).filter(
|
||
ContentMetrics.topic_id == topic_id
|
||
).order_by(ContentMetrics.created_at.desc()).all()
|
||
|
||
|
||
@router.get("/by-platform")
|
||
def get_metrics_by_platform(
|
||
db: Session = Depends(get_db),
|
||
current_user=Depends(get_current_user)
|
||
):
|
||
q = db.query(
|
||
ContentMetrics.platform,
|
||
func.sum(ContentMetrics.views).label("total_views"),
|
||
func.sum(ContentMetrics.likes).label("total_likes"),
|
||
func.sum(ContentMetrics.comments).label("total_comments"),
|
||
func.count(ContentMetrics.id).label("count")
|
||
).join(Topic, ContentMetrics.topic_id == Topic.id)
|
||
of_m = org_filter(current_user, Topic)
|
||
if of_m is not True:
|
||
q = q.filter(of_m)
|
||
rows = q.group_by(ContentMetrics.platform).all()
|
||
|
||
return [
|
||
{
|
||
"platform": row.platform,
|
||
"total_views": row.total_views or 0,
|
||
"total_likes": row.total_likes or 0,
|
||
"total_comments": row.total_comments or 0,
|
||
"count": row.count or 0,
|
||
"avg_views": (row.total_views or 0) / (row.count or 1),
|
||
"avg_likes": (row.total_likes or 0) / (row.count or 1),
|
||
}
|
||
for row in rows
|
||
]
|
||
|
||
|
||
@router.get("/recommend-topics")
|
||
def recommend_topics_from_metrics(
|
||
limit: int = Query(10, le=50),
|
||
db: Session = Depends(get_db),
|
||
current_user=Depends(get_current_user)
|
||
):
|
||
try:
|
||
hp_q = db.query(
|
||
ContentMetrics.topic_id,
|
||
func.avg(ContentMetrics.engagement_rate).label("avg_engagement"),
|
||
func.max(ContentMetrics.views).label("max_views")
|
||
).join(Topic, ContentMetrics.topic_id == Topic.id)
|
||
of_m = org_filter(current_user, Topic)
|
||
if of_m is not True:
|
||
hp_q = hp_q.filter(of_m)
|
||
high_performing = hp_q.group_by(ContentMetrics.topic_id).order_by(desc("avg_engagement")).limit(20).all()
|
||
|
||
recommendations = []
|
||
for row in high_performing:
|
||
topic = db.query(Topic).filter(Topic.id == row.topic_id).first()
|
||
if not topic:
|
||
continue
|
||
metrics = db.query(ContentMetrics).filter(
|
||
ContentMetrics.topic_id == row.topic_id
|
||
).all()
|
||
recommendations.append({
|
||
"topic_id": row.topic_id,
|
||
"title": topic.title,
|
||
"field": topic.field_name,
|
||
"status": topic.status,
|
||
"avg_engagement": round(float(row.avg_engagement), 2) if row.avg_engagement else 0,
|
||
"max_views": int(row.max_views) if row.max_views else 0,
|
||
"platforms": list(set(m.platform for m in metrics)),
|
||
"reason": f"平均互动率 {round(float(row.avg_engagement), 2) if row.avg_engagement else 0}%，最高阅读 {int(row.max_views) if row.max_views else 0}"
|
||
})
|
||
|
||
return recommendations[:limit]
|
||
except Exception as e:
|
||
logging.getLogger(__name__).exception(f"推荐选题失败: {e}")
|
||
return []
|
||
|
||
|
||
_ZHIHU_UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
|
||
|
||
|
||
def _extract_zhihu_post_id(url: str) -> str:
|
||
m = re.search(r'zhuanlan\.zhihu\.com/p/(\d+)', url)
|
||
if m:
|
||
return m.group(1)
|
||
m = re.search(r'zhihu\.com/question/\d+/answer/(\d+)', url)
|
||
if m:
|
||
return m.group(1)
|
||
raise ValueError("无法从URL中提取知乎文章/回答ID")
|
||
|
||
|
||
def _fetch_zhihu_page(post_id: str) -> dict:
|
||
"""多种方式尝试获取知乎文章数据，返回原始 JSON"""
|
||
UA = _ZHIHU_UA
|
||
# 方式1：专栏 API
|
||
urls = [
|
||
f"https://zhuanlan.zhihu.com/api/posts/{post_id}",
|
||
f"https://www.zhihu.com/api/v4/posts/{post_id}",
|
||
f"https://www.zhihu.com/api/v4/answers/{post_id}",
|
||
]
|
||
exceptions = []
|
||
for url in urls:
|
||
try:
|
||
resp = http_requests.get(url, headers={"User-Agent": UA}, timeout=10)
|
||
if resp.status_code == 200:
|
||
return resp.json()
|
||
exceptions.append(f"{url} → {resp.status_code}")
|
||
except Exception as e:
|
||
exceptions.append(f"{url} → {e}")
|
||
raise RuntimeError(f"知乎API已封禁，无法获取数据（{'; '.join(exceptions)}）")
|
||
|
||
|
||
class ZhihuFetchRequest(BaseModel):
|
||
topic_id: str
|
||
zhihu_url: str
|
||
|
||
|
||
@router.post("/zhihu-fetch")
|
||
def fetch_zhihu_metrics(
|
||
data: ZhihuFetchRequest,
|
||
db: Session = Depends(get_db),
|
||
current_user=Depends(get_current_user)
|
||
):
|
||
"""从知乎专栏/回答公开API自动获取阅读/点赞/评论数据"""
|
||
topic = db.query(Topic).filter(Topic.id == data.topic_id).first()
|
||
if not topic:
|
||
raise HTTPException(status_code=404, detail="选题不存在")
|
||
if current_user.role != "admin" and topic.org_id != current_user.org_id:
|
||
raise HTTPException(status_code=404, detail="选题不存在")
|
||
|
||
post_id = _extract_zhihu_post_id(data.zhihu_url)
|
||
|
||
try:
|
||
raw = _fetch_zhihu_page(post_id)
|
||
platform = "zhihu"
|
||
|
||
existing = db.query(ContentMetrics).filter(
|
||
ContentMetrics.topic_id == data.topic_id,
|
||
ContentMetrics.platform == platform
|
||
).first()
|
||
|
||
metric_data = {
|
||
"views": raw.get("voteup_count", raw.get("views_count", 0)),
|
||
"likes": raw.get("voteup_count", 0),
|
||
"favorites": raw.get("favorite_count", 0),
|
||
"comments": raw.get("comment_count", raw.get("comments_count", 0)),
|
||
"shares": raw.get("share_count", 0) or raw.get("shared_count", 0),
|
||
"publish_url": data.zhihu_url,
|
||
"data_snapshot": raw,
|
||
}
|
||
|
||
if existing:
|
||
for k, v in metric_data.items():
|
||
setattr(existing, k, v)
|
||
existing.last_fetched = datetime.now()
|
||
db.commit()
|
||
db.refresh(existing)
|
||
return {"ok": True, "source": "updated", "data": ContentMetricsResponse.model_validate(existing)}
|
||
else:
|
||
metric = ContentMetrics(
|
||
topic_id=data.topic_id,
|
||
platform=platform,
|
||
**metric_data,
|
||
last_fetched=datetime.now()
|
||
)
|
||
db.add(metric)
|
||
db.commit()
|
||
db.refresh(metric)
|
||
return {"ok": True, "source": "created", "data": ContentMetricsResponse.model_validate(metric)}
|
||
|
||
except HTTPException:
|
||
raise
|
||
except Exception as e:
|
||
raise HTTPException(status_code=502, detail=f"获取知乎数据失败: {str(e)}") |