571 lines
21 KiB
Python
571 lines
21 KiB
Python
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
from dataclasses import dataclass, field
|
||
from typing import Any
|
||
|
||
from app.config import settings
|
||
from app.services.scoring.scoring_service import ScoringResultV2
|
||
from app.utils.json_extractor import extract_json
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
@dataclass
|
||
class GeoPlanActionItem:
|
||
action_type: str
|
||
title: str
|
||
description: str
|
||
reason: str
|
||
priority: str
|
||
target_keyword: str | None = None
|
||
target_platform: str | None = None
|
||
content_style: str | None = None
|
||
estimated_impact: str | None = None
|
||
difficulty: str = "medium"
|
||
execution_params: dict[str, Any] | None = None
|
||
|
||
|
||
@dataclass
|
||
class GeoPlanData:
|
||
title: str
|
||
estimated_weeks: int
|
||
actions: list[GeoPlanActionItem] = field(default_factory=list)
|
||
weekly_plan: list[dict[str, Any]] = field(default_factory=list)
|
||
|
||
|
||
def _get_weakest_dimensions(
|
||
mention_rate_pct: float,
|
||
rank_pct: float,
|
||
sentiment_pct: float,
|
||
citation_pct: float,
|
||
competitive_pct: float,
|
||
) -> list[tuple[str, float]]:
|
||
dimensions = [
|
||
("提及率", mention_rate_pct),
|
||
("推荐排名", rank_pct),
|
||
("情感倾向", sentiment_pct),
|
||
("引用质量", citation_pct),
|
||
("竞品对比", competitive_pct),
|
||
]
|
||
return sorted(dimensions, key=lambda x: x[1])
|
||
|
||
|
||
def _generate_rule_based_plan(
|
||
brand_name: str,
|
||
overall_score: float,
|
||
target_score: int,
|
||
mention_rate_pct: float,
|
||
rank_pct: float,
|
||
sentiment_pct: float,
|
||
citation_pct: float,
|
||
competitive_pct: float,
|
||
total_queries: int,
|
||
platform_scores: dict[str, float],
|
||
competitor_data: dict[str, Any],
|
||
) -> GeoPlanData:
|
||
actions: list[GeoPlanActionItem] = []
|
||
score_gap = target_score - overall_score
|
||
estimated_weeks = min(12, max(4, int(score_gap / 5) + 4))
|
||
|
||
if mention_rate_pct < 50:
|
||
actions.append(GeoPlanActionItem(
|
||
action_type="content_creation",
|
||
title=f"提升{brand_name}在AI平台的提及率",
|
||
description=(
|
||
f"当前提及率仅{mention_rate_pct:.0f}%，品牌在AI回答中被提及的频率较低。"
|
||
f"需要创建高质量内容提高品牌在AI搜索结果中的出现频率。"
|
||
),
|
||
reason=f"提及率得分率{mention_rate_pct:.0f}%，低于50%阈值，是最需要优先改善的维度",
|
||
priority="high",
|
||
target_keyword=f"{brand_name}+行业关键词",
|
||
target_platform="知乎",
|
||
content_style="专业严谨",
|
||
estimated_impact="预计可将提及率提升15-25个百分点",
|
||
difficulty="medium",
|
||
execution_params={
|
||
"keyword": f"{brand_name} 行业解决方案",
|
||
"platform": "知乎",
|
||
"style": "专业严谨",
|
||
"word_count": 2000,
|
||
},
|
||
))
|
||
|
||
if citation_pct < 40:
|
||
actions.append(GeoPlanActionItem(
|
||
action_type="content_creation",
|
||
title=f"提升{brand_name}引用内容质量",
|
||
description=(
|
||
f"当前引用质量得分率仅{citation_pct:.0f}%，AI对品牌的引用多为浅层提及。"
|
||
f"需要创建深度评测和对比内容，增加被深度引用的概率。"
|
||
),
|
||
reason=f"引用质量得分率{citation_pct:.0f}%，低于40%阈值，引用缺乏深度正面描述",
|
||
priority="high",
|
||
target_keyword=f"{brand_name}评测/对比",
|
||
target_platform="通用",
|
||
content_style="专业严谨",
|
||
estimated_impact="预计可将引用质量得分率提升15-25个百分点",
|
||
difficulty="medium",
|
||
execution_params={
|
||
"keyword": f"{brand_name} 深度评测对比",
|
||
"platform": "通用",
|
||
"style": "专业严谨",
|
||
"word_count": 3000,
|
||
},
|
||
))
|
||
|
||
if rank_pct < 40:
|
||
actions.append(GeoPlanActionItem(
|
||
action_type="content_optimization",
|
||
title=f"提升{brand_name}在AI推荐中的排名",
|
||
description=(
|
||
f"当前推荐排名得分率仅{rank_pct:.0f}%，品牌在AI推荐列表中排名靠后。"
|
||
f"需要优化现有内容，增加品牌在推荐场景中的出现概率。"
|
||
),
|
||
reason=f"推荐排名得分率{rank_pct:.0f}%，低于40%阈值，排名靠后用户看到概率低",
|
||
priority="high",
|
||
target_keyword=f"最佳{brand_name}推荐",
|
||
target_platform="通用",
|
||
content_style="专业严谨",
|
||
estimated_impact="预计可将推荐排名提升2-3位",
|
||
difficulty="medium",
|
||
execution_params={
|
||
"keyword": f"最佳{brand_name}推荐",
|
||
"platform": "通用",
|
||
"style": "专业严谨",
|
||
"word_count": 2000,
|
||
},
|
||
))
|
||
|
||
if competitive_pct < 40:
|
||
ahead_competitors = []
|
||
if competitor_data:
|
||
brand_mentions = competitor_data.get("brand_mentions", 0)
|
||
for name, count in competitor_data.get("competitor_mentions", {}).items():
|
||
if count > brand_mentions:
|
||
ahead_competitors.append(name)
|
||
ahead_str = "、".join(ahead_competitors[:3]) if ahead_competitors else "竞品"
|
||
|
||
actions.append(GeoPlanActionItem(
|
||
action_type="content_creation",
|
||
title=f"缩小与{ahead_str}的差距",
|
||
description=(
|
||
f"当前竞品对比得分率仅{competitive_pct:.0f}%，"
|
||
f"品牌在AI引用中落后于主要竞品。需要创建对比内容突出品牌优势。"
|
||
),
|
||
reason=f"竞品对比得分率{competitive_pct:.0f}%，低于40%阈值，品牌落后于主要竞品",
|
||
priority="high",
|
||
target_keyword=f"{brand_name} vs {ahead_competitors[0] if ahead_competitors else '竞品'}",
|
||
target_platform="知乎",
|
||
content_style="专业严谨",
|
||
estimated_impact="预计3-6个月内可将竞品对比得分率提升15-25个百分点",
|
||
difficulty="hard",
|
||
execution_params={
|
||
"keyword": f"{brand_name} vs {ahead_competitors[0] if ahead_competitors else '竞品'} 对比评测",
|
||
"platform": "知乎",
|
||
"style": "专业严谨",
|
||
"word_count": 2500,
|
||
},
|
||
))
|
||
|
||
if sentiment_pct < 40:
|
||
actions.append(GeoPlanActionItem(
|
||
action_type="content_optimization",
|
||
title=f"改善AI平台对{brand_name}的情感倾向",
|
||
description=(
|
||
f"当前情感倾向得分率仅{sentiment_pct:.0f}%，AI在引用品牌时倾向使用负面或中性表述。"
|
||
f"需要优化内容以增加正面引用比例。"
|
||
),
|
||
reason=f"情感倾向得分率{sentiment_pct:.0f}%，低于40%阈值，负面引用影响品牌形象",
|
||
priority="medium",
|
||
target_keyword=f"{brand_name}优势/正面评价",
|
||
target_platform="通用",
|
||
content_style="轻松活泼",
|
||
estimated_impact="减少负面引用比例10-20个百分点",
|
||
difficulty="medium",
|
||
execution_params={
|
||
"keyword": f"{brand_name} 优势 正面评价",
|
||
"platform": "通用",
|
||
"style": "轻松活泼",
|
||
"word_count": 1500,
|
||
},
|
||
))
|
||
|
||
if total_queries < 10:
|
||
suggested_queries = [
|
||
f"{brand_name}怎么样",
|
||
f"{brand_name}推荐",
|
||
f"最佳{brand_name}",
|
||
f"{brand_name}评测",
|
||
f"{brand_name}对比",
|
||
]
|
||
actions.append(GeoPlanActionItem(
|
||
action_type="query_expansion",
|
||
title="扩展查询词覆盖范围",
|
||
description=(
|
||
f"当前仅有{total_queries}个查询词，覆盖范围不足，"
|
||
f"无法全面反映品牌在AI搜索中的表现。"
|
||
),
|
||
reason=f"查询词数量仅{total_queries}个，低于10个阈值，分析结果不够全面",
|
||
priority="high" if total_queries < 3 else "medium",
|
||
target_keyword=None,
|
||
target_platform=None,
|
||
content_style=None,
|
||
estimated_impact="更多查询词可提升评分准确度，发现更多优化机会",
|
||
difficulty="easy",
|
||
execution_params={
|
||
"suggested_queries": suggested_queries,
|
||
},
|
||
))
|
||
|
||
schema_score = 0
|
||
if schema_score == 0:
|
||
actions.append(GeoPlanActionItem(
|
||
action_type="schema_optimization",
|
||
title="添加FAQ结构化数据",
|
||
description=(
|
||
"当前网站缺少结构化数据(Schema)，AI搜索引擎无法有效提取品牌信息。"
|
||
"添加FAQ Schema可以显著提升品牌在AI回答中的引用概率。"
|
||
),
|
||
reason="网站Schema标记缺失，AI搜索引擎无法高效提取品牌关键信息",
|
||
priority="medium",
|
||
target_keyword=None,
|
||
target_platform=None,
|
||
content_style=None,
|
||
estimated_impact="添加Schema后预计可提升引用率10-15个百分点",
|
||
difficulty="easy",
|
||
execution_params={
|
||
"optimization_type": "add_faq_schema",
|
||
},
|
||
))
|
||
|
||
weak_platforms = sorted(
|
||
platform_scores.items(),
|
||
key=lambda x: x[1],
|
||
)[:3]
|
||
weak_platform_names = [p[0] for p in weak_platforms if p[1] < 40]
|
||
if weak_platform_names:
|
||
actions.append(GeoPlanActionItem(
|
||
action_type="platform_targeting",
|
||
title=f"重点优化{'、'.join(weak_platform_names)}平台表现",
|
||
description=(
|
||
f"在这些平台上品牌评分低于40分，AI引用率极低。"
|
||
f"需要针对性优化各平台的内容策略。"
|
||
),
|
||
reason=f"平台{'、'.join(weak_platform_names)}评分低于40分，AI引用率极低",
|
||
priority="high",
|
||
target_keyword=None,
|
||
target_platform=weak_platform_names[0],
|
||
content_style=None,
|
||
estimated_impact="预计可将弱平台评分提升20-30分",
|
||
difficulty="hard",
|
||
execution_params={
|
||
"target_platforms": weak_platform_names,
|
||
},
|
||
))
|
||
|
||
priority_order = {"high": 0, "medium": 1, "low": 2}
|
||
actions.sort(key=lambda a: priority_order.get(a.priority, 1))
|
||
actions = actions[:8]
|
||
|
||
weekly_plan = _generate_weekly_plan(actions, estimated_weeks)
|
||
|
||
title = f"{brand_name} GEO优化方案 - 从{overall_score:.0f}分提升至{target_score}分"
|
||
|
||
return GeoPlanData(
|
||
title=title,
|
||
estimated_weeks=estimated_weeks,
|
||
actions=actions,
|
||
weekly_plan=weekly_plan,
|
||
)
|
||
|
||
|
||
def _generate_weekly_plan(
|
||
actions: list[GeoPlanActionItem],
|
||
estimated_weeks: int,
|
||
) -> list[dict[str, Any]]:
|
||
weekly_plan: list[dict[str, Any]] = []
|
||
high_actions = [i for i, a in enumerate(actions) if a.priority == "high"]
|
||
medium_actions = [i for i, a in enumerate(actions) if a.priority == "medium"]
|
||
low_actions = [i for i, a in enumerate(actions) if a.priority == "low"]
|
||
|
||
weeks_per_high = max(1, (estimated_weeks // 2) // max(len(high_actions), 1)) if high_actions else 0
|
||
week_idx = 0
|
||
for i, action_idx in enumerate(high_actions):
|
||
week_num = week_idx + 1
|
||
impact_str = actions[action_idx].estimated_impact or ""
|
||
try:
|
||
num_part = impact_str.split("-")[-1].replace("个百分点", "").strip()
|
||
expected_val = max(3, int(num_part)) if num_part.isdigit() else 5
|
||
except (ValueError, IndexError):
|
||
expected_val = 5
|
||
expected = f"+{expected_val}"
|
||
weekly_plan.append({
|
||
"week": week_num,
|
||
"action_indices": [action_idx],
|
||
"expected_score_change": expected,
|
||
})
|
||
week_idx += weeks_per_high
|
||
|
||
remaining_weeks = estimated_weeks - week_idx
|
||
medium_per_week = max(1, len(medium_actions) // max(remaining_weeks, 1)) if medium_actions and remaining_weeks > 0 else len(medium_actions)
|
||
batch: list[int] = []
|
||
for i, action_idx in enumerate(medium_actions):
|
||
batch.append(action_idx)
|
||
if len(batch) >= medium_per_week or i == len(medium_actions) - 1:
|
||
week_idx += 1
|
||
weekly_plan.append({
|
||
"week": week_idx,
|
||
"action_indices": batch[:],
|
||
"expected_score_change": "+3",
|
||
})
|
||
batch = []
|
||
|
||
for action_idx in low_actions:
|
||
week_idx += 1
|
||
weekly_plan.append({
|
||
"week": week_idx,
|
||
"action_indices": [action_idx],
|
||
"expected_score_change": "+2",
|
||
})
|
||
|
||
return weekly_plan
|
||
|
||
|
||
GEO_PLAN_PROMPT = """你是一个GEO（生成式引擎优化）策略专家。基于以下品牌诊断数据，制定一个8周GEO优化方案。
|
||
|
||
品牌: {brand_name}
|
||
当前评分: {overall_score}/100
|
||
目标评分: {target_score}/100
|
||
|
||
评分维度:
|
||
- 提及率: {mention_rate_percentage}%
|
||
- 推荐排名: {rank_percentage}%
|
||
- 情感倾向: {sentiment_percentage}%
|
||
- 引用质量: {citation_percentage}%
|
||
- 竞品对比: {competitive_percentage}%
|
||
|
||
竞品数据: {competitor_data}
|
||
平台评分: {platform_scores}
|
||
|
||
请返回JSON格式:
|
||
{{
|
||
"title": "方案标题",
|
||
"estimated_weeks": 8,
|
||
"actions": [
|
||
{{
|
||
"action_type": "content_creation",
|
||
"title": "行动标题",
|
||
"description": "详细描述",
|
||
"reason": "基于诊断数据的原因",
|
||
"priority": "high",
|
||
"target_keyword": "推荐关键词",
|
||
"target_platform": "推荐平台",
|
||
"content_style": "推荐风格",
|
||
"estimated_impact": "预期效果",
|
||
"difficulty": "medium",
|
||
"execution_params": {{
|
||
"keyword": "关键词",
|
||
"platform": "平台",
|
||
"style": "风格",
|
||
"word_count": 2000
|
||
}}
|
||
}}
|
||
],
|
||
"weekly_plan": [
|
||
{{
|
||
"week": 1,
|
||
"action_indices": [0, 1],
|
||
"expected_score_change": "+5"
|
||
}}
|
||
]
|
||
}}
|
||
|
||
要求:
|
||
1. 行动项必须基于诊断数据，优先解决最弱维度
|
||
2. 每个行动项必须有 execution_params，可直接传给内容生成API
|
||
3. 行动项按优先级排序
|
||
4. 生成5-8个行动项
|
||
5. 周计划要合理分配任务"""
|
||
|
||
|
||
async def _generate_llm_plan(
|
||
brand_name: str,
|
||
overall_score: float,
|
||
target_score: int,
|
||
mention_rate_pct: float,
|
||
rank_pct: float,
|
||
sentiment_pct: float,
|
||
citation_pct: float,
|
||
competitive_pct: float,
|
||
total_queries: int,
|
||
platform_scores: dict[str, float],
|
||
competitor_data: dict[str, Any],
|
||
) -> GeoPlanData:
|
||
if not settings.ENABLE_LLM or not settings.DEEPSEEK_API_KEY:
|
||
logger.info("LLM未启用或API Key未配置，使用规则生成方案")
|
||
return _generate_rule_based_plan(
|
||
brand_name=brand_name,
|
||
overall_score=overall_score,
|
||
target_score=target_score,
|
||
mention_rate_pct=mention_rate_pct,
|
||
rank_pct=rank_pct,
|
||
sentiment_pct=sentiment_pct,
|
||
citation_pct=citation_pct,
|
||
competitive_pct=competitive_pct,
|
||
total_queries=total_queries,
|
||
platform_scores=platform_scores,
|
||
competitor_data=competitor_data,
|
||
)
|
||
|
||
try:
|
||
prompt = GEO_PLAN_PROMPT.format(
|
||
brand_name=brand_name,
|
||
overall_score=round(overall_score, 1),
|
||
target_score=target_score,
|
||
mention_rate_percentage=round(mention_rate_pct, 1),
|
||
rank_percentage=round(rank_pct, 1),
|
||
sentiment_percentage=round(sentiment_pct, 1),
|
||
citation_percentage=round(citation_pct, 1),
|
||
competitive_percentage=round(competitive_pct, 1),
|
||
competitor_data=json.dumps(competitor_data, ensure_ascii=False, indent=2),
|
||
platform_scores=json.dumps(platform_scores, ensure_ascii=False, indent=2),
|
||
)
|
||
|
||
from openai import OpenAI
|
||
|
||
client = OpenAI(
|
||
api_key=settings.DEEPSEEK_API_KEY,
|
||
base_url="https://api.deepseek.com",
|
||
)
|
||
|
||
response = await asyncio.to_thread(
|
||
client.chat.completions.create,
|
||
model="deepseek-chat",
|
||
messages=[{"role": "user", "content": prompt}],
|
||
temperature=0.3,
|
||
max_tokens=3000,
|
||
)
|
||
|
||
content = response.choices[0].message.content
|
||
if not content:
|
||
raise ValueError("LLM返回空响应")
|
||
|
||
json_str = extract_json(content)
|
||
result = json.loads(json_str)
|
||
|
||
valid_action_types = {
|
||
"content_creation", "content_optimization",
|
||
"query_expansion", "schema_optimization", "platform_targeting",
|
||
}
|
||
valid_priorities = {"high", "medium", "low"}
|
||
valid_difficulties = {"easy", "medium", "hard"}
|
||
|
||
actions: list[GeoPlanActionItem] = []
|
||
for item in result.get("actions", []):
|
||
action_type = item.get("action_type", "content_creation")
|
||
if action_type not in valid_action_types:
|
||
action_type = "content_creation"
|
||
|
||
priority = item.get("priority", "medium")
|
||
if priority not in valid_priorities:
|
||
priority = "medium"
|
||
|
||
difficulty = item.get("difficulty", "medium")
|
||
if difficulty not in valid_difficulties:
|
||
difficulty = "medium"
|
||
|
||
actions.append(GeoPlanActionItem(
|
||
action_type=action_type,
|
||
title=item.get("title", "优化行动"),
|
||
description=item.get("description", ""),
|
||
reason=item.get("reason", ""),
|
||
priority=priority,
|
||
target_keyword=item.get("target_keyword"),
|
||
target_platform=item.get("target_platform"),
|
||
content_style=item.get("content_style"),
|
||
estimated_impact=item.get("estimated_impact"),
|
||
difficulty=difficulty,
|
||
execution_params=item.get("execution_params"),
|
||
))
|
||
|
||
if not actions:
|
||
logger.warning("LLM未返回有效行动项，回退到规则生成")
|
||
return _generate_rule_based_plan(
|
||
brand_name=brand_name,
|
||
overall_score=overall_score,
|
||
target_score=target_score,
|
||
mention_rate_pct=mention_rate_pct,
|
||
rank_pct=rank_pct,
|
||
sentiment_pct=sentiment_pct,
|
||
citation_pct=citation_pct,
|
||
competitive_pct=competitive_pct,
|
||
total_queries=total_queries,
|
||
platform_scores=platform_scores,
|
||
competitor_data=competitor_data,
|
||
)
|
||
|
||
weekly_plan = result.get("weekly_plan", [])
|
||
|
||
return GeoPlanData(
|
||
title=result.get("title", f"{brand_name} GEO优化方案"),
|
||
estimated_weeks=result.get("estimated_weeks", 8),
|
||
actions=actions[:8],
|
||
weekly_plan=weekly_plan,
|
||
)
|
||
|
||
except Exception as e:
|
||
logger.error(f"LLM生成方案失败: {e}，回退到规则生成")
|
||
return _generate_rule_based_plan(
|
||
brand_name=brand_name,
|
||
overall_score=overall_score,
|
||
target_score=target_score,
|
||
mention_rate_pct=mention_rate_pct,
|
||
rank_pct=rank_pct,
|
||
sentiment_pct=sentiment_pct,
|
||
citation_pct=citation_pct,
|
||
competitive_pct=competitive_pct,
|
||
total_queries=total_queries,
|
||
platform_scores=platform_scores,
|
||
competitor_data=competitor_data,
|
||
)
|
||
|
||
|
||
async def generate_geo_plan(
|
||
brand_name: str,
|
||
scoring_result: ScoringResultV2,
|
||
target_score: int,
|
||
total_queries: int = 0,
|
||
platform_scores: dict[str, float] | None = None,
|
||
competitor_data: dict[str, Any] | None = None,
|
||
) -> GeoPlanData:
|
||
if settings.ENABLE_LLM and settings.DEEPSEEK_API_KEY:
|
||
return await _generate_llm_plan(
|
||
brand_name=brand_name,
|
||
overall_score=scoring_result.overall_score,
|
||
target_score=target_score,
|
||
mention_rate_pct=scoring_result.mention_rate.percentage,
|
||
rank_pct=scoring_result.recommendation_rank.percentage,
|
||
sentiment_pct=scoring_result.sentiment_score.percentage,
|
||
citation_pct=scoring_result.citation_quality.percentage,
|
||
competitive_pct=scoring_result.competitive_position.percentage,
|
||
total_queries=total_queries,
|
||
platform_scores=platform_scores or {},
|
||
competitor_data=competitor_data or {},
|
||
)
|
||
return _generate_rule_based_plan(
|
||
brand_name=brand_name,
|
||
overall_score=scoring_result.overall_score,
|
||
target_score=target_score,
|
||
mention_rate_pct=scoring_result.mention_rate.percentage,
|
||
rank_pct=scoring_result.recommendation_rank.percentage,
|
||
sentiment_pct=scoring_result.sentiment_score.percentage,
|
||
citation_pct=scoring_result.citation_quality.percentage,
|
||
competitive_pct=scoring_result.competitive_position.percentage,
|
||
total_queries=total_queries,
|
||
platform_scores=platform_scores or {},
|
||
competitor_data=competitor_data or {},
|
||
)
|