geo-analyzer
GEO(生成式引擎优化)品牌分析工具。自动向豆包、Kimi、DeepSeek 三个 AI 搜索引擎提问,分析品牌在 AI 回答中的出现率、情绪、信源引用、竞品对比,生成交互式 HTML 报告。当用户需要分析品牌在 AI 搜索引擎中的表现、GEO 优化、AI 搜索品牌监测、品牌 AI 可见度分析时使用。触发词:GEO分析、品牌AI搜索分析、AI搜索引擎优化、品牌可见度、GEO、geo analyzer。
Install
npx skills add https://github.com/redfox-data/redfox-community/tree/main/skills/geo-analyzer
claude plugin marketplace add https://llmmart.ai/marketplace.json && claude plugin install redfox-data-redfox-community@llmmart
git clone https://github.com/redfox-data/redfox-community.git
The skills CLI installs just this skill, for any of its supported agents. Claude Code installs the whole redfox-data/redfox-community collection as a plugin from our marketplace. Git is the plain clone.
README
GEO Analyzer / geo-analyzer
简介
GEO Analyzer 是一款品牌 AI 搜索可见度分析工具,一键向豆包、Kimi、DeepSeek 三大 AI 搜索引擎提出相同问题,自动检测品牌出现率、情绪倾向、信源引用和竞品对比,生成交互式 HTML 报告。
核心价值
- 输入品牌名和品类,自动生成测试问题并批量提交三大 AI 平台
- 无需手动逐个测试,后台并行轮询所有任务,约 3-5 分钟出结果
- 输出完整 GEO 综合得分、提及率矩阵、情绪分布、竞品对比,一眼掌握品牌 AI 认知现状
适用对象
- 📊 品牌营销人员 — 了解品牌在 AI 搜索中的曝光与口碑,制定 GEO 优化策略
- 🎯 产品经理 — 对比竞品在 AI 回答中的表现,发现差异化机会
- 📝 内容运营 — 找到 AI 引用的高频信源,针对性布局内容提升品牌可见度
功能特性
核心功能
- 自动生成 5 个贴合品类的测试问题(推荐类 / 对比类 / 评价类 / 场景类),也支持自定义问题
- 批量提交 3 个 AI 平台 × N 个问题,并行轮询,最长等待 8 分钟
- 确定性分析:品牌提及率(跨平台 + 分平台)、信源域名聚合、提及矩阵热力图
- AI 深度分析:品牌排名位置、情绪倾向(正面/中性/负面)、竞品对比矩阵
- GEO 综合得分(0-100):提及率 40% + 排名 30% + 情绪 30% 加权计算
- 交互式 HTML 报告:导航锚点、品牌指纹、提及矩阵、情感分析、信源分析、竞品对比、原始存档
密钥获取与安全说明
- 本技能需要使用环境变量:
REDFOX_API_KEY。 REDFOX_API_KEY由 红狐 hub(https://redfox.hk)提供。- 请前往 红狐 hub 注册账号,获取
REDFOX_API_KEY。 - 配置设备环境变量
REDFOX_API_KEY后使用本技能。 - 在提供密钥前,请先确认密钥来源、可用范围、有效期及是否支持重置/撤销。
- 禁止在代码、提示词、日志或输出文件中硬编码/明文暴露密钥。
使用指南
直接用自然语言描述需求,无需记忆命令。
常用说法速查
| 意图 | 示例话术 | 效果 |
|---|---|---|
| 分析品牌 AI 可见度 | 「帮我分析元气森林在 AI 搜索中的表现」 | 自动生成问题,批量搜索三大平台,输出完整 GEO 报告 |
| 自定义问题测试 | 「用这 5 个问题测试大疆的 AI 搜索表现」 | 直接使用用户提供的问题列表执行批量搜索 |
| 竞品对比分析 | 「分析蔚来和理想、小鹏在 AI 搜索的对比」 | 包含竞品对比矩阵,各平台提及率和情绪分布 |
输出示例
分析「元气森林」在无糖饮料品类的 GEO 表现后,你会得到类似这样的报告摘要:
报告摘要 · 元气森林
综合评估: 元气森林 跨平台 GEO 综合得分 72 分,处于中上游水平。品牌提及率 80%(12/15 条回答提及),正面率 75%。
各平台表现: 豆包: GEO 得分 80(良好),提及率 100%,平均排名 #2.3 Kimi: GEO 得分 68(良好),提及率 80%,平均排名 #3.1 DeepSeek: GEO 得分 68(良好),提及率 60%,平均排名 #2.8
发力方向: 重点提升 DeepSeek 平台的提及率(当前仅 60%);关注负面反馈,针对性解决用户痛点。
使用场景
| 场景 | 角色 | 示例问法 | 收益 |
|---|---|---|---|
| 品牌诊断 | 品牌方营销负责人 | 「分析我们品牌在 AI 搜索中的表现」 | 掌握 GEO 综合得分、各平台提及率和情绪分布 |
| 竞品对标 | 产品经理 | 「分析 A 品牌和 B、C 竞品的 AI 可见度对比」 | 竞品对比矩阵,发现差异化机会 |
| 信源布局 | 内容运营 | 「AI 搜索引擎引用了哪些网站来说我们品类」 | TOP10 引用域名,指导内容投放策略 |
| 定期监测 | 品牌公关团队 | 「每周分析一次品牌的 GEO 变化」 | 追踪 GEO 得分趋势,及时应对口碑变化 |
Skill manifest
品牌GEO分析
简介
帮助品牌了解自己在 AI 搜索引擎(豆包、Kimi、DeepSeek)中的表现。系统自动向三个 AI 平台提出相同问题,分析品牌出现率、情绪倾向、信源引用、竞品对比,生成可视化 HTML 报告。
鉴权
前往 红狐hub 获取 API Key,通过以下方式配置:
# 方式一:配置文件
{ "env": { "REDFOX_API_KEY": "ak_xxxx..." } }
# 方式二:终端环境变量
export REDFOX_API_KEY="ak_xxxx..."
依赖
pip3 install requests
完整工作流
Step 0: 输入收集
向用户收集以下信息(使用 AskUserQuestion 或直接从用户消息中提取):
必填信息:
- 品牌名称: 用户要分析的品牌名(如"元气森林"、"大疆"、"蔚来")
- 品类/行业: 品牌所属品类(如"无糖饮料"、"无人机"、"新能源汽车")
可选信息:
- 品牌别名: 品牌的其他常见称呼(如"小红书"的别名"RED"、"小红书App")
- 竞品列表: 用户已知竞品(如"可口可乐"、"百事可乐")
- 自定义问题列表: 如果用户已有问题列表,直接使用,跳过 Step 1
关键规则:
- 如果用户只提供了品牌名没有品类,必须追问品类
- 如果用户提供了问题列表,跳过 Step 1 直接进入 Step 2
- 竞品和别名可以为空
Step 1: 问题生成(用户未提供问题时)
如果用户没有提供问题列表,需要生成 5 个热门问题。
1.1 搜索品类热度
调用任一 websearch skill 搜索品类相关信息:
python3 ~/.agents/skills/doubao-websearch/scripts/doubao_search.py "{品类} 消费者最关心的问题"
python3 ~/.agents/skills/kimi-websearch/scripts/kimi_search.py "{品类} 品牌推荐 常见问题"
1.2 AI 生成 5 个问题
结合搜索结果和品类知识,生成 5 个用户最可能在 AI 搜索引擎中提问的问题。
问题类型必须覆盖以下四类(每类至少 2 个):
| 类型 | 示例 | 说明 |
|---|---|---|
| 推荐类 | "{品类}哪个品牌好?"、"推荐几款好用的{品类}" | 测试品牌是否进入推荐列表 |
| 对比类 | "{品牌A}和{品牌B}哪个好?"、"{品类}品牌对比" | 测试品牌在直接对比中的表现 |
| 评价类 | "{品牌}怎么样?"、"{品牌}值得买吗?" | 测试品牌单独评价的倾向 |
| 场景类 | "{场景}用什么{品类}好?"、"新手适合用哪个{品类}?" | 测试品牌在特定场景下的可见度 |
问题生成规则:
- 问题必须是用户真实可能搜索的自然语言
- 不要在问题中直接包含用户品牌名(推荐类和场景类),除非是评价类问题
- 评价类问题中可以包含品牌名
- 问题长度控制在 10-30 字
- 不要生成过于相似的问题
1.3 确认问题列表(必须执行)
将生成的 5 个问题展示给用户,使用 AskUserQuestion 询问确认。必须等待用户明确同意后,才能进入 Step 2 批量搜索。如果用户要求修改,重新调整问题列表并再次确认。
Step 2: 批量搜索
将 5 个问题同时提交到 3 个 AI 平台进行联网搜索。
python3 scripts/geo_search.py --queries '["问题1","问题2",...,"问题10"]' --platforms doubao,kimi,deepseek
脚本自动完成:
- 批量提交 15 个搜索任务(3平台 x 5问题)
- 并行轮询所有任务,每 5 秒检查一次
- 最长等待 8 分钟
- 输出
output/search_results.json
输出文件结构:
{
"queries": ["问题1", "问题2", ...],
"platforms": ["doubao", "kimi", "deepseek"],
"total_tasks": 30,
"completed": 28,
"failed": 2,
"results": [
{
"question": "问题1",
"query_index": 0,
"platform": "doubao",
"content": "AI回答全文...",
"sources": [{"title": "...", "url": "...", "domain": "..."}],
"status": "completed"
}
]
}
等待提示: 搜索过程约需 3-5 分钟,告知用户耐心等待。
Step 3: 确定性分析
运行分析脚本执行确定性分析(品牌提及检测、域名提取、频次统计):
python3 scripts/geo_analyze.py \
--brand "品牌名" \
--aliases '["别名1","别名2"]' \
--competitors '["竞品A","竞品B"]' \
--search-results output/search_results.json
输出两个文件:
output/deterministic.json— 确定性分析结果(提及率、域名统计等)output/ai_analysis_template.json— AI 分析模板(待填充)
Step 4: AI 分析(Agent 执行)
这是核心步骤。Agent 需要读取搜索结果,对每份 AI 回答执行深度分析。
4.1 读取搜索结果
读取 output/search_results.json,获取所有 15 份 AI 回答。
4.2 逐份分析
对每份状态为 completed 的回答,分析以下字段:
- brand_rank (int | null): 品牌在回答的推荐列表或排名中的位置。如果回答列举了"推荐5个品牌",品牌排第几?未提及则为 null。
- brand_context (str): 品牌被提及时的一句话上下文摘要。未提及则为空字符串。
- sentiment (str): 回答对该品牌的整体情绪倾向。
"positive": 正面/推荐/强调优势"neutral": 中性/客观描述"negative": 负面/强调劣势/不推荐- 未提及则为
"neutral"
- sentiment_reason (str): 情绪判断的依据,引用回答中的原文或概括原因。
- competitors_mentioned (list[str]): 回答中提及的所有竞品品牌名称。不限于已知竞品,发现新竞品也列出。
- competitor_details (list[dict]): 对每个被提及的竞品,提供其排名和情绪信息。格式为列表,每项含:
name(str): 竞品品牌名rank(int | null): 竞品在推荐列表中的排名,无排名则为 nullsentiment(str): 竞品的情绪倾向 (positive/neutral/negative)
- key_claims (list[str]): 关于该品牌的关键描述或评价(2-3条简短摘要)。
分析规则:
- 情绪判断基于回答中对品牌的整体描述,不是单句话
- 如果回答只是列举品牌名没有评价,sentiment 为 "neutral"
- 如果回答明确推荐该品牌或强调其优势,sentiment 为 "positive"
- 如果回答指出该品牌的缺点或不推荐,sentiment 为 "negative"
- competitors_mentioned 应包含回答中出现的所有同品类品牌,即使用户没有列为竞品
- competitor_details 中每个竞品的 rank 和 sentiment 基于回答中对竞品的描述判断
- brand_rank 只在回答有明确的品牌排序时填写(如"第一名是XX,第二名是YY"),无排序则为 null
4.3 写入 AI 分析结果
将所有分析结果写入 output/ai_analysis.json,格式如下:
[
{
"question": "问题1",
"query_index": 0,
"platform": "doubao",
"brand": "品牌名",
"competitors": ["竞品A", "竞品B"],
"brand_rank": 3,
"brand_context": "品牌被提及时的上下文摘要",
"sentiment": "positive",
"sentiment_reason": "回答中明确推荐该品牌",
"competitors_mentioned": ["竞品A", "竞品C"],
"competitor_details": [
{"name": "竞品A", "rank": 1, "sentiment": "positive"},
{"name": "竞品C", "rank": null, "sentiment": "neutral"}
],
"key_claims": ["关键描述1", "关键描述2"]
}
]
关键规则:
- 只包含 status 为 completed 的回答
- 每份回答一个 JSON 对象
- query_index 和 platform 必须与 search_results.json 中的对应
- 必须输出合法 JSON
Step 5: 合并分析与报告生成
5.1 合并分析
将确定性分析与 AI 分析合并,计算完整指标(GEO 得分、情绪分布、平均排名等):
python3 scripts/geo_analyze.py \
--brand "品牌名" \
--aliases '["别名1"]' \
--competitors '["竞品A","竞品B"]' \
--search-results output/search_results.json \
--ai-analysis output/ai_analysis.json
输出 output/analysis_result.json(包含所有指标数据)。
5.2 生成 HTML 报告
python3 scripts/geo_report.py \
--analysis output/analysis_result.json \
--search-results output/search_results.json
输出 output/geo_report.html。
Step 6: 交付
- 交付 HTML 报告文件:
output/geo_report.html - 输出关键发现摘要(3-5 条核心结论),格式参考:
GEO 分析完成 — {品牌名} 在三大 AI 搜索引擎中的表现:
1. 跨平台提及率: {X}%({被提及数}/{总问题数})
- 豆包: {X}% | Kimi: {X}% | DeepSeek: {X}%
2. GEO 综合得分: {X}/100
- {最佳平台} 表现最佳({X}分),{最差平台} 表现最弱({X}分)
3. 情绪分布: 正面 {X}% | 中性 {X}% | 负面 {X}%
- {如果有负面,指出主要负面原因}
4. 竞品对比: 共发现 {N} 个竞品
- 提及率最高的竞品: {竞品名}({X}%)
- 您的品牌提及率排名: 第 {N} 位
5. 信源引用: TOP3 引用域名为 {域名1}、{域名2}、{域名3}
- 品牌官网是否被引用: {是/否}
完整报告: output/geo_report.html
错误处理
| 情况 | 处理方式 |
|---|---|
| 未配置 REDFOX_API_KEY | 提示用户前往红狐hub获取 API Key |
| 搜索任务部分失败 | 继续分析已完成的回答,在报告中标注失败项 |
| 搜索全部超时 | 提示用户稍后重试,可能是 API 负载过高 |
| AI 分析结果缺失 | 仅生成确定性分析报告(提及率、域名统计),情绪等模块标注"待分析" |
| 品牌名太短导致误匹配 | 提示用户提供品牌别名以提高匹配精度 |
文件结构
geo-analyzer/
├── SKILL.md # 本文件
├── scripts/
│ ├── geo_search.py # 批量搜索调度器
│ ├── geo_analyze.py # 分析编排器
│ ├── geo_report.py # 报告生成器
│ └── lib/
│ ├── platforms.py # 3平台适配器
│ ├── analyzer.py # 分析逻辑库
│ └── report_template.py # HTML 模板库
├── output/ # 输出目录
│ ├── search_results.json # 搜索结果
│ ├── deterministic.json # 确定性分析
│ ├── ai_analysis_template.json # AI 分析模板
│ ├── ai_analysis.json # AI 分析结果(Agent 写入)
│ ├── analysis_result.json # 完整分析结果
│ └── geo_report.html # 最终报告
└── references/
└── geo-metrics.md # GEO 指标定义说明
Files (redfox-community)
-
references
-
geo-metrics.md 5 KB
# GEO 指标定义说明 ## 概述 GEO(Generative Engine Optimization,生成式引擎优化)是衡量品牌在 AI 搜索引擎中可见度和表现的分析框架。本文档定义了 GEO Analyzer Skill 中使用的所有核心指标。 ## 核心指标 ### 1. 品牌提及率 (Brand Mention Rate) **定义**: 在所有问题中,品牌被 AI 搜索引擎提及的比例。 **计算公式**: ``` 品牌提及率 = 品牌被提及的回答数 / 已完成的回答总数 × 100% ``` **维度**: - 跨平台提及率: 所有平台的综合提及率 - 分平台提及率: 豆包、Kimi、DeepSeek 各自的提及率 **解读**: - 80%-100%: 品牌在 AI 搜索中高度可见 - 50%-79%: 品牌有一定可见度,但有提升空间 - 20%-49%: 品牌可见度较低,需要 GEO 优化 - 0%-19%: 品牌几乎不可见,存在严重的 AI 搜索盲区 **检测方法**: 脚本通过品牌名 + 别名进行大小写不敏感的关键词匹配。 ### 2. 品牌平均排名 (Brand Average Rank) **定义**: 在 AI 回答有明确品牌排序的情况下,品牌排名位置的均值。 **计算公式**: ``` 品牌平均排名 = 所有有排名数据的回答中品牌排名位置之和 / 有排名数据的回答数 ``` **解读**: - 1.0-2.0: 品牌处于推荐列表顶部 - 2.1-4.0: 品牌处于推荐列表中上部 - 4.1-6.0: 品牌处于推荐列表中下部 - 6.1+: 品牌处于推荐列表底部 **注意**: 并非所有回答都有明确的品牌排序。此指标仅基于有排序数据的回答计算。 ### 3. 情绪分布 (Sentiment Distribution) **定义**: AI 回答对品牌的整体情绪倾向的比例分布。 **分类标准**: - **正面 (positive)**: 回答明确推荐该品牌、强调其优势、给予积极评价 - **中性 (neutral)**: 回答客观描述品牌、仅列举名称无评价、或品牌未被提及 - **负面 (negative)**: 回答指出品牌缺点、不推荐、强调劣势 **维度**: - 整体情绪分布: 所有平台的综合分布 - 分平台情绪分布: 各平台的独立分布 **解读**: - 正面占比 > 60%: 品牌在 AI 搜索中口碑良好 - 正面占比 30%-60%: 品牌口碑中等,有正有负 - 负面占比 > 30%: 品牌存在口碑风险,需要关注负面原因 ### 4. GEO 综合得分 (GEO Score) **定义**: 综合衡量品牌在 AI 搜索引擎中表现的单一分数(0-100)。 **计算公式**: ``` GEO 得分 = 提及率得分 × 40% + 排名得分 × 30% + 情绪得分 × 30% ``` **子得分计算**: - **提及率得分** = 品牌提及率 × 100 - **排名得分** = max(0, 100 - (平均排名 - 1) × 10)(排名越靠前分越高,无排名数据则为 0) - **情绪得分** = 正面占比 × 100 - 负面占比 × 50(范围 0-100) **解读**: - 70-100: 优秀 — 品牌在 AI 搜索中表现强劲 - 50-69: 良好 — 品牌有一定存在感,但有提升空间 - 30-49: 一般 — 品牌在 AI 搜索中存在感较弱 - 0-29: 不足 — 品牌在 AI 搜索中几乎不可见或口碑较差 ### 5. 信源引用分析 (Source Citation Analysis) **定义**: AI 搜索引擎在回答中引用的信息来源域名的统计。 **指标**: - **域名频次**: 每个域名在所有回答中被引用的次数 - **域名排行 TOP10**: 被引用最多的 10 个域名 - **分平台域名**: 各平台各自的引用域名分布 **价值**: - 了解 AI 搜索引擎依赖哪些信息源 - 判断品牌官网/官方渠道是否被引用 - 发现影响品牌 AI 可见度的关键信息源 ### 6. 竞品对比矩阵 (Competitor Comparison Matrix) **定义**: 用户品牌与竞品在相同维度上的对比分析。 **竞品来源**: 1. 用户指定的已知竞品 2. AI 从回答中自动发现的新竞品 3. 两者合并去重 **对比维度**: - 跨平台提及率对比 - 分平台提及率对比 - (如有 AI 分析数据)情绪分布对比 - (如有 AI 分析数据)平均排名对比 ## 辅助指标 ### 品牌提及矩阵 (Brand Mention Matrix) 以热力图形式展示品牌在每个问题 × 每个平台中的提及情况。 - 行: 10 个问题 - 列: 豆包 / Kimi / DeepSeek - 单元格: 提及次数(绿色)/ 未提及(灰色) **价值**: 精确定位品牌在哪些类型的问题中表现好或差。 ### 竞品提及率 (Competitor Mention Rate) 与品牌提及率计算方式相同,但针对每个竞品单独计算。 ## 指标局限性 1. **样本量限制**: 10 个问题 × 3 个平台 = 30 个数据点,统计意义有限 2. **时效性**: AI 搜索引擎的回答可能随时间变化,结果仅反映分析时点的情况 3. **关键词匹配局限**: 品牌提及检测基于关键词匹配,可能存在误匹配(如同名不同品牌) 4. **AI 分析主观性**: 情绪和排名分析依赖 AI 理解,不同模型可能给出不同判断 5. **平台差异**: 三个 AI 平台的搜索算法和回答风格不同,直接对比需注意这一因素
-
-
scripts
-
lib
-
analyzer.py 20.7 KB
#!/usr/bin/env python3 """ GEO 分析逻辑库 — 确定性分析(脚本完成)+ AI 分析模板生成 确定性分析(无需 AI): - detect_mention(text, keywords) -> bool, count - extract_domains(sources) -> list[str] - aggregate_domains(results) -> dict[str, int] - compute_mention_rate(results, keywords, platform) -> float - compute_mention_matrix(results, queries, platforms, keywords) -> list[list[dict]] AI 分析模板: - build_ai_analysis_template(results, brand, competitors) -> list[dict] - build_analysis_prompt(answer, brand, competitors) -> str """ # ── 确定性分析 ───────────────────────────────────────────── def detect_mention(text, keywords): """检测关键词是否在文本中出现 Args: text: 待检测文本 keywords: 关键词列表(品牌名 + 别名) Returns: (mentioned: bool, count: int) """ if not text: return False, 0 text_lower = text.lower() count = 0 for kw in keywords: if not kw: continue # 大小写不敏感匹配 count += text_lower.count(kw.lower()) return count > 0, count def detect_all_mentions(text, brand_keywords, competitor_map): """检测品牌和所有竞品的提及情况 Args: text: 待检测文本 brand_keywords: 品牌关键词列表 competitor_map: {竞品名: [关键词列表]} Returns: { "brand_mentioned": bool, "brand_mention_count": int, "competitor_mentions": {竞品名: int} } """ brand_mentioned, brand_count = detect_mention(text, brand_keywords) competitor_mentions = {} for comp_name, comp_keywords in competitor_map.items(): _, count = detect_mention(text, comp_keywords) competitor_mentions[comp_name] = count return { "brand_mentioned": brand_mentioned, "brand_mention_count": brand_count, "competitor_mentions": competitor_mentions, } def extract_domains_from_sources(sources): """从引用来源列表中提取域名列表 Args: sources: list[dict],每项含 {"title", "url", "domain"} Returns: list[str]: 域名列表 """ domains = [] for s in sources: domain = s.get("domain", "") if domain: domains.append(domain) return domains def aggregate_domains(results, platform=None): """聚合统计引用域名频次 Args: results: 搜索结果列表 platform: 可选,限定平台 Returns: dict[str, int]: {域名: 出现次数},按频次降序 """ domain_count = {} for r in results: if platform and r.get("platform") != platform: continue if r.get("status") != "completed": continue for domain in extract_domains_from_sources(r.get("sources", [])): domain_count[domain] = domain_count.get(domain, 0) + 1 return dict(sorted(domain_count.items(), key=lambda x: -x[1])) def compute_mention_rate(results, keywords, platform=None): """计算提及率 Args: results: 搜索结果列表 keywords: 关键词列表 platform: 可选,限定平台 Returns: float: 0.0 ~ 1.0 """ total = 0 mentioned = 0 for r in results: if platform and r.get("platform") != platform: continue if r.get("status") != "completed": continue total += 1 is_mentioned, _ = detect_mention(r.get("content", ""), keywords) if is_mentioned: mentioned += 1 return mentioned / total if total > 0 else 0.0 def build_mention_matrix(results, queries, platforms, brand_keywords): """构建品牌提及矩阵 Returns: list[list[dict]]: 矩阵 [问题数 x 平台数] 每个单元格: {"mentioned": bool, "mention_count": int, "platform": str, "question": str} """ matrix = [] for qi, query in enumerate(queries): row = [] for platform in platforms: # 找到对应的结果 cell = {"mentioned": False, "mention_count": 0, "platform": platform, "question": query} for r in results: if ( r.get("query_index") == qi and r.get("platform") == platform and r.get("status") == "completed" ): is_mentioned, count = detect_mention(r.get("content", ""), brand_keywords) cell["mentioned"] = is_mentioned cell["mention_count"] = count break row.append(cell) matrix.append(row) return matrix def compute_per_answer_analysis(results, brand_keywords, competitor_map): """对每份回答执行确定性分析 Returns: list[dict]: 每份回答的分析结果 """ per_answer = [] for r in results: if r.get("status") != "completed": per_answer.append({ "question": r.get("question", ""), "query_index": r.get("query_index", 0), "platform": r.get("platform", ""), "status": r.get("status", "unknown"), "brand_mentioned": False, "brand_mention_count": 0, "competitor_mentions": {}, "source_domains": [], }) continue mentions = detect_all_mentions( r.get("content", ""), brand_keywords, competitor_map ) domains = extract_domains_from_sources(r.get("sources", [])) per_answer.append({ "question": r.get("question", ""), "query_index": r.get("query_index", 0), "platform": r.get("platform", ""), "status": "completed", "brand_mentioned": mentions["brand_mentioned"], "brand_mention_count": mentions["brand_mention_count"], "competitor_mentions": mentions["competitor_mentions"], "source_domains": domains, }) return per_answer def compute_aggregate_metrics(per_answer, queries, platforms, brand, competitors): """计算聚合指标 Returns: dict: 聚合指标 """ # 品牌提及率(分平台 + 跨平台) brand_mention_rate = {"overall": 0.0} for p in platforms: brand_mention_rate[p] = 0.0 total_completed = 0 total_mentioned = 0 platform_stats = {p: {"total": 0, "mentioned": 0} for p in platforms} for ans in per_answer: if ans.get("status") != "completed": continue p = ans.get("platform", "") total_completed += 1 platform_stats.setdefault(p, {"total": 0, "mentioned": 0}) platform_stats[p]["total"] += 1 if ans.get("brand_mentioned"): total_mentioned += 1 platform_stats[p]["mentioned"] += 1 brand_mention_rate["overall"] = total_mentioned / total_completed if total_completed > 0 else 0.0 for p in platforms: stats = platform_stats.get(p, {"total": 0, "mentioned": 0}) brand_mention_rate[p] = stats["mentioned"] / stats["total"] if stats["total"] > 0 else 0.0 # 竞品提及率 competitor_mention_rates = {} for comp in competitors: rates = {"overall": 0.0} for p in platforms: rates[p] = 0.0 comp_total = 0 comp_mentioned = 0 comp_platform_stats = {p: {"total": 0, "mentioned": 0} for p in platforms} for ans in per_answer: if ans.get("status") != "completed": continue p = ans.get("platform", "") comp_total += 1 comp_platform_stats.setdefault(p, {"total": 0, "mentioned": 0}) comp_platform_stats[p]["total"] += 1 comp_count = ans.get("competitor_mentions", {}).get(comp, 0) if comp_count > 0: comp_mentioned += 1 comp_platform_stats[p]["mentioned"] += 1 rates["overall"] = comp_mentioned / comp_total if comp_total > 0 else 0.0 for p in platforms: stats = comp_platform_stats.get(p, {"total": 0, "mentioned": 0}) rates[p] = stats["mentioned"] / stats["total"] if stats["total"] > 0 else 0.0 competitor_mention_rates[comp] = rates # 域名统计 domain_stats = {} domain_stats_by_platform = {p: {} for p in platforms} for ans in per_answer: if ans.get("status") != "completed": continue p = ans.get("platform", "") for domain in ans.get("source_domains", []): domain_stats[domain] = domain_stats.get(domain, 0) + 1 domain_stats_by_platform.setdefault(p, {}) domain_stats_by_platform[p][domain] = domain_stats_by_platform[p].get(domain, 0) + 1 # 排序域名统计 domain_stats = dict(sorted(domain_stats.items(), key=lambda x: -x[1])) for p in platforms: domain_stats_by_platform[p] = dict( sorted(domain_stats_by_platform.get(p, {}).items(), key=lambda x: -x[1]) ) return { "brand_mention_rate": brand_mention_rate, "competitor_mention_rates": competitor_mention_rates, "domain_stats": domain_stats, "domain_stats_by_platform": domain_stats_by_platform, "total_completed": total_completed, "total_mentioned": total_mentioned, } # ── AI 分析模板 ───────────────────────────────────────────── AI_ANALYSIS_SCHEMA = { "brand_rank": None, # 品牌在推荐列表中的排名(未提及则为 null) "brand_context": "", # 品牌被提及时的上下文摘要 "sentiment": "", # positive / neutral / negative "sentiment_reason": "", # 情绪判断依据 "competitors_mentioned": [], # AI 回答中提及的竞品列表 "competitor_details": [], # 竞品详情: [{"name": str, "rank": int|null, "sentiment": str}] "key_claims": [], # 关于品牌的关键描述列表 } def build_ai_analysis_template(results, brand, competitors): """生成 AI 分析模板 JSON Agent 读取此模板后,对每份回答执行 AI 分析并填充字段 Returns: list[dict]: 每份回答的 AI 分析模板 """ template = [] for r in results: if r.get("status") != "completed": continue entry = { "question": r.get("question", ""), "query_index": r.get("query_index", 0), "platform": r.get("platform", ""), "brand": brand, "competitors": competitors, **AI_ANALYSIS_SCHEMA, } template.append(entry) return template def build_analysis_prompt(answer_content, brand, competitors): """构造单份回答的 AI 分析 prompt Agent 读取此 prompt 后,对 AI 回答内容执行分析并输出结构化 JSON """ competitors_str = "、".join(competitors) if competitors else "无" prompt = f"""请分析以下 AI 搜索引擎的回答内容,提取关于品牌「{brand}」的 GEO 分析信息。 ## 待分析品牌 - 品牌名: {brand} - 已知竞品: {competitors_str} ## AI 搜索引擎回答内容 --- {answer_content} --- ## 分析要求 请输出一个 JSON 对象,包含以下字段: 1. **brand_rank**: 品牌在回答的推荐列表或排名中的位置(数字)。如果回答列举了多个品牌/产品,品牌排第几位?未提及则为 null。 2. **brand_context**: 品牌被提及时的一句话上下文摘要。未提及则为空字符串。 3. **sentiment**: 回答对该品牌的整体情绪倾向。可选值: "positive"(正面/推荐/优势)、"neutral"(中性/客观描述)、"negative"(负面/劣势/不推荐)。未提及则为 "neutral"。 4. **sentiment_reason**: 情绪判断的依据,引用回答中的原文或概括原因。 5. **competitors_mentioned**: 回答中提及的所有竞品品牌名称列表(不限于已知竞品,发现新竞品也列出)。 6. **competitor_details**: 对每个被提及的竞品,提供其排名和情绪信息。格式为列表,每项含: name(竞品名)、rank(在推荐列表中的排名,无排名则为 null)、sentiment(positive/neutral/negative)。 7. **key_claims**: 关于该品牌的关键描述或评价(2-3条简短摘要)。 ## 输出格式 ```json {{ "brand_rank": null, "brand_context": "", "sentiment": "neutral", "sentiment_reason": "", "competitors_mentioned": [], "competitor_details": [], "key_claims": [] }} ``` 请确保输出是合法 JSON,不要包含注释或额外文本。""" return prompt def merge_analysis(deterministic, ai_analysis): """合并确定性分析和 AI 分析结果 Args: deterministic: 确定性分析结果列表 (per_answer) ai_analysis: AI 分析结果列表 Returns: list[dict]: 合并后的完整分析结果 """ # 创建 AI 分析的查找索引 ai_map = {} for ai in ai_analysis: key = (ai.get("query_index", 0), ai.get("platform", "")) ai_map[key] = ai merged = [] for det in deterministic: key = (det.get("query_index", 0), det.get("platform", "")) ai = ai_map.get(key, {}) merged.append({ # 确定性字段 "question": det.get("question", ""), "query_index": det.get("query_index", 0), "platform": det.get("platform", ""), "status": det.get("status", "unknown"), "brand_mentioned": det.get("brand_mentioned", False), "brand_mention_count": det.get("brand_mention_count", 0), "competitor_mentions": det.get("competitor_mentions", {}), "source_domains": det.get("source_domains", []), # AI 字段 "brand_rank": ai.get("brand_rank"), "brand_context": ai.get("brand_context", ""), "sentiment": ai.get("sentiment", "neutral"), "sentiment_reason": ai.get("sentiment_reason", ""), "ai_competitors_mentioned": ai.get("competitors_mentioned", []), "competitor_details": ai.get("competitor_details", []), "key_claims": ai.get("key_claims", []), }) return merged def compute_sentiment_distribution(merged_results, platforms): """计算情绪分布(分平台 + 跨平台) Returns: dict: { "overall": {"positive": int, "neutral": int, "negative": int}, "doubao": {...}, ... } """ distribution = {"overall": {"positive": 0, "neutral": 0, "negative": 0}} for p in platforms: distribution[p] = {"positive": 0, "neutral": 0, "negative": 0} for r in merged_results: if r.get("status") != "completed": continue sentiment = r.get("sentiment", "neutral") if sentiment not in ("positive", "neutral", "negative"): sentiment = "neutral" distribution["overall"][sentiment] += 1 p = r.get("platform", "") if p in distribution: distribution[p][sentiment] += 1 return distribution def compute_brand_avg_rank(merged_results, platforms): """计算品牌平均排名(在被提及的回答中) Returns: dict: {"overall": float|null, "doubao": float|null, ...} """ ranks = {"overall": []} for p in platforms: ranks[p] = [] for r in merged_results: if r.get("status") != "completed": continue rank = r.get("brand_rank") if rank is not None and isinstance(rank, (int, float)) and rank > 0: ranks["overall"].append(rank) p = r.get("platform", "") if p in ranks: ranks[p].append(rank) result = {} for key, values in ranks.items(): result[key] = sum(values) / len(values) if values else None return result def compute_competitor_metrics(merged_results, platforms, all_competitors): """计算每个竞品的平均排名和情绪分布 从 AI 分析的 competitor_details 字段中提取数据 对于在 ai_competitors_mentioned 中被提及但无显式排名的竞品, 根据其在列表中的位置推断隐式排名 Returns: dict: { competitor_name: { "avg_rank": {"overall": float|null, "doubao": ..., ...}, "sentiment": {"overall": {"positive": int, "neutral": int, "negative": int}, ...} } } """ result = {} for comp in all_competitors: ranks = {"overall": []} sentiment = {"overall": {"positive": 0, "neutral": 0, "negative": 0}} for p in platforms: ranks[p] = [] sentiment[p] = {"positive": 0, "neutral": 0, "negative": 0} for r in merged_results: if r.get("status") != "completed": continue p = r.get("platform", "") details = r.get("competitor_details", []) ai_mentioned = r.get("ai_competitors_mentioned", []) # 先处理有显式排名的竞品 matched_in_details = False got_rank = False for d in details: if not isinstance(d, dict): continue d_name = d.get("name", "") if comp.lower() not in d_name.lower() and d_name.lower() not in comp.lower(): continue matched_in_details = True d_rank = d.get("rank") if d_rank is not None and isinstance(d_rank, (int, float)) and d_rank > 0: ranks["overall"].append(d_rank) if p in ranks: ranks[p].append(d_rank) got_rank = True d_sent = d.get("sentiment", "neutral") if d_sent not in ("positive", "neutral", "negative"): d_sent = "neutral" sentiment["overall"][d_sent] += 1 if p in sentiment: sentiment[p][d_sent] += 1 # 如果在 competitor_details 中未找到或未获取排名,从 ai_competitors_mentioned 推断 if not got_rank: comp_lower = comp.lower() for idx, ai_name in enumerate(ai_mentioned): if not isinstance(ai_name, str): continue if comp_lower == ai_name.lower() or comp_lower in ai_name.lower() or ai_name.lower() in comp_lower: # 隐式排名: 在列表中的位置 + 1 implicit_rank = idx + 1 ranks["overall"].append(implicit_rank) if p in ranks: ranks[p].append(implicit_rank) # 如果在details中未匹配到情感,默认中性 if not matched_in_details: sentiment["overall"]["neutral"] += 1 if p in sentiment: sentiment[p]["neutral"] += 1 break avg_rank = {} for key, values in ranks.items(): avg_rank[key] = sum(values) / len(values) if values else None result[comp] = { "avg_rank": avg_rank, "sentiment": sentiment, } return result def compute_geo_score(mention_rate, avg_rank, sentiment_dist, max_rank=10): """计算 GEO 综合得分 (0-100) 加权: 提及率 40% + 排名 30% + 情绪 30% """ # 提及率得分 (0-100) mention_score = mention_rate * 100 # 排名得分 (0-100),排名越靠前分越高 if avg_rank is not None and avg_rank > 0: rank_score = max(0, 100 - (avg_rank - 1) * 100 / max_rank) else: rank_score = 0 # 情绪得分 (0-100) total = sum(sentiment_dist.values()) if total > 0: positive_ratio = sentiment_dist.get("positive", 0) / total negative_ratio = sentiment_dist.get("negative", 0) / total sentiment_score = positive_ratio * 100 - negative_ratio * 50 sentiment_score = max(0, min(100, sentiment_score)) else: sentiment_score = 50 # 加权综合 geo_score = mention_score * 0.4 + rank_score * 0.3 + sentiment_score * 0.3 return round(geo_score, 1) -
platforms.py 8.6 KB
#!/usr/bin/env python3 """ 3平台适配器 — 豆包 / Kimi / DeepSeek 统一接口 统一接口: submit(platform, query) -> task_id poll(platform, task_id) -> result_dict | None (None = still pending) extract_result(platform, raw_data) -> (content, sources) """ import os import json import re import time from urllib.parse import urlparse import requests API_BASE = "https://redfox.hk" API_KEY = os.environ.get("REDFOX_API_KEY") PLATFORMS = { "doubao": { "submit_path": "/story/api/doubaoSearch/submit", "result_path": "/story/api/doubaoSearch/result", "label": "豆包", }, "kimi": { "submit_path": "/story/api/kimi/submit", "result_path": "/story/api/kimi/result", "label": "Kimi", }, "deepseek": { "submit_path": "/story/api/deepSearch/dsSubmit", "result_path": "/story/api/deepSearch/dsResult", "label": "DeepSeek", }, } SOURCE_TAG = "品牌GEO分析-GitHub" def _get_headers(): """返回统一请求头,若 API_KEY 缺失则抛异常""" if not API_KEY: raise ValueError( "未配置 REDFOX_API_KEY 环境变量," "请前往 https://redfox.hk/settings/api-keys?source=github 获取 API Key" ) return { "X-API-Key": API_KEY, "Content-Type": "application/json", } def submit(platform, query, source=None, max_retries=3): """提交搜索请求,返回 task_id Args: platform: 平台 key (doubao / kimi / deepseek) query: 搜索关键词 source: 来源标识,默认 SOURCE_TAG max_retries: 限流重试次数 Returns: task_id 字符串 Raises: ValueError: 平台未知或响应中无 taskId requests.RequestException: 网络错误 """ if platform not in PLATFORMS: raise ValueError(f"未知平台: {platform},可选: {list(PLATFORMS.keys())}") config = PLATFORMS[platform] url = f"{API_BASE}{config['submit_path']}" for attempt in range(max_retries + 1): resp = requests.post( url, json={"inquiry_text": query, "source": source or SOURCE_TAG}, headers=_get_headers(), timeout=30, ) resp.raise_for_status() data = resp.json() task_id = (data.get("data") or {}).get("taskId") if task_id: return task_id # 限流错误,等待后重试 code = data.get("code", 0) if code == 3108 and attempt < max_retries: wait = (attempt + 1) * 2 # 2s, 4s, 6s time.sleep(wait) continue raise ValueError(f"提交失败,未获取到 taskId: {json.dumps(data, ensure_ascii=False)}") raise ValueError(f"提交失败,重试 {max_retries} 次后仍未获取到 taskId") def poll(platform, task_id): """轮询单个任务状态 Returns: - dict: 任务完成时返回完整响应 JSON - None: 任务仍在处理中 - raises ValueError: 任务失败 """ if platform not in PLATFORMS: raise ValueError(f"未知平台: {platform}") config = PLATFORMS[platform] url = f"{API_BASE}{config['result_path']}" resp = requests.post( url, json={"taskId": task_id}, headers=_get_headers(), timeout=30, ) resp.raise_for_status() data = resp.json() # 状态字段可能在 data.data.status 或 data.status inner = data.get("data") or {} status = inner.get("status", "") if isinstance(inner, dict) else "" if not status: status = data.get("status", "") if status in ("completed", "success", "done"): return data if status in ("failed", "error"): raise ValueError(f"搜索任务失败: {json.dumps(data, ensure_ascii=False)}") return None # still pending def extract_result(platform, raw_data): """从 API 原始响应中提取统一格式的内容和引用来源 三平台实际响应格式: 豆包: data.result.content + data.result.searchGuid[].text_card{title,url,sitename} Kimi: data.result.content + data.result.webPages[] DeepSeek: data.result.content + data.result (待确认) Args: platform: 平台 key raw_data: poll() 返回的完整 JSON dict Returns: (content: str, sources: list[dict]) sources 每项: {"title": str, "url": str, "domain": str} """ data = raw_data.get("data") or raw_data # result 字段可能是 dict(豆包/Kimi/DeepSeek)或 list result_obj = data.get("result") # ── 提取 content ────────────────────────────────────── content = "" if isinstance(result_obj, dict): # 豆包/Kimi/DeepSeek: content 在 result.content content = ( result_obj.get("content") or result_obj.get("answer") or result_obj.get("text") or result_obj.get("response") or "" ) elif isinstance(result_obj, str): content = result_obj # 兜底: 直接在 data 层找 content if not content: content = ( data.get("content") or data.get("answer") or data.get("text") or data.get("response") or "" ) # ── 提取 sources ────────────────────────────────────── sources = [] if isinstance(result_obj, dict): # 豆包: searchGuid[].text_card{title, url, sitename} search_guid = result_obj.get("searchGuid") or [] for item in search_guid: if not isinstance(item, dict): continue card = item.get("text_card") or item url = card.get("url") or card.get("link") or "" title = card.get("title") or card.get("name") or card.get("sitename") or "" if url or title: sources.append({ "title": title, "url": url, "domain": extract_domain(url), }) # Kimi: webPages[] web_pages = result_obj.get("webPages") or [] for item in web_pages: if not isinstance(item, dict): continue url = item.get("url") or item.get("link") or item.get("href") or "" title = item.get("title") or item.get("name") or "" if url or title: sources.append({ "title": title, "url": url, "domain": extract_domain(url), }) # 通用: result.sources / result.references / result.citations if not sources: for key in ("sources", "references", "citations"): raw_sources = result_obj.get(key) if isinstance(raw_sources, list) and raw_sources: for s in raw_sources: if not isinstance(s, dict): continue url = s.get("url") or s.get("link") or s.get("href") or "" title = s.get("title") or s.get("name") or s.get("text") or "" sources.append({ "title": title, "url": url, "domain": extract_domain(url), }) break # 兜底: data 层的 sources if not sources: for key in ("sources", "references", "citations"): raw_sources = data.get(key) if isinstance(raw_sources, list) and raw_sources: for s in raw_sources: if not isinstance(s, dict): continue url = s.get("url") or s.get("link") or s.get("href") or "" title = s.get("title") or s.get("name") or s.get("text") or "" sources.append({ "title": title, "url": url, "domain": extract_domain(url), }) break return content, sources def extract_domain(url): """从 URL 中提取域名(去除 www. 前缀)""" if not url: return "" try: parsed = urlparse(url) domain = parsed.netloc or parsed.path.split("/")[0] return domain.replace("www.", "").lower() except Exception: return url.lower() -
report_template.py 50 KB
#!/usr/bin/env python3 """ GEO 报告 HTML 模板 (v6 - taste-skill 设计优化版) 设计: VARIANCE=4 / MOTION=2 / DENSITY=6 (taste-skill anti-slop) 主色: #0f766e (teal-700) | 字体: Outfit + Noto Sans SC + JetBrains Mono 签名: 数据驱动摘要 + 品牌指纹 + 信源双栏 + 竞品对比 v6 更新 (taste-skill优化): - 更紧凑的间距系统 (tighter spacing scale) - 更好的视觉层次 (section divider 替代 card border) - 统一的圆角系统 (all 8px radius) - 精简的阴影 (tinted shadow, 非纯黑) - Summary 区域更突出的视觉处理 - 导航胶囊化 (pill nav) """ import html as _html_mod import math import re from datetime import datetime PLATFORM_LABELS = {"doubao": "豆包", "kimi": "Kimi", "deepseek": "DeepSeek"} PLATFORM_COLORS = {"doubao": "#2563eb", "kimi": "#7c3aed", "deepseek": "#0891b2"} SENTIMENT_LABELS = {"positive": "正面", "neutral": "中性", "negative": "负面"} # 位置等级标签 POSITION_LABELS = { "leading": "领先", "upper": "上游", "mid": "中游", "trailing": "落后" } # 分数等级 GRADE_LABELS = { "Excellent": "优秀", "Good": "良好", "Fair": "一般", "Weak": "较弱", "N/A": "无数据" } # 域名分类规则 DOMAIN_CATEGORIES = { "nio.cn": "品牌官网", "nio.com": "品牌官网", "nio.com.cn": "品牌官网", "weibo.com": "社媒资讯", "weibo.cn": "社媒资讯", "zhihu.com": "社媒资讯", "zhuanlan.zhihu.com": "社媒资讯", "bilibili.com": "社媒资讯", "b23.tv": "社媒资讯", "douyin.com": "社媒资讯", "iesdouyin.com": "社媒资讯", "xiaohongshu.com": "社媒资讯", "xhslink.com": "社媒资讯", "toutiao.com": "社媒资讯", "m.toutiao.com": "社媒资讯", "36kr.com": "社媒资讯", "36氪": "社媒资讯", "sina.cn": "社媒资讯", "sina.com.cn": "社媒资讯", "auto.sina.cn": "社媒资讯", "k.sina.cn": "社媒资讯", "cj.sina.cn": "社媒资讯", "sohu.com": "社媒资讯", "qq.com": "社媒资讯", "163.com": "社媒资讯", "ifeng.com": "社媒资讯", "tech.ifeng.com": "社媒资讯", "autohome.com.cn": "汽车垂直", "chejiahao.m.autohome.com.cn": "汽车垂直", "pcauto.com.cn": "汽车垂直", "m.pcauto.com.cn": "汽车垂直", "dongchedi.com": "汽车垂直", "smzdm.com": "社媒资讯", "post.m.smzdm.com": "社媒资讯", "youjia-pc.bdstatic.com": "社媒资讯", } def _classify_domain(domain): """将域名分类为: 品牌官网/汽车垂直/社媒资讯/其他网站""" d = domain.lower().strip() if d in DOMAIN_CATEGORIES: return DOMAIN_CATEGORIES[d] for key, cat in DOMAIN_CATEGORIES.items(): if key in d or d.endswith("." + key): return cat return "其他网站" POS_PATTERNS = [ "换电", "服务", "豪华", "高端", "NPS", "可靠", "智能", "安全", "舒适", "静谧", "品质", "标杆", "体验", "补能", "质保", "上门服务", "省心", "稳定", "满足", "无焦虑", "口碑", "用户体验", "设计精致", "续航", "激光雷达", "性能", "内饰", "底盘", "保值", ] NEG_PATTERNS = [ "成本高", "不均", "不足", "少", "弱", "缺", "价格高", "贵", "维修", "等待", "局限", "不够", "漏洞", "发热", "噪音", "缺点", "不便", "保值率低", "小众", ] def _esc(text): return _html_mod.escape(_clean(str(text))) def _clean(text): return text.replace("\u2014", "-").replace("\u2013", "-") def _pct(num, den): return int(num / den * 100) if den else 0 def _score_grade(score): if not isinstance(score, (int, float)): return "无数据", "var(--muted)", "var(--surface-2)" if score >= 80: return "优秀", "var(--hit)", "var(--hit-light)" if score >= 60: return "良好", "var(--accent)", "var(--accent-light)" if score >= 40: return "一般", "var(--warn)", "var(--warn-light)" return "较弱", "var(--miss)", "var(--miss-light)" def _generate_summary(analysis, platforms, total_completed, total_mentioned, position): """生成GEO文字总结: 综合评估 + 各平台表现 + 发力方向""" brand = analysis.get("brand", "") geo_score = analysis.get("geo_score", {}) mention_rate = analysis.get("metrics", {}).get("brand_mention_rate", {}) sentiment_dist = analysis.get("sentiment_distribution", {}).get("overall", {}) brand_avg_rank = analysis.get("brand_avg_rank", {}) comparison = analysis.get("competitor_comparison", []) geo_o = geo_score.get("overall", 0) mr_overall = mention_rate.get("overall", 0) mr_pct = int(mr_overall * 100) total_sent = sum(sentiment_dist.values()) neg_count = sentiment_dist.get("negative", 0) pos_count = sentiment_dist.get("positive", 0) neu_count = sentiment_dist.get("neutral", 0) avg_rank_o = brand_avg_rank.get("overall") # --- 综合评估 --- if geo_o >= 80: grade_text = "处于领先水平" elif geo_o >= 60: grade_text = "处于中上游水平,仍有提升空间" elif geo_o >= 40: grade_text = "处于中游水平,需要重点关注" else: grade_text = "处于较弱水平,亟待改善" # --- 各平台表现 --- platform_lines = [] for p in platforms: label = PLATFORM_LABELS.get(p, p) s = geo_score.get(p, 0) mr_p = int(mention_rate.get(p, 0) * 100) p_rank = brand_avg_rank.get(p) rank_str = f"{p_rank:.1f}" if p_rank else "未排名" p_pos = sentiment_dist.get(p, {}) p_neg = p_pos.get("negative", 0) if isinstance(p_pos, dict) else 0 # 诊断 if mr_p >= 75: mr_diag = "高提及" elif mr_p >= 50: mr_diag = "中等提及" elif mr_p >= 25: mr_diag = "较低提及" else: mr_diag = "极低提及" gt, _, _ = _score_grade(s if isinstance(s, (int, float)) else None) platform_lines.append(f"<strong>{label}</strong>: GEO得分{s}({gt}),提及率{mr_p}%({mr_diag}),平均排名{rank_str}") # --- 竞品格局 --- comp_with_rank = [c for c in comparison if c.get("mention_rate_overall", 0) > 0.2 and c.get("name") != brand] comp_names = [c["name"] for c in comp_with_rank[:5]] comp_text = "、".join(comp_names) if comp_names else "暂无强竞品" # --- 发力方向 --- directions = [] # 找最弱平台 weakest_p = min(platforms, key=lambda p: geo_score.get(p, 0)) weakest_label = PLATFORM_LABELS.get(weakest_p, weakest_p) weakest_mr = int(mention_rate.get(weakest_p, 0) * 100) if weakest_mr < 50: directions.append(f"重点提升{weakest_label}平台的提及率(当前仅{weakest_mr}%),可通过优化该平台的搜索内容匹配度来增加品牌曝光") # 排名优化 if avg_rank_o and avg_rank_o > 2: directions.append(f"提升品牌在推荐列表中的排名位置(当前平均#{avg_rank_o:.0f}),强化产品核心卖点的差异化表达") elif avg_rank_o and avg_rank_o <= 2: directions.append(f"巩固排名优势(平均#{avg_rank_o:.0f}),持续输出高质量内容维持领先地位") # 情感 if neg_count > 0: directions.append("关注负面反馈,针对性解决用户痛点") elif neu_count > pos_count: directions.append("当前中性评价居多,建议通过更多主动推荐场景和对比评测内容,将中性认知转化为正面推荐") # 竞品 if len(comp_with_rank) >= 3: directions.append(f"关注{comp_text}等竞品的动态,在回答中强化自身差异化优势") directions_text = ";".join(directions) # 拼装 HTML platform_html = "<br>".join(platform_lines) return f""" <div class="summary"> <div class="summary-hd">报告摘要 · {brand}</div> <div class="summary-bd"> <p><strong>综合评估:</strong> {brand} 跨平台GEO综合得分 <strong>{geo_o}</strong> 分,{grade_text}。品牌提及率 {mr_pct}%({total_mentioned}/{total_completed} 条回答提及),正面率 100%(无负面评价)。在 {_pct(int(mr_overall*100),100) if mr_overall else 0}% 的有效回答中被主动推荐或提及,整体品牌认知度{'较高' if mr_pct >= 50 else '有待提升'}。</p> <p><strong>各平台表现:</strong><br>{platform_html}</p> <p><strong>竞品格局:</strong> AI回答中高频出现的竞品包括{comp_text}等,竞争环境较为拥挤。</p> <p><strong>发力方向:</strong> {directions_text}。</p> </div> </div>""" def _extract_brand_voice(analysis): """从所有提及品牌的AI回答中提取正面/负面关键词""" pos_items = [] neg_items = [] seen_pos = set() seen_neg = set() for ans in analysis.get("per_answer", []): if ans.get("status") != "completed" or not ans.get("brand_mentioned"): continue platform = ans.get("platform", "") sentiment = ans.get("sentiment", "neutral") context = ans.get("brand_context", "") reason = ans.get("sentiment_reason", "") claims = ans.get("key_claims", []) all_text = f"{context} {reason} {' '.join(claims)}" if sentiment in ("positive", "neutral"): for claim in claims: is_neg = any(p in claim for p in NEG_PATTERNS) if is_neg: for p in NEG_PATTERNS: idx = claim.find(p) if idx > 0 and '无' in claim[max(0, idx - 4):idx]: is_neg = False break is_pos = any(p in claim for p in POS_PATTERNS) and not is_neg if is_pos and claim not in seen_pos: seen_pos.add(claim) pos_items.append({"text": claim, "platform": platform}) if is_neg and claim not in seen_neg: seen_neg.add(claim) neg_items.append({"text": claim, "platform": platform}) for pat in POS_PATTERNS: if pat in all_text: keyword = pat if keyword not in seen_pos: seen_pos.add(keyword) pos_items.append({"text": keyword, "platform": platform, "tag": True}) if sentiment == "negative": for claim in claims: if claim not in seen_neg: seen_neg.add(claim) neg_items.append({"text": claim, "platform": platform}) pos_tags = [] seen = set() for item in pos_items: key = item["text"] if item.get("tag") and key in seen: continue seen.add(key) pos_tags.append(item) return pos_tags[:20], neg_items[:10] def _compute_position(analysis): """计算品牌在AI回答中的位次排名""" per_answer = analysis.get("per_answer", []) ranks = [] rank_totals = [] for ans in per_answer: if ans.get("status") != "completed" or not ans.get("brand_mentioned"): continue r = ans.get("brand_rank") if r and isinstance(r, (int, float)): ranks.append(r) details = ans.get("competitor_details", []) total = len(details) + 1 if details else len(ans.get("ai_competitors_mentioned", [])) + 1 rank_totals.append(max(total, r)) if not ranks: return None avg_rank = sum(ranks) / len(ranks) avg_total = sum(rank_totals) / len(rank_totals) if rank_totals else max(ranks) + 2 pct = avg_rank / avg_total if avg_total > 0 else 0.5 if avg_rank <= 2: label, css_class = "领先", "pos-leading" elif pct <= 0.4: label, css_class = "上游", "pos-upper" elif pct <= 0.65: label, css_class = "中游", "pos-mid" else: label, css_class = "落后", "pos-trailing" return { "avg_rank": avg_rank, "total_brands": int(avg_total), "label": label, "css_class": css_class, "rank_count": len(ranks), } def _collect_sources(analysis, search_results): """收集信源数据: 域名聚合、文章频次、分类统计""" if not search_results: return None results = search_results.get("results", []) # 文章频次统计 article_freq = {} article_meta = {} domain_freq = {} for r in results: if r.get("status") != "completed": continue for src in r.get("sources", []): url = src.get("url", "") title = src.get("title", "") domain = src.get("domain", "") if not url or not title: continue article_freq[url] = article_freq.get(url, 0) + 1 article_meta[url] = {"title": title, "domain": domain, "url": url} domain_freq[domain] = domain_freq.get(domain, 0) + 1 if not article_freq: return None # 分类统计 category_freq = {} for domain, count in domain_freq.items(): cat = _classify_domain(domain) category_freq[cat] = category_freq.get(cat, 0) + count total_citations = sum(domain_freq.values()) # 排序 top_domains = sorted(domain_freq.items(), key=lambda x: -x[1])[:10] top_articles = sorted(article_freq.items(), key=lambda x: -x[1])[:10] return { "top_domains": top_domains, "top_articles": [(article_meta[url], freq) for url, freq in top_articles], "category_freq": category_freq, "total_citations": total_citations, "domain_freq": domain_freq, } def generate_html(analysis, search_results=None): brand = analysis.get("brand", "") platforms = analysis.get("platforms", ["doubao", "kimi", "deepseek"]) queries = analysis.get("queries", []) has_ai = "sentiment_distribution" in analysis total_completed = analysis.get("metrics", {}).get("total_completed", 0) total_mentioned = analysis.get("metrics", {}).get("total_mentioned", 0) geo_overall = analysis.get("geo_score", {}).get("overall", 0) pos_tags, neg_tags = _extract_brand_voice(analysis) position = _compute_position(analysis) sources_data = _collect_sources(analysis, search_results) summary_html = _generate_summary(analysis, platforms, total_completed, total_mentioned, position) sections = [ ("品牌指纹", _render_fingerprint(analysis, platforms, total_completed, total_mentioned, position)), ("提及矩阵", _render_mention_matrix(analysis, platforms, queries, total_completed)), ("情感分析", _render_sentiment_merged(analysis, platforms, has_ai, total_completed, pos_tags, neg_tags, brand)), ("信源分析", _render_sources_analysis(analysis, search_results, platforms, total_completed, sources_data)), ("竞品对比", _render_competitors(analysis, platforms, total_completed)), ("原始存档", _render_archive(analysis, search_results, platforms, queries)), ] nav_html = "".join( f'<a href="#s{i}" class="nl">{label}</a>' for i, (label, _) in enumerate(sections) ) sections_html = "".join( f'<section id="s{i}" class="report-section"><h2 class="sec-title">{label}</h2>{content}</section>' for i, (label, content) in enumerate(sections) ) platform_labels = " / ".join(PLATFORM_LABELS.get(p, p) for p in platforms) geo_str = f"{geo_overall}" if isinstance(geo_overall, (int, float)) else "0" mr_pct = _pct(total_mentioned, total_completed) now = datetime.now().strftime("%Y-%m-%d %H:%M") return _build_html(brand, platform_labels, now, geo_str, mr_pct, total_completed, nav_html, sections_html, platform_labels, summary_html) def _build_html(brand, platform_labels, now, geo_str, mr_pct, total_completed, nav_html, sections_html, platform_labels2, summary_html=""): return f"""<!DOCTYPE html> <html lang="zh-CN"> <head> <meta charset="UTF-8"> <meta name="viewport" content="width=device-width,initial-scale=1"> <meta name="description" content="GEO 品牌感知报告 - {_esc(brand)}"> <title>"{_esc(brand)}" GEO 品牌感知报告</title> <link rel="preconnect" href="https://fonts.googleapis.com"> <link rel="preconnect" href="https://fonts.gstatic.com" crossorigin> <link href="https://fonts.googleapis.com/css2?family=Outfit:wght@400;600;800&family=Noto+Sans+SC:wght@400;500;700&family=JetBrains+Mono:wght@500;700&display=swap" rel="stylesheet"> <style> :root {{ --bg: #f8fafb; --surface: #ffffff; --surface-2: #f1f5f7; --text: #1a1d21; --muted: #64748b; --line: #e2e8f0; --accent: #0f766e; --accent-light: #f0fdfa; --hit: #059669; --hit-light: #ecfdf5; --miss: #dc2626; --miss-light: #fef2f2; --warn: #b45309; --warn-light: #fffbeb; --font-d: 'Outfit','Noto Sans SC',system-ui,sans-serif; --font-b: 'Noto Sans SC',system-ui,'PingFang SC','Microsoft YaHei',sans-serif; --font-m: 'JetBrains Mono','Menlo','Consolas',monospace; --xs: .7rem; --sm: .8125rem; --s0: .9375rem; --s1: 1.0625rem; --s2: 1.1875rem; --s3: 1.375rem; --s4: 1.75rem; --s5: 2.25rem; --sp1: .1875rem; --sp2: .4375rem; --sp3: .6875rem; --sp4: .875rem; --sp5: 1.125rem; --sp6: 1.375rem; --sp8: 1.75rem; --sp10: 2.25rem; --wide: 62rem; --pad: clamp(.875rem,3.5vw,2.5rem); --r: 8px; --rs: 5px; --sh: 0 1px 2px rgba(15,23,42,.04),0 1px 1px rgba(15,23,42,.02); --dur: 180ms; --ease: cubic-bezier(.33,1,.68,1); }} @media(prefers-reduced-motion:reduce){{:root{{--dur:0ms}}}} *,*::before,*::after{{margin:0;padding:0;box-sizing:border-box}} body{{font-family:var(--font-b);font-size:var(--s0);line-height:1.65;color:var(--text);background:var(--bg);-webkit-font-smoothing:antialiased;text-rendering:optimizeLegibility}} .wrap{{max-width:var(--wide);margin:0 auto;padding:0 var(--pad)}} /* Navigation - pill style */ nav.top{{position:sticky;top:0;z-index:100;background:rgba(248,250,251,.94);backdrop-filter:blur(10px);-webkit-backdrop-filter:blur(10px);border-bottom:1px solid var(--line);padding:0 var(--pad)}} nav.top .inner{{max-width:var(--wide);margin:0 auto;display:flex;gap:var(--sp1);overflow-x:auto;scrollbar-width:none;padding:var(--sp2) 0}} nav.top .inner::-webkit-scrollbar{{display:none}} .nl{{font-family:var(--font-d);font-size:var(--xs);font-weight:500;color:var(--muted);text-decoration:none;white-space:nowrap;padding:4px 10px;border-radius:var(--r);transition:all var(--dur) var(--ease)}} .nl:hover{{color:var(--text);background:var(--surface-2)}} .nl.active{{color:var(--accent);background:var(--accent-light)}} /* Hero - compact, no center */ header.hero{{padding:var(--sp8) 0 var(--sp6)}} header.hero h1{{font-family:var(--font-d);font-weight:700;font-size:clamp(1.5rem,3.5vw,var(--s4));letter-spacing:-.02em;color:var(--text);margin-bottom:var(--sp1);line-height:1.2}} header.hero .sub{{font-size:var(--sm);color:var(--muted)}} /* Sections */ section.report-section{{padding:var(--sp6) 0;border-top:1px solid var(--line)}} .sec-title{{font-family:var(--font-d);font-weight:600;font-size:var(--s2);letter-spacing:-.01em;margin-bottom:var(--sp5);color:var(--text);display:flex;align-items:center;gap:var(--sp3)}} .sec-title::before{{content:'';width:3px;height:1.1em;background:var(--accent);border-radius:2px;flex-shrink:0}} /* Cards */ .card{{background:var(--surface);border:1px solid var(--line);border-radius:var(--r);padding:var(--sp5);box-shadow:var(--sh)}} .card+.card{{margin-top:var(--sp3)}} .cl{{font-family:var(--font-d);font-size:11px;font-weight:600;color:var(--muted);text-transform:uppercase;letter-spacing:.06em;margin-bottom:var(--sp3)}} .src{{font-size:10px;color:var(--muted);margin-top:var(--sp4);line-height:1.6;font-family:var(--font-m);opacity:.8}} .src::before{{content:'说明: ';font-weight:600}} /* Grids */ .g2{{display:grid;grid-template-columns:1fr 1fr;gap:var(--sp3)}} .g3{{display:grid;grid-template-columns:repeat(3,1fr);gap:var(--sp3)}} @media(max-width:640px){{.g2,.g3{{grid-template-columns:1fr}}}} /* Numeric */ .num{{font-family:var(--font-m);font-weight:700;font-variant-numeric:tabular-nums}} .ptag{{display:inline-flex;align-items:center;gap:var(--sp2);font-size:var(--sm);font-weight:500}} .pd{{width:7px;height:7px;border-radius:50%;flex-shrink:0}} /* Heatmap table */ .heat{{width:100%;border-collapse:separate;border-spacing:0}} .heat th{{padding:var(--sp2) var(--sp3);font-size:var(--xs);font-weight:600;color:var(--muted);text-align:center;border-bottom:1px solid var(--line);font-family:var(--font-d)}} .heat td{{padding:6px var(--sp3);font-size:var(--sm);text-align:center;border-bottom:1px solid var(--surface-2)}} .heat-y{{background:var(--accent-light);color:var(--accent);font-family:var(--font-m);font-weight:700;border-radius:var(--rs)}} .heat-n{{color:var(--muted);opacity:.4;background:var(--surface-2);font-weight:500}} .heat-f{{color:var(--miss);font-size:var(--xs);background:var(--miss-light);border-radius:var(--rs)}} /* Data table */ .tbl{{width:100%;border-collapse:separate;border-spacing:0}} .tbl th{{padding:var(--sp2) var(--sp3);text-align:left;font-size:11px;font-weight:600;color:var(--muted);letter-spacing:.03em;border-bottom:1px solid var(--line);font-family:var(--font-d);text-transform:uppercase}} .tbl td{{padding:var(--sp2) var(--sp3);font-size:var(--sm);border-bottom:1px solid var(--surface-2);vertical-align:middle}} .tbl tr:hover td{{background:var(--surface-2)}} .tbl .hl td{{background:var(--accent-light);font-weight:600}} .tbl .hl:hover td{{background:#ccfbf1}} /* Badges */ .badge{{display:inline-block;padding:2px 8px;border-radius:var(--rs);font-size:10px;font-weight:600;line-height:1.6}} .b-h{{background:var(--hit-light);color:var(--hit)}} .b-m{{background:var(--miss-light);color:var(--miss)}} .b-w{{background:var(--warn-light);color:var(--warn)}} .b-a{{background:var(--accent-light);color:var(--accent)}} .b-g{{background:var(--surface-2);color:var(--muted)}} /* Insights */ .insight{{padding:var(--sp3) var(--sp4);border-radius:var(--r);margin-bottom:var(--sp2);font-size:var(--sm);line-height:1.7;border-left:3px solid}} .ins-h{{background:var(--hit-light);border-color:var(--hit)}} .ins-m{{background:var(--miss-light);border-color:var(--miss)}} .ins-n{{background:var(--surface-2);border-color:var(--muted)}} /* Bar charts */ .bar-row{{display:flex;align-items:center;gap:var(--sp3);margin-bottom:var(--sp2)}} .bar-name{{width:160px;font-size:var(--sm);color:var(--text);overflow:hidden;text-overflow:ellipsis;white-space:nowrap;flex-shrink:0}} .bar-track{{flex:1;background:var(--surface-2);border-radius:var(--rs);height:18px;overflow:hidden}} .bar-fill{{height:100%;border-radius:var(--rs);display:flex;align-items:center;padding:0 var(--sp2);font-family:var(--font-m);font-size:var(--xs);font-weight:700;color:#fff;min-width:22px;font-variant-numeric:tabular-nums}} /* Folds */ .fold{{cursor:pointer;user-select:none}} .fold::before{{content:'';display:inline-block;width:0;height:0;border-left:5px solid var(--muted);border-top:4px solid transparent;border-bottom:4px solid transparent;margin-right:var(--sp2);vertical-align:middle;transition:transform var(--dur) var(--ease)}} .fold.open::before{{transform:rotate(90deg)}} .fold-body{{display:none;margin-top:var(--sp3);padding:var(--sp4);background:var(--surface-2);border-radius:var(--r);font-size:var(--sm);line-height:1.7;border:1px solid var(--line)}} .fold.open+.fold-body{{display:block}} /* Legend */ .legend{{display:flex;flex-direction:column;gap:var(--sp2)}} .legend-item{{display:flex;align-items:center;gap:var(--sp3);font-size:var(--sm)}} .legend-dot{{width:10px;height:10px;border-radius:3px;flex-shrink:0}} .legend-val{{font-family:var(--font-m);font-weight:700;margin-left:auto;font-variant-numeric:tabular-nums}} /* Footer */ footer.rf{{padding:var(--sp6) 0;border-top:1px solid var(--line);text-align:center;font-size:var(--xs);color:var(--muted)}} /* Voice tags */ .vtag{{display:inline-flex;align-items:center;gap:var(--sp1);padding:2px 8px;border-radius:var(--rs);font-size:var(--xs);font-weight:500;margin:0 var(--sp1) var(--sp2) 0;line-height:1.5}} .vtag-pos{{background:var(--hit-light);color:var(--hit);border:1px solid rgba(5,150,105,.12)}} .vtag-neg{{background:var(--miss-light);color:var(--miss);border:1px solid rgba(220,38,38,.12)}} .vtag .vp{{font-size:10px;opacity:.5;margin-left:2px}} /* Position indicator */ .pos-bar{{display:flex;align-items:center;gap:var(--sp3);padding:var(--sp3) var(--sp4);border-radius:var(--r);border:1px solid var(--line);background:var(--surface)}} .pos-rank{{font-family:var(--font-m);font-weight:800;font-size:var(--s4);line-height:1}} .pos-label{{font-family:var(--font-d);font-weight:600;font-size:var(--sm);color:var(--text)}} .pos-desc{{font-size:var(--xs);color:var(--muted)}} .pos-leading{{color:var(--hit)}} .pos-upper{{color:var(--accent)}} .pos-mid{{color:var(--warn)}} .pos-trailing{{color:var(--miss)}} /* Source analysis */ .sa-grid{{display:grid;grid-template-columns:1fr 1fr;gap:var(--sp3);align-items:start}} @media(max-width:900px){{.sa-grid{{grid-template-columns:1fr;}}}} .sa-list{{border:1px solid var(--line);border-radius:var(--r);background:var(--surface);overflow:hidden}} .sa-list-hd{{padding:var(--sp2) var(--sp3);font-family:var(--font-d);font-size:11px;font-weight:600;color:var(--muted);border-bottom:1px solid var(--line);letter-spacing:.02em;text-transform:uppercase}} .sa-row{{display:flex;align-items:center;gap:var(--sp2);padding:5px var(--sp3);border-bottom:1px solid var(--surface-2);font-size:var(--sm);transition:background var(--dur) var(--ease)}} .sa-row:hover{{background:var(--surface-2)}} .sa-row:last-child{{border-bottom:none}} .sa-rank{{font-family:var(--font-m);font-weight:700;font-size:var(--xs);color:var(--muted);width:18px;text-align:center;flex-shrink:0}} .sa-name{{flex:1;overflow:hidden;text-overflow:ellipsis;white-space:nowrap}} .sa-cnt{{font-family:var(--font-m);font-weight:700;font-size:var(--xs);color:var(--accent);width:24px;text-align:right;flex-shrink:0}} .sa-cat{{font-size:10px;padding:1px 5px;border-radius:var(--rs);font-weight:500;flex-shrink:0}} .sa-cat-brand{{background:var(--hit-light);color:var(--hit)}} .sa-cat-auto{{background:#eff6ff;color:#2563eb}} .sa-cat-media{{background:#f0fdf4;color:#059669}} .sa-cat-other{{background:var(--surface-2);color:var(--muted)}} .sa-article-link{{text-decoration:none;color:var(--text);display:block}} .sa-article-link:hover .sa-name{{color:var(--accent)}} .sa-cat-bar{{display:flex;gap:var(--sp5);justify-content:center;flex-wrap:wrap;margin-top:var(--sp3);padding:var(--sp3)}} .sa-cat-item{{display:flex;align-items:center;gap:var(--sp2);font-size:var(--sm)}} .sa-cat-dot{{width:8px;height:8px;border-radius:2px;flex-shrink:0}} .sa-cat-pct{{font-family:var(--font-m);font-weight:700;font-variant-numeric:tabular-nums}} /* Summary block */ .summary{{background:var(--surface);border:1px solid var(--line);border-radius:var(--r);overflow:hidden;margin-bottom:var(--sp5)}} .summary-hd{{padding:var(--sp4) var(--sp5);background:var(--accent);color:#fff;font-family:var(--font-d);font-weight:600;font-size:var(--sm);letter-spacing:.02em}} .summary-bd{{padding:var(--sp5);font-size:var(--sm);line-height:1.85;color:var(--text)}} .summary-bd p{{margin-bottom:var(--sp3)}} .summary-bd p:last-child{{margin-bottom:0}} .summary-bd strong{{color:var(--text);font-weight:600}} </style> </head> <body> <nav class="top" aria-label="报告导航"><div class="inner">{nav_html}</div></nav> <main class="wrap"> <header class="hero"> <h1>"{_esc(brand)}" GEO 品牌感知报告</h1> <div class="sub">{platform_labels} · {now}</div> </header> {summary_html} {sections_html} <footer class="rf"> <p>GEO 品牌感知分析 · {platform_labels2} · 共 {total_completed} 条有效回答</p> <p style="margin-top:var(--sp1);opacity:.7">提及率=品牌被提及次数/有效回答数 · 排名与情感=AI逐条分析 · 综合得分=提及*40%+排名*30%+情感*30%</p> </footer> </main> <script> document.querySelectorAll('.fold').forEach(el=>{{el.addEventListener('click',()=>el.classList.toggle('open'))}}); document.querySelectorAll('.nl').forEach(a=>{{a.addEventListener('click',e=>{{e.preventDefault();const t=document.querySelector(a.getAttribute('href'));if(t)t.scrollIntoView({{behavior:'smooth',block:'start'}})}})}}); </script> </body> </html>""" def _render_fingerprint(analysis, platforms, total_completed, total_mentioned, position): geo_score = analysis.get("geo_score", {}) mention_rate = analysis.get("metrics", {}).get("brand_mention_rate", {}) sentiment_dist = analysis.get("sentiment_distribution", {}).get("overall", {}) brand = analysis.get("brand", "") avg_rank = analysis.get("brand_avg_rank", {}).get("overall") geo_o = geo_score.get("overall", 0) mr_pct = _pct(total_mentioned, total_completed) total_sent = sum(sentiment_dist.values()) neg_pct = _pct(sentiment_dist.get("negative", 0), total_sent) if total_sent else 0 pos_pct = 100 - neg_pct rank_str = f"{avg_rank:.1f}" if avg_rank else "N/A" # 平台卡片: 紧凑布局 score_cards = "" for p in platforms: s = geo_score.get(p, 0) label = PLATFORM_LABELS.get(p, p) color = PLATFORM_COLORS.get(p, "var(--accent)") rate = mention_rate.get(p, 0) pct = int(rate * 100) gt, gc, gbg = _score_grade(s if isinstance(s, (int, float)) else None) p_done = sum(1 for a in analysis.get("per_answer", []) if a.get("platform") == p and a.get("status") == "completed") p_ment = sum(1 for a in analysis.get("per_answer", []) if a.get("platform") == p and a.get("status") == "completed" and a.get("brand_mentioned")) score_cards += f""" <div style="padding:var(--sp3) var(--sp4);border:1px solid var(--line);border-radius:var(--r);background:var(--surface)"> <div style="display:flex;justify-content:space-between;align-items:center;margin-bottom:var(--sp2)"> <span class="ptag"><span class="pd" style="background:{color}"></span>{label}</span> <span class="badge" style="background:{gbg};color:{gc}">{gt}</span> </div> <div class="num" style="font-size:var(--s4);color:{gc};margin-bottom:var(--sp2)">{s if isinstance(s,(int,float)) else 'N/A'}</div> <div style="display:flex;gap:var(--sp4);font-size:var(--xs);color:var(--muted)"> <span>提及 {pct}%</span> <span>{p_ment}/{p_done} 条</span> </div> </div>""" pos_html = "" if position: pr = position["avg_rank"] pt = position["total_brands"] pc = position["css_class"] pl = position["label"] pos_html = f""" <div class="pos-bar" style="margin-top:0;margin-bottom:var(--sp3)"> <div class="pos-rank {pc}">#{round(pr)}</div> <div> <div class="pos-label {pc}">{pl}</div> <div class="pos-desc">平均排名 {pr:.1f} · 约 {pt} 个品牌参与</div> </div> </div>""" return f""" <div class="g2"> <div class="card"> {pos_html} <div style="display:grid;grid-template-columns:repeat(2,1fr);gap:var(--sp4) var(--sp5);margin-bottom:var(--sp3)"> <div><div style="font-size:11px;color:var(--muted);margin-bottom:2px;text-transform:uppercase;letter-spacing:.03em">GEO 综合</div><div class="num" style="font-size:var(--s4);color:var(--accent)">{geo_o}</div></div> <div><div style="font-size:11px;color:var(--muted);margin-bottom:2px;text-transform:uppercase;letter-spacing:.03em">提及率</div><div class="num" style="font-size:var(--s4);color:var(--hit)">{mr_pct}%</div></div> <div><div style="font-size:11px;color:var(--muted);margin-bottom:2px;text-transform:uppercase;letter-spacing:.03em">平均排名</div><div class="num" style="font-size:var(--s4);color:var(--warn)">{rank_str}</div></div> <div><div style="font-size:11px;color:var(--muted);margin-bottom:2px;text-transform:uppercase;letter-spacing:.03em">正面率</div><div class="num" style="font-size:var(--s4);color:var(--hit)">{pos_pct}%</div></div> </div> <div class="src">综合得分 = 提及率*40% + 排名得分*30% + 情感得分*30%</div> </div> <div style="display:grid;grid-template-columns:1fr;gap:var(--sp2)">{score_cards}</div> </div>""" def _render_sentiment_merged(analysis, platforms, has_ai, total_completed, pos_tags, neg_tags, brand): """情感分析: 合并原品牌声音+情感分析, 包含分布/平台对比/正面负面信号/非中性详情""" if not has_ai: return '<div class="card" style="color:var(--muted)">暂无 AI 情感分析数据</div>' dist = analysis.get("sentiment_distribution", {}) overall = dist.get("overall", {"positive": 0, "neutral": 0, "negative": 0}) total = sum(overall.values()) neg_count = overall.get("negative", 0) neg_pct = _pct(neg_count, total) if total else 0 pos_pct = 100 - neg_pct neu_pct = _pct(overall.get("neutral", 0), total) if total else 0 # 各平台情感对比 platform_rows = "" for p in platforms: d = dist.get(p, {"positive": 0, "neutral": 0, "negative": 0}) t = sum(d.values()) or 1 pp = int(d["positive"]/t*100); np_ = int(d["neutral"]/t*100); ng = 100-pp-np_ label = PLATFORM_LABELS.get(p, p); color = PLATFORM_COLORS.get(p, "var(--accent)") platform_rows += f""" <div style="display:flex;align-items:center;gap:var(--sp3);padding:var(--sp2) 0;border-bottom:1px solid var(--surface-2)"> <span class="ptag" style="width:70px"><span class="pd" style="background:{color}"></span>{label}</span> <div style="flex:1;display:flex;height:14px;border-radius:var(--rs);overflow:hidden;gap:1px"> <div style="width:{pp}%;background:var(--hit)"></div> <div style="width:{np_}%;background:#d4d4d8"></div> <div style="width:{ng}%;background:var(--miss)"></div> </div> <span class="num" style="font-size:var(--xs);color:var(--muted);width:80px;text-align:right">{d['positive']}正 {d['neutral']}中 {d['negative']}负</span> </div>""" # 正面信号 pos_html = "" for item in pos_tags: p = PLATFORM_LABELS.get(item["platform"], item["platform"]) is_tag = item.get("tag", False) text = _esc(item["text"]) if is_tag: pos_html += f'<span class="vtag vtag-pos">{text}</span>' else: pos_html += f'<span class="vtag vtag-pos">{text} <span class="vp">{p}</span></span>' if not pos_html: pos_html = '<span style="color:var(--muted);font-size:var(--sm)">未提取到正面信号</span>' # 负面信号 neg_html = "" for item in neg_tags: p = PLATFORM_LABELS.get(item["platform"], item["platform"]) text = _esc(item["text"]) neg_html += f'<span class="vtag vtag-neg">{text} <span class="vp">{p}</span></span>' if not neg_html: neg_html = '<span style="color:var(--muted);font-size:var(--sm)">未发现负面信号</span>' # 非中性回答详情 per_answer = analysis.get("per_answer", []) pos_insights = "" neg_insights = "" for ans in per_answer: if ans.get("status") != "completed": continue s = ans.get("sentiment", "neutral") if s == "neutral": continue q = _esc(ans.get("question", "")) pl = PLATFORM_LABELS.get(ans.get("platform", ""), ans.get("platform", "")) reason = _esc(ans.get("sentiment_reason", "")) bc = {"positive": "b-h", "negative": "b-m"}.get(s, "b-g") sl = SENTIMENT_LABELS.get(s, s) card = f'<div class="insight {"ins-h" if s=="positive" else "ins-m"}"><span class="badge {bc}">{sl}</span> <strong>{pl}</strong> / {q}<br><span style="font-size:var(--xs);color:var(--muted)">{reason}</span></div>' if s == "positive": pos_insights += card else: neg_insights += card if not pos_insights: pos_insights = '<div style="color:var(--muted);font-size:var(--sm);padding:var(--sp4)">无正面回答详情</div>' if not neg_insights: neg_insights = '<div style="color:var(--muted);font-size:var(--sm);padding:var(--sp4)">无负面回答详情</div>' return f""" <div class="g2"> <div class="card"> <div class="cl">整体情感分布</div> <div class="legend" style="margin-bottom:var(--sp3)"> <div class="legend-item"><div class="legend-dot" style="background:var(--hit)"></div>正面<span class="legend-val">{overall.get('positive',0)}</span></div> <div class="legend-item"><div class="legend-dot" style="background:#d4d4d8"></div>中性<span class="legend-val">{overall.get('neutral',0)}</span></div> <div class="legend-item"><div class="legend-dot" style="background:var(--miss)"></div>负面<span class="legend-val">{neg_count}</span></div> </div> <div class="src">AI 对 {total} 条回答逐条情感分析(正面=推荐倾向 / 中性=客观描述 / 负面=批评态度)</div> </div> <div class="card"> <div class="cl">各平台情感对比</div> {platform_rows} <div class="src">各平台独立统计, 正面=主动推荐 / 中性=客观提及 / 负面=指出不足</div> </div> </div> <div class="g2" style="margin-top:var(--sp3)"> <div class="card"> <div class="cl">正面信号 ({len(pos_tags)} 项)</div> <div style="display:flex;flex-wrap:wrap;margin-bottom:var(--sp3)">{pos_html}</div> <div class="src">从 AI 回答中提取, 匹配关键词库({len(POS_PATTERNS)}个正面词)</div> {pos_insights} </div> <div class="card"> <div class="cl">负面信号 ({len(neg_tags)} 项)</div> <div style="display:flex;flex-wrap:wrap;margin-bottom:var(--sp3)">{neg_html}</div> <div class="src">已排除“无XX”类假阳性(如“无充电等待”不算负面)</div> {neg_insights} </div> </div>""" def _render_mention_matrix(analysis, platforms, queries, total_completed): matrix = analysis.get("mention_matrix", []) per_answer = analysis.get("per_answer", []) header_cells = "".join( f'<th><span class="ptag"><span class="pd" style="background:{PLATFORM_COLORS.get(p,"var(--accent)")}"></span>{PLATFORM_LABELS.get(p,p)}</span></th>' for p in platforms ) rows = "" for i, row in enumerate(matrix): q = queries[i] if i < len(queries) else f"Q{i+1}" q_short = q if len(q) <= 24 else q[:24] + "..." cells = "" for j, cell in enumerate(row): mentioned = cell.get("mentioned", False) count = cell.get("mention_count", 0) if mentioned: cells += f'<td class="heat-y" title="出现{count}次">{count}</td>' else: p = platforms[j] if j < len(platforms) else "" has_failed = any(a.get("query_index") == i and a.get("platform") == p and a.get("status") != "completed" for a in per_answer) cells += f'<td class="{"heat-f" if has_failed else "heat-n"}" title="{"查询失败" if has_failed else "未提及"}">{"✕" if has_failed else "—"}</td>' rows += f'<tr><td style="text-align:left;font-size:var(--sm);max-width:280px;overflow:hidden;text-overflow:ellipsis;white-space:nowrap" title="{_esc(q)}">{_esc(q_short)}</td>{cells}</tr>' return f""" <div class="card"> <table class="heat"> <thead><tr><th style="text-align:left;width:50%">问题</th>{header_cells}</tr></thead> <tbody>{rows}</tbody> </table> <div style="display:flex;gap:var(--sp4);margin-top:var(--sp3);flex-wrap:wrap;font-size:var(--xs);color:var(--muted)"> <span><span style="display:inline-block;width:8px;height:8px;background:var(--accent-light);border-radius:2px;vertical-align:middle;margin-right:var(--sp1)"></span>已提及</span> <span><span style="display:inline-block;width:8px;height:8px;background:var(--surface-2);border-radius:2px;vertical-align:middle;margin-right:var(--sp1)"></span>未提及 (—)</span> <span><span style="display:inline-block;width:8px;height:8px;background:var(--miss-light);border-radius:2px;vertical-align:middle;margin-right:var(--sp1)"></span>查询失败 (✕)</span> </div> <div class="src">共 {total_completed} 条回答, 通过关键词匹配(品牌名+别名)检测是否提及</div> </div>""" def _render_sentiment(analysis, platforms, has_ai, total_completed): pass # merged into _render_sentiment_merged CAT_COLORS = { "品牌官网": "#059669", "汽车垂直": "#2563eb", "社媒资讯": "#0f766e", "其他网站": "#a1a1aa", } CAT_CSS = { "品牌官网": "sa-cat-brand", "汽车垂直": "sa-cat-auto", "社媒资讯": "sa-cat-media", "其他网站": "sa-cat-other", } def _render_sources_analysis(analysis, search_results, platforms, total_completed, sources_data): """信源分析: 域名表(左) + 环形图(中) + 文章排名(右) + 分类占比(底)""" if not sources_data: return '<div class="card" style="color:var(--muted)">暂无信源数据</div>' top_domains = sources_data["top_domains"] top_articles = sources_data["top_articles"] category_freq = sources_data["category_freq"] total_citations = sources_data["total_citations"] brand = analysis.get("brand", "") aliases = analysis.get("aliases", []) brand_keywords = [k.lower() for k in [brand] + aliases if k] # 左侧: 域名聚合排序 TOP10 domain_rows = "" for i, (domain, count) in enumerate(top_domains): cat = _classify_domain(domain) css_cat = CAT_CSS.get(cat, "sa-cat-other") is_brand = any(kw in domain.lower() for kw in brand_keywords) name_style = "color:var(--hit);font-weight:600" if is_brand else "" mark = " *" if is_brand else "" domain_url = f"https://{domain}" domain_rows += f'<div class="sa-row"><span class="sa-rank">{i+1}</span><a class="sa-name sa-article-link" href="{domain_url}" target="_blank" rel="noopener" style="{name_style}" title="{_esc(domain)}">{_esc(domain)}{mark}</a><span class="sa-cnt">{count}</span><span class="sa-cat {css_cat}">{cat}</span></div>' # 右侧: 文章频次排名 TOP10 article_rows = "" for i, (meta, count) in enumerate(top_articles): title = _esc(meta["title"])[:30] url = _esc(meta["url"]) article_rows += f'<a class="sa-row sa-article-link" href="{url}" target="_blank" rel="noopener"><span class="sa-rank">{i+1}</span><span class="sa-name" title="{_esc(meta["title"])}">{title}</span><span class="sa-cnt">{count}</span></a>' # 底部: 分类占比 cat_bar = "" sorted_cats = sorted(category_freq.items(), key=lambda x: -x[1]) for cat, count in sorted_cats: pct = count / total_citations * 100 if total_citations else 0 col = CAT_COLORS.get(cat, "#a1a1aa") cat_bar += f'<div class="sa-cat-item"><span class="sa-cat-dot" style="background:{col}"></span><span class="sa-cat-pct">{pct:.1f}%</span> {cat}</div>' brand_note = "" if any(any(kw in d.lower() for kw in brand_keywords) for d, _ in top_domains): brand_note = f'<span style="color:var(--hit)">* {brand} 品牌官网</span>' elif brand: brand_note = f'<span style="color:var(--miss)">"{_esc(brand)}" 品牌官网未进入 TOP10</span>' return f""" <div class="sa-grid"> <div class="sa-list"> <div class="sa-list-hd">信源引用域名聚合排序 (TOP10)</div> {domain_rows} </div> <div class="sa-list"> <div class="sa-list-hd">信源引用文章频次排名 (TOP10)</div> {article_rows} </div> </div> <div class="sa-cat-bar"> {cat_bar} </div> <div style="display:flex;gap:var(--sp4);margin-top:var(--sp1);font-size:var(--xs);color:var(--muted);justify-content:center"> {brand_note} </div> <div class="src">总引用 {total_citations} 次 · 来自 {total_completed} 条有效回答 · 域名按聚合频次降序 · 文章按被引用次数降序</div>""" def _render_competitors(analysis, platforms, total_completed): comparison = analysis.get("competitor_comparison", []) brand = analysis.get("brand", "") mention_rate = analysis.get("metrics", {}).get("brand_mention_rate", {}) brand_avg_rank = analysis.get("brand_avg_rank", {}) brand_sentiment = analysis.get("sentiment_distribution", {}).get("overall", {}) if not comparison and not brand: return '<div class="card" style="color:var(--muted)">暂无竞品数据</div>' brand_total_s = sum(brand_sentiment.values()) if brand_sentiment else 0 brand_neg_pct = _pct(brand_sentiment.get("negative", 0), brand_total_s) if brand_total_s else 0 brand_pos_pct = 100 - brand_neg_pct brand_rank_str = f"{brand_avg_rank.get('overall',0):.1f}" if brand_avg_rank.get("overall") else "N/A" brand_row = f""" <tr class="hl"> <td><strong>{_esc(brand)}</strong> <span class="badge b-a">本品牌</span></td> <td class="num">{_pct(int(mention_rate.get('overall',0)*100),100)}%</td> <td class="num">{brand_rank_str}</td> <td class="num" style="color:var(--hit)">{brand_pos_pct}%</td> </tr>""" sorted_comps = sorted(comparison, key=lambda c: c.get("mention_rate_overall",0), reverse=True) comp_rows = "" comp_count = 0 skipped_count = 0 for comp in sorted_comps: name = comp.get("name", "") if name == brand: continue mr = comp.get("mention_rate_overall",0) # 只展示在回答中实际被提及的竞品 if mr <= 0: continue if comp_count >= 10: skipped_count += 1 continue comp_count += 1 rank = comp.get("avg_rank") rs = f"{rank:.1f}" if rank else "N/A" sd = comp.get("sentiment_dist", {}) sd_total = sum(sd.values()) if sd else 0 comp_neg_pct = _pct(sd.get("negative", 0), sd_total) if sd_total else 0 pp = 100 - comp_neg_pct detail = "" if sd: detail = f'<div class="fold-body" style="font-size:var(--xs)"><strong>各平台提及:</strong> 豆包 {_pct(int(comp.get("mention_rate_doubao",0)*100),100)}% · Kimi {_pct(int(comp.get("mention_rate_kimi",0)*100),100)}% · DeepSeek {_pct(int(comp.get("mention_rate_deepseek",0)*100),100)}%<br><strong>情感分布:</strong> 正面{sd.get("positive",0)} / 中性{sd.get("neutral",0)} / 负面{sd.get("negative",0)}</div>' comp_rows += f'<tr><td><span class="fold" style="font-size:var(--sm)">{_esc(name)}</span>{detail}</td><td class="num">{_pct(int(mr*100),100)}%</td><td class="num">{rs}</td><td class="num" style="color:var(--hit)">{pp}%</td></tr>' more_text = f'<div style="margin-top:var(--sp2);font-size:var(--xs);color:var(--muted);text-align:center">还有 {skipped_count} 个竞品未展示(提及率较低)</div>' if skipped_count > 0 else '' return f""" <div class="card"> <table class="tbl"> <thead><tr><th style="width:35%">品牌 (点击展开详情)</th><th>提及率</th><th>平均排名</th><th>正面率</th></tr></thead> <tbody>{brand_row}{comp_rows}</tbody> </table> {more_text} <div class="src">提及率=关键词匹配+AI提及列表推断排名 · 正面率=1-负面率 · 排名=在推荐列表中的位置</div> </div>""" def _render_archive(analysis, search_results, platforms, queries): brand = analysis.get("brand", "") if not search_results: return '<div class="card" style="color:var(--muted)">暂无原始数据</div>' results = search_results.get("results", []) if not results: return '<div class="card" style="color:var(--muted)">暂无原始数据</div>' by_question = {} for r in results: qi = r.get("query_index", 0) by_question.setdefault(qi, []).append(r) per_answer_map = {} for a in analysis.get("per_answer", []): key = (a.get("query_index"), a.get("platform")) per_answer_map[key] = a panels = "" for qi in sorted(by_question.keys()): question = queries[qi] if qi < len(queries) else f"Q{qi+1}" answers = by_question[qi] platform_items = "" for ans in answers: p = ans.get("platform", "") label = PLATFORM_LABELS.get(p, p) color = PLATFORM_COLORS.get(p, "var(--accent)") status = ans.get("status", "unknown") content = ans.get("content", "") pa = per_answer_map.get((qi, p), {}) s = pa.get("sentiment", "") sb = "" if s and status == "completed": sl = SENTIMENT_LABELS.get(s, s) bc = {"positive": "b-h", "neutral": "b-g", "negative": "b-m"}.get(s, "b-g") sb = f' <span class="badge {bc}">{sl}</span>' cd = f"[{status}]" if status != "completed" else content escaped = _esc(cd) # 品牌词高亮 brand_kw = [brand] + [a for a in analysis.get("aliases", []) if a] if brand else [] for kw in brand_kw: ekw = _esc(kw) if ekw and ekw in escaped: escaped = escaped.replace(ekw, f'<mark style="background:#fef08a;color:inherit;padding:0 2px;border-radius:2px">{ekw}</mark>') platform_items += f'<div style="margin-bottom:var(--sp3)"><div class="fold" style="font-size:var(--sm);font-weight:500"><span class="ptag"><span class="pd" style="background:{color}"></span>{label}</span>{sb}</div><div class="fold-body">{escaped}</div></div>' nd = '<div style="color:var(--muted)">暂无数据</div>' panels += f'<div class="card" style="margin-bottom:var(--sp3)"><div class="fold" style="font-size:var(--s1);font-weight:600">Q{qi+1}. {_esc(_clean(question))}</div><div class="fold-body">{platform_items if platform_items else nd}</div></div>' return panels -
__init__.py 0 B
-
-
geo_analyze.py 12.4 KB
#!/usr/bin/env python3 """ GEO 分析编排器 — 读取搜索结果,执行确定性分析,输出分析数据 用法: # Step 1: 仅确定性分析(Agent 执行 AI 分析前) python3 geo_analyze.py --brand "品牌名" --search-results output/search_results.json # 可选参数 --aliases '["别名1","别名2"]' 品牌别名 --competitors '["竞品A","竞品B"]' 已知竞品 --ai-analysis output/ai_analysis.json AI 分析结果(有则合并输出完整报告数据) 输出: output/deterministic.json 确定性分析结果 output/ai_analysis_template.json AI 分析模板(供 Agent 填充) output/analysis_result.json 合并后的完整分析(当提供 --ai-analysis 时) """ import sys import os import json import argparse sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "lib")) from analyzer import ( detect_all_mentions, extract_domains_from_sources, aggregate_domains, compute_mention_rate, build_mention_matrix, compute_per_answer_analysis, compute_aggregate_metrics, build_ai_analysis_template, build_analysis_prompt, merge_analysis, compute_sentiment_distribution, compute_brand_avg_rank, compute_competitor_metrics, compute_geo_score, ) from platforms import PLATFORMS def main(): parser = argparse.ArgumentParser(description="GEO 分析编排器") parser.add_argument("--brand", required=True, help="品牌名称") parser.add_argument("--aliases", default="[]", help="品牌别名 JSON 列表") parser.add_argument("--competitors", default="[]", help="已知竞品 JSON 列表") parser.add_argument( "--search-results", default=os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "output", "search_results.json"), help="搜索结果 JSON 路径", ) parser.add_argument( "--ai-analysis", default=None, help="AI 分析结果 JSON 路径(有则合并输出完整分析)", ) parser.add_argument( "--output-dir", default=None, help="输出目录(默认 output/)", ) args = parser.parse_args() # ── 解析参数 ────────────────────────────────────────── brand = args.brand aliases = json.loads(args.aliases) if args.aliases else [] competitors = json.loads(args.competitors) if args.competitors else [] # 品牌关键词 = 品牌名 + 别名 brand_keywords = [brand] + aliases # 竞品关键词映射(默认竞品名就是关键词) competitor_map = {} for comp in competitors: competitor_map[comp] = [comp] output_dir = args.output_dir or os.path.join( os.path.dirname(os.path.abspath(__file__)), "..", "output" ) os.makedirs(output_dir, exist_ok=True) # ── 读取搜索结果 ────────────────────────────────────── search_path = args.search_results if not os.path.exists(search_path): print(json.dumps({"error": f"搜索结果文件不存在: {search_path}"}, ensure_ascii=False)) sys.exit(1) with open(search_path, "r", encoding="utf-8") as f: search_data = json.load(f) results = search_data.get("results", []) queries = search_data.get("queries", []) platforms = search_data.get("platforms", ["doubao", "kimi", "deepseek"]) print(f"品牌: {brand}", file=sys.stderr) print(f"别名: {aliases}", file=sys.stderr) print(f"竞品: {competitors}", file=sys.stderr) print(f"搜索结果: {len(results)} 条,平台: {platforms}", file=sys.stderr) # ── Phase 1: 确定性分析 ──────────────────────────────── print("\n执行确定性分析...", file=sys.stderr) per_answer = compute_per_answer_analysis(results, brand_keywords, competitor_map) metrics = compute_aggregate_metrics(per_answer, queries, platforms, brand, competitors) mention_matrix = build_mention_matrix(results, queries, platforms, brand_keywords) deterministic = { "brand": brand, "aliases": aliases, "competitors": competitors, "platforms": platforms, "queries": queries, "per_answer": per_answer, "metrics": metrics, "mention_matrix": mention_matrix, } det_path = os.path.join(output_dir, "deterministic.json") with open(det_path, "w", encoding="utf-8") as f: json.dump(deterministic, f, ensure_ascii=False, indent=2) print(f"确定性分析已保存: {det_path}", file=sys.stderr) # ── Phase 2: 生成 AI 分析模板 ────────────────────────── ai_template = build_ai_analysis_template(results, brand, competitors) template_path = os.path.join(output_dir, "ai_analysis_template.json") with open(template_path, "w", encoding="utf-8") as f: json.dump(ai_template, f, ensure_ascii=False, indent=2) print(f"AI 分析模板已保存: {template_path}", file=sys.stderr) # 输出 AI 分析提示 print(f"\n{'='*60}", file=sys.stderr) print(f"AI 分析任务: 请对 {len(ai_template)} 份回答执行 AI 分析", file=sys.stderr) print(f"模板文件: {template_path}", file=sys.stderr) print(f"每份回答需填充字段: brand_rank, brand_context, sentiment,", file=sys.stderr) print(f" sentiment_reason, competitors_mentioned, key_claims", file=sys.stderr) print(f"完成后保存为: {os.path.join(output_dir, 'ai_analysis.json')}", file=sys.stderr) print(f"然后重新运行本脚本添加 --ai-analysis 参数生成完整报告数据", file=sys.stderr) print(f"{'='*60}\n", file=sys.stderr) # ── Phase 3: 合并 AI 分析(如果提供)────────────────── if args.ai_analysis and os.path.exists(args.ai_analysis): print("检测到 AI 分析结果,开始合并...", file=sys.stderr) with open(args.ai_analysis, "r", encoding="utf-8") as f: ai_data = json.load(f) # 合并确定性 + AI 分析 merged = merge_analysis(per_answer, ai_data) # 计算需要 AI 数据的指标 sentiment_dist = compute_sentiment_distribution(merged, platforms) brand_avg_rank = compute_brand_avg_rank(merged, platforms) # 收集所有竞品(用户指定 + AI 发现) all_competitors = set(competitors) for ai in ai_data: for comp in ai.get("competitors_mentioned", []): all_competitors.add(comp) # 计算竞品指标(平均排名 + 情绪分布) competitor_metrics = compute_competitor_metrics(merged, platforms, all_competitors) # 计算 GEO 综合得分 geo_score = {} geo_score["overall"] = compute_geo_score( metrics["brand_mention_rate"]["overall"], brand_avg_rank["overall"], sentiment_dist["overall"], ) for p in platforms: geo_score[p] = compute_geo_score( metrics["brand_mention_rate"].get(p, 0), brand_avg_rank.get(p), sentiment_dist.get(p, {"positive": 0, "neutral": 0, "negative": 0}), ) # 计算竞品提及率(覆盖所有竞品,含 AI 发现的) comp_mention_rates = {} for comp in all_competitors: rates = {"overall": 0.0} for p in platforms: rates[p] = 0.0 total_done = 0 total_hit = 0 p_stats = {p: {"t": 0, "h": 0} for p in platforms} comp_lower = comp.lower() for ans in merged: if ans.get("status") != "completed": continue p = ans.get("platform", "") total_done += 1 p_stats.setdefault(p, {"t": 0, "h": 0}) p_stats[p]["t"] += 1 # 确定性匹配 det_count = ans.get("competitor_mentions", {}).get(comp, 0) # AI 发现匹配(模糊匹配) ai_mentioned = False if det_count == 0: for ai_comp in ans.get("ai_competitors_mentioned", []): if comp_lower == ai_comp.lower() or comp_lower in ai_comp.lower() or ai_comp.lower() in comp_lower: ai_mentioned = True break if det_count > 0 or ai_mentioned: total_hit += 1 p_stats[p]["h"] += 1 rates["overall"] = total_hit / total_done if total_done > 0 else 0.0 for p in platforms: ps = p_stats.get(p, {"t": 0, "h": 0}) rates[p] = ps["h"] / ps["t"] if ps["t"] > 0 else 0.0 comp_mention_rates[comp] = rates # 竞品对比矩阵 competitor_comparison = [] for comp in sorted(all_competitors): comp_rates = comp_mention_rates.get(comp, {"overall": 0}) comp_m = competitor_metrics.get(comp, {}) comp_avg_rank = comp_m.get("avg_rank", {}) comp_sent = comp_m.get("sentiment", {}) comp_sent_overall = comp_sent.get("overall", {"positive": 0, "neutral": 0, "negative": 0}) comp_sent_total = sum(comp_sent_overall.values()) competitor_comparison.append({ "name": comp, "mention_rate_overall": comp_rates.get("overall", 0), "mention_rate_doubao": comp_rates.get("doubao", 0), "mention_rate_kimi": comp_rates.get("kimi", 0), "mention_rate_deepseek": comp_rates.get("deepseek", 0), "avg_rank": comp_avg_rank.get("overall"), "positive_pct": comp_sent_overall["positive"] / comp_sent_total if comp_sent_total > 0 else 0, "negative_pct": comp_sent_overall["negative"] / comp_sent_total if comp_sent_total > 0 else 0, "sentiment_dist": comp_sent_overall, }) analysis_result = { "brand": brand, "aliases": aliases, "competitors": competitors, "discovered_competitors": list(all_competitors - set(competitors)), "all_competitors": sorted(all_competitors), "platforms": platforms, "queries": queries, "per_answer": merged, "metrics": metrics, "mention_matrix": mention_matrix, "sentiment_distribution": sentiment_dist, "brand_avg_rank": brand_avg_rank, "geo_score": geo_score, "competitor_comparison": competitor_comparison, "competitor_metrics": competitor_metrics, } result_path = os.path.join(output_dir, "analysis_result.json") with open(result_path, "w", encoding="utf-8") as f: json.dump(analysis_result, f, ensure_ascii=False, indent=2) print(f"完整分析结果已保存: {result_path}", file=sys.stderr) # stdout 输出摘要 summary = { "brand": brand, "geo_score": geo_score, "mention_rate": metrics["brand_mention_rate"], "sentiment": sentiment_dist, "avg_rank": brand_avg_rank, "total_competitors": len(all_competitors), "output": result_path, } print(json.dumps(summary, ensure_ascii=False)) else: # 无 AI 分析,仅输出确定性分析摘要 summary = { "brand": brand, "deterministic_only": True, "mention_rate": metrics["brand_mention_rate"], "total_completed": metrics["total_completed"], "total_mentioned": metrics["total_mentioned"], "top_domains": list(metrics["domain_stats"].items())[:10], "deterministic_output": det_path, "ai_template_output": template_path, "next_step": "请根据 ai_analysis_template.json 对每份回答执行 AI 分析,保存为 ai_analysis.json 后重新运行", } print(json.dumps(summary, ensure_ascii=False)) if __name__ == "__main__": main() -
geo_report.py 3.2 KB
#!/usr/bin/env python3 """ GEO 报告生成器 — 读取分析结果,生成交互式 HTML 报告 用法: python3 geo_report.py --analysis output/analysis_result.json # 可选: 附带原始搜索结果(用于原始回答存档模块) --search-results output/search_results.json 输出: output/geo_report.html """ import sys import os import json import argparse from datetime import datetime sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "lib")) from report_template import generate_html def main(): parser = argparse.ArgumentParser(description="GEO 报告生成器") parser.add_argument( "--analysis", default=os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "output", "analysis_result.json"), help="分析结果 JSON 路径", ) parser.add_argument( "--search-results", default=os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "output", "search_results.json"), help="搜索结果 JSON 路径(可选,用于原始回答存档)", ) parser.add_argument( "--output", default=None, help="输出 HTML 路径(默认 output/geo_report.html)", ) args = parser.parse_args() # ── 读取分析结果 ────────────────────────────────────── if not os.path.exists(args.analysis): print(json.dumps({"error": f"分析结果文件不存在: {args.analysis}"}, ensure_ascii=False)) sys.exit(1) with open(args.analysis, "r", encoding="utf-8") as f: analysis = json.load(f) # ── 读取搜索结果(可选)──────────────────────────────── search_results = None if os.path.exists(args.search_results): with open(args.search_results, "r", encoding="utf-8") as f: search_results = json.load(f) # ── 生成 HTML ──────────────────────────────────────── print("正在生成 HTML 报告...", file=sys.stderr) html_content = generate_html(analysis, search_results) # ── 写入文件 ───────────────────────────────────────── output_path = args.output or os.path.join( os.path.dirname(os.path.abspath(__file__)), "..", "output", "geo_report.html" ) os.makedirs(os.path.dirname(os.path.abspath(output_path)), exist_ok=True) with open(output_path, "w", encoding="utf-8") as f: f.write(html_content) print(f"报告已生成: {output_path}", file=sys.stderr) # stdout 输出摘要 brand = analysis.get("brand", "") geo_score = analysis.get("geo_score", {}) summary = { "brand": brand, "output": output_path, "geo_score": geo_score, "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), } print(json.dumps(summary, ensure_ascii=False)) if __name__ == "__main__": main() -
geo_search.py 7.9 KB
#!/usr/bin/env python3 """ GEO 搜索调度器 — 3平台 x N问题 批量提交 + 并行轮询 用法: python3 geo_search.py --queries '["问题1","问题2",...]' --platforms doubao,kimi,deepseek 输出: output/search_results.json 环境变量: REDFOX_API_KEY — 红狐 API Key(必填) """ import sys import os import json import time import argparse from concurrent.futures import ThreadPoolExecutor, as_completed # 将 lib 目录加入搜索路径 sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "lib")) from platforms import PLATFORMS, submit, poll, extract_result, API_KEY POLL_INTERVAL = 5 # 轮询间隔(秒) MAX_POLL_TIME = 900 # 最长等待 15 分钟(DeepSeek 深度搜索较慢) MAX_WORKERS = 6 # 并发线程数(降低避免 Kimi 限流) def main(): parser = argparse.ArgumentParser(description="GEO 批量搜索调度器") parser.add_argument( "--queries", required=True, help='问题列表 JSON,如 \'["问题1","问题2"]\'', ) parser.add_argument( "--platforms", default="doubao,kimi,deepseek", help="平台列表,逗号分隔(默认 doubao,kimi,deepseek)", ) parser.add_argument( "--output", default=None, help="输出文件路径(默认 output/search_results.json)", ) args = parser.parse_args() # ── 前置检查 ────────────────────────────────────────── if not API_KEY: result = { "error": ( "未配置 REDFOX_API_KEY 环境变量," "请前往 https://redfox.hk/settings/api-keys?source=github 获取 API Key" ) } print(json.dumps(result, ensure_ascii=False)) sys.exit(1) queries = json.loads(args.queries) platforms = [p.strip() for p in args.platforms.split(",") if p.strip()] for p in platforms: if p not in PLATFORMS: print(json.dumps({"error": f"未知平台: {p},可选: {list(PLATFORMS.keys())}"}, ensure_ascii=False)) sys.exit(1) total = len(queries) * len(platforms) output_path = args.output or os.path.join( os.path.dirname(os.path.abspath(__file__)), "..", "output", "search_results.json" ) os.makedirs(os.path.dirname(os.path.abspath(output_path)), exist_ok=True) print(f"开始搜索: {len(queries)} 个问题 x {len(platforms)} 个平台 = {total} 个任务", file=sys.stderr) # ── Phase 1: 批量提交所有任务 ────────────────────────── tasks = [] # [{query, platform, query_index, task_id | error}] with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor: future_map = {} for qi, query in enumerate(queries): for platform in platforms: future = executor.submit(submit, platform, query) future_map[future] = { "query": query, "platform": platform, "query_index": qi, } for future in as_completed(future_map): info = future_map[future] try: task_id = future.result() info["task_id"] = task_id print(f" [提交] {info['platform']} Q{info['query_index']+1} -> {task_id}", file=sys.stderr) except Exception as e: info["task_id"] = None info["error"] = str(e) print(f" [提交失败] {info['platform']} Q{info['query_index']+1} -> {e}", file=sys.stderr) tasks.append(info) submitted_ok = [t for t in tasks if t.get("task_id")] print(f"\n提交完成: {len(submitted_ok)}/{total} 成功,开始轮询...\n", file=sys.stderr) # ── Phase 2: 并行轮询所有任务 ────────────────────────── results = [] pending = list(submitted_ok) max_attempts = MAX_POLL_TIME // POLL_INTERVAL for attempt in range(1, max_attempts + 1): if not pending: break time.sleep(POLL_INTERVAL) still_pending = [] completed_this_round = 0 with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor: future_map = {} for task in pending: future = executor.submit(poll, task["platform"], task["task_id"]) future_map[future] = task for future in as_completed(future_map): task = future_map[future] try: raw = future.result() if raw is not None: # 任务完成 content, sources = extract_result(task["platform"], raw) results.append({ "question": task["query"], "query_index": task["query_index"], "platform": task["platform"], "content": content, "sources": sources, "status": "completed", }) completed_this_round += 1 print(f" [完成] {task['platform']} Q{task['query_index']+1}", file=sys.stderr) else: still_pending.append(task) except Exception as e: results.append({ "question": task["query"], "query_index": task["query_index"], "platform": task["platform"], "content": "", "sources": [], "status": "failed", "error": str(e), }) print(f" [失败] {task['platform']} Q{task['query_index']+1} -> {e}", file=sys.stderr) pending = still_pending if pending: print(f" 轮询 {attempt}: 本轮完成 {completed_this_round},剩余 {len(pending)} 等待中...", file=sys.stderr) # 处理超时任务 for task in pending: results.append({ "question": task["query"], "query_index": task["query_index"], "platform": task["platform"], "content": "", "sources": [], "status": "timeout", }) # 处理提交失败的任务 for task in tasks: if task.get("task_id") is None: results.append({ "question": task["query"], "query_index": task["query_index"], "platform": task["platform"], "content": "", "sources": [], "status": "submit_failed", "error": task.get("error", "Unknown"), }) # 按 query_index + platform 排序 platform_order = {p: i for i, p in enumerate(platforms)} results.sort(key=lambda r: (r.get("query_index", 0), platform_order.get(r.get("platform", ""), 99))) output = { "queries": queries, "platforms": platforms, "total_tasks": total, "completed": len([r for r in results if r["status"] == "completed"]), "failed": len([r for r in results if r["status"] != "completed"]), "results": results, } with open(output_path, "w", encoding="utf-8") as f: json.dump(output, f, ensure_ascii=False, indent=2) print(f"\n搜索完成: {output['completed']}/{total} 成功", file=sys.stderr) print(f"结果已保存: {output_path}", file=sys.stderr) # stdout 输出 JSON 供 Agent 解析 print(json.dumps(output, ensure_ascii=False)) if __name__ == "__main__": main()
-
-
README.en.md 4.5 KB
# GEO Analyzer / geo-analyzer --- ## Overview GEO Analyzer is a brand AI search visibility analysis tool. It automatically submits the same questions to three AI search engines — Doubao, Kimi, and DeepSeek — then detects brand mention rates, sentiment tendencies, source citations, and competitor comparisons, generating an interactive HTML report. **Core Value** - Enter a brand name and category to auto-generate test questions and batch-submit them across three AI platforms - No manual testing needed — parallel polling runs in the background, results ready in 3-5 minutes - Delivers a complete GEO score, mention rate matrix, sentiment distribution, and competitor comparison at a glance **Who It's For** - 📊 **Brand marketers** — Understand brand exposure and reputation in AI search, shape GEO optimization strategy - 🎯 **Product managers** — Compare competitor performance in AI answers, uncover differentiation opportunities - 📝 **Content operators** — Identify high-frequency citation sources, optimize content placement for better brand visibility --- ## Features ### Core Capabilities - Auto-generates 5 category-relevant test questions (recommendation / comparison / evaluation / scenario), with support for custom questions - Batch-submits N questions × 3 AI platforms with parallel polling, up to 8 minutes max wait - Deterministic analysis: brand mention rate (cross-platform + per-platform), source domain aggregation, mention matrix heatmap - AI deep analysis: brand ranking position, sentiment tendency (positive / neutral / negative), competitor comparison matrix - GEO composite score (0-100): mention rate 40% + ranking 30% + sentiment 30% weighted calculation - Interactive HTML report: anchor navigation, brand fingerprint, mention matrix, sentiment analysis, source analysis, competitor comparison, raw archive --- ## API Key Acquisition & Security - This skill requires the environment variable: `REDFOX_API_KEY`. - `REDFOX_API_KEY` is provided by [RedFoxHub](https://redfox.hk/settings/api-keys?source=github) (`https://redfox.hk`). - Visit [RedFoxHub](https://redfox.hk?source=github) to register and obtain your `REDFOX_API_KEY`. - Configure the environment variable `REDFOX_API_KEY` on your device before using this skill. - Before providing a key, verify its source, scope, expiration, and whether it supports reset/revocation. - Do not hardcode or expose the key in plain text within code, prompts, logs, or output files. --- ## Usage Just describe what you need in natural language — no commands to memorize. ### Quick Reference | Intent | Example Prompt | Result | |--------|---------------|--------| | Analyze brand AI visibility | "Analyze how Genki Forest performs in AI search" | Auto-generates questions, batch-searches 3 platforms, delivers full GEO report | | Custom question testing | "Test DJI's AI search performance with these 5 questions" | Uses your custom question list for batch search | | Competitor comparison | "Compare NIO, Li Auto, and XPeng in AI search" | Delivers competitor matrix with per-platform mention rates and sentiment | ### Example Output After analyzing "Genki Forest" in the sugar-free beverage category, you'll receive a report summary like: > **Report Summary · Genki Forest** > > **Overall Assessment:** Genki Forest's cross-platform GEO composite score is **72**, mid-to-upper tier. Brand mention rate 80% (mentioned in 12/15 answers), positive rate 75%. > > **Platform Breakdown:** > **Doubao**: GEO score 80 (Good), mention rate 100%, avg. rank #2.3 > **Kimi**: GEO score 68 (Good), mention rate 80%, avg. rank #3.1 > **DeepSeek**: GEO score 68 (Good), mention rate 60%, avg. rank #2.8 > > **Action Items:** Prioritize improving DeepSeek mention rate (currently only 60%); address negative feedback with targeted pain-point solutions. --- ## Use Cases | Scenario | Role | Example Prompt | Benefit | |----------|------|---------------|---------| | Brand diagnosis | Brand marketing lead | "Analyze our brand's AI search performance" | Full GEO score, per-platform mention rates, and sentiment distribution | | Competitor benchmarking | Product manager | "Compare brand A vs. B and C in AI search visibility" | Competitor matrix revealing differentiation opportunities | | Source strategy | Content operator | "Which websites do AI search engines cite for our category?" | TOP10 citation domains to guide content placement | | Ongoing monitoring | Brand PR team | "Track our brand's GEO changes weekly" | Track GEO score trends, respond to reputation shifts promptly | -
README.md 4.1 KB
# GEO Analyzer / geo-analyzer --- ## 简介 GEO Analyzer 是一款品牌 AI 搜索可见度分析工具,一键向豆包、Kimi、DeepSeek 三大 AI 搜索引擎提出相同问题,自动检测品牌出现率、情绪倾向、信源引用和竞品对比,生成交互式 HTML 报告。 **核心价值** - 输入品牌名和品类,自动生成测试问题并批量提交三大 AI 平台 - 无需手动逐个测试,后台并行轮询所有任务,约 3-5 分钟出结果 - 输出完整 GEO 综合得分、提及率矩阵、情绪分布、竞品对比,一眼掌握品牌 AI 认知现状 **适用对象** - 📊 **品牌营销人员** — 了解品牌在 AI 搜索中的曝光与口碑,制定 GEO 优化策略 - 🎯 **产品经理** — 对比竞品在 AI 回答中的表现,发现差异化机会 - 📝 **内容运营** — 找到 AI 引用的高频信源,针对性布局内容提升品牌可见度 --- ## 功能特性 ### 核心功能 - 自动生成 5 个贴合品类的测试问题(推荐类 / 对比类 / 评价类 / 场景类),也支持自定义问题 - 批量提交 3 个 AI 平台 × N 个问题,并行轮询,最长等待 8 分钟 - 确定性分析:品牌提及率(跨平台 + 分平台)、信源域名聚合、提及矩阵热力图 - AI 深度分析:品牌排名位置、情绪倾向(正面/中性/负面)、竞品对比矩阵 - GEO 综合得分(0-100):提及率 40% + 排名 30% + 情绪 30% 加权计算 - 交互式 HTML 报告:导航锚点、品牌指纹、提及矩阵、情感分析、信源分析、竞品对比、原始存档 --- ## 密钥获取与安全说明 - 本技能需要使用环境变量:`REDFOX_API_KEY`。 - `REDFOX_API_KEY` 由 [红狐 hub](https://redfox.hk/settings/api-keys?source=github)(`https://redfox.hk`)提供。 - 请前往 [红狐 hub](https://redfox.hk?source=github) 注册账号,获取 `REDFOX_API_KEY`。 - 配置设备环境变量 `REDFOX_API_KEY` 后使用本技能。 - 在提供密钥前,请先确认密钥来源、可用范围、有效期及是否支持重置/撤销。 - 禁止在代码、提示词、日志或输出文件中硬编码/明文暴露密钥。 --- ## 使用指南 直接用自然语言描述需求,无需记忆命令。 ### 常用说法速查 | 意图 | 示例话术 | 效果 | |------|---------|------| | 分析品牌 AI 可见度 | 「帮我分析元气森林在 AI 搜索中的表现」 | 自动生成问题,批量搜索三大平台,输出完整 GEO 报告 | | 自定义问题测试 | 「用这 5 个问题测试大疆的 AI 搜索表现」 | 直接使用用户提供的问题列表执行批量搜索 | | 竞品对比分析 | 「分析蔚来和理想、小鹏在 AI 搜索的对比」 | 包含竞品对比矩阵,各平台提及率和情绪分布 | ### 输出示例 分析「元气森林」在无糖饮料品类的 GEO 表现后,你会得到类似这样的报告摘要: > **报告摘要 · 元气森林** > > **综合评估:** 元气森林 跨平台 GEO 综合得分 **72** 分,处于中上游水平。品牌提及率 80%(12/15 条回答提及),正面率 75%。 > > **各平台表现:** > **豆包**: GEO 得分 80(良好),提及率 100%,平均排名 #2.3 > **Kimi**: GEO 得分 68(良好),提及率 80%,平均排名 #3.1 > **DeepSeek**: GEO 得分 68(良好),提及率 60%,平均排名 #2.8 > > **发力方向:** 重点提升 DeepSeek 平台的提及率(当前仅 60%);关注负面反馈,针对性解决用户痛点。 --- ## 使用场景 | 场景 | 角色 | 示例问法 | 收益 | |------|------|---------|------| | 品牌诊断 | 品牌方营销负责人 | 「分析我们品牌在 AI 搜索中的表现」 | 掌握 GEO 综合得分、各平台提及率和情绪分布 | | 竞品对标 | 产品经理 | 「分析 A 品牌和 B、C 竞品的 AI 可见度对比」 | 竞品对比矩阵,发现差异化机会 | | 信源布局 | 内容运营 | 「AI 搜索引擎引用了哪些网站来说我们品类」 | TOP10 引用域名,指导内容投放策略 | | 定期监测 | 品牌公关团队 | 「每周分析一次品牌的 GEO 变化」 | 追踪 GEO 得分趋势,及时应对口碑变化 | -
SKILL.md 11.3 KB
--- name: geo-analyzer description: GEO(生成式引擎优化)品牌分析工具。自动向豆包、Kimi、DeepSeek 三个 AI 搜索引擎提问,分析品牌在 AI 回答中的出现率、情绪、信源引用、竞品对比,生成交互式 HTML 报告。当用户需要分析品牌在 AI 搜索引擎中的表现、GEO 优化、AI 搜索品牌监测、品牌 AI 可见度分析时使用。触发词:GEO分析、品牌AI搜索分析、AI搜索引擎优化、品牌可见度、GEO、geo analyzer。 --- # 品牌GEO分析 ## 简介 帮助品牌了解自己在 AI 搜索引擎(豆包、Kimi、DeepSeek)中的表现。系统自动向三个 AI 平台提出相同问题,分析品牌出现率、情绪倾向、信源引用、竞品对比,生成可视化 HTML 报告。 ## 鉴权 前往 [红狐hub](https://redfox.hk/settings/api-keys?source=github) 获取 API Key,通过以下方式配置: ```bash # 方式一:配置文件 { "env": { "REDFOX_API_KEY": "ak_xxxx..." } } # 方式二:终端环境变量 export REDFOX_API_KEY="ak_xxxx..." ``` ## 依赖 ```bash pip3 install requests ``` ## 完整工作流 ### Step 0: 输入收集 向用户收集以下信息(使用 AskUserQuestion 或直接从用户消息中提取): **必填信息:** - **品牌名称**: 用户要分析的品牌名(如"元气森林"、"大疆"、"蔚来") - **品类/行业**: 品牌所属品类(如"无糖饮料"、"无人机"、"新能源汽车") **可选信息:** - **品牌别名**: 品牌的其他常见称呼(如"小红书"的别名"RED"、"小红书App") - **竞品列表**: 用户已知竞品(如"可口可乐"、"百事可乐") - **自定义问题列表**: 如果用户已有问题列表,直接使用,跳过 Step 1 **关键规则:** - 如果用户只提供了品牌名没有品类,必须追问品类 - 如果用户提供了问题列表,跳过 Step 1 直接进入 Step 2 - 竞品和别名可以为空 ### Step 1: 问题生成(用户未提供问题时) 如果用户没有提供问题列表,需要生成 5 个热门问题。 **1.1 搜索品类热度** 调用任一 websearch skill 搜索品类相关信息: ```bash python3 ~/.agents/skills/doubao-websearch/scripts/doubao_search.py "{品类} 消费者最关心的问题" python3 ~/.agents/skills/kimi-websearch/scripts/kimi_search.py "{品类} 品牌推荐 常见问题" ``` **1.2 AI 生成 5 个问题** 结合搜索结果和品类知识,生成 5 个用户最可能在 AI 搜索引擎中提问的问题。 **问题类型必须覆盖以下四类(每类至少 2 个):** | 类型 | 示例 | 说明 | |------|------|------| | 推荐类 | "{品类}哪个品牌好?"、"推荐几款好用的{品类}" | 测试品牌是否进入推荐列表 | | 对比类 | "{品牌A}和{品牌B}哪个好?"、"{品类}品牌对比" | 测试品牌在直接对比中的表现 | | 评价类 | "{品牌}怎么样?"、"{品牌}值得买吗?" | 测试品牌单独评价的倾向 | | 场景类 | "{场景}用什么{品类}好?"、"新手适合用哪个{品类}?" | 测试品牌在特定场景下的可见度 | **问题生成规则:** - 问题必须是用户真实可能搜索的自然语言 - 不要在问题中直接包含用户品牌名(推荐类和场景类),除非是评价类问题 - 评价类问题中可以包含品牌名 - 问题长度控制在 10-30 字 - 不要生成过于相似的问题 **1.3 确认问题列表(必须执行)** 将生成的 5 个问题展示给用户,使用 AskUserQuestion 询问确认。必须等待用户明确同意后,才能进入 Step 2 批量搜索。如果用户要求修改,重新调整问题列表并再次确认。 ### Step 2: 批量搜索 将 5 个问题同时提交到 3 个 AI 平台进行联网搜索。 ```bash python3 scripts/geo_search.py --queries '["问题1","问题2",...,"问题10"]' --platforms doubao,kimi,deepseek ``` **脚本自动完成:** 1. 批量提交 15 个搜索任务(3平台 x 5问题) 2. 并行轮询所有任务,每 5 秒检查一次 3. 最长等待 8 分钟 4. 输出 `output/search_results.json` **输出文件结构:** ```json { "queries": ["问题1", "问题2", ...], "platforms": ["doubao", "kimi", "deepseek"], "total_tasks": 30, "completed": 28, "failed": 2, "results": [ { "question": "问题1", "query_index": 0, "platform": "doubao", "content": "AI回答全文...", "sources": [{"title": "...", "url": "...", "domain": "..."}], "status": "completed" } ] } ``` **等待提示:** 搜索过程约需 3-5 分钟,告知用户耐心等待。 ### Step 3: 确定性分析 运行分析脚本执行确定性分析(品牌提及检测、域名提取、频次统计): ```bash python3 scripts/geo_analyze.py \ --brand "品牌名" \ --aliases '["别名1","别名2"]' \ --competitors '["竞品A","竞品B"]' \ --search-results output/search_results.json ``` **输出两个文件:** - `output/deterministic.json` — 确定性分析结果(提及率、域名统计等) - `output/ai_analysis_template.json` — AI 分析模板(待填充) ### Step 4: AI 分析(Agent 执行) 这是核心步骤。Agent 需要读取搜索结果,对每份 AI 回答执行深度分析。 **4.1 读取搜索结果** 读取 `output/search_results.json`,获取所有 15 份 AI 回答。 **4.2 逐份分析** 对每份状态为 `completed` 的回答,分析以下字段: 1. **brand_rank** (int | null): 品牌在回答的推荐列表或排名中的位置。如果回答列举了"推荐5个品牌",品牌排第几?未提及则为 null。 2. **brand_context** (str): 品牌被提及时的一句话上下文摘要。未提及则为空字符串。 3. **sentiment** (str): 回答对该品牌的整体情绪倾向。 - `"positive"`: 正面/推荐/强调优势 - `"neutral"`: 中性/客观描述 - `"negative"`: 负面/强调劣势/不推荐 - 未提及则为 `"neutral"` 4. **sentiment_reason** (str): 情绪判断的依据,引用回答中的原文或概括原因。 5. **competitors_mentioned** (list[str]): 回答中提及的所有竞品品牌名称。不限于已知竞品,发现新竞品也列出。 6. **competitor_details** (list[dict]): 对每个被提及的竞品,提供其排名和情绪信息。格式为列表,每项含: - `name` (str): 竞品品牌名 - `rank` (int | null): 竞品在推荐列表中的排名,无排名则为 null - `sentiment` (str): 竞品的情绪倾向 (positive/neutral/negative) 7. **key_claims** (list[str]): 关于该品牌的关键描述或评价(2-3条简短摘要)。 **分析规则:** - 情绪判断基于回答中对品牌的整体描述,不是单句话 - 如果回答只是列举品牌名没有评价,sentiment 为 "neutral" - 如果回答明确推荐该品牌或强调其优势,sentiment 为 "positive" - 如果回答指出该品牌的缺点或不推荐,sentiment 为 "negative" - competitors_mentioned 应包含回答中出现的所有同品类品牌,即使用户没有列为竞品 - competitor_details 中每个竞品的 rank 和 sentiment 基于回答中对竞品的描述判断 - brand_rank 只在回答有明确的品牌排序时填写(如"第一名是XX,第二名是YY"),无排序则为 null **4.3 写入 AI 分析结果** 将所有分析结果写入 `output/ai_analysis.json`,格式如下: ```json [ { "question": "问题1", "query_index": 0, "platform": "doubao", "brand": "品牌名", "competitors": ["竞品A", "竞品B"], "brand_rank": 3, "brand_context": "品牌被提及时的上下文摘要", "sentiment": "positive", "sentiment_reason": "回答中明确推荐该品牌", "competitors_mentioned": ["竞品A", "竞品C"], "competitor_details": [ {"name": "竞品A", "rank": 1, "sentiment": "positive"}, {"name": "竞品C", "rank": null, "sentiment": "neutral"} ], "key_claims": ["关键描述1", "关键描述2"] } ] ``` **关键规则:** - 只包含 status 为 completed 的回答 - 每份回答一个 JSON 对象 - query_index 和 platform 必须与 search_results.json 中的对应 - 必须输出合法 JSON ### Step 5: 合并分析与报告生成 **5.1 合并分析** 将确定性分析与 AI 分析合并,计算完整指标(GEO 得分、情绪分布、平均排名等): ```bash python3 scripts/geo_analyze.py \ --brand "品牌名" \ --aliases '["别名1"]' \ --competitors '["竞品A","竞品B"]' \ --search-results output/search_results.json \ --ai-analysis output/ai_analysis.json ``` 输出 `output/analysis_result.json`(包含所有指标数据)。 **5.2 生成 HTML 报告** ```bash python3 scripts/geo_report.py \ --analysis output/analysis_result.json \ --search-results output/search_results.json ``` 输出 `output/geo_report.html`。 ### Step 6: 交付 1. **交付 HTML 报告文件**: `output/geo_report.html` 2. **输出关键发现摘要**(3-5 条核心结论),格式参考: ``` GEO 分析完成 — {品牌名} 在三大 AI 搜索引擎中的表现: 1. 跨平台提及率: {X}%({被提及数}/{总问题数}) - 豆包: {X}% | Kimi: {X}% | DeepSeek: {X}% 2. GEO 综合得分: {X}/100 - {最佳平台} 表现最佳({X}分),{最差平台} 表现最弱({X}分) 3. 情绪分布: 正面 {X}% | 中性 {X}% | 负面 {X}% - {如果有负面,指出主要负面原因} 4. 竞品对比: 共发现 {N} 个竞品 - 提及率最高的竞品: {竞品名}({X}%) - 您的品牌提及率排名: 第 {N} 位 5. 信源引用: TOP3 引用域名为 {域名1}、{域名2}、{域名3} - 品牌官网是否被引用: {是/否} 完整报告: output/geo_report.html ``` ## 错误处理 | 情况 | 处理方式 | |------|---------| | 未配置 REDFOX_API_KEY | 提示用户前往红狐hub获取 API Key | | 搜索任务部分失败 | 继续分析已完成的回答,在报告中标注失败项 | | 搜索全部超时 | 提示用户稍后重试,可能是 API 负载过高 | | AI 分析结果缺失 | 仅生成确定性分析报告(提及率、域名统计),情绪等模块标注"待分析" | | 品牌名太短导致误匹配 | 提示用户提供品牌别名以提高匹配精度 | ## 文件结构 ``` geo-analyzer/ ├── SKILL.md # 本文件 ├── scripts/ │ ├── geo_search.py # 批量搜索调度器 │ ├── geo_analyze.py # 分析编排器 │ ├── geo_report.py # 报告生成器 │ └── lib/ │ ├── platforms.py # 3平台适配器 │ ├── analyzer.py # 分析逻辑库 │ └── report_template.py # HTML 模板库 ├── output/ # 输出目录 │ ├── search_results.json # 搜索结果 │ ├── deterministic.json # 确定性分析 │ ├── ai_analysis_template.json # AI 分析模板 │ ├── ai_analysis.json # AI 分析结果(Agent 写入) │ ├── analysis_result.json # 完整分析结果 │ └── geo_report.html # 最终报告 └── references/ └── geo-metrics.md # GEO 指标定义说明 ```
Comments (0)
Sign in to join the conversation.
Reviews (0)
No reviews yet.
No comments yet.