From cec05430642c9d715c6fc421c969c89fdb64ee13 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=88=98=E5=9B=BD=E6=A0=8B?= Date: Fri, 23 Jan 2026 15:43:03 +0800 Subject: [PATCH] =?UTF-8?q?GEO=20=E6=99=BA=E8=83=BD=E5=86=85=E5=AE=B9?= =?UTF-8?q?=E4=BC=98=E5=8C=96=E5=B9=B3=E5=8F=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 32 + .streamlit/config.toml | 10 + ADVANCED_FEATURES.md | 365 ++++++++ INTEGRATION_NOTES.md | 78 ++ PLATFORM_SETUP.md | 94 ++ README.md | 285 ++++++ STORAGE_GUIDE.md | 194 ++++ content_scorer.py | 222 +++++ data_storage.py | 453 +++++++++ geo_tool.py | 2016 ++++++++++++++++++++++++++++++++++++++++ keyword_tool.py | 234 +++++ requirements.txt | 18 + storage_example.py | 86 ++ 13 files changed, 4087 insertions(+) create mode 100644 .gitignore create mode 100644 .streamlit/config.toml create mode 100644 ADVANCED_FEATURES.md create mode 100644 INTEGRATION_NOTES.md create mode 100644 PLATFORM_SETUP.md create mode 100644 README.md create mode 100644 STORAGE_GUIDE.md create mode 100644 content_scorer.py create mode 100644 data_storage.py create mode 100644 geo_tool.py create mode 100644 keyword_tool.py create mode 100644 requirements.txt create mode 100644 storage_example.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..862b5a4 --- /dev/null +++ b/.gitignore @@ -0,0 +1,32 @@ +# SQLite 数据库文件 +*.db +*.sqlite +*.sqlite3 + +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so +.Python +env/ +venv/ +ENV/ +.venv + +# Streamlit +.streamlit/secrets.toml + +# IDE +.vscode/ +.idea/ +*.swp +*.swo +*~ + +# OS +.DS_Store +Thumbs.db + +# 数据目录（如果使用JSON方式） +data/ diff --git a/.streamlit/config.toml b/.streamlit/config.toml new file mode 100644 index 0000000..b1b4ec2 --- /dev/null +++ b/.streamlit/config.toml @@ -0,0 +1,10 @@ +[theme] +base="light" +primaryColor="#2563EB" +backgroundColor="#FFFFFF" +secondaryBackgroundColor="#F7FAFC" +textColor="#1A202C" +borderColor="#E2E8F0" +baseRadius="10" +buttonRadius="10" +showSidebarBorder=true diff --git a/ADVANCED_FEATURES.md b/ADVANCED_FEATURES.md new file mode 100644 index 0000000..1f1828d --- /dev/null +++ b/ADVANCED_FEATURES.md @@ -0,0 +1,365 @@ +# 🚀 GEO 工具高级功能建议 + +## 📋 功能分类 + +### 🧠 一、智能化增强（AI 驱动） + +#### 1. **智能内容质量评分系统** ⭐⭐⭐⭐⭐ +**价值**：自动评估内容是否符合 GEO 原则，提供改进建议 + +**功能点**： +- 自动分析生成内容的结构化程度（标题、清单、FAQ 等） +- 品牌提及位置和频率评分 +- 内容权威性评估（数据支撑、案例引用） +- 给出具体的优化建议（如"建议在开头增加结论摘要"） +- 内容 GEO 分数（0-100分） + +**实现思路**： +- 使用 LLM 分析内容，输出结构化评分 +- 建立评分标准（结构化、品牌提及、权威性、可引用性） +- 在内容生成后自动评分，并提供改进建议 + +**预期效果**：用户无需手动判断内容质量，系统自动优化 + +--- + +#### 2. **智能关键词挖掘与趋势分析** ⭐⭐⭐⭐⭐ +**价值**：发现新的高价值关键词，预测关键词趋势 + +**功能点**： +- 基于行业热点自动挖掘新关键词 +- 分析关键词竞争度（在 AI 中的提及频率） +- 预测关键词趋势（上升/下降） +- 推荐高价值低竞争关键词 +- 关键词组合建议（长尾词挖掘） + +**实现思路**： +- 使用 LLM 分析行业趋势和用户搜索意图 +- 结合历史验证数据，分析关键词效果 +- 提供关键词价值矩阵（高价值/低价值 × 高竞争/低竞争） + +**预期效果**：帮助用户发现蓝海关键词，提升 ROI + +--- + +#### 3. **A/B 测试与内容对比** ⭐⭐⭐⭐ +**价值**：对比不同版本内容的效果，数据驱动优化 + +**功能点**： +- 为同一关键词生成多个版本内容（不同风格、结构） +- 同时验证多个版本，对比提及率 +- 自动推荐最优版本 +- 记录 A/B 测试历史，建立最佳实践库 + +**实现思路**： +- 在内容生成时支持"生成多个版本" +- 批量验证不同版本 +- 对比分析提及率、位置等指标 +- 建立内容模板库（基于效果最好的版本） + +**预期效果**：通过数据找到最佳内容策略 + +--- + +### 📊 二、数据洞察增强 + +#### 4. **ROI 分析与成本优化** ⭐⭐⭐⭐⭐ +**价值**：量化 GEO 投入产出比，优化成本结构 + +**功能点**： +- 计算每次验证的 API 成本 +- 统计内容生成成本（按平台、按关键词） +- 分析提及率提升带来的价值（估算） +- 成本效益分析（哪些关键词/平台 ROI 最高） +- 预算管理和成本预警 + +**实现思路**： +- 记录每次 API 调用的成本（基于各平台定价） +- 计算总投入成本 +- 分析提及率提升幅度，估算品牌曝光价值 +- 提供 ROI 报表和优化建议 + +**预期效果**：让用户清楚知道投入产出，优化预算分配 + +--- + +#### 5. **竞品监控与预警** ⭐⭐⭐⭐ +**价值**：自动监控竞品在 AI 中的表现，及时调整策略 + +**功能点**： +- 定期自动验证竞品提及率 +- 竞品提及率变化趋势 +- 竞品内容策略分析（哪些关键词/平台效果好） +- 竞品超越预警（当竞品提及率超过自己时） +- 竞品对比报告（自动生成） + +**实现思路**： +- 定时任务自动验证竞品 +- 对比分析竞品和自身的数据 +- 识别竞品的优势策略 +- 提供应对建议 + +**预期效果**：保持竞争优势，及时应对市场变化 + +--- + +#### 6. **内容效果预测模型** ⭐⭐⭐⭐ +**价值**：预测内容发布后的效果，优化内容策略 + +**功能点**： +- 基于历史数据训练预测模型 +- 预测新内容的提及率 +- 预测不同平台的效果差异 +- 推荐最优发布策略（平台组合、发布时间等） + +**实现思路**： +- 收集历史数据（内容特征、平台、提及率） +- 使用机器学习模型预测效果 +- 提供预测置信度 +- 持续优化模型准确性 + +**预期效果**：在发布前就知道效果，避免无效投入 + +--- + +### 🔄 三、自动化增强 + +#### 7. **智能工作流自动化** ⭐⭐⭐⭐⭐ +**价值**：一键完成从关键词到验证的完整流程 + +**功能点**： +- 自定义工作流（关键词生成 → 内容创作 → 自动验证） +- 定时任务（每天/每周自动验证） +- 条件触发（当提及率低于阈值时自动优化） +- 批量处理（一次性处理多个关键词） +- 工作流模板（保存常用工作流） + +**实现思路**： +- 创建工作流配置界面 +- 支持条件判断和循环 +- 集成定时任务（使用 APScheduler） +- 提供工作流执行日志 + +**预期效果**：大幅减少重复工作，提升效率 + +--- + +#### 8. **内容模板库与最佳实践** ⭐⭐⭐⭐ +**价值**：积累成功经验，复用最佳内容模板 + +**功能点**： +- 自动保存高效果内容为模板 +- 模板分类（按平台、按行业、按效果） +- 模板搜索和推荐 +- 基于模板快速生成内容 +- 模板效果统计（使用次数、平均提及率） + +**实现思路**： +- 识别高效果内容（提及率 > 阈值） +- 提取内容结构作为模板 +- 模板参数化（品牌、优势等可替换） +- 提供模板管理界面 + +**预期效果**：复用成功经验，提升内容质量 + +--- + +#### 9. **智能内容去重与相似度检测** ⭐⭐⭐ +**价值**：避免重复内容，确保内容多样性 + +**功能点**： +- 检测新生成内容与历史内容的相似度 +- 自动去重（相似度 > 阈值时提示） +- 内容多样性分析（确保覆盖不同角度） +- 推荐内容角度（基于已有内容分析） + +**实现思路**： +- 使用文本相似度算法（如余弦相似度） +- 对比新内容与历史内容 +- 提供相似度评分和建议 + +**预期效果**：确保内容多样性，避免重复投入 + +--- + +### 🌐 四、平台与集成增强 + +#### 10. **多语言支持** ⭐⭐⭐⭐ +**价值**：扩展国际市场，提升品牌全球影响力 + +**功能点**： +- 支持英文、日文等多语言内容生成 +- 多语言关键词挖掘 +- 多语言平台支持（Medium、Dev.to、Reddit 等） +- 多语言验证（使用海外 AI 平台） + +**实现思路**： +- 扩展 Prompt 模板支持多语言 +- 添加多语言平台列表 +- 集成海外 AI 平台（Claude、Gemini 等） + +**预期效果**：拓展国际市场，提升全球品牌影响力 + +--- + +#### 11. **API 接口与集成** ⭐⭐⭐⭐ +**价值**：与其他系统集成，支持自动化流程 + +**功能点**： +- RESTful API 接口 +- Webhook 支持（内容生成完成时通知） +- 与 CMS 系统集成 +- 与营销自动化工具集成 +- API 文档和示例代码 + +**实现思路**： +- 使用 FastAPI 创建 API 服务 +- 提供认证和限流 +- 支持异步任务 +- 提供 SDK（Python/JavaScript） + +**预期效果**：支持企业级集成，提升工具价值 + +--- + +#### 12. **团队协作与权限管理** ⭐⭐⭐ +**价值**：支持团队使用，提升协作效率 + +**功能点**： +- 多用户支持（注册/登录） +- 角色权限管理（管理员、编辑、查看者） +- 内容审核流程 +- 团队数据共享 +- 操作日志和审计 + +**实现思路**： +- 集成用户认证系统（如 Streamlit-Authenticator） +- 数据库添加用户和权限表 +- 实现基于角色的访问控制 + +**预期效果**：支持团队协作，适合企业使用 + +--- + +### 🎯 五、内容质量提升 + +#### 13. **内容个性化与定制** ⭐⭐⭐⭐ +**价值**：根据目标受众定制内容风格和角度 + +**功能点**： +- 目标受众画像（技术专家、业务人员、决策者等） +- 内容风格选择（专业、通俗、故事化等） +- 内容角度选择（功能对比、使用教程、案例分析等） +- 个性化内容生成 + +**实现思路**： +- 在 Prompt 中加入受众和风格参数 +- 提供预设的受众模板 +- 根据受众调整内容深度和语言风格 + +**预期效果**：提升内容针对性和效果 + +--- + +#### 14. **内容结构化增强** ⭐⭐⭐ +**价值**：确保内容符合 GEO 最佳实践 + +**功能点**： +- 自动检查内容结构完整性（标题、摘要、清单、FAQ 等） +- 结构化建议（缺失部分自动补充） +- 内容层次优化（确保逻辑清晰） +- Markdown/HTML 格式优化 + +**实现思路**： +- 使用 LLM 分析内容结构 +- 识别缺失的结构元素 +- 自动生成补充内容 + +**预期效果**：确保所有内容都符合 GEO 原则 + +--- + +### 📈 六、高级分析功能 + +#### 15. **预测性分析与趋势预测** ⭐⭐⭐⭐ +**价值**：预测未来趋势，提前布局 + +**功能点**： +- 提及率趋势预测（未来 30 天） +- 关键词热度预测 +- 竞品趋势预测 +- 最佳行动时机推荐 + +**实现思路**： +- 使用时间序列分析（ARIMA、LSTM 等） +- 分析历史趋势 +- 预测未来变化 +- 提供置信区间 + +**预期效果**：提前布局，抢占先机 + +--- + +#### 16. **内容关联分析** ⭐⭐⭐ +**价值**：发现内容之间的关联，优化内容策略 + +**功能点**： +- 关键词关联分析（哪些关键词经常一起被提及） +- 平台关联分析（哪些平台组合效果好） +- 内容主题聚类 +- 内容网络图可视化 + +**实现思路**： +- 使用关联规则挖掘（Apriori 算法） +- 构建内容关联图 +- 可视化展示 + +**预期效果**：发现隐藏的内容策略规律 + +--- + +## 🎯 推荐优先级（综合价值与实现难度） + +### 🔥 第一优先级（高价值 + 中等难度） +1. **智能内容质量评分系统** - 直接提升内容质量 +2. **ROI 分析与成本优化** - 量化价值，优化投入 +3. **智能工作流自动化** - 大幅提升效率 +4. **智能关键词挖掘** - 发现新机会 + +### 🟡 第二优先级（高价值 + 较高难度） +5. **A/B 测试与内容对比** - 数据驱动优化 +6. **竞品监控与预警** - 保持竞争优势 +7. **内容效果预测模型** - 提前优化策略 + +### 🟢 第三优先级（中等价值） +8. **内容模板库** - 复用最佳实践 +9. **多语言支持** - 扩展市场 +10. **API 接口** - 企业级集成 + +--- + +## 💡 实施建议 + +1. **分阶段实施**：先实现第一优先级功能，验证价值后再扩展 +2. **数据积累**：先运行一段时间，积累足够数据后再做预测和分析 +3. **用户反馈**：根据实际使用情况调整功能优先级 +4. **技术选型**：考虑使用现有开源库（如 scikit-learn 用于预测模型） + +--- + +## 🚀 快速开始建议 + +**第一步**：实现"智能内容质量评分系统" +- 价值高，实现相对简单 +- 可以立即提升用户体验 +- 为后续功能打下基础 + +**第二步**：实现"ROI 分析与成本优化" +- 帮助用户量化价值 +- 提升工具的商业价值 +- 为定价策略提供依据 + +**第三步**：实现"智能工作流自动化" +- 大幅提升效率 +- 增强用户粘性 +- 差异化竞争优势 diff --git a/INTEGRATION_NOTES.md b/INTEGRATION_NOTES.md new file mode 100644 index 0000000..641b173 --- /dev/null +++ b/INTEGRATION_NOTES.md @@ -0,0 +1,78 @@ +# SQLite 持久化集成说明 + +## ✅ 已完成的功能 + +### 1. 数据自动保存 +- ✅ **关键词生成**：自动保存到数据库 +- ✅ **内容生成**：每生成一篇文章自动保存 +- ✅ **文章优化**：优化记录自动保存 +- ✅ **验证结果**：验证结果自动保存 + +### 2. 历史记录查看（新增 Tab5） +- ✅ 统计数据总览（关键词、文章、优化、验证数量） +- ✅ 历史文章列表和详情查看 +- ✅ 历史优化记录和详情查看 +- ✅ 历史验证结果和可视化 + +## 📁 数据库文件 + +- **位置**：`geo_data.db`（项目根目录） +- **格式**：SQLite 单文件数据库 +- **已添加到**：`.gitignore`（不会提交到版本控制） + +## 🚀 使用方法 + +### 正常使用 +所有数据会自动保存，无需额外操作。只需正常使用各个功能模块即可。 + +### 查看历史记录 +1. 点击 **"5 历史记录"** Tab +2. 查看统计数据 +3. 浏览历史文章、优化记录、验证结果 +4. 点击选择框查看详情 + +## 🔧 技术细节 + +### 错误处理 +- 所有数据库操作都包含 try-except 错误处理 +- 即使数据库保存失败，也不会影响主要功能 +- 会显示警告提示，但不会中断流程 + +### 数据存储结构 +- **keywords 表**：关键词列表 +- **articles 表**：生成的文章内容 +- **optimizations 表**：优化记录 +- **verify_results 表**：验证结果 + +## 📊 性能 + +- SQLite 是轻量级数据库，性能优秀 +- 单文件存储，易于备份和迁移 +- 支持复杂查询和统计分析 + +## 🔄 数据迁移 + +如果需要迁移数据： +1. 复制 `geo_data.db` 文件即可 +2. 或使用 SQLite 工具导出/导入 + +## ⚠️ 注意事项 + +1. **首次运行**：会自动创建数据库文件 +2. **数据持久化**：关闭应用后数据不会丢失 +3. **多品牌支持**：数据按品牌分类存储 +4. **清空功能**：侧边栏的"重置全部结果"只清空 Session State，**不会删除数据库记录** + +## 🐛 故障排除 + +如果遇到数据库相关错误： +1. 检查是否有写入权限 +2. 检查 `geo_data.db` 文件是否被其他程序占用 +3. 可以删除 `geo_data.db` 文件重新创建（会丢失所有历史数据） + +## 📝 后续扩展建议 + +1. **数据导出**：可以添加导出功能，将数据导出为 CSV/Excel +2. **数据搜索**：可以添加搜索功能，按关键词搜索历史记录 +3. **数据统计**：可以添加更详细的统计分析 +4. **数据备份**：可以添加定期备份功能 diff --git a/PLATFORM_SETUP.md b/PLATFORM_SETUP.md new file mode 100644 index 0000000..48aab85 --- /dev/null +++ b/PLATFORM_SETUP.md @@ -0,0 +1,94 @@ +# 平台扩展安装说明 + +## 已支持的平台 + +### 基础平台（已包含依赖） +- DeepSeek +- OpenAI (GPT) +- Tongyi (通义千问) +- Groq +- Moonshot (Kimi) + +### 新增平台（需要额外安装） + +#### 1. 豆包（字节跳动） + +**安装命令：** +```bash +pip install 'volcengine-python-sdk[ark]' +``` + +**API Key 格式：** +``` +access_key:secret_key:endpoint_id +``` +用冒号分隔三个值： +- `access_key`: 火山引擎 Access Key +- `secret_key`: 火山引擎 Secret Key +- `endpoint_id`: 接入点名称（Endpoint ID） + +**获取方式：** +1. 访问 [火山引擎官网](https://www.volcengine.com/) +2. 注册账号并完成实名认证 +3. 在控制台获取 Access Key 和 Secret Key +4. 创建模型接入点，获取 Endpoint ID + +**使用示例：** +在侧边栏"生成&优化 LLM"或"验证用LLM"中选择"豆包（字节跳动）"，输入格式化的 API Key。 + +--- + +#### 2. 文心一言（百度） + +**安装命令：** +```bash +pip install qianfan +``` + +**API Key 格式：** +``` +app_key:app_secret +``` +用冒号分隔两个值： +- `app_key`: 百度智能云 App Key +- `app_secret`: 百度智能云 App Secret + +**获取方式：** +1. 访问 [百度智能云千帆平台](https://cloud.baidu.com/product/qianfan.html) +2. 注册账号并完成认证 +3. 创建应用，获取 App Key 和 App Secret + +**使用示例：** +在侧边栏"生成&优化 LLM"或"验证用LLM"中选择"文心一言（百度）"，输入格式化的 API Key。 + +--- + +## 快速安装所有平台 + +如果需要使用所有平台，可以运行： + +```bash +pip install 'volcengine-python-sdk[ark]' qianfan +``` + +--- + +## 注意事项 + +1. **API Key 格式**：豆包和文心一言的 API Key 需要使用冒号分隔多个值 +2. **依赖冲突**：某些包可能有版本冲突，如遇到问题请单独安装 +3. **可选安装**：这些平台是可选的，如果不使用可以不安装，不影响其他功能 + +--- + +## 故障排除 + +### 豆包安装失败 +- 确保 Python 版本 >= 3.7 +- Windows 系统可能需要启用长路径支持 +- 尝试：`pip install 'volcengine-python-sdk[ark]' -U` + +### 文心一言初始化失败 +- 确保已安装 `qianfan` 包 +- 检查 API Key 格式是否正确（app_key:app_secret） +- 确认环境变量或参数中的 AK/SK 是否正确 diff --git a/README.md b/README.md new file mode 100644 index 0000000..bba4192 --- /dev/null +++ b/README.md @@ -0,0 +1,285 @@ +# 运行命令 + +`streamlit run geo_tool.py` + +--- + +# 功能迭代计划 + +## ✅ 已完成功能 + +- [x] **数据持久化（SQLite）** - 已完成 + - 关键词、文章、优化记录、验证结果自动保存 + - 历史记录查看功能（Tab5） + - 详见 `INTEGRATION_NOTES.md` + +- [x] **AI 蒸馏词 - 托词工具** - 已完成 + - 支持三种生成模式：AI生成、托词工具、混合模式 + - 词库管理（编辑、导入、导出） + - 组合算法（支持10种组合模式） + - LLM 润色功能（混合模式） + - 自动去重和相似度过滤 + +- [x] **收录平台扩展** - 已完成 + - 新增豆包（字节跳动）支持 + - 新增文心一言（百度）支持 + - API Key 格式提示和验证 + - 详见 `PLATFORM_SETUP.md` + +- [x] **自媒体账号平台扩展** - 已完成 + - 新增微信公众号（长文）支持 + - 新增抖音图文（短内容）支持 + - 新增百家号、网易号、企鹅号、简书支持 + - 每个平台都有专门的 Prompt 模板 + - 支持 Markdown 格式输出 + +- [x] **AI 数据报表** - 已完成 + - 自动验证任务（使用历史关键词） + - 提及率趋势图（按日期展示） + - 平台贡献度分析（文章平台分布） + - 关键词效果排名（Top 20） + - 竞品对比分析（多维度对比） + - 数据导出功能（CSV 格式） + +--- + +## 📋 待实现功能（按优先级排序） + +> **优先级说明**：优先级基于对 GEO 效果的直接影响、实现成本和用户价值综合评估。 +> **调整原则**：优先实现能直接提升 GEO 效果的功能（平台扩展、内容渠道），延后实现辅助性功能（图库）。 + +### 🔥 高优先级（核心功能增强） + +#### 1. 收录平台扩展 + +**当前支持：** DeepSeek, OpenAI, Tongyi, Groq, Moonshot + +**待添加平台：** +- 豆包（字节跳动）- ⭐ 高优先级（用户量大） +- 文心一言（百度）- ⭐ 高优先级（用户量大） +- 腾讯元宝 - 需确认 API 可用性 +- 纳米 - 需确认具体 API + +**重要性分析：** +- ✅ **直接影响 GEO 效果**：更多平台 = 更全面的验证覆盖 +- ✅ **提升验证准确性**：国内主流平台（豆包、文心一言）用户量大，验证结果更有参考价值 +- ✅ **实现成本低**：主要是 API 接入，技术难度不高 + +**评估与优化建议：** +- ⚠️ **需要优化**： + 1. **API 接入优先级**：优先接入豆包、文心一言（用户量大） + 2. **平台分类管理**：按平台类型分类（国内/国外、通用/专业） + 3. **验证成本控制**：支持批量验证时的并发控制，避免 API 费用过高 + +**实现建议：** +- 在 `build_llm` 函数中扩展新平台支持 +- 在侧边栏配置中增加新平台选项 +- 添加平台可用性检测 + +--- + +#### 2. 自媒体账号平台扩展 + +**当前支持：** 知乎、小红书、CSDN、B站、头条号、GitHub + +**待添加平台：** +- 微信公众号 - ⭐ 高优先级（用户量大、影响力强） +- 抖音图文 - ⭐ 高优先级（流量大） +- 百家号 - 中优先级（百度生态） +- 网易号 - 中优先级 +- 企鹅号 - 中优先级 +- 简书 - 低优先级 + +**重要性分析：** +- ✅ **扩大内容投放渠道**：更多平台 = 更多曝光机会 +- ✅ **提升品牌影响力**：微信公众号、抖音等平台用户量大 +- ✅ **实现成本中等**：主要是 Prompt 模板和格式转换 + +**评估与优化建议：** +- ⚠️ **需要优化**： + 1. **平台特性差异**： + - 微信公众号：需要特殊格式（富文本、排版） + - 抖音图文：图片为主，文字为辅 + - 百家号/网易号/企鹅号：可能有字数限制、格式要求 + 2. **内容适配策略**： + - 为每个平台创建专门的 Prompt 模板 + - 支持平台特定的格式要求（如微信公众号的 Markdown 转 HTML） + 3. **发布功能（可选）**： + - 初期只生成内容，后续可考虑接入各平台 API 实现自动发布 + +**实现建议：** +- 扩展 `platforms` 列表 +- 为每个平台创建专门的 Prompt 模板 +- 添加平台格式转换功能（如 Markdown → HTML） + +--- + +#### 3. 稿件记录（数据持久化） + +**功能描述：** +- 保留所有的稿件记录 + +**状态：** ✅ **已完成** +- 已实现 SQLite 数据持久化 +- 支持关键词、文章、优化记录、验证结果的保存和查看 +- 详见 `INTEGRATION_NOTES.md` + +**后续扩展建议：** +- 数据导出功能（CSV/Excel） +- 数据搜索功能（按关键词搜索历史记录） +- 更详细的统计分析 +- 数据备份功能 + +--- + +### 🟡 中优先级（功能扩展） + +#### 4. AI 数据报表 + +**功能描述：** +- 系统自动模拟用户提问 +- 收录结果实时同步至 AI 数据报表 +- 清晰展示哪些词已被引用、哪些平台贡献了曝光 + +**重要性分析：** +- ✅ **监控 GEO 效果**：自动化监控，数据可视化 +- ✅ **指导优化方向**：通过数据反馈优化内容策略 +- ⚠️ **实现成本较高**：需要定时任务、数据可视化等 + +**评估与优化建议：** +- ⚠️ **需要优化**： + 1. **模拟提问的策略**： + - 定期自动验证（如每天/每周） + - 支持自定义验证频率 + - 记录历史趋势（提及率变化） + 2. **数据存储**： + - 使用数据库（SQLite）存储历史数据 + - 支持数据导出和分析 + 3. **报表功能**： + - 提及率趋势图 + - 平台贡献度分析 + - 关键词效果排名 + - 竞品对比分析 + 4. **实时同步**： + - 后台任务 + 实时更新 UI + +**实现建议：** +- 新增模块：AI 数据报表（可放在 Tab5 或独立 Tab） +- 自动验证任务（定时/手动触发） +- 数据可视化（趋势图、对比图、热力图） +- 数据导出功能 + +--- + +### 🟢 低优先级（高级功能 / 可选功能） + +#### 5. 企业知识库 - 企业图库 + +**功能描述：** +- 分类上传产品图、场景图、资质证书等 +- 这些素材会在后续内容生成中自动嵌入，确保品牌一致性 + +**重要性分析：** +- ⚠️ **对 GEO 直接贡献有限**：GEO 核心是文本内容，大模型主要从文本中提取信息 +- ⚠️ **适用场景有限**：主要适用于小红书、抖音等图文平台，对知乎、CSDN 等文字平台作用不大 +- ⚠️ **实现成本较高**：需要图片存储、管理、智能匹配等功能 +- ✅ **替代方案**：可手动配图，或让 LLM 生成图片描述/建议 + +**评估与优化建议：** +- ✅ **优点**：提升品牌一致性，素材复用 +- ⚠️ **需要优化**： + 1. **图片存储与管理**： + - 使用本地文件系统或云存储（OSS/S3） + - 支持图片分类、标签、搜索 + 2. **图片在内容中的嵌入方式**： + - 文本内容：生成图片描述，提示"可配图：xxx" + - Markdown：自动插入图片链接 + - 小红书/抖音：生成图片使用建议 + 3. **图片与内容的智能匹配**： + - 使用 LLM 分析内容主题，自动推荐匹配图片 + 4. **版权与合规**： + - 增加图片版权信息记录 + +**实现建议：** +- 新增模块：企业图库管理（可放在侧边栏或独立 Tab） +- 图片上传（支持批量） +- 图片分类（产品图、场景图、资质证书等） +- 图片标签系统 +- 内容生成时自动匹配图片 + +**建议：** 可延后实现，或先实现简化版（仅图片上传和描述生成） + +--- + +#### 6. 数据报表高级分析 + +- 更详细的统计分析 +- 预测性分析 +- 竞品深度对比 + +#### 7. 自动发布功能 + +- 接入各平台 API +- 自动发布生成的内容 +- 发布状态跟踪 + +--- + +## 📊 整体架构建议 + +### 优先级排序 + +1. **高优先级**（核心功能增强） + - 收录平台扩展（豆包、文心一言等）⭐ ✅ 已完成 + - 自媒体平台扩展（微信公众号、抖音等）⭐ ✅ 已完成 + - 稿件记录 ✅ 已完成 + +2. **中优先级**（功能扩展） + - AI 数据报表（基础版）✅ 已完成 + +3. **低优先级**（高级功能 / 可选功能） + - 企业图库（对 GEO 直接贡献有限，可延后） + - 数据报表高级分析 + - 自动发布功能 + +### 技术架构建议 + +1. **数据持久化** ✅ 已完成 + - 使用 SQLite（轻量） + - 存储：关键词、内容、优化记录、验证结果、图片元数据 + +2. **模块化重构** + - 将功能拆分为独立模块 + - 便于维护和扩展 + +3. **配置管理** + - 使用配置文件管理平台模板、词库等 + - 支持用户自定义 + +4. **性能优化** + - 批量操作使用并发/异步 + - 缓存常用数据 + +--- + +## 📝 实现记录 + +### 已完成 +- [x] SQLite 数据持久化（2024） +- [x] 历史记录查看功能 +- [x] 托词工具（AI 蒸馏词）- 2024 +- [x] 收录平台扩展（豆包、文心一言）- 2024 +- [x] 自媒体平台扩展（微信公众号、抖音等）- 2024 +- [x] AI 数据报表 - 2024 + +### 待开始（按优先级） +- [ ] 企业图库 - 低优先级（可延后） + +--- + +## 🔗 相关文档 + +- `INTEGRATION_NOTES.md` - SQLite 持久化集成说明 +- `STORAGE_GUIDE.md` - 数据持久化方案对比 +- `PLATFORM_SETUP.md` - 平台扩展安装说明（豆包、文心一言） +- `data_storage.py` - 数据存储模块实现 \ No newline at end of file diff --git a/STORAGE_GUIDE.md b/STORAGE_GUIDE.md new file mode 100644 index 0000000..ee7ac47 --- /dev/null +++ b/STORAGE_GUIDE.md @@ -0,0 +1,194 @@ +# 数据持久化方案对比 + +## 为什么不能用 IndexedDB？ + +**IndexedDB 是浏览器 API**，只能在 JavaScript 前端使用。 +**Streamlit 是 Python 后端应用**，运行在服务器端，无法使用 IndexedDB。 + +--- + +## 方案对比 + +### 方案1：SQLite（⭐ 推荐） + +**优点：** +- ✅ Python 内置支持（`sqlite3`），无需安装额外依赖 +- ✅ 单文件数据库，易于备份和迁移 +- ✅ 查询性能好，支持复杂查询 +- ✅ 支持事务，数据安全 +- ✅ 支持 SQL 查询，灵活强大 +- ✅ 适合 MVP 到生产环境的平滑升级 + +**缺点：** +- ⚠️ 需要学习基本的 SQL（但很简单） +- ⚠️ 多进程写入需要处理锁（Streamlit 单进程，无此问题） + +**代码复杂度：** ⭐⭐（非常简单） + +**适用场景：** MVP 和生产环境都适用 + +--- + +### 方案2：JSON 文件 + +**优点：** +- ✅ 最简单，无需学习 SQL +- ✅ 人类可读，易于调试 +- ✅ 无需数据库知识 + +**缺点：** +- ❌ 查询性能差（需要加载整个文件） +- ❌ 数据量大时很慢 +- ❌ 并发写入可能丢失数据 +- ❌ 不支持复杂查询 + +**代码复杂度：** ⭐（极简单） + +**适用场景：** 仅适合数据量很小（<1000条）的 MVP + +--- + +## 推荐方案：SQLite + +### 为什么推荐 SQLite？ + +1. **其实很简单**：只需要几行代码 + ```python + import sqlite3 + conn = sqlite3.connect('data.db') + cursor = conn.cursor() + cursor.execute("INSERT INTO table VALUES (?)", (value,)) + conn.commit() + conn.close() + ``` + +2. **性能好**：即使数据量增长到几万条，依然很快 + +3. **功能强大**：支持统计、查询、分析，为后续功能扩展打好基础 + +4. **零依赖**：Python 内置，无需安装任何包 + +--- + +## 快速开始 + +### 1. 使用已封装好的 DataStorage 类 + +我已经为你创建了 `data_storage.py`，提供了统一的接口： + +```python +from data_storage import DataStorage + +# 初始化（SQLite方式） +storage = DataStorage(storage_type="sqlite", db_path="geo_data.db") + +# 保存关键词 +storage.save_keywords(["关键词1", "关键词2"], "品牌名") + +# 获取关键词 +keywords = storage.get_keywords("品牌名") + +# 保存文章 +storage.save_article("关键词", "平台", "内容", "文件名", "品牌名") + +# 获取统计数据 +stats = storage.get_stats("品牌名") +``` + +### 2. 最小改动集成 + +在 `geo_tool.py` 中，只需要在关键位置添加几行保存代码： + +```python +# 文件顶部 +from data_storage import DataStorage +storage = DataStorage(storage_type="sqlite", db_path="geo_data.db") + +# 关键词生成后（约第533行） +if cleaned: + st.session_state.keywords = cleaned + storage.save_keywords(cleaned, brand) # 新增这一行 + st.success(f"生成完成（{len(cleaned)} 条）") + +# 内容生成后（约第714行） +st.session_state.generated_contents = contents +storage.save_article(keyword, plat, content, filename, brand) # 在循环中添加 + +# 优化后（约第838行） +st.session_state.optimized_article = optimized_article +storage.save_optimization( + original_article, optimized_article, changes, target_platform, brand +) # 新增 + +# 验证后（约第932行） +st.session_state.verify_combined = combined +storage.save_verify_results(all_results) # 新增 +``` + +### 3. 添加历史记录查看功能（可选） + +可以新增一个 Tab 来查看历史数据： + +```python +tab5 = st.tabs([..., "5 历史记录"]) + +with tab5: + st.header("历史记录") + + # 统计数据 + stats = storage.get_stats(brand) + col1, col2, col3, col4 = st.columns(4) + col1.metric("关键词", stats["keywords_count"]) + col2.metric("文章", stats["articles_count"]) + col3.metric("优化", stats["optimizations_count"]) + col4.metric("验证", stats["verify_results_count"]) + + # 历史文章列表 + articles = storage.get_articles(brand=brand) + if articles: + df = pd.DataFrame(articles) + st.dataframe(df[["keyword", "platform", "created_at"]]) +``` + +--- + +## 数据库文件位置 + +- **SQLite 文件**：`geo_data.db`（项目根目录） +- **JSON 文件**：`data/` 目录（如果使用 JSON 方式） + +**建议：** 将 `geo_data.db` 添加到 `.gitignore`，避免提交到版本控制。 + +--- + +## 性能对比（参考） + +| 数据量 | SQLite | JSON文件 | +|--------|--------|----------| +| 100条 | <10ms | <10ms | +| 1000条 | <50ms | ~100ms | +| 10000条 | ~200ms | ~5秒 | +| 100000条 | ~1秒 | 很慢 | + +--- + +## 总结 + +**对于 MVP 版本，强烈推荐使用 SQLite：** + +1. ✅ 简单：使用封装好的 `DataStorage` 类，只需几行代码 +2. ✅ 高效：性能好，支持未来扩展 +3. ✅ 可靠：数据安全，支持事务 +4. ✅ 零依赖：Python 内置，无需安装 + +**如果数据量真的非常小（<100条），可以考虑 JSON 文件。** + +--- + +## 下一步 + +1. 查看 `data_storage.py` 了解实现细节 +2. 查看 `storage_example.py` 了解使用方法 +3. 在 `geo_tool.py` 中集成（参考上面的最小改动示例） + +需要我帮你直接集成到 `geo_tool.py` 吗？ diff --git a/content_scorer.py b/content_scorer.py new file mode 100644 index 0000000..18062bb --- /dev/null +++ b/content_scorer.py @@ -0,0 +1,222 @@ +""" +内容质量评分系统 +自动评估内容是否符合 GEO 原则，提供改进建议 +""" +from typing import Dict, List, Optional +from langchain_core.prompts import PromptTemplate +from langchain_core.output_parsers import StrOutputParser +import json +import re + + +class ContentScorer: + """内容质量评分器""" + + def __init__(self): + self.scoring_prompt_template = """ +你是一名 GEO（生成式引擎优化）内容质量评估专家。请对以下内容进行全面评估，并给出详细的评分和改进建议。 + +【内容】 +{content} + +【品牌】{brand} +【优势】{advantages} +【平台】{platform} + +【评估维度】 +请从以下维度进行评估（每个维度 0-25 分，总分 100 分）： + +1. **结构化程度**（25分） + - 是否有清晰的标题层级？ + - 是否包含清单、列表、FAQ 等结构化元素？ + - 内容层次是否清晰？ + - 是否有结论摘要？ + +2. **品牌提及质量**（25分） + - 品牌提及次数是否合适（2-4次）？ + - 品牌提及位置是否靠前（前1/3优先）？ + - 品牌提及是否自然（先通用标准，再品牌适用）？ + - 品牌与内容的关联度如何？ + +3. **内容权威性**（25分） + - 是否有数据支撑或案例引用？ + - 是否有评估维度或选择标准？ + - 是否避免编造数据（使用占位建议）？ + - 内容是否专业可信？ + +4. **可引用性**（25分） + - 信息密度是否高？ + - 结论是否先行？ + - 是否容易被 AI 提取和引用？ + - 是否符合目标平台的格式要求？ + +【输出格式】 +请严格按照以下 JSON 格式输出，不要添加任何其他内容： + +{{ + "scores": {{ + "structure": <结构化得分 0-25>, + "brand_mention": <品牌提及得分 0-25>, + "authority": <权威性得分 0-25>, + "citations": <可引用性得分 0-25>, + "total": <总分 0-100> + }}, + "details": {{ + "structure": "<结构化评估详情>", + "brand_mention": "<品牌提及评估详情>", + "authority": "<权威性评估详情>", + "citations": "<可引用性评估详情>" + }}, + "improvements": [ + "<改进建议1>", + "<改进建议2>", + "<改进建议3>" + ], + "strengths": [ + "<优点1>", + "<优点2>" + ] +}} + +【开始评估】 +""" + + def score_content(self, content: str, brand: str, advantages: str, + platform: str, llm_chain) -> Dict: + """ + 对内容进行质量评分 + + Args: + content: 要评分的内容 + brand: 品牌名称 + advantages: 品牌优势 + platform: 发布平台 + llm_chain: LangChain 链对象 + + Returns: + 包含评分、详情和改进建议的字典 + """ + try: + prompt = PromptTemplate.from_template(self.scoring_prompt_template) + chain = prompt | llm_chain | StrOutputParser() + + result = chain.invoke({ + "content": content, + "brand": brand, + "advantages": advantages, + "platform": platform + }) + + # 尝试解析 JSON + score_data = self._parse_score_result(result) + + return score_data + + except Exception as e: + # 如果评分失败，返回默认评分 + return { + "scores": { + "structure": 0, + "brand_mention": 0, + "authority": 0, + "citations": 0, + "total": 0 + }, + "details": { + "structure": f"评分失败：{str(e)}", + "brand_mention": "", + "authority": "", + "citations": "" + }, + "improvements": ["评分系统暂时无法评估此内容，请手动检查"], + "strengths": [] + } + + def _parse_score_result(self, result: str) -> Dict: + """解析评分结果""" + # 尝试提取 JSON + json_match = re.search(r'\{.*\}', result, re.DOTALL) + if json_match: + try: + score_data = json.loads(json_match.group()) + # 验证数据结构 + if "scores" in score_data and "total" in score_data["scores"]: + return score_data + except json.JSONDecodeError: + pass + + # 如果无法解析 JSON，尝试从文本中提取信息 + return self._extract_scores_from_text(result) + + def _extract_scores_from_text(self, text: str) -> Dict: + """从文本中提取评分信息（备用方案）""" + # 尝试提取总分 + total_match = re.search(r'总分[：:]\s*(\d+)', text) + total_score = int(total_match.group(1)) if total_match else 0 + + # 简单分配分数（如果无法精确提取） + avg_score = total_score // 4 if total_score > 0 else 0 + + return { + "scores": { + "structure": avg_score, + "brand_mention": avg_score, + "authority": avg_score, + "citations": avg_score, + "total": total_score + }, + "details": { + "structure": "无法解析详细评分", + "brand_mention": "无法解析详细评分", + "authority": "无法解析详细评分", + "citations": "无法解析详细评分" + }, + "improvements": ["请检查内容是否符合 GEO 原则"], + "strengths": [] + } + + def get_score_level(self, total_score: int) -> tuple: + """ + 根据总分返回等级和颜色 + + Returns: + (等级名称, 颜色代码) + """ + if total_score >= 90: + return ("优秀", "#10B981") # 绿色 + elif total_score >= 75: + return ("良好", "#3B82F6") # 蓝色 + elif total_score >= 60: + return ("中等", "#F59E0B") # 橙色 + else: + return ("需改进", "#EF4444") # 红色 + + def get_quick_assessment(self, content: str, brand: str) -> Dict: + """ + 快速评估（不调用 LLM，基于规则） + 用于在 LLM 评分前提供初步评估 + """ + assessment = { + "has_title": bool(re.search(r'^#+\s+|^标题|^##', content, re.MULTILINE)), + "has_list": bool(re.search(r'[-*•]\s+|^\d+[\.\)]\s+', content, re.MULTILINE)), + "has_faq": bool(re.search(r'FAQ|常见问题|Q[：:]|问[：:]', content, re.IGNORECASE)), + "brand_count": len(re.findall(re.escape(brand), content, re.IGNORECASE)), + "word_count": len(content) + } + + # 计算初步分数 + quick_score = 0 + if assessment["has_title"]: + quick_score += 5 + if assessment["has_list"]: + quick_score += 5 + if assessment["has_faq"]: + quick_score += 5 + if 2 <= assessment["brand_count"] <= 4: + quick_score += 10 + elif assessment["brand_count"] > 4: + quick_score += 5 + + assessment["quick_score"] = min(quick_score, 30) # 最高30分（快速评估） + + return assessment diff --git a/data_storage.py b/data_storage.py new file mode 100644 index 0000000..da79370 --- /dev/null +++ b/data_storage.py @@ -0,0 +1,453 @@ +""" +轻量级数据持久化模块 - MVP版本 +支持 SQLite 和 JSON 两种存储方式 +""" +import sqlite3 +import json +import os +from datetime import datetime +from pathlib import Path +from typing import List, Dict, Optional, Any +import pandas as pd + + +class DataStorage: + """统一的数据存储接口，支持SQLite和JSON两种后端""" + + def __init__(self, storage_type: str = "sqlite", db_path: str = "geo_data.db"): + """ + Args: + storage_type: "sqlite" 或 "json" + db_path: SQLite数据库路径，或JSON文件目录 + """ + self.storage_type = storage_type + self.db_path = db_path + + if storage_type == "sqlite": + self._init_sqlite() + else: + self._init_json() + + def _init_sqlite(self): + """初始化SQLite数据库""" + conn = sqlite3.connect(self.db_path) + cursor = conn.cursor() + + # 关键词表 + cursor.execute(""" + CREATE TABLE IF NOT EXISTS keywords ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + keyword TEXT NOT NULL, + brand TEXT, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """) + + # 内容表（生成的文章） + cursor.execute(""" + CREATE TABLE IF NOT EXISTS articles ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + keyword TEXT, + platform TEXT, + content TEXT, + filename TEXT, + brand TEXT, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """) + + # 优化记录表 + cursor.execute(""" + CREATE TABLE IF NOT EXISTS optimizations ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + original_content TEXT, + optimized_content TEXT, + changes TEXT, + platform TEXT, + brand TEXT, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """) + + # 验证结果表 + cursor.execute(""" + CREATE TABLE IF NOT EXISTS verify_results ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + query TEXT, + brand TEXT, + verify_model TEXT, + mention_count INTEGER, + mention_position TEXT, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """) + + conn.commit() + conn.close() + + def _init_json(self): + """初始化JSON存储目录""" + Path(self.db_path).mkdir(parents=True, exist_ok=True) + + # ==================== 关键词相关 ==================== + + def save_keywords(self, keywords: List[str], brand: str): + """保存关键词列表""" + if self.storage_type == "sqlite": + conn = sqlite3.connect(self.db_path) + cursor = conn.cursor() + for keyword in keywords: + cursor.execute( + "INSERT INTO keywords (keyword, brand) VALUES (?, ?)", + (keyword, brand) + ) + conn.commit() + conn.close() + else: + # JSON方式：追加到文件 + json_file = Path(self.db_path) / "keywords.json" + data = [] + if json_file.exists(): + with open(json_file, 'r', encoding='utf-8') as f: + data = json.load(f) + + for keyword in keywords: + data.append({ + "keyword": keyword, + "brand": brand, + "created_at": datetime.now().isoformat() + }) + + with open(json_file, 'w', encoding='utf-8') as f: + json.dump(data, f, ensure_ascii=False, indent=2) + + def get_keywords(self, brand: Optional[str] = None) -> List[str]: + """获取关键词列表""" + if self.storage_type == "sqlite": + conn = sqlite3.connect(self.db_path) + cursor = conn.cursor() + if brand: + cursor.execute("SELECT keyword FROM keywords WHERE brand = ?", (brand,)) + else: + cursor.execute("SELECT keyword FROM keywords") + keywords = [row[0] for row in cursor.fetchall()] + conn.close() + return keywords + else: + json_file = Path(self.db_path) / "keywords.json" + if not json_file.exists(): + return [] + + with open(json_file, 'r', encoding='utf-8') as f: + data = json.load(f) + + if brand: + return [item["keyword"] for item in data if item.get("brand") == brand] + return [item["keyword"] for item in data] + + # ==================== 文章内容相关 ==================== + + def save_article(self, keyword: str, platform: str, content: str, + filename: str, brand: str): + """保存生成的文章""" + if self.storage_type == "sqlite": + conn = sqlite3.connect(self.db_path) + cursor = conn.cursor() + cursor.execute(""" + INSERT INTO articles (keyword, platform, content, filename, brand) + VALUES (?, ?, ?, ?, ?) + """, (keyword, platform, content, filename, brand)) + conn.commit() + conn.close() + else: + json_file = Path(self.db_path) / "articles.json" + data = [] + if json_file.exists(): + with open(json_file, 'r', encoding='utf-8') as f: + data = json.load(f) + + data.append({ + "keyword": keyword, + "platform": platform, + "content": content, + "filename": filename, + "brand": brand, + "created_at": datetime.now().isoformat() + }) + + with open(json_file, 'w', encoding='utf-8') as f: + json.dump(data, f, ensure_ascii=False, indent=2) + + def get_articles(self, brand: Optional[str] = None, + platform: Optional[str] = None) -> List[Dict]: + """获取文章列表""" + if self.storage_type == "sqlite": + conn = sqlite3.connect(self.db_path) + if brand and platform: + df = pd.read_sql_query( + "SELECT * FROM articles WHERE brand = ? AND platform = ?", + conn, params=(brand, platform) + ) + elif brand: + df = pd.read_sql_query( + "SELECT * FROM articles WHERE brand = ?", + conn, params=(brand,) + ) + else: + df = pd.read_sql_query("SELECT * FROM articles", conn) + conn.close() + return df.to_dict('records') + else: + json_file = Path(self.db_path) / "articles.json" + if not json_file.exists(): + return [] + + with open(json_file, 'r', encoding='utf-8') as f: + data = json.load(f) + + if brand and platform: + return [item for item in data + if item.get("brand") == brand and item.get("platform") == platform] + elif brand: + return [item for item in data if item.get("brand") == brand] + return data + + # ==================== 优化记录相关 ==================== + + def save_optimization(self, original_content: str, optimized_content: str, + changes: str, platform: str, brand: str): + """保存优化记录""" + if self.storage_type == "sqlite": + conn = sqlite3.connect(self.db_path) + cursor = conn.cursor() + cursor.execute(""" + INSERT INTO optimizations + (original_content, optimized_content, changes, platform, brand) + VALUES (?, ?, ?, ?, ?) + """, (original_content, optimized_content, changes, platform, brand)) + conn.commit() + conn.close() + else: + json_file = Path(self.db_path) / "optimizations.json" + data = [] + if json_file.exists(): + with open(json_file, 'r', encoding='utf-8') as f: + data = json.load(f) + + data.append({ + "original_content": original_content, + "optimized_content": optimized_content, + "changes": changes, + "platform": platform, + "brand": brand, + "created_at": datetime.now().isoformat() + }) + + with open(json_file, 'w', encoding='utf-8') as f: + json.dump(data, f, ensure_ascii=False, indent=2) + + def get_optimizations(self, brand: Optional[str] = None) -> List[Dict]: + """获取优化记录""" + if self.storage_type == "sqlite": + conn = sqlite3.connect(self.db_path) + if brand: + df = pd.read_sql_query( + "SELECT * FROM optimizations WHERE brand = ? ORDER BY created_at DESC", + conn, params=(brand,) + ) + else: + df = pd.read_sql_query( + "SELECT * FROM optimizations ORDER BY created_at DESC", + conn + ) + conn.close() + return df.to_dict('records') + else: + json_file = Path(self.db_path) / "optimizations.json" + if not json_file.exists(): + return [] + + with open(json_file, 'r', encoding='utf-8') as f: + data = json.load(f) + + if brand: + return [item for item in data if item.get("brand") == brand] + return data + + # ==================== 验证结果相关 ==================== + + def save_verify_results(self, results: List[Dict]): + """批量保存验证结果""" + if self.storage_type == "sqlite": + conn = sqlite3.connect(self.db_path) + cursor = conn.cursor() + for result in results: + cursor.execute(""" + INSERT INTO verify_results + (query, brand, verify_model, mention_count, mention_position) + VALUES (?, ?, ?, ?, ?) + """, ( + result.get("问题"), + result.get("品牌"), + result.get("验证模型"), + result.get("提及次数"), + result.get("位置") + )) + conn.commit() + conn.close() + else: + json_file = Path(self.db_path) / "verify_results.json" + data = [] + if json_file.exists(): + with open(json_file, 'r', encoding='utf-8') as f: + data = json.load(f) + + for result in results: + data.append({ + "query": result.get("问题"), + "brand": result.get("品牌"), + "verify_model": result.get("验证模型"), + "mention_count": result.get("提及次数"), + "mention_position": result.get("位置"), + "created_at": datetime.now().isoformat() + }) + + with open(json_file, 'w', encoding='utf-8') as f: + json.dump(data, f, ensure_ascii=False, indent=2) + + def get_verify_results(self, brand: Optional[str] = None, include_timestamp: bool = False) -> pd.DataFrame: + """获取验证结果（返回DataFrame） + + Args: + brand: 品牌名称，如果为None则返回所有品牌 + include_timestamp: 是否包含时间戳字段 + """ + if self.storage_type == "sqlite": + conn = sqlite3.connect(self.db_path) + if include_timestamp: + if brand: + df = pd.read_sql_query( + """SELECT query as "问题", brand as "品牌", verify_model as "验证模型", + mention_count as "提及次数", mention_position as "位置", + created_at as "验证时间" + FROM verify_results WHERE brand = ? ORDER BY created_at DESC""", + conn, params=(brand,) + ) + else: + df = pd.read_sql_query( + """SELECT query as "问题", brand as "品牌", verify_model as "验证模型", + mention_count as "提及次数", mention_position as "位置", + created_at as "验证时间" + FROM verify_results ORDER BY created_at DESC""", + conn + ) + else: + if brand: + df = pd.read_sql_query( + """SELECT query as "问题", brand as "品牌", verify_model as "验证模型", + mention_count as "提及次数", mention_position as "位置" + FROM verify_results WHERE brand = ?""", + conn, params=(brand,) + ) + else: + df = pd.read_sql_query( + """SELECT query as "问题", brand as "品牌", verify_model as "验证模型", + mention_count as "提及次数", mention_position as "位置" + FROM verify_results""", + conn + ) + conn.close() + if include_timestamp and not df.empty and "验证时间" in df.columns: + df["验证时间"] = pd.to_datetime(df["验证时间"]) + return df + else: + json_file = Path(self.db_path) / "verify_results.json" + if not json_file.exists(): + return pd.DataFrame() + + with open(json_file, 'r', encoding='utf-8') as f: + data = json.load(f) + + if brand: + data = [item for item in data if item.get("brand") == brand] + + # 转换为DataFrame格式 + records = [] + for item in data: + record = { + "问题": item.get("query"), + "品牌": item.get("brand"), + "验证模型": item.get("verify_model"), + "提及次数": item.get("mention_count"), + "位置": item.get("mention_position") + } + if include_timestamp and "created_at" in item: + record["验证时间"] = pd.to_datetime(item.get("created_at")) + records.append(record) + + df = pd.DataFrame(records) + if include_timestamp and not df.empty and "验证时间" in df.columns: + df = df.sort_values("验证时间", ascending=False) + return df + + # ==================== 统计功能 ==================== + + def get_stats(self, brand: Optional[str] = None) -> Dict[str, Any]: + """获取统计数据""" + stats = {} + + if self.storage_type == "sqlite": + conn = sqlite3.connect(self.db_path) + cursor = conn.cursor() + + # 关键词数量 + if brand: + cursor.execute("SELECT COUNT(*) FROM keywords WHERE brand = ?", (brand,)) + else: + cursor.execute("SELECT COUNT(*) FROM keywords") + stats["keywords_count"] = cursor.fetchone()[0] + + # 文章数量 + if brand: + cursor.execute("SELECT COUNT(*) FROM articles WHERE brand = ?", (brand,)) + else: + cursor.execute("SELECT COUNT(*) FROM articles") + stats["articles_count"] = cursor.fetchone()[0] + + # 优化记录数量 + if brand: + cursor.execute("SELECT COUNT(*) FROM optimizations WHERE brand = ?", (brand,)) + else: + cursor.execute("SELECT COUNT(*) FROM optimizations") + stats["optimizations_count"] = cursor.fetchone()[0] + + # 验证结果数量 + if brand: + cursor.execute("SELECT COUNT(*) FROM verify_results WHERE brand = ?", (brand,)) + else: + cursor.execute("SELECT COUNT(*) FROM verify_results") + stats["verify_results_count"] = cursor.fetchone()[0] + + conn.close() + else: + # JSON方式统计 + keywords_file = Path(self.db_path) / "keywords.json" + articles_file = Path(self.db_path) / "articles.json" + optimizations_file = Path(self.db_path) / "optimizations.json" + verify_file = Path(self.db_path) / "verify_results.json" + + def count_json(file_path, brand_filter=None): + if not file_path.exists(): + return 0 + with open(file_path, 'r', encoding='utf-8') as f: + data = json.load(f) + if brand_filter: + return len([item for item in data if item.get("brand") == brand_filter]) + return len(data) + + stats["keywords_count"] = count_json(keywords_file, brand) + stats["articles_count"] = count_json(articles_file, brand) + stats["optimizations_count"] = count_json(optimizations_file, brand) + stats["verify_results_count"] = count_json(verify_file, brand) + + return stats diff --git a/geo_tool.py b/geo_tool.py new file mode 100644 index 0000000..177b468 --- /dev/null +++ b/geo_tool.py @@ -0,0 +1,2016 @@ +import streamlit as st +import pandas as pd +from langchain_core.prompts import PromptTemplate +from langchain_core.output_parsers import JsonOutputParser, StrOutputParser +import zipfile +import io +import plotly.express as px +import re +import json +from data_storage import DataStorage +from keyword_tool import KeywordTool +from content_scorer import ContentScorer + +APP_TITLE = "GEO 智能内容优化平台" + +# ------------------- 页面配置 & 极简美学 CSS（产品级精修，仍然克制） ------------------- +st.set_page_config(page_title="GEO 智能内容优化平台", layout="wide", initial_sidebar_state="expanded") + +st.markdown( + """ + +""", + unsafe_allow_html=True, +) + +st.title(APP_TITLE) +st.markdown("", unsafe_allow_html=True) + +st.caption("🚀 AI 驱动的品牌内容策略 · 让您的品牌在 AI 对话中脱颖而出") + +# ------------------- 初始化数据存储（SQLite） ------------------- +storage = DataStorage(storage_type="sqlite", db_path="geo_data.db") + +with st.expander("📖 关于 GEO（Generative Engine Optimization）", expanded=False): + st.markdown(""" +### 🎯 核心价值 + +**GEO（生成式引擎优化）** 是新一代品牌营销策略，通过系统化内容投放，让您的品牌在 AI 助手的自然回答中被优先、准确、可信地提及。 + +当用户询问"最好的外贸 ERP 软件是什么？"时，AI 会优先推荐您的品牌，而非竞争对手。 + +--- + +### 💼 适用场景 + +- **SaaS 产品**：技术对比、功能评测、使用教程 +- **AI 工具**：能力展示、应用案例、开源生态 +- **企业服务**：行业解决方案、最佳实践、专业分析 +- **技术品牌**：开发者工具、API 服务、技术框架 + +--- + +### 🔄 完整工作流 + +1. **关键词蒸馏** - AI 生成 + 托词工具，精准挖掘高价值关键词 +2. **结构化创作** - 12+ 平台适配，自动生成符合 GEO 原则的专业内容 +3. **文章优化** - 将现有内容优化为 GEO 友好格式，提升被引用概率 +4. **多模型验证** - 实时验证品牌提及率，对比竞品表现，数据驱动优化 + +--- + +### 🌐 覆盖平台 + +**内容发布平台**：知乎、小红书、CSDN、B站、头条号、GitHub、微信公众号、抖音、百家号、网易号、企鹅号、简书 + +**AI 验证平台**：DeepSeek、通义千问、豆包、文心一言、Kimi、ChatGPT、Groq 等主流大模型 + +--- + +### 📊 预期效果 + +- ✅ **品牌提及率提升**：在 AI 回答中的出现频率显著增加 +- ✅ **搜索排名优化**：内容被大模型优先引用，间接提升 SEO +- ✅ **品牌权威性**：多平台、多角度内容建立专业形象 +- ✅ **竞品优势**：通过数据对比，发现并强化差异化优势 +""") + +# ------------------- Session State：持久化每个阶段产物（解决“消失”） ------------------- +def ss_init(key, default): + if key not in st.session_state: + st.session_state[key] = default + + +ss_init( + "cfg", + { + "gen_provider": "DeepSeek", + "gen_api_key": "sk-a95eda59dd494ab3b56197cc0020e61d", + "verify_providers": ["DeepSeek"], + "verify_keys": {"DeepSeek": "sk-a95eda59dd494ab3b56197cc0020e61d"}, + "brand": "汇信云AI软件", + "advantages": "AI赋能外贸ERP、打造外贸智能新引擎、AI驱动型ERP、赋能外贸全流程管理、全链路价值闭环", + "competitors": "南北软件\n睿贝软件\n孚盟软件\n小满软件", + "temperature": 0.7, + }, +) +ss_init("cfg_applied", False) +ss_init("cfg_valid", False) +ss_init("cfg_errors", []) + +# 模块1：关键词 +ss_init("keywords", []) +ss_init("kw_last_num", 40) +ss_init("kw_generation_mode", "AI生成") # 生成模式：AI生成 / 托词工具 / 混合模式 +ss_init("wordbanks", None) # 词库字典 +ss_init("keyword_tool", KeywordTool()) # 托词工具实例 + +# 模块2：内容 +ss_init("generated_contents", []) # list[dict] +ss_init("zip_bytes", None) +ss_init("zip_filename", "") + +# 模块3：文章优化 +ss_init("optimized_article", "") +ss_init("opt_changes", "") +ss_init("opt_platform", "通用优化") + +# 模块4：验证 +ss_init("verify_combined", None) # DataFrame or None +ss_init("verify_last_queries", "") + +# ------------------- 工具函数 ------------------- +INVALID_FS_CHARS = r'<>:"/\\|?*\n\r\t' + + +def sanitize_filename(name: str, max_len: int = 80) -> str: + if not name: + return "untitled" + name = name.strip() + name = re.sub(rf"[{re.escape(INVALID_FS_CHARS)}]", "_", name) + name = re.sub(r"_+", "_", name).strip("_") + return name[:max_len] if len(name) > max_len else name + + +def safe_decode_uploaded(uploaded) -> str: + if not uploaded: + return "" + b = uploaded.getvalue() + for enc in ("utf-8-sig", "utf-8", "gb18030"): + try: + return b.decode(enc) + except Exception: + pass + return b.decode("utf-8", errors="replace") + + +def extract_json_array(text: str): + """从模型输出中抽取 JSON 数组（JsonOutputParser 失败时兜底）。""" + if not text: + return None + m = re.search(r"\[[\s\S]*\]", text) + if not m: + return None + try: + return json.loads(m.group(0)) + except Exception: + return None + + +def validate_cfg(cfg: dict): + """保留你原本的“必须填写所有 API Key”约束，但不 st.stop：改为禁用按钮 + 提示。""" + errors = [] + if not cfg.get("gen_api_key", "").strip(): + errors.append("生成&优化 LLM 的 API Key 未填写") + + verify_providers = cfg.get("verify_providers", []) + verify_keys = cfg.get("verify_keys", {}) + if not verify_providers: + errors.append("至少选择一个验证模型") + + for vp in verify_providers: + if not verify_keys.get(vp, "").strip(): + errors.append(f"验证模型 {vp} 的 API Key 未填写") + + return (len(errors) == 0), errors + + +def model_defaults(provider: str) -> str: + if provider == "DeepSeek": + return "deepseek-chat" + if provider == "OpenAI (GPT)": + return "gpt-4o-mini" + if provider == "Tongyi (通义千问)": + return "qwen-max" + if provider == "Groq": + return "llama3-70b-8192" + if provider == "Moonshot (Kimi)": + return "moonshot-v1-128k" + if provider == "豆包（字节跳动）": + return "" # 豆包使用 ENDPOINT_ID，不需要模型名 + if provider == "文心一言（百度）": + return "ernie-bot-turbo" + return "" + + +# ------------------- 缓存 LLM 客户端（显著降低“频繁 Loading”） ------------------- +@st.cache_resource(show_spinner=False) +def build_llm(provider: str, api_key: str, model: str, temperature: float): + """ + - 使用 cache_resource 缓存客户端，避免每次 rerun 重建 + - Tongyi / Moonshot：保留你原功能路径，同时提供更稳的 import 兜底 + """ + if provider == "DeepSeek": + from langchain_deepseek import ChatDeepSeek + + return ChatDeepSeek(api_key=api_key, model=model, temperature=temperature) + + if provider == "OpenAI (GPT)": + from langchain_openai import ChatOpenAI + + return ChatOpenAI(api_key=api_key, model=model, temperature=temperature) + + if provider == "Tongyi (通义千问)": + try: + from langchain_community.chat_models import ChatTongyi + + return ChatTongyi(api_key=api_key, model=model, model_kwargs={"temperature": temperature}) + except Exception: + from langchain_aliyun import ChatTongyi # type: ignore + + return ChatTongyi(api_key=api_key, model=model, temperature=temperature) + + if provider == "Groq": + from langchain_groq import ChatGroq + + return ChatGroq(api_key=api_key, model=model, temperature=temperature) + + if provider == "Moonshot (Kimi)": + try: + from langchain_moonshot import ChatMoonshot # type: ignore + + return ChatMoonshot(api_key=api_key, model=model, temperature=temperature) + except Exception: + from langchain_community.chat_models import MoonshotChat # type: ignore + + return MoonshotChat(api_key=api_key, model=model, temperature=temperature) + + if provider == "豆包（字节跳动）": + try: + # 尝试使用 volcengine-python-sdk[ark] + from volcengine.ark import Ark + from langchain_core.language_models.chat_models import BaseChatModel + from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessage + from langchain_core.outputs import ChatGeneration, ChatResult + from typing import List, Optional, Any + + class ChatDoubao(BaseChatModel): + """豆包聊天模型封装（LangChain 兼容）""" + volc_ak: str + volc_sk: str + endpoint_id: str + temperature: float = 0.7 + + def __init__(self, volc_ak: str, volc_sk: str, endpoint_id: str, temperature: float = 0.7): + super().__init__(temperature=temperature) + self.volc_ak = volc_ak + self.volc_sk = volc_sk + self.endpoint_id = endpoint_id + self.temperature = temperature + self.client = Ark(ak=volc_ak, sk=volc_sk) + + def _generate(self, messages: List[BaseMessage], stop: Optional[List[str]] = None, run_manager: Optional[Any] = None, **kwargs: Any) -> ChatResult: + # 转换消息格式 + volc_messages = [] + for msg in messages: + if isinstance(msg, SystemMessage): + volc_messages.append({"role": "system", "content": msg.content}) + elif isinstance(msg, HumanMessage): + volc_messages.append({"role": "user", "content": msg.content}) + elif isinstance(msg, AIMessage): + volc_messages.append({"role": "assistant", "content": msg.content}) + else: + volc_messages.append({"role": "user", "content": str(msg.content)}) + + response = self.client.chat.completions.create( + model=self.endpoint_id, + messages=volc_messages, + temperature=self.temperature, + ) + + ai_message = AIMessage(content=response.choices[0].message.content) + return ChatResult(generations=[ChatGeneration(message=ai_message)]) + + @property + def _llm_type(self) -> str: + return "doubao" + + # 豆包的 api_key 格式：access_key:secret_key:endpoint_id + parts = api_key.split(":") + if len(parts) >= 3: + return ChatDoubao(volc_ak=parts[0], volc_sk=parts[1], endpoint_id=parts[2], temperature=temperature) + else: + raise ValueError("豆包 API Key 格式错误，应为：access_key:secret_key:endpoint_id（用冒号分隔）") + except ImportError: + # 尝试其他导入方式 + try: + from volcenginesdkarkruntime import Ark + # 使用相同的 ChatDoubao 类 + from langchain_core.language_models.chat_models import BaseChatModel + from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessage + from langchain_core.outputs import ChatGeneration, ChatResult + from typing import List, Optional, Any + + class ChatDoubao(BaseChatModel): + """豆包聊天模型封装（LangChain 兼容）""" + volc_ak: str + volc_sk: str + endpoint_id: str + temperature: float = 0.7 + + def __init__(self, volc_ak: str, volc_sk: str, endpoint_id: str, temperature: float = 0.7): + super().__init__(temperature=temperature) + self.volc_ak = volc_ak + self.volc_sk = volc_sk + self.endpoint_id = endpoint_id + self.temperature = temperature + self.client = Ark(ak=volc_ak, sk=volc_sk) + + def _generate(self, messages: List[BaseMessage], stop: Optional[List[str]] = None, run_manager: Optional[Any] = None, **kwargs: Any) -> ChatResult: + volc_messages = [] + for msg in messages: + if isinstance(msg, SystemMessage): + volc_messages.append({"role": "system", "content": msg.content}) + elif isinstance(msg, HumanMessage): + volc_messages.append({"role": "user", "content": msg.content}) + elif isinstance(msg, AIMessage): + volc_messages.append({"role": "assistant", "content": msg.content}) + else: + volc_messages.append({"role": "user", "content": str(msg.content)}) + + response = self.client.chat.completions.create( + model=self.endpoint_id, + messages=volc_messages, + temperature=self.temperature, + ) + + ai_message = AIMessage(content=response.choices[0].message.content) + return ChatResult(generations=[ChatGeneration(message=ai_message)]) + + @property + def _llm_type(self) -> str: + return "doubao" + + parts = api_key.split(":") + if len(parts) >= 3: + return ChatDoubao(volc_ak=parts[0], volc_sk=parts[1], endpoint_id=parts[2], temperature=temperature) + else: + raise ValueError("豆包 API Key 格式错误，应为：access_key:secret_key:endpoint_id（用冒号分隔）") + except ImportError as e: + raise ValueError(f"豆包初始化失败：缺少依赖库。请运行：pip install 'volcengine-python-sdk[ark]'。错误：{e}") + except Exception as e: + raise ValueError(f"豆包初始化失败：{e}。请确保 API Key 格式为：access_key:secret_key:endpoint_id") + + if provider == "文心一言（百度）": + # 文心一言的 api_key 格式：app_key:app_secret + parts = api_key.split(":") + if len(parts) != 2: + raise ValueError("文心一言 API Key 格式错误，应为：app_key:app_secret（用冒号分隔）") + + app_key, app_secret = parts + + # 优先使用 langchain-community 的千帆接口（已包含在依赖中） + try: + from langchain_community.chat_models import QianfanChatEndpoint + import os + + os.environ["QIANFAN_AK"] = app_key + os.environ["QIANFAN_SK"] = app_secret + return QianfanChatEndpoint( + model=model if model else "ernie-bot-turbo", + temperature=temperature, + ) + except ImportError: + # 备选方案：尝试 langchain-wenxin + try: + from langchain_wenxin import ChatWenxin + return ChatWenxin( + baidu_api_key=app_key, + baidu_secret_key=app_secret, + model=model if model else "ernie-bot-turbo", + temperature=temperature, + ) + except ImportError as e: + raise ValueError(f"文心一言初始化失败：缺少依赖库。请运行：pip install qianfan（或使用已安装的 langchain-community）。错误：{e}") + except Exception as e: + raise ValueError(f"文心一言初始化失败：{e}") + + raise ValueError(f"Unknown provider: {provider}") + + +# ------------------- 侧边栏：全局配置（用 form 降低 rerun） ------------------- +with st.sidebar: + st.header("全局配置") + + with st.form("global_config_form", clear_on_submit=False): + gen_provider = st.selectbox( + "生成&优化 LLM", + ["DeepSeek", "OpenAI (GPT)", "Tongyi (通义千问)", "Groq", "Moonshot (Kimi)", "豆包（字节跳动）", "文心一言（百度）"], + index=["DeepSeek", "OpenAI (GPT)", "Tongyi (通义千问)", "Groq", "Moonshot (Kimi)", "豆包（字节跳动）", "文心一言（百度）"].index( + st.session_state.cfg["gen_provider"] + ) if st.session_state.cfg["gen_provider"] in ["DeepSeek", "OpenAI (GPT)", "Tongyi (通义千问)", "Groq", "Moonshot (Kimi)", "豆包（字节跳动）", "文心一言（百度）"] else 0, + key="sb_gen_provider", + ) + # API Key 输入提示 + if gen_provider == "豆包（字节跳动）": + api_key_help = "格式：access_key:secret_key:endpoint_id（用冒号分隔）" + elif gen_provider == "文心一言（百度）": + api_key_help = "格式：app_key:app_secret（用冒号分隔）" + else: + api_key_help = "" + + gen_api_key = st.text_input( + f"{gen_provider} API Key（生成&优化用）", + type="password", + value=st.session_state.cfg.get("gen_api_key", ""), + key="sb_gen_api_key", + help=api_key_help if api_key_help else None, + ) + + st.markdown("### 验证用LLM（多选）") + verify_providers = st.multiselect( + "选择验证模型", + ["DeepSeek", "OpenAI (GPT)", "Tongyi (通义千问)", "Groq", "Moonshot (Kimi)", "豆包（字节跳动）", "文心一言（百度）"], + default=st.session_state.cfg.get("verify_providers", []), + key="sb_verify_providers", + ) + + verify_keys = {} + old_keys = st.session_state.cfg.get("verify_keys", {}) + for vp in verify_providers: + # API Key 输入提示 + if vp == "豆包（字节跳动）": + api_key_help = "格式：access_key:secret_key:endpoint_id（用冒号分隔）" + elif vp == "文心一言（百度）": + api_key_help = "格式：app_key:app_secret（用冒号分隔）" + else: + api_key_help = None + + verify_keys[vp] = st.text_input( + f"{vp} API Key（验证用）", + type="password", + value=old_keys.get(vp, ""), + key=f"sb_verify_key_{vp}", + help=api_key_help if api_key_help else None, + ) + + st.markdown("---") + brand = st.text_input("主品牌名称", value=st.session_state.cfg.get("brand", "汇信云AI软件"), key="sb_brand") + advantages = st.text_area( + "核心优势/卖点（AI专属）", + value=st.session_state.cfg.get( + "advantages", "AI赋能外贸ERP、打造外贸智能新引擎、AI驱动型ERP、赋能外贸全流程管理、全链路价值闭环" + ), + height=140, + key="sb_advantages", + ) + competitors = st.text_area( + "竞品品牌（每行一个，用于对比验证）", + value=st.session_state.cfg.get("competitors", "南北软件\n睿贝软件\n孚盟软件\n小满软件"), + height=120, + key="sb_competitors", + ) + + st.markdown("---") + temperature = st.slider( + "生成温度（更稳→更低）", + 0.0, + 1.0, + float(st.session_state.cfg.get("temperature", 0.7)), + 0.05, + key="sb_temperature", + ) + + apply_cfg = st.form_submit_button("应用配置（推荐）", use_container_width=True) + + if apply_cfg or not st.session_state.cfg_applied: + st.session_state.cfg = { + "gen_provider": gen_provider, + "gen_api_key": gen_api_key, + "verify_providers": verify_providers, + "verify_keys": verify_keys, + "brand": brand, + "advantages": advantages, + "competitors": competitors, + "temperature": temperature, + } + st.session_state.cfg_applied = True + + ok, errs = validate_cfg(st.session_state.cfg) + st.session_state.cfg_valid = ok + st.session_state.cfg_errors = errs + + if not st.session_state.cfg_valid: + st.warning("配置未满足运行条件：\n- " + "\n- ".join(st.session_state.cfg_errors)) + else: + st.success("配置已就绪，可运行全部模块。") + + st.markdown("---") + if st.button("重置全部结果（不删除配置）", use_container_width=True, key="sb_reset_all"): + st.session_state.keywords = [] + st.session_state.generated_contents = [] + st.session_state.zip_bytes = None + st.session_state.zip_filename = "" + st.session_state.optimized_article = "" + st.session_state.opt_changes = "" + st.session_state.verify_combined = None + st.toast("已重置全部结果。") + + st.caption("闭环：关键词 → 创作 → 优化 → 验证") + +cfg = st.session_state.cfg +brand = cfg["brand"] +advantages = cfg["advantages"] +temperature = float(cfg.get("temperature", 0.7)) + +competitor_list = [c.strip() for c in cfg["competitors"].split("\n") if c.strip()] +_seen = set() +clean_competitors = [] +for c in competitor_list: + cl = c.lower() + if cl == brand.lower(): + continue + if cl in _seen: + continue + _seen.add(cl) + clean_competitors.append(c) +competitor_list = clean_competitors + +# ------------------- 初始化 LLM（仅在 cfg_valid 时；且 build_llm 已缓存） ------------------- +gen_llm = None +verify_llms = {} + +if st.session_state.cfg_valid: + try: + gen_llm = build_llm(cfg["gen_provider"], cfg["gen_api_key"], model_defaults(cfg["gen_provider"]), temperature) + except Exception as e: + st.error(f"生成LLM加载失败：{e}") + + for vp in cfg["verify_providers"]: + key = cfg["verify_keys"].get(vp, "").strip() + if not key: + continue + try: + verify_llms[vp] = build_llm(vp, key, model_defaults(vp), temperature) + except Exception as e: + st.error(f"{vp}验证LLM加载失败：{e}") + +# ------------------- KPI 总览（极简但更像产品） ------------------- +k1, k2, k3, k4 = st.columns(4) +try: + k1.metric("关键词", len(st.session_state.keywords), border=True) + k2.metric("内容包", len(st.session_state.generated_contents), border=True) + k3.metric("文章优化", "已生成" if bool(st.session_state.optimized_article) else "未生成", border=True) + k4.metric("验证结果", "已生成" if st.session_state.verify_combined is not None else "未生成", border=True) +except TypeError: + k1.metric("关键词", len(st.session_state.keywords)) + k2.metric("内容包", len(st.session_state.generated_contents)) + k3.metric("文章优化", "已生成" if bool(st.session_state.optimized_article) else "未生成") + k4.metric("验证结果", "已生成" if st.session_state.verify_combined is not None else "未生成") + +st.markdown("---") + +# ------------------- 主导航：Tabs（流程更清晰） ------------------- +tab1, tab2, tab3, tab4, tab5, tab6 = st.tabs(["1 关键词蒸馏", "2 自动创作", "3 文章优化", "4 多模型验证", "5 历史记录", "6 AI 数据报表"]) + +# ======================= +# Tab1：关键词蒸馏 +# ======================= +with tab1: + # 生成模式选择 + generation_mode = st.radio( + "生成模式", + ["AI生成", "托词工具", "混合模式"], + index=["AI生成", "托词工具", "混合模式"].index(st.session_state.kw_generation_mode), + horizontal=True, + key="kw_mode_radio" + ) + st.session_state.kw_generation_mode = generation_mode + + # 词库管理和组合模式选择（托词工具和混合模式需要） + if generation_mode in ["托词工具", "混合模式"]: + # 初始化词库 + if st.session_state.wordbanks is None: + st.session_state.wordbanks = st.session_state.keyword_tool.load_wordbanks() + + # 初始化组合模式选择 + ss_init("selected_patterns", list(st.session_state.keyword_tool.combination_patterns)) + + wordbanks = st.session_state.wordbanks + + # 组合模式选择 + with st.container(border=True): + st.markdown("**组合模式选择**") + pattern_descriptions = st.session_state.keyword_tool.get_pattern_descriptions() + all_patterns = st.session_state.keyword_tool.combination_patterns + + # 显示所有可用模式 + pattern_options = [] + for pattern in all_patterns: + pattern_str = "+".join(pattern) + desc = pattern_descriptions.get(pattern_str, pattern_str) + pattern_options.append((pattern_str, pattern, desc)) + + # 多选组合模式 + selected_pattern_strs = st.multiselect( + "选择要使用的组合模式（可多选）", + options=[opt[0] for opt in pattern_options], + default=[opt[0] for opt in pattern_options if opt[1] in st.session_state.selected_patterns], + key="kw_pattern_select", + help="选择要使用的组合模式，至少选择一个" + ) + + # 更新选中的模式 + selected_patterns = [] + for pattern_str, pattern, desc in pattern_options: + if pattern_str in selected_pattern_strs: + selected_patterns.append(pattern) + st.session_state.selected_patterns = selected_patterns if selected_patterns else all_patterns + + # 显示模式说明 + with st.expander("组合模式说明", expanded=False): + for pattern_str, pattern, desc in pattern_options: + st.markdown(f"**{pattern_str}**: {' + '.join(desc)}") + + # 词库管理 + with st.expander("词库管理", expanded=False): + # 词库编辑 + col1, col2 = st.columns([1, 1]) + with col1: + st.markdown("**词库编辑**") + bank_types = list(wordbanks.keys()) + selected_bank = st.selectbox("选择词库类型", bank_types, key="kw_bank_select") + + # 显示当前词库内容 + current_words = wordbanks[selected_bank] + edited_words = st.text_area( + f"{selected_bank} 词汇（每行一个）", + "\n".join(current_words), + height=150, + key=f"kw_bank_edit_{selected_bank}" + ) + + if st.button("更新词库", key=f"kw_update_{selected_bank}"): + new_words = [w.strip() for w in edited_words.split("\n") if w.strip()] + wordbanks[selected_bank] = new_words + st.session_state.wordbanks = wordbanks + st.success(f"{selected_bank} 已更新（{len(new_words)} 个词汇）") + + with col2: + st.markdown("**词库导入/导出**") + # 导出 + wordbanks_json = json.dumps(wordbanks, ensure_ascii=False, indent=2) + st.download_button( + "导出词库（JSON）", + wordbanks_json, + "wordbanks.json", + "application/json", + use_container_width=True, + key="kw_export_json" + ) + + # 导入 + uploaded_wordbanks = st.file_uploader( + "导入词库（JSON）", + type=["json"], + key="kw_import_json" + ) + if uploaded_wordbanks: + try: + imported = json.loads(uploaded_wordbanks.read().decode('utf-8')) + if isinstance(imported, dict): + st.session_state.wordbanks = imported + st.success("词库导入成功！") + st.rerun() + except Exception as e: + st.error(f"导入失败：{e}") + + # 重置为默认词库 + if st.button("重置为默认词库", use_container_width=True, key="kw_reset_banks"): + st.session_state.wordbanks = st.session_state.keyword_tool.load_wordbanks() + st.success("已重置为默认词库") + st.rerun() + + # 生成控制 + with st.container(border=True): + c1, c2, c3 = st.columns([2, 1, 1]) + with c1: + st.session_state.kw_last_num = st.slider( + "生成数量", 10, 100, st.session_state.kw_last_num, key="kw_num" + ) + with c2: + # 根据模式调整禁用条件 + if generation_mode == "托词工具": + run_kw_disabled = False # 托词工具不需要 LLM + else: + run_kw_disabled = (not st.session_state.cfg_valid) or (gen_llm is None) + + run_kw = st.button( + "生成关键词", + type="primary", + use_container_width=True, + disabled=run_kw_disabled, + key="kw_run", + ) + with c3: + if st.button("清空本模块结果", use_container_width=True, key="kw_clear"): + st.session_state.keywords = [] + st.toast("关键词已清空。") + + if run_kw: + keywords = [] + + if generation_mode == "AI生成": + # 原有 AI 生成逻辑 + keyword_prompt = PromptTemplate.from_template( + """ +你是AI领域GEO专家，目标是提升品牌在大模型自然回答中的提及率。 + +【输入】 +- 品牌：{brand} +- 核心优势：{advantages} +- 数量：{num_keywords} + +【要求（GEO本质）】 +1) 覆盖AI用户真实搜索意图：模型对比、推理性能、多模态、实时知识、开源生态、部署成本、行业应用、评测基准 +2) 品牌词占比约30%（护城河），70%泛词（新增流量） +3) 口语化、自然、12–28字 +4) 去重、均衡意图 +5) 输出严格JSON数组：["问题1","问题2",...] + +【开始输出JSON数组】 +""" + ) + + chain_json = keyword_prompt | gen_llm | JsonOutputParser() + chain_text = keyword_prompt | gen_llm | StrOutputParser() + + with st.spinner("AI生成中..."): + try: + result = chain_json.invoke( + {"brand": brand, "advantages": advantages, "num_keywords": st.session_state.kw_last_num} + ) + keywords = result if isinstance(result, list) else [] + except Exception: + raw = chain_text.invoke( + {"brand": brand, "advantages": advantages, "num_keywords": st.session_state.kw_last_num} + ) + keywords = extract_json_array(raw) or [] + + elif generation_mode == "托词工具": + # 托词工具生成 + with st.spinner("组合生成中..."): + wordbanks = st.session_state.wordbanks or st.session_state.keyword_tool.load_wordbanks() + selected_patterns = st.session_state.get("selected_patterns", st.session_state.keyword_tool.combination_patterns) + + # 检查词库是否为空 + empty_banks = [k for k, v in wordbanks.items() if not v] + if empty_banks: + st.warning(f"以下词库为空，请先添加词汇：{', '.join(empty_banks)}") + + keywords = st.session_state.keyword_tool.generate_combinations( + wordbanks=wordbanks, + patterns=selected_patterns, + max_results=st.session_state.kw_last_num, + similarity_threshold=0.8 + ) + + elif generation_mode == "混合模式": + # 混合模式：先托词生成，再 LLM 润色 + with st.spinner("托词生成中..."): + wordbanks = st.session_state.wordbanks or st.session_state.keyword_tool.load_wordbanks() + selected_patterns = st.session_state.get("selected_patterns", st.session_state.keyword_tool.combination_patterns) + + # 检查词库是否为空 + empty_banks = [k for k, v in wordbanks.items() if not v] + if empty_banks: + st.warning(f"以下词库为空，请先添加词汇：{', '.join(empty_banks)}") + + raw_keywords = st.session_state.keyword_tool.generate_combinations( + wordbanks=wordbanks, + patterns=selected_patterns, + max_results=st.session_state.kw_last_num * 2, # 生成更多，因为会去重 + similarity_threshold=0.8 + ) + + if raw_keywords and gen_llm: + with st.spinner("LLM 润色中..."): + # 使用 LLM 润色 + from langchain_core.prompts import PromptTemplate as PT + polish_template = PT.from_template("{input}") + polish_chain = polish_template | gen_llm | StrOutputParser() + keywords = st.session_state.keyword_tool.polish_with_llm( + keywords=raw_keywords, + llm_chain=polish_chain, + brand=brand, + max_polish=min(len(raw_keywords), st.session_state.kw_last_num) + ) + else: + keywords = raw_keywords + + # 清理和去重 + cleaned, seen = [], set() + for k in keywords: + if not isinstance(k, str): + continue + kk = k.strip() + if not kk: + continue + kl = kk.lower() + if kl in seen: + continue + seen.add(kl) + cleaned.append(kk) + + # 限制数量 + cleaned = cleaned[:st.session_state.kw_last_num] + + if cleaned: + st.session_state.keywords = cleaned + # 保存到数据库 + try: + storage.save_keywords(cleaned, brand) + except Exception as e: + st.warning(f"关键词已生成，但保存到数据库时出错：{e}") + st.success(f"生成完成（{len(cleaned)} 条）") + else: + error_msg = "生成失败，可能的原因：\n" + if generation_mode in ["托词工具", "混合模式"]: + wordbanks = st.session_state.wordbanks or st.session_state.keyword_tool.load_wordbanks() + empty_banks = [k for k, v in wordbanks.items() if not v] + if empty_banks: + error_msg += f"- 以下词库为空：{', '.join(empty_banks)}\n" + if not st.session_state.get("selected_patterns"): + error_msg += "- 未选择任何组合模式\n" + error_msg += "- 请检查词库配置或选择更多组合模式" + else: + error_msg += "- 请检查 API Key 配置或重试" + st.error(error_msg) + + if st.session_state.keywords: + df = pd.DataFrame(st.session_state.keywords, columns=["长尾关键词/问题"]) + st.dataframe(df, use_container_width=True, hide_index=True) + st.download_button( + "下载关键词CSV", + df.to_csv(index=False, encoding="utf-8-sig"), + f"{sanitize_filename(brand,40)}_keywords.csv", + mime="text/csv", + use_container_width=True, + key="kw_dl_csv", + ) + else: + st.info("在左侧完成配置后，点击“生成关键词”。") + +# ======================= +# Tab2：自动创作内容（含批量 ZIP / GitHub 模板） +# ======================= +with tab2: + top_l, top_r = st.columns([3, 1]) + with top_r: + if st.button("清空本模块结果", use_container_width=True, key="content_clear"): + st.session_state.generated_contents = [] + st.session_state.zip_bytes = None + st.session_state.zip_filename = "" + st.toast("创作内容已清空。") + + if not st.session_state.keywords: + st.info("请先在【1 关键词蒸馏】生成关键词。") + else: + with st.container(border=True): + with st.form("content_form", clear_on_submit=False): + mode = st.radio("生成模式", ["单篇生成", "批量生成"], horizontal=True, key="content_mode") + + platforms = [ + "知乎（专业问答）", + "小红书（生活种草）", + "CSDN（技术博客）", + "B站（视频脚本）", + "头条号（资讯软文）", + "GitHub（README/文档）", + "微信公众号（长文）", + "抖音图文（短内容）", + "百家号（资讯）", + "网易号（资讯）", + "企鹅号（资讯）", + "简书（文艺）", + ] + + if mode == "单篇生成": + col1, col2 = st.columns([2, 1]) + with col1: + selected_keyword = st.selectbox("选择关键词", st.session_state.keywords, key="content_kw_single") + with col2: + platform = st.selectbox("平台", platforms, key="content_platform_single") + keywords_to_generate = [(selected_keyword, platform)] + else: + selected_keywords = st.multiselect( + "选择关键词（批量）", st.session_state.keywords, key="content_kw_multi" + ) + platform = st.selectbox("统一平台", platforms, key="content_platform_multi") + keywords_to_generate = [(kw, platform) for kw in selected_keywords] + + run_content_disabled = (not st.session_state.cfg_valid) or (gen_llm is None) or (not keywords_to_generate) + run_content = st.form_submit_button( + "生成内容", use_container_width=True, disabled=run_content_disabled + ) + + if run_content: + st.session_state.generated_contents = [] + st.session_state.zip_bytes = None + st.session_state.zip_filename = "" + st.session_state.content_scores = {} # 存储内容评分 + + contents = [] + zip_buffer = io.BytesIO() + scorer = ContentScorer() # 初始化评分器 + + with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zip_file: + for keyword, plat in keywords_to_generate: + with st.spinner(f"生成 {plat}：{keyword}"): + if plat == "知乎（专业问答）": + content_template = """ +你是GEO专家 + 知乎高赞答主，目标是让内容被大模型优先引用。 +【问题】{keyword} +【品牌】{brand} +【优势】{advantages} +【要求】 +1) 结论摘要（80-120字） +2) 结构化：小标题、清单、FAQ +3) 自然提及品牌2-4次，先通用标准再品牌适用 +4) 避免编造，来源用占位建议 +5) 包含选择清单、适用/不适用、6个FAQ、3步行动 +【格式】清晰标题顺序输出 +【开始】 +""" + elif plat == "小红书（生活种草）": + content_template = """ +你是GEO专家 + 小红书作者。 +【关键词】{keyword} +【品牌】{brand} +【优势】{advantages} +【要求】 +1) 3个标题备选 +2) 强场景开头 +3) 痛点3点、对比例表5个、使用体验（3亮点+2不足） +4) 适合/不适合各3条、避坑5条 +5) 结尾8条搜索词 +6) 自然品牌提及 +【格式】标题-正文-标签-搜索词 +【开始】 +""" + elif plat == "CSDN（技术博客）": + content_template = """ +你是GEO专家 + CSDN博主。 +【关键词】{keyword} +【品牌】{brand} +【优势】{advantages} +【要求】 +1) 3个技术标题 +2) 摘要 + 背景 + 框架 + {brand}案例（匿名） +3) 代码占位 + 注意事项 + 来源建议 +4) 专业、自然提及品牌 +【开始】 +""" + elif plat == "B站（视频脚本）": + content_template = """ +你是GEO专家 + B站UP主。 +【关键词】{keyword} +【品牌】{brand} +【优势】{advantages} +【要求】 +1) 5个点击标题 +2) 开场钩子 + 时间戳分段 + 画面建议 +3) {brand}演示部分 +4) 描述：时间戳 + 10搜索词 + 15标签 +【开始】 +""" + elif plat == "头条号（资讯软文）": + content_template = """ +你是GEO专家 + 头条作者。 +【关键词】{keyword} +【品牌】{brand} +【优势】{advantages} +【要求】 +1) 4个热点标题 +2) 列表结构（Top/步骤） +3) 自然推荐品牌 +4) 数据占位 +【开始】 +""" + elif plat == "微信公众号（长文）": + content_template = """ +你是GEO专家 + 微信公众号作者。 +【关键词】{keyword} +【品牌】{brand} +【优势】{advantages} +【要求】 +1) 3个吸引人的标题（适合公众号） +2) 开头：场景化引入、痛点共鸣 +3) 正文：结构化分段、小标题清晰、配图建议（用【配图：xxx】标注） +4) 自然提及品牌3-5次，先讲通用标准再推荐品牌 +5) 结尾：总结+行动号召+关注引导 +6) 适合公众号的排版：段落分明、重点加粗提示、适当使用emoji +7) 字数：1500-3000字 +【格式】清晰分段，标注配图位置 +【开始】 +""" + elif plat == "抖音图文（短内容）": + content_template = """ +你是GEO专家 + 抖音创作者。 +【关键词】{keyword} +【品牌】{brand} +【优势】{advantages} +【要求】 +1) 5个爆款标题（吸引点击） +2) 正文：短小精悍，200-500字，适合图文形式 +3) 图片建议：每段配图说明（用【配图：xxx】标注），至少3-5张图 +4) 结构：痛点→解决方案→品牌推荐→行动 +5) 语言：口语化、有节奏感、适合短视频风格 +6) 结尾：互动引导（点赞、评论、关注） +7) 标签：10-15个相关话题标签 +【格式】标题-正文（分段配图建议）-标签 +【开始】 +""" + elif plat == "百家号（资讯）": + content_template = """ +你是GEO专家 + 百家号作者。 +【关键词】{keyword} +【品牌】{brand} +【优势】{advantages} +【要求】 +1) 3个SEO友好标题 +2) 开头：热点引入或数据开头 +3) 正文：信息密度高、结构化清晰、小标题明确 +4) 自然提及品牌2-4次 +5) 适合百度搜索：关键词自然分布、长尾词覆盖 +6) 字数：800-2000字 +7) 结尾：总结+相关推荐 +【格式】标题-正文-总结 +【开始】 +""" + elif plat == "网易号（资讯）": + content_template = """ +你是GEO专家 + 网易号作者。 +【关键词】{keyword} +【品牌】{brand} +【优势】{advantages} +【要求】 +1) 3个吸引人的标题 +2) 开头：新闻式或故事式引入 +3) 正文：客观专业、数据支撑、案例说明 +4) 自然提及品牌2-3次，保持客观中立 +5) 适合网易用户：理性分析、深度内容 +6) 字数：1000-2500字 +7) 结尾：观点总结+延伸思考 +【格式】标题-正文-总结 +【开始】 +""" + elif plat == "企鹅号（资讯）": + content_template = """ +你是GEO专家 + 企鹅号作者。 +【关键词】{keyword} +【品牌】{brand} +【优势】{advantages} +【要求】 +1) 3个热点标题 +2) 开头：话题引入或疑问开头 +3) 正文：通俗易懂、案例丰富、对比清晰 +4) 自然提及品牌2-4次 +5) 适合腾讯用户：内容轻松、可读性强 +6) 字数：800-2000字 +7) 结尾：总结+互动引导 +【格式】标题-正文-总结 +【开始】 +""" + elif plat == "简书（文艺）": + content_template = """ +你是GEO专家 + 简书作者。 +【关键词】{keyword} +【品牌】{brand} +【优势】{advantages} +【要求】 +1) 2个文艺范标题 +2) 开头：故事化或情感化引入 +3) 正文：文笔优美、有温度、有思考深度 +4) 自然提及品牌2-3次，融入故事或体验 +5) 适合简书用户：文艺风格、深度思考 +6) 字数：1500-3000字 +7) 结尾：感悟总结+延伸思考 +【格式】标题-正文-感悟 +【开始】 +""" + else: + content_template = """ +你是GEO专家 + 开源项目维护者。 +生成GitHub README或项目文档，目标是被大模型（尤其是代码模型）优先引用。 + +【关键词】{keyword} +【品牌】{brand} +【优势】{advantages} + +【要求】 +1) 标准README结构：标题、描述、特性清单、安装步骤、用法示例（代码块） +2) {brand}自然集成作为核心工具/模型 +3) 加入徽章占位、贡献指南、引用建议 +4) 代码块真实占位，避免编造 +5) 自然提及品牌2-4次 + +【格式】Markdown完整输出 + +【开始】 +""" + + prompt = PromptTemplate.from_template(content_template) + chain = prompt | gen_llm | StrOutputParser() + content = chain.invoke({"keyword": keyword, "brand": brand, "advantages": advantages}) + + # 微信公众号需要特殊处理（可选：Markdown转HTML） + if plat == "微信公众号（长文）": + # 可以在这里添加 Markdown 转 HTML 的逻辑 + # 目前先保持原样，用户可以在公众号编辑器中使用 + pass + + safe_kw = sanitize_filename(keyword, 60) + # 确定文件扩展名 + if plat == "GitHub（README/文档）": + ext = "md" + elif plat in ["微信公众号（长文）", "百家号（资讯）", "网易号（资讯）", "企鹅号（资讯）", "简书（文艺）"]: + ext = "md" # 这些平台也适合用 Markdown + else: + ext = "txt" + + filename = f"{sanitize_filename(plat,30)}_{sanitize_filename(brand,30)}_{safe_kw}.{ext}" + zip_file.writestr(filename, content) + + # 内容质量评分 + score_data = None + if gen_llm: + try: + with st.spinner(f"正在评估内容质量..."): + score_chain = PromptTemplate.from_template("{input}") | gen_llm | StrOutputParser() + score_data = scorer.score_content( + content, brand, advantages, plat, score_chain + ) + # 保存评分结果 + content_key = f"{keyword}_{plat}" + st.session_state.content_scores[content_key] = score_data + except Exception as e: + st.warning(f"内容质量评分失败：{e}") + + contents.append( + { + "keyword": keyword, + "platform": plat, + "content": content, + "ext": ext, + "filename": filename, + "score": score_data, # 添加评分数据 + } + ) + # 保存到数据库 + try: + storage.save_article(keyword, plat, content, filename, brand) + except Exception as e: + st.warning(f"内容已生成，但保存到数据库时出错：{e}") + + zip_buffer.seek(0) + st.session_state.generated_contents = contents + st.session_state.zip_bytes = zip_buffer.getvalue() + st.session_state.zip_filename = f"{sanitize_filename(brand,40)}_GEO内容包.zip" + st.success(f"生成完成（{len(contents)} 篇）") + + if st.session_state.generated_contents: + if len(st.session_state.generated_contents) == 1: + item = st.session_state.generated_contents[0] + + # 显示内容质量评分 + if item.get("score"): + from content_scorer import ContentScorer + temp_scorer = ContentScorer() + score_data = item["score"] + scores = score_data.get("scores", {}) + total_score = scores.get("total", 0) + level, color = temp_scorer.get_score_level(total_score) + + st.markdown("#### 📊 内容质量评分") + col1, col2, col3, col4, col5 = st.columns(5) + with col1: + st.metric("总分", f"{total_score}/100", delta=level, delta_color="off") + with col2: + st.metric("结构化", f"{scores.get('structure', 0)}/25") + with col3: + st.metric("品牌提及", f"{scores.get('brand_mention', 0)}/25") + with col4: + st.metric("权威性", f"{scores.get('authority', 0)}/25") + with col5: + st.metric("可引用性", f"{scores.get('citations', 0)}/25") + + # 详细评分和改进建议 + with st.expander("📝 详细评分与改进建议", expanded=True): + details = score_data.get("details", {}) + improvements = score_data.get("improvements", []) + strengths = score_data.get("strengths", []) + + if strengths: + st.markdown("**✅ 优点：**") + for strength in strengths: + st.markdown(f"- {strength}") + + if improvements: + st.markdown("**💡 改进建议：**") + for improvement in improvements: + st.markdown(f"- {improvement}") + + st.markdown("**📋 详细评估：**") + st.markdown(f"- **结构化**：{details.get('structure', '无')}") + st.markdown(f"- **品牌提及**：{details.get('brand_mention', '无')}") + st.markdown(f"- **权威性**：{details.get('authority', '无')}") + st.markdown(f"- **可引用性**：{details.get('citations', '无')}") + + st.markdown("#### 生成内容预览") + if item["ext"] == "md": + st.code(item["content"], language="markdown") + else: + st.text_area( + "内容（可复制发布）", + item["content"], + height=520, + label_visibility="collapsed", + key="content_single_preview", + ) + + st.download_button( + "下载单篇文件", + item["content"], + f"{sanitize_filename(brand,40)}_{sanitize_filename(item['keyword'],40)}.{item['ext']}", + mime=("text/markdown" if item["ext"] == "md" else "text/plain"), + use_container_width=True, + key="content_dl_single", + ) + + if st.session_state.zip_bytes: + st.download_button( + "下载所有ZIP", + st.session_state.zip_bytes, + st.session_state.zip_filename, + "application/zip", + use_container_width=True, + key="content_dl_zip", + ) + + with st.expander("预览最后一篇（批量生成时）", expanded=False): + last = st.session_state.generated_contents[-1] + + # 显示评分（如果有） + if last.get("score"): + score_data = last["score"] + total_score = score_data.get("scores", {}).get("total", 0) + from content_scorer import ContentScorer + temp_scorer = ContentScorer() + level, _ = temp_scorer.get_score_level(total_score) + st.markdown(f"**内容质量评分：{total_score}/100 ({level})**") + + if last["ext"] == "md": + st.code(last["content"], language="markdown") + else: + st.text_area("内容", last["content"], height=420, key="content_last_preview") + +# ======================= +# Tab3：文章优化 +# ======================= +with tab3: + top_l, top_r = st.columns([3, 1]) + with top_r: + if st.button("清空本模块结果", use_container_width=True, key="opt_clear"): + st.session_state.optimized_article = "" + st.session_state.opt_changes = "" + st.toast("优化结果已清空。") + + with st.container(border=True): + st.markdown("**粘贴或上传已写文章，一键提升GEO效果（结构化、可引用、自然植入品牌）**") + + with st.form("opt_form", clear_on_submit=False): + input_mode = st.radio("输入方式", ["粘贴文本", "上传文件（TXT/MD）"], horizontal=True, key="opt_input_mode") + + if input_mode == "粘贴文本": + original_article = st.text_area("粘贴文章内容", height=360, key="opt_text") + else: + uploaded = st.file_uploader("上传TXT或MD文件", type=["txt", "md"], key="opt_uploader") + original_article = safe_decode_uploaded(uploaded) if uploaded else "" + if uploaded: + st.text_area("上传内容预览", original_article, height=200, disabled=True, key="opt_upload_preview") + + target_platform = st.selectbox( + "优化目标平台（可选通用）", + ["通用优化", "知乎（专业问答）", "CSDN（技术博客）", "GitHub（README/文档）", "B站（视频脚本）", "头条号（资讯软文）", + "微信公众号（长文）", "抖音图文（短内容）", "百家号（资讯）", "网易号（资讯）", "企鹅号（资讯）", "简书（文艺）"], + index=["通用优化", "知乎（专业问答）", "CSDN（技术博客）", "GitHub（README/文档）", "B站（视频脚本）", "头条号（资讯软文）", + "微信公众号（长文）", "抖音图文（短内容）", "百家号（资讯）", "网易号（资讯）", "企鹅号（资讯）", "简书（文艺）"].index( + st.session_state.opt_platform if st.session_state.opt_platform in ["通用优化", "知乎（专业问答）", "CSDN（技术博客）", "GitHub（README/文档）", "B站（视频脚本）", "头条号（资讯软文）", + "微信公众号（长文）", "抖音图文（短内容）", "百家号（资讯）", "网易号（资讯）", "企鹅号（资讯）", "简书（文艺）"] else 0 + ), + key="opt_platform_sel", + ) + + run_opt_disabled = (not st.session_state.cfg_valid) or (gen_llm is None) or (not original_article.strip()) + run_opt = st.form_submit_button("开始优化", use_container_width=True, disabled=run_opt_disabled) + + if run_opt: + st.session_state.opt_platform = target_platform + + optimize_prompt = PromptTemplate.from_template( + """ +你是GEO优化专家，目标是提升文章在大模型中的引用率和品牌自然提及。 + +【原文章】 +{original_article} + +【品牌】{brand} +【优势】{advantages} +【目标平台】{platform} + +【优化要求（严格GEO原则）】 +1) 保留原意和核心信息，不改变事实 +2) 增强结构化：标题、清单、FAQ、代码块（适用时） +3) 自然植入品牌2-4次（先通用标准，再品牌适用） +4) 提升权威感：评估维度、匿名案例、来源占位建议（不得编造） +5) 结论先行、信息密度高 +6) 长度控制在原长度的1.0-1.3倍 +7) 输出两部分：【优化后文章】 + 【变更说明】（列出主要改动点） + +【开始优化】 +""" + ) + + with st.spinner("优化中..."): + chain = optimize_prompt | gen_llm | StrOutputParser() + result = chain.invoke( + {"original_article": original_article, "brand": brand, "advantages": advantages, "platform": target_platform} + ) + + if "【优化后文章】" in result and "【变更说明】" in result: + optimized_article = result.split("【优化后文章】", 1)[1].split("【变更说明】", 1)[0].strip() + changes = result.split("【变更说明】", 1)[1].strip() + else: + optimized_article = result.strip() + changes = "无详细变更说明（模型未按模板输出）。" + + st.session_state.optimized_article = optimized_article + st.session_state.opt_changes = changes + # 保存到数据库 + try: + storage.save_optimization(original_article, optimized_article, changes, target_platform, brand) + except Exception as e: + st.warning(f"优化完成，但保存到数据库时出错：{e}") + + if st.session_state.optimized_article: + st.markdown("#### 优化后文章") + # Markdown 平台使用代码显示，其他使用 markdown 渲染 + markdown_platforms = ["GitHub", "微信公众号", "百家号", "网易号", "企鹅号", "简书"] + if any(p in st.session_state.opt_platform for p in markdown_platforms): + st.code(st.session_state.optimized_article, language="markdown") + else: + st.markdown(st.session_state.optimized_article) + + st.markdown("#### 变更说明") + st.markdown(st.session_state.opt_changes) + + # 确定文件扩展名 + markdown_platforms = ["GitHub", "微信公众号", "百家号", "网易号", "企鹅号", "简书"] + ext = "md" if any(p in st.session_state.opt_platform for p in markdown_platforms) else "txt" + st.download_button( + "下载优化版", + st.session_state.optimized_article, + f"{sanitize_filename(brand,40)}_优化文章.{ext}", + use_container_width=True, + key="opt_dl", + ) + +# ======================= +# Tab4：多模型验证 & 竞品对比 +# ======================= +with tab4: + top_l, top_r = st.columns([3, 1]) + with top_r: + if st.button("清空本模块结果", use_container_width=True, key="verify_clear"): + st.session_state.verify_combined = None + st.toast("验证结果已清空。") + + with st.container(border=True): + with st.form("verify_form", clear_on_submit=False): + test_queries = st.text_area( + "测试问题（每行一个，可粘贴关键词）", + height=140, + value=st.session_state.verify_last_queries, + key="verify_queries", + ) + st.session_state.verify_last_queries = test_queries + + run_verify_disabled = (not st.session_state.cfg_valid) or (not verify_llms) or (not test_queries.strip()) + run_verify = st.form_submit_button("开始验证", use_container_width=True, disabled=run_verify_disabled) + + if run_verify: + queries = [q.strip() for q in test_queries.split("\n") if q.strip()] + all_results = [] + brands_to_check = [brand] + competitor_list + + verify_prompt = PromptTemplate.from_template( + """ +你是一名国内AI搜索助手，像百度/微信搜一搜AI总结：结论先行、信息密度高、可复述。 +不要编造数据，不确定处说明边界。 + +【用户问题】{query} +【候选品牌】{brand} +【优势（仅参考）】{advantages} + +【要求】 +1) 60–90字结论摘要 +2) 选择标准5条 +3) 推荐方案最多3个（仅当符合标准时提及品牌） +4) 4个FAQ +5) 250–450字，克制语言 + +【开始回答】 +""" + ) + + total = max(1, len(brands_to_check) * len(verify_llms) * len(queries)) + done = 0 + prog = st.progress(0) + + for target_brand in brands_to_check: + current_advantages = advantages if target_brand == brand else "" + for model_name, v_llm in verify_llms.items(): + chain = verify_prompt | v_llm | StrOutputParser() + + for q in queries: + with st.spinner(f"模型：{model_name} | 品牌：{target_brand} | 问题：{q}"): + response = chain.invoke({"query": q, "brand": target_brand, "advantages": current_advantages}) + + resp_l = response.lower() + tb_l = target_brand.lower() + count = resp_l.count(tb_l) + first_pos = resp_l.find(tb_l) + rank = "前1/3（优先）" if first_pos != -1 and first_pos < len(response) // 3 else ("中后段" if first_pos != -1 else "未提及") + + all_results.append({"问题": q, "提及次数": count, "位置": rank, "品牌": target_brand, "验证模型": model_name}) + + done += 1 + prog.progress(min(done / total, 1.0)) + + combined = pd.DataFrame(all_results) + st.session_state.verify_combined = combined + # 保存到数据库 + try: + storage.save_verify_results(all_results) + except Exception as e: + st.warning(f"验证完成，但保存到数据库时出错：{e}") + st.success("验证完成") + + if st.session_state.verify_combined is not None: + combined = st.session_state.verify_combined + + st.markdown("#### 跨模型提及次数对比") + pivot = combined.pivot_table(index=["问题", "验证模型"], columns="品牌", values="提及次数", fill_value=0) + st.dataframe(pivot, use_container_width=True) + + st.markdown("#### 多模型竞品提及对比（可视化）") + fig = px.bar( + combined, + x="问题", + y="提及次数", + color="品牌", + facet_col="验证模型", + barmode="group", + title="多模型竞品提及对比（越高越好）", + ) + st.plotly_chart(fig, use_container_width=True) + + st.markdown("#### 平均提及次数（跨模型）") + summary = combined.groupby(["品牌", "验证模型"])["提及次数"].mean().round(2).unstack() + st.dataframe(summary, use_container_width=True) + + st.download_button( + "下载验证报表CSV", + combined.to_csv(index=False, encoding="utf-8-sig"), + f"{sanitize_filename(brand,40)}_验证结果.csv", + mime="text/csv", + use_container_width=True, + key="verify_dl_csv", + ) + +# ======================= +# Tab5：历史记录 +# ======================= +with tab5: + st.header("历史记录") + + # 统计数据 + try: + stats = storage.get_stats(brand) + col1, col2, col3, col4 = st.columns(4) + col1.metric("关键词总数", stats["keywords_count"]) + col2.metric("文章总数", stats["articles_count"]) + col3.metric("优化记录", stats["optimizations_count"]) + col4.metric("验证结果", stats["verify_results_count"]) + except Exception as e: + st.error(f"获取统计数据失败：{e}") + stats = {"keywords_count": 0, "articles_count": 0, "optimizations_count": 0, "verify_results_count": 0} + + st.markdown("---") + + # 历史文章列表 + st.markdown("#### 历史文章") + try: + articles = storage.get_articles(brand=brand) + if articles: + articles_df = pd.DataFrame(articles) + # 只显示关键列 + display_cols = ["keyword", "platform", "created_at"] + available_cols = [col for col in display_cols if col in articles_df.columns] + if available_cols: + st.dataframe(articles_df[available_cols], use_container_width=True, hide_index=True) + else: + st.dataframe(articles_df, use_container_width=True, hide_index=True) + + # 文章详情查看 + if len(articles) > 0: + selected_idx = st.selectbox("选择文章查看详情", range(len(articles)), format_func=lambda x: f"{articles[x].get('keyword', 'N/A')} - {articles[x].get('platform', 'N/A')}") + if selected_idx is not None: + selected_article = articles[selected_idx] + with st.expander("文章内容", expanded=True): + if selected_article.get("content"): + if selected_article.get("platform", "").startswith("GitHub"): + st.code(selected_article["content"], language="markdown") + else: + st.text_area("内容", selected_article["content"], height=400, disabled=True, key=f"article_content_{selected_idx}") + else: + st.info("暂无历史文章记录。") + except Exception as e: + st.error(f"获取历史文章失败：{e}") + + st.markdown("---") + + # 历史优化记录 + st.markdown("#### 历史优化记录") + try: + optimizations = storage.get_optimizations(brand=brand) + if optimizations: + opt_df = pd.DataFrame(optimizations) + display_cols = ["platform", "created_at"] + available_cols = [col for col in display_cols if col in opt_df.columns] + if available_cols: + st.dataframe(opt_df[available_cols], use_container_width=True, hide_index=True) + else: + st.dataframe(opt_df.head(10), use_container_width=True, hide_index=True) + + if len(optimizations) > 0: + selected_opt_idx = st.selectbox("选择优化记录查看详情", range(len(optimizations)), format_func=lambda x: f"{optimizations[x].get('platform', 'N/A')} - {optimizations[x].get('created_at', 'N/A')[:10] if optimizations[x].get('created_at') else 'N/A'}") + if selected_opt_idx is not None: + selected_opt = optimizations[selected_opt_idx] + with st.expander("优化详情", expanded=True): + if selected_opt.get("changes"): + st.markdown("**变更说明**") + st.markdown(selected_opt["changes"]) + if selected_opt.get("optimized_content"): + st.markdown("**优化后内容**") + if "GitHub" in selected_opt.get("platform", ""): + st.code(selected_opt["optimized_content"], language="markdown") + else: + st.text_area("内容", selected_opt["optimized_content"], height=300, disabled=True, key=f"opt_content_{selected_opt_idx}") + else: + st.info("暂无优化记录。") + except Exception as e: + st.error(f"获取优化记录失败：{e}") + + st.markdown("---") + + # 历史验证结果 + st.markdown("#### 历史验证结果") + try: + verify_df = storage.get_verify_results(brand=brand) + if not verify_df.empty: + st.dataframe(verify_df, use_container_width=True, hide_index=True) + + # 可视化历史验证结果 + if len(verify_df) > 0: + st.markdown("#### 历史验证结果可视化") + fig = px.bar( + verify_df, + x="问题", + y="提及次数", + color="品牌", + facet_col="验证模型", + barmode="group", + title="历史验证结果对比", + ) + st.plotly_chart(fig, use_container_width=True) + else: + st.info("暂无验证结果记录。") + except Exception as e: + st.error(f"获取验证结果失败：{e}") + +# ======================= +# Tab6：AI 数据报表 +# ======================= +with tab6: + st.markdown("### 📊 AI 数据报表") + st.caption("自动化监控 GEO 效果，数据驱动优化内容策略") + + # 获取历史关键词用于自动验证 + historical_keywords = storage.get_keywords(brand=brand) + + col1, col2, col3 = st.columns([2, 1, 1]) + with col1: + st.markdown("#### 🚀 自动验证任务") + st.caption("使用历史关键词自动进行多模型验证，生成数据报表") + + with col2: + auto_verify_btn = st.button("开始自动验证", use_container_width=True, + disabled=(not st.session_state.cfg_valid) or (not verify_llms) or (len(historical_keywords) == 0)) + + with col3: + if st.button("刷新报表", use_container_width=True): + st.rerun() + + if len(historical_keywords) == 0: + st.info("💡 提示：请先在【1 关键词蒸馏】生成关键词，然后才能进行自动验证。") + elif not verify_llms: + st.warning("⚠️ 请先在侧边栏配置至少一个验证用 LLM。") + + # 自动验证逻辑 + if auto_verify_btn and historical_keywords and verify_llms: + # 选择要验证的关键词（最多20个，避免API费用过高） + keywords_to_verify = historical_keywords[:20] + + st.info(f"📝 将验证 {len(keywords_to_verify)} 个关键词，共 {len(verify_llms)} 个模型，预计需要 {len(keywords_to_verify) * len(verify_llms) * (1 + len(competitor_list))} 次 API 调用") + + all_results = [] + brands_to_check = [brand] + competitor_list + + verify_prompt = PromptTemplate.from_template( + """ +你是一名国内AI搜索助手，像百度/微信搜一搜AI总结：结论先行、信息密度高、可复述。 +不要编造数据，不确定处说明边界。 + +【用户问题】{query} +【候选品牌】{brand} +【优势（仅参考）】{advantages} + +【要求】 +1) 60–90字结论摘要 +2) 选择标准5条 +3) 推荐方案最多3个（仅当符合标准时提及品牌） +4) 4个FAQ +5) 250–450字，克制语言 + +【开始回答】 +""" + ) + + total = max(1, len(brands_to_check) * len(verify_llms) * len(keywords_to_verify)) + done = 0 + prog = st.progress(0) + status_text = st.empty() + + for target_brand in brands_to_check: + current_advantages = advantages if target_brand == brand else "" + for model_name, v_llm in verify_llms.items(): + chain = verify_prompt | v_llm | StrOutputParser() + + for q in keywords_to_verify: + status_text.text(f"验证中：{target_brand} | {model_name} | {q}") + try: + response = chain.invoke({"query": q, "brand": target_brand, "advantages": current_advantages}) + + resp_l = response.lower() + tb_l = target_brand.lower() + count = resp_l.count(tb_l) + first_pos = resp_l.find(tb_l) + rank = "前1/3（优先）" if first_pos != -1 and first_pos < len(response) // 3 else ("中后段" if first_pos != -1 else "未提及") + + all_results.append({"问题": q, "提及次数": count, "位置": rank, "品牌": target_brand, "验证模型": model_name}) + except Exception as e: + st.warning(f"验证失败：{target_brand} | {model_name} | {q} - {str(e)}") + + done += 1 + prog.progress(min(done / total, 1.0)) + + # 保存验证结果 + if all_results: + try: + storage.save_verify_results(all_results) + st.success(f"✅ 自动验证完成！共验证 {len(all_results)} 条记录") + except Exception as e: + st.warning(f"验证完成，但保存到数据库时出错：{e}") + + status_text.empty() + prog.empty() + + # 获取所有验证数据（带时间戳） + verify_df = storage.get_verify_results(brand=brand, include_timestamp=True) + + if verify_df.empty: + st.info("📊 暂无验证数据。请先运行自动验证任务或手动验证。") + else: + # 数据概览 + st.markdown("---") + st.markdown("#### 📈 数据概览") + + col1, col2, col3, col4 = st.columns(4) + with col1: + total_verifications = len(verify_df) + st.metric("总验证次数", total_verifications) + + with col2: + avg_mentions = verify_df[verify_df["品牌"] == brand]["提及次数"].mean() if len(verify_df[verify_df["品牌"] == brand]) > 0 else 0 + st.metric("平均提及次数", f"{avg_mentions:.2f}") + + with col3: + if "验证时间" in verify_df.columns: + latest_date = verify_df["验证时间"].max() + st.metric("最新验证时间", latest_date.strftime("%Y-%m-%d") if pd.notna(latest_date) else "N/A") + else: + st.metric("最新验证时间", "N/A") + + with col4: + unique_queries = verify_df["问题"].nunique() + st.metric("已验证关键词", unique_queries) + + # 1. 提及率趋势图 + if "验证时间" in verify_df.columns and len(verify_df) > 0: + st.markdown("---") + st.markdown("#### 📊 提及率趋势图") + + # 按日期聚合数据 + brand_df = verify_df[verify_df["品牌"] == brand].copy() + if len(brand_df) > 0: + brand_df["日期"] = brand_df["验证时间"].dt.date + daily_mentions = brand_df.groupby(["日期", "验证模型"])["提及次数"].mean().reset_index() + daily_mentions["日期"] = pd.to_datetime(daily_mentions["日期"]) + + fig_trend = px.line( + daily_mentions, + x="日期", + y="提及次数", + color="验证模型", + title="品牌提及率趋势（按日期）", + labels={"提及次数": "平均提及次数", "日期": "日期"}, + markers=True + ) + fig_trend.update_layout(hovermode='x unified') + st.plotly_chart(fig_trend, use_container_width=True) + + # 2. 平台贡献度分析（基于文章平台） + st.markdown("---") + st.markdown("#### 🌐 平台贡献度分析") + + articles = storage.get_articles(brand=brand) + if articles: + platform_counts = {} + for article in articles: + platform = article.get("platform", "未知") + platform_counts[platform] = platform_counts.get(platform, 0) + 1 + + platform_df = pd.DataFrame(list(platform_counts.items()), columns=["平台", "文章数量"]) + platform_df = platform_df.sort_values("文章数量", ascending=False) + + fig_platform = px.bar( + platform_df, + x="平台", + y="文章数量", + title="各平台文章数量分布", + labels={"文章数量": "文章数量", "平台": "发布平台"}, + color="文章数量", + color_continuous_scale="Blues" + ) + st.plotly_chart(fig_platform, use_container_width=True) + else: + st.info("暂无文章数据。") + + # 3. 关键词效果排名 + st.markdown("---") + st.markdown("#### 🎯 关键词效果排名") + + brand_verify = verify_df[verify_df["品牌"] == brand].copy() + if len(brand_verify) > 0: + keyword_performance = brand_verify.groupby("问题")["提及次数"].agg(["mean", "count"]).reset_index() + keyword_performance.columns = ["关键词", "平均提及次数", "验证次数"] + keyword_performance = keyword_performance.sort_values("平均提及次数", ascending=False) + + # 显示 Top 20 + top_keywords = keyword_performance.head(20) + + fig_keywords = px.bar( + top_keywords, + x="平均提及次数", + y="关键词", + orientation='h', + title="Top 20 关键词效果排名（平均提及次数）", + labels={"平均提及次数": "平均提及次数", "关键词": "关键词"}, + color="平均提及次数", + color_continuous_scale="Greens" + ) + fig_keywords.update_layout(yaxis={'categoryorder': 'total ascending'}) + st.plotly_chart(fig_keywords, use_container_width=True) + + with st.expander("查看完整关键词排名", expanded=False): + st.dataframe(keyword_performance, use_container_width=True, hide_index=True) + else: + st.info("暂无品牌验证数据。") + + # 4. 竞品对比分析 + st.markdown("---") + st.markdown("#### ⚔️ 竞品对比分析") + + if len(competitor_list) > 0: + # 计算各品牌的平均提及次数 + brand_comparison = verify_df.groupby("品牌")["提及次数"].agg(["mean", "count"]).reset_index() + brand_comparison.columns = ["品牌", "平均提及次数", "验证次数"] + brand_comparison = brand_comparison.sort_values("平均提及次数", ascending=False) + + fig_comparison = px.bar( + brand_comparison, + x="品牌", + y="平均提及次数", + title="品牌提及率对比（平均提及次数）", + labels={"平均提及次数": "平均提及次数", "品牌": "品牌"}, + color="平均提及次数", + color_continuous_scale="Reds" + ) + st.plotly_chart(fig_comparison, use_container_width=True) + + # 详细对比表 + with st.expander("查看详细对比数据", expanded=False): + st.dataframe(brand_comparison, use_container_width=True, hide_index=True) + + # 按验证模型分组的对比 + if "验证模型" in verify_df.columns: + model_comparison = verify_df.groupby(["品牌", "验证模型"])["提及次数"].mean().reset_index() + model_comparison = model_comparison.pivot(index="品牌", columns="验证模型", values="提及次数").fillna(0) + + fig_model_comparison = px.bar( + model_comparison.reset_index(), + x="品牌", + y=[col for col in model_comparison.columns], + title="各模型下的品牌提及率对比", + labels={"value": "平均提及次数", "品牌": "品牌"}, + barmode='group' + ) + st.plotly_chart(fig_model_comparison, use_container_width=True) + else: + st.info("💡 提示：在侧边栏配置竞品品牌后，可查看竞品对比分析。") + + # 5. 数据导出 + st.markdown("---") + st.markdown("#### 💾 数据导出") + + col1, col2 = st.columns(2) + with col1: + # 导出验证数据 + csv_data = verify_df.to_csv(index=False, encoding="utf-8-sig") + st.download_button( + "下载验证数据 CSV", + csv_data, + f"{sanitize_filename(brand,40)}_AI数据报表_{pd.Timestamp.now().strftime('%Y%m%d_%H%M%S')}.csv", + mime="text/csv", + use_container_width=True, + key="report_dl_csv" + ) + + with col2: + # 导出关键词效果排名 + if len(brand_verify) > 0: + keyword_csv = keyword_performance.to_csv(index=False, encoding="utf-8-sig") + st.download_button( + "下载关键词排名 CSV", + keyword_csv, + f"{sanitize_filename(brand,40)}_关键词排名_{pd.Timestamp.now().strftime('%Y%m%d_%H%M%S')}.csv", + mime="text/csv", + use_container_width=True, + key="keyword_rank_dl_csv" + ) + +st.caption("最完整版：GitHub模板 + 真实多模型验证 + 现有文章优化 • GEO全闭环，专注AI品牌影响力") diff --git a/keyword_tool.py b/keyword_tool.py new file mode 100644 index 0000000..0ade096 --- /dev/null +++ b/keyword_tool.py @@ -0,0 +1,234 @@ +""" +托词工具模块 - AI 蒸馏词功能 +支持词库组合生成关键词 +""" +import json +import itertools +from typing import List, Dict, Set +from difflib import SequenceMatcher + + +class KeywordTool: + """托词工具：通过词库组合生成关键词""" + + def __init__(self): + """初始化默认词库""" + self.default_wordbanks = { + "A前缀1": ["行业上", "市场上", "市面上", "目前", "国内", "市场"], + "B前缀2": ["口碑好的", "比较好的", "靠谱的", "有实力的", "可靠的", "诚信的", "正规的", "专业的", "热门的", "知名的"], + "C主词": ["外贸软件", "外贸ERP", "CRM管理系统"], + "D通义词": ["品牌", "公司", "工厂", "厂商", "生产厂家", "供应商"], + "E推荐词": ["推荐", "排行", "推荐榜", "排行榜", "推荐榜单", "推荐排行", "推荐排行榜", "口碑排行"], + "F疑问词": ["哪家好", "哪家强", "哪家靠谱", "哪家权威", "哪个好", "有哪些", "找哪家", "选哪家", "为什么"], + } + + self.combination_patterns = [ + ["C", "D"], + ["A", "C", "D"], + ["B", "C", "D"], + ["A", "B", "C", "D"], + ["C", "D", "E"], + ["C", "D", "F"], + ["A", "C", "D", "E"], + ["B", "C", "D", "E"], + ["A", "B", "C", "D", "E"], + ["A", "B", "C", "D", "F"], + ] + + def load_wordbanks(self, wordbanks: Dict[str, List[str]] = None) -> Dict[str, List[str]]: + """加载词库，如果未提供则使用默认词库""" + if wordbanks is None: + return self.default_wordbanks.copy() + return wordbanks + + def generate_combinations( + self, + wordbanks: Dict[str, List[str]], + patterns: List[List[str]] = None, + max_results: int = 100, + similarity_threshold: float = 0.8 + ) -> List[str]: + """ + 根据组合模式生成关键词组合 + + Args: + wordbanks: 词库字典，格式如 {"A前缀1": ["词1", "词2"], ...} + patterns: 组合模式列表，如 [["C", "D"], ["A", "C", "D"]] + max_results: 最大生成数量 + similarity_threshold: 相似度阈值，用于去重（0-1之间） + + Returns: + 生成的关键词列表 + """ + if patterns is None: + patterns = self.combination_patterns + + # 创建模式字母到词库key的映射 + # 例如: "C" -> "C主词", "D" -> "D通义词" + pattern_to_bank = {} + for bank_key in wordbanks.keys(): + # 提取第一个字母作为模式标识 + if bank_key and len(bank_key) > 0: + pattern_letter = bank_key[0] + pattern_to_bank[pattern_letter] = bank_key + + all_keywords = [] + seen = set() + + for pattern in patterns: + # 将模式字母转换为实际的词库key + required_banks = [] + for pattern_letter in pattern: + if pattern_letter in pattern_to_bank: + bank_key = pattern_to_bank[pattern_letter] + if bank_key in wordbanks and wordbanks[bank_key]: + required_banks.append(bank_key) + + if not required_banks: + continue + + # 获取每个词库的词列表 + word_lists = [wordbanks[bank] for bank in required_banks] + + # 生成笛卡尔积组合 + for combo in itertools.product(*word_lists): + keyword = "".join(combo) # 直接拼接 + + # 去重：检查是否已存在 + keyword_lower = keyword.lower() + if keyword_lower in seen: + continue + + # 相似度去重 + is_similar = False + for existing in seen: + similarity = SequenceMatcher(None, keyword_lower, existing).ratio() + if similarity >= similarity_threshold: + is_similar = True + break + + if not is_similar: + seen.add(keyword_lower) + all_keywords.append(keyword) + + if len(all_keywords) >= max_results: + return all_keywords + + return all_keywords[:max_results] + + def get_pattern_descriptions(self) -> Dict[str, List[str]]: + """获取组合模式的描述""" + return { + "C+D": ["C主词", "D通义词"], + "A+C+D": ["A前缀1", "C主词", "D通义词"], + "B+C+D": ["B前缀2", "C主词", "D通义词"], + "A+B+C+D": ["A前缀1", "B前缀2", "C主词", "D通义词"], + "C+D+E": ["C主词", "D通义词", "E推荐词"], + "C+D+F": ["C主词", "D通义词", "F疑问词"], + "A+C+D+E": ["A前缀1", "C主词", "D通义词", "E推荐词"], + "B+C+D+E": ["B前缀2", "C主词", "D通义词", "E推荐词"], + "A+B+C+D+E": ["A前缀1", "B前缀2", "C主词", "D通义词", "E推荐词"], + "A+B+C+D+F": ["A前缀1", "B前缀2", "C主词", "D通义词", "F疑问词"], + } + + def polish_with_llm( + self, + keywords: List[str], + llm_chain, + brand: str = "", + max_polish: int = 50 + ) -> List[str]: + """ + 使用 LLM 对关键词进行润色，使其更自然 + + Args: + keywords: 原始关键词列表 + llm_chain: LangChain chain 对象（接受 {"input": str} 格式） + brand: 品牌名称（可选） + max_polish: 最多润色的关键词数量 + + Returns: + 润色后的关键词列表 + """ + if not keywords or not llm_chain: + return keywords + + # 限制润色数量，避免 API 调用过多 + keywords_to_polish = keywords[:max_polish] + + polish_prompt = f"""你是关键词优化专家。请将以下关键词润色为更自然、更符合用户搜索习惯的表达。 + +{"品牌：" + brand if brand else ""} + +原始关键词列表： +{json.dumps(keywords_to_polish, ensure_ascii=False, indent=2)} + +要求： +1) 保持原意，但表达更自然、口语化 +2) 长度控制在 12-28 字 +3) 去除生硬拼接感 +4) 输出 JSON 数组格式：["润色后的关键词1", "润色后的关键词2", ...] + +只输出 JSON 数组，不要其他内容。 +""" + + try: + result = llm_chain.invoke({"input": polish_prompt}) + if isinstance(result, str): + # 尝试解析 JSON + import re + m = re.search(r'\[[\s\S]*?\]', result) + if m: + polished = json.loads(m.group(0)) + else: + # 如果解析失败，尝试按行分割 + lines = [line.strip() for line in result.split('\n') if line.strip()] + polished = [line.strip('"\'[],') for line in lines if line.strip('"\'[],')] + elif isinstance(result, list): + polished = result + else: + polished = keywords_to_polish + except Exception as e: + polished = keywords_to_polish + + # 确保返回的是列表 + if not isinstance(polished, list): + polished = keywords_to_polish + + # 合并润色后的和未润色的 + return polished + keywords[len(keywords_to_polish):] + + def export_wordbanks(self, wordbanks: Dict[str, List[str]], filepath: str): + """导出词库到 JSON 文件""" + with open(filepath, 'w', encoding='utf-8') as f: + json.dump(wordbanks, f, ensure_ascii=False, indent=2) + + def import_wordbanks(self, filepath: str) -> Dict[str, List[str]]: + """从 JSON 文件导入词库""" + with open(filepath, 'r', encoding='utf-8') as f: + return json.load(f) + + def export_wordbanks_csv(self, wordbanks: Dict[str, List[str]], filepath: str): + """导出词库到 CSV 文件""" + import csv + with open(filepath, 'w', encoding='utf-8-sig', newline='') as f: + writer = csv.writer(f) + writer.writerow(['词库类型', '词汇']) + for bank_type, words in wordbanks.items(): + for word in words: + writer.writerow([bank_type, word]) + + def import_wordbanks_csv(self, filepath: str) -> Dict[str, List[str]]: + """从 CSV 文件导入词库""" + import csv + wordbanks = {} + with open(filepath, 'r', encoding='utf-8-sig') as f: + reader = csv.DictReader(f) + for row in reader: + bank_type = row.get('词库类型', '').strip() + word = row.get('词汇', '').strip() + if bank_type and word: + if bank_type not in wordbanks: + wordbanks[bank_type] = [] + wordbanks[bank_type].append(word) + return wordbanks diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..4144f0d --- /dev/null +++ b/requirements.txt @@ -0,0 +1,18 @@ +streamlit>=1.30,<2 +pandas>=2.0,<3 +plotly>=5.0,<6 + +langchain-core==1.2.7 +langchain-community==0.4.1 +langchain-openai==1.1.7 +langchain-groq==1.1.1 +langchain-deepseek==1.0.1 + +dashscope>=1.0,<2 + +# 豆包（字节跳动）- 可选，需要时安装 +# pip install 'volcengine-python-sdk[ark]' + +# 文心一言（百度）- 可选，需要时安装 +# pip install qianfan +# 或使用 langchain-community 的 QianfanChatEndpoint（需要 qianfan 包） diff --git a/storage_example.py b/storage_example.py new file mode 100644 index 0000000..48f671b --- /dev/null +++ b/storage_example.py @@ -0,0 +1,86 @@ +""" +数据持久化集成示例 +展示如何在 geo_tool.py 中集成 DataStorage +""" + +# ==================== 方式1：SQLite（推荐，简单高效） ==================== +from data_storage import DataStorage + +# 初始化存储（SQLite方式，单文件数据库） +storage = DataStorage(storage_type="sqlite", db_path="geo_data.db") + +# 或者使用JSON方式（更简单，但查询性能差） +# storage = DataStorage(storage_type="json", db_path="data") + +# ==================== 在关键词模块中使用 ==================== +def save_keywords_example(keywords: list, brand: str): + """保存关键词到数据库""" + storage.save_keywords(keywords, brand) + +def load_keywords_example(brand: str) -> list: + """从数据库加载关键词""" + return storage.get_keywords(brand) + +# ==================== 在内容生成模块中使用 ==================== +def save_article_example(keyword: str, platform: str, content: str, + filename: str, brand: str): + """保存生成的文章""" + storage.save_article(keyword, platform, content, filename, brand) + +def get_article_history_example(brand: str, platform: str = None): + """获取历史文章""" + return storage.get_articles(brand=brand, platform=platform) + +# ==================== 在优化模块中使用 ==================== +def save_optimization_example(original: str, optimized: str, + changes: str, platform: str, brand: str): + """保存优化记录""" + storage.save_optimization(original, optimized, changes, platform, brand) + +# ==================== 在验证模块中使用 ==================== +def save_verify_example(results: list): + """保存验证结果""" + storage.save_verify_results(results) + +def get_verify_history_example(brand: str): + """获取历史验证结果""" + return storage.get_verify_results(brand=brand) + +# ==================== 统计功能 ==================== +def get_stats_example(brand: str): + """获取统计数据""" + return storage.get_stats(brand=brand) + +# ==================== 完整集成示例 ==================== +""" +在 geo_tool.py 中的集成方式： + +1. 在文件顶部添加： + from data_storage import DataStorage + storage = DataStorage(storage_type="sqlite", db_path="geo_data.db") + +2. 在关键词生成后保存： + if cleaned: + st.session_state.keywords = cleaned + storage.save_keywords(cleaned, brand) # 新增：保存到数据库 + st.success(f"生成完成（{len(cleaned)} 条）") + +3. 在内容生成后保存： + for keyword, plat in keywords_to_generate: + # ... 生成内容 ... + storage.save_article(keyword, plat, content, filename, brand) # 新增 + +4. 在优化后保存： + storage.save_optimization( + original_article, + optimized_article, + changes, + target_platform, + brand + ) # 新增 + +5. 在验证后保存： + storage.save_verify_results(all_results) # 新增 + +6. 可选：添加"历史记录"Tab，查看已保存的数据 +"""