313 lines
12 KiB
Python
313 lines
12 KiB
Python
"""智能洞察生成 - 调用LLM分析数据"""
|
||
import json
|
||
import logging
|
||
import uuid
|
||
from datetime import datetime, timedelta
|
||
|
||
from sqlalchemy import select, func
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.models.analytics import PublishRecord, ContentMetrics, OptimizationInsight
|
||
from app.services.llm import LLMFactory
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
_INSIGHT_SYSTEM_PROMPT = """你是一位专业的内容营销数据分析师，擅长GEO（Generative Engine Optimization）领域。
|
||
请根据提供的内容数据分析报告，生成结构化的洞察与优化建议。
|
||
|
||
输出格式必须是合法JSON数组，每个洞察对象包含以下字段：
|
||
- insight_type: "trend" | "anomaly" | "opportunity" | "suggestion"
|
||
- title: 简短标题（不超过50字）
|
||
- description: 详细说明（100-200字）
|
||
- recommendation: 可执行的具体建议（50-150字）
|
||
- severity: "info" | "warning" | "success"
|
||
|
||
输出示例：
|
||
[
|
||
{
|
||
"insight_type": "trend",
|
||
"title": "AI引用量持续增长",
|
||
"description": "过去30天内容被AI模型引用次数增长了35%，...",
|
||
"recommendation": "继续保持结构化写作风格，增加Q&A格式内容...",
|
||
"severity": "success"
|
||
}
|
||
]
|
||
仅输出JSON数组，不要包含其他文字。"""
|
||
|
||
|
||
class InsightGenerator:
|
||
|
||
async def generate_insights(self, organization_id: str, session: AsyncSession) -> list[dict]:
|
||
"""
|
||
分析近30天数据，生成智能洞察
|
||
- 对比各平台表现
|
||
- 识别内容表现模式（什么类型内容互动率高）
|
||
- 发现AI引用趋势
|
||
- 给出可执行建议
|
||
"""
|
||
# 1. 汇总近30天数据
|
||
summary = await self._build_data_summary(organization_id, session)
|
||
if not summary["records"]:
|
||
logger.info("组织 %s 没有发布记录，跳过洞察生成", organization_id)
|
||
return []
|
||
|
||
# 2. 构造分析prompt
|
||
user_prompt = self._build_analysis_prompt(summary)
|
||
|
||
# 3. 调用LLM生成洞察
|
||
llm = LLMFactory.get_default()
|
||
try:
|
||
response = await llm.chat(
|
||
messages=[
|
||
{"role": "system", "content": _INSIGHT_SYSTEM_PROMPT},
|
||
{"role": "user", "content": user_prompt},
|
||
]
|
||
)
|
||
raw_content = response.content.strip()
|
||
# 尝试提取JSON数组
|
||
if "```" in raw_content:
|
||
raw_content = raw_content.split("```")[1]
|
||
if raw_content.startswith("json"):
|
||
raw_content = raw_content[4:]
|
||
insights_data: list[dict] = json.loads(raw_content)
|
||
except Exception as e:
|
||
logger.error("LLM生成洞察失败: %s", e)
|
||
insights_data = self._fallback_insights(summary)
|
||
|
||
# 4. 解析并存储到OptimizationInsight表
|
||
saved = []
|
||
for item in insights_data:
|
||
insight = OptimizationInsight(
|
||
id=str(uuid.uuid4()),
|
||
organization_id=organization_id,
|
||
insight_type=item.get("insight_type", "suggestion"),
|
||
title=item.get("title", "")[:200],
|
||
description=item.get("description", ""),
|
||
recommendation=item.get("recommendation", ""),
|
||
severity=item.get("severity", "info"),
|
||
)
|
||
session.add(insight)
|
||
saved.append({
|
||
"id": insight.id,
|
||
"insight_type": insight.insight_type,
|
||
"title": insight.title,
|
||
"description": insight.description,
|
||
"recommendation": insight.recommendation,
|
||
"severity": insight.severity,
|
||
"applied": insight.applied,
|
||
"created_at": insight.created_at.isoformat() if isinstance(insight.created_at, datetime) else None,
|
||
})
|
||
|
||
await session.commit()
|
||
logger.info("组织 %s 成功生成 %d 条洞察", organization_id, len(saved))
|
||
return saved
|
||
|
||
async def analyze_single_content(self, publish_record_id: str, session: AsyncSession) -> dict:
|
||
"""分析单篇内容表现"""
|
||
pr_stmt = select(PublishRecord).where(PublishRecord.id == publish_record_id)
|
||
pr_result = await session.execute(pr_stmt)
|
||
record = pr_result.scalar_one_or_none()
|
||
if not record:
|
||
return {"error": "发布记录不存在"}
|
||
|
||
metrics_stmt = (
|
||
select(ContentMetrics)
|
||
.where(ContentMetrics.publish_record_id == publish_record_id)
|
||
.order_by(ContentMetrics.recorded_at.desc())
|
||
.limit(1)
|
||
)
|
||
metrics_result = await session.execute(metrics_stmt)
|
||
latest = metrics_result.scalar_one_or_none()
|
||
|
||
if not latest:
|
||
return {"error": "暂无效果数据"}
|
||
|
||
prompt = f"""请分析以下单篇内容的表现数据，给出优化建议：
|
||
|
||
内容标题：{record.content_title}
|
||
发布平台：{record.platform}
|
||
发布时间：{record.published_at}
|
||
|
||
最新效果数据：
|
||
- 浏览量：{latest.views}
|
||
- 点赞：{latest.likes}，评论：{latest.comments}，分享：{latest.shares}
|
||
- 收藏：{latest.bookmarks}
|
||
- AI引用次数：{latest.ai_citation_count}
|
||
- 搜索曝光：{latest.search_impressions}，搜索点击：{latest.search_clicks}
|
||
- 平均阅读时长：{latest.avg_read_duration}秒
|
||
- 完读率：{latest.read_completion_rate:.1%}
|
||
|
||
请输出JSON：{{"analysis": "...", "strengths": ["...", "..."], "improvements": ["...", "..."], "geo_suggestions": ["..."]}}"""
|
||
|
||
llm = LLMFactory.get_default()
|
||
try:
|
||
response = await llm.chat(
|
||
messages=[
|
||
{"role": "system", "content": "你是GEO内容优化专家。请输出合法JSON，不含其他文字。"},
|
||
{"role": "user", "content": prompt},
|
||
]
|
||
)
|
||
raw = response.content.strip()
|
||
if "```" in raw:
|
||
raw = raw.split("```")[1]
|
||
if raw.startswith("json"):
|
||
raw = raw[4:]
|
||
return json.loads(raw)
|
||
except Exception as e:
|
||
logger.error("单内容分析失败: %s", e)
|
||
return {
|
||
"analysis": "数据分析服务暂时不可用，请稍后重试。",
|
||
"strengths": [],
|
||
"improvements": [],
|
||
"geo_suggestions": [],
|
||
}
|
||
|
||
# ------------------------------------------------------------------ #
|
||
# 内部辅助方法
|
||
# ------------------------------------------------------------------ #
|
||
|
||
async def _build_data_summary(self, organization_id: str, session: AsyncSession) -> dict:
|
||
"""汇总近30天发布与指标数据"""
|
||
since = datetime.utcnow() - timedelta(days=30)
|
||
|
||
pr_stmt = select(PublishRecord).where(
|
||
PublishRecord.organization_id == organization_id,
|
||
PublishRecord.created_at >= since,
|
||
)
|
||
pr_result = await session.execute(pr_stmt)
|
||
records = pr_result.scalars().all()
|
||
|
||
if not records:
|
||
return {"records": [], "platform_stats": {}, "totals": {}}
|
||
|
||
record_ids = [r.id for r in records]
|
||
|
||
# 每条记录最新快照
|
||
subq = (
|
||
select(
|
||
ContentMetrics.publish_record_id,
|
||
func.max(ContentMetrics.recorded_at).label("latest"),
|
||
)
|
||
.where(ContentMetrics.publish_record_id.in_(record_ids))
|
||
.group_by(ContentMetrics.publish_record_id)
|
||
.subquery()
|
||
)
|
||
metrics_stmt = select(ContentMetrics).join(
|
||
subq,
|
||
(ContentMetrics.publish_record_id == subq.c.publish_record_id)
|
||
& (ContentMetrics.recorded_at == subq.c.latest),
|
||
)
|
||
metrics_result = await session.execute(metrics_stmt)
|
||
latest_map: dict[str, ContentMetrics] = {
|
||
m.publish_record_id: m for m in metrics_result.scalars().all()
|
||
}
|
||
|
||
# 平台维度统计
|
||
platform_stats: dict[str, dict] = {}
|
||
for r in records:
|
||
m = latest_map.get(r.id)
|
||
p = r.platform
|
||
if p not in platform_stats:
|
||
platform_stats[p] = {
|
||
"count": 0, "views": 0, "interactions": 0, "ai_citations": 0
|
||
}
|
||
platform_stats[p]["count"] += 1
|
||
if m:
|
||
platform_stats[p]["views"] += m.views
|
||
platform_stats[p]["interactions"] += m.likes + m.comments + m.shares
|
||
platform_stats[p]["ai_citations"] += m.ai_citation_count
|
||
|
||
totals = {
|
||
"total_records": len(records),
|
||
"total_views": sum(m.views for m in latest_map.values()),
|
||
"total_interactions": sum(
|
||
m.likes + m.comments + m.shares for m in latest_map.values()
|
||
),
|
||
"total_ai_citations": sum(m.ai_citation_count for m in latest_map.values()),
|
||
"avg_read_completion": (
|
||
sum(m.read_completion_rate for m in latest_map.values()) / len(latest_map)
|
||
if latest_map else 0.0
|
||
),
|
||
}
|
||
|
||
record_details = []
|
||
for r in records:
|
||
m = latest_map.get(r.id)
|
||
record_details.append({
|
||
"title": r.content_title,
|
||
"platform": r.platform,
|
||
"views": m.views if m else 0,
|
||
"interactions": (m.likes + m.comments + m.shares) if m else 0,
|
||
"ai_citations": m.ai_citation_count if m else 0,
|
||
"read_completion_rate": m.read_completion_rate if m else 0.0,
|
||
})
|
||
|
||
return {
|
||
"records": record_details,
|
||
"platform_stats": platform_stats,
|
||
"totals": totals,
|
||
}
|
||
|
||
def _build_analysis_prompt(self, summary: dict) -> str:
|
||
totals = summary["totals"]
|
||
platform_stats = summary["platform_stats"]
|
||
records = summary["records"]
|
||
|
||
top_by_views = sorted(records, key=lambda x: x["views"], reverse=True)[:5]
|
||
top_by_citations = sorted(records, key=lambda x: x["ai_citations"], reverse=True)[:3]
|
||
|
||
prompt = f"""以下是近30天内容运营数据摘要，请生成3-5条洞察与建议：
|
||
|
||
【总览】
|
||
- 发布内容数：{totals['total_records']}
|
||
- 总浏览量：{totals['total_views']}
|
||
- 总互动量（点赞+评论+分享）：{totals['total_interactions']}
|
||
- 总AI引用次数：{totals['total_ai_citations']}
|
||
- 平均完读率：{totals['avg_read_completion']:.1%}
|
||
|
||
【各平台表现】
|
||
"""
|
||
for platform, stats in platform_stats.items():
|
||
prompt += (
|
||
f"- {platform}：{stats['count']}篇，"
|
||
f"浏览{stats['views']}，互动{stats['interactions']}，"
|
||
f"AI引用{stats['ai_citations']}\n"
|
||
)
|
||
|
||
prompt += "\n【浏览量Top5内容】\n"
|
||
for item in top_by_views:
|
||
prompt += f"- 《{item['title']}》({item['platform']}) 浏览:{item['views']} 互动:{item['interactions']}\n"
|
||
|
||
prompt += "\n【AI引用量Top3内容】\n"
|
||
for item in top_by_citations:
|
||
prompt += f"- 《{item['title']}》({item['platform']}) AI引用:{item['ai_citations']}\n"
|
||
|
||
return prompt
|
||
|
||
def _fallback_insights(self, summary: dict) -> list[dict]:
|
||
"""LLM失败时的降级洞察"""
|
||
totals = summary.get("totals", {})
|
||
insights = []
|
||
|
||
if totals.get("total_ai_citations", 0) > 0:
|
||
insights.append({
|
||
"insight_type": "trend",
|
||
"title": "内容已获得AI引用",
|
||
"description": f"近30天共获得 {totals['total_ai_citations']} 次AI引用，说明内容质量已获得生成式引擎认可。",
|
||
"recommendation": "继续保持结构化、权威性的写作风格，增加数据支撑和专业术语密度，提升AI引用率。",
|
||
"severity": "success",
|
||
})
|
||
|
||
if totals.get("total_records", 0) > 0:
|
||
avg_views = totals.get("total_views", 0) / totals["total_records"]
|
||
if avg_views < 100:
|
||
insights.append({
|
||
"insight_type": "suggestion",
|
||
"title": "内容分发有待加强",
|
||
"description": f"当前平均浏览量为 {avg_views:.0f}，建议优化内容分发渠道和发布时机。",
|
||
"recommendation": "尝试在多平台同步发布，配合社群推广，并选择目标平台的流量高峰时段发布。",
|
||
"severity": "warning",
|
||
})
|
||
|
||
return insights
|