【实战】从零搭建GEO多平台监控系统：支持ChatGPT、豆包、Kimi、文心一言

2. 具体平台实现

ChatGPT平台

import os import requests classChatGPTPlatform(BasePlatform):"""ChatGPT平台"""def__init__(self, api_key:str=None):super().__init__("ChatGPT", api_key or os.getenv("OPENAI_API_KEY")) self.api_base ="https://api.openai.com/v1" self.model ="gpt-4"defquery(self, search_query:str)->str:"""查询ChatGPT""" url =f"{self.api_base}/chat/completions" headers ={"Authorization":f"Bearer {self.api_key}","Content-Type":"application/json"} payload ={"model": self.model,"messages":[{"role":"user","content":f"作为一个专业的AI助手，请回答：{search_query}"}],"temperature":0.7,"max_tokens":500} response = requests.post(url, headers=headers, json=payload, timeout=30) response.raise_for_status() result = response.json()return result["choices"][0]["message"]["content"]defextract_brand_info(self, response:str, brand:str)-> Dict:"""提取品牌信息""" mentioned = brand in response ifnot mentioned:return{"mentioned":False,"position":0,"context":""}# 查找品牌在响应中的位置 position = response.index(brand)# 提取上下文(前后各50字符) start =max(0, position -50) end =min(len(response), position +len(brand)+50) context = response[start:end]return{"mentioned":True,"position":1,# 简化：出现即为位置1"context":f"...{context}..."}

豆包平台

classDoubaoPlatform(BasePlatform):"""豆包平台(字节跳动)"""def__init__(self, api_key:str=None):super().__init__("豆包", api_key or os.getenv("DOUBAO_API_KEY")) self.api_base ="https://ark.cn-beijing.volces.com/api/v3" self.model ="doubao-seed-250615"defquery(self, search_query:str)->str:"""查询豆包""" url =f"{self.api_base}/chat/completions" headers ={"Authorization":f"Bearer {self.api_key}","Content-Type":"application/json"} payload ={"model": self.model,"messages":[{"role":"user","content": search_query}]} response = requests.post(url, headers=headers, json=payload, timeout=30) response.raise_for_status() result = response.json()return result["choices"][0]["message"]["content"]defextract_brand_info(self, response:str, brand:str)-> Dict:"""提取品牌信息""" mentioned = brand in response ifnot mentioned:return{"mentioned":False,"position":0,"context":""} position = response.index(brand) start =max(0, position -50) end =min(len(response), position +len(brand)+50)return{"mentioned":True,"position":1,"context":f"...{response[start:end]}..."}

Kimi平台

classKimiPlatform(BasePlatform):"""Kimi平台(月之暗面)"""def__init__(self, api_key:str=None):super().__init__("Kimi", api_key or os.getenv("KIMI_API_KEY")) self.api_base ="https://api.moonshot.cn/v1" self.model ="kimi-flash"defquery(self, search_query:str)->str:"""查询Kimi""" url =f"{self.api_base}/chat/completions" headers ={"Authorization":f"Bearer {self.api_key}","Content-Type":"application/json"} payload ={"model": self.model,"messages":[{"role":"user","content": search_query}]} response = requests.post(url, headers=headers, json=payload, timeout=30) response.raise_for_status() result = response.json()return result["choices"][0]["message"]["content"]defextract_brand_info(self, response:str, brand:str)-> Dict:"""提取品牌信息""" mentioned = brand in response ifnot mentioned:return{"mentioned":False,"position":0,"context":""}return{"mentioned":True,"position":1,"context":f"...{response[max(0, response.index(brand)-50):response.index(brand)+len(brand)+50]}..."}

3. 多平台监控器

from typing import List, Dict import json classGEOMultiPlatformMonitor:"""GEO多平台监控器"""def__init__(self, brand:str): self.brand = brand self.platforms: List[BasePlatform]=[] self.results: List[GEOResult]=[]defadd_platform(self, platform: BasePlatform):"""添加监控平台""" self.platforms.append(platform)defcheck_all_platforms(self, queries: List[str])-> List[GEOResult]:"""检查所有平台""" all_results =[]for platform in self.platforms:for query in queries:print(f"正在检查 {platform.name} - {query}...") result = platform.check_brand(query, self.brand) all_results.append(result)# 添加延迟，避免请求过快 time.sleep(random.uniform(1.0,2.0)) self.results = all_results return all_results defgenerate_report(self)->str:"""生成监控报告"""ifnot self.results:return"暂无监控数据" report =f"""# GEO多平台监控报告 生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} 监控品牌: {self.brand} --- ## 📊 总体概览 """ total =len(self.results) mentioned =sum(1for r in self.results if r.mentioned) success =sum(1for r in self.results ifnot r.error) report +=f"- 总查询次数: {total}\n" report +=f"- 品牌出现次数: {mentioned}\n" report +=f"- **整体可见性: {mentioned/total*100:.1f}%**\n" report +=f"- 查询成功率: {success/total*100:.1f}%\n\n"# 分平台统计 report +="## 🏠 分平台数据\n\n"for platform_name inset(r.platform for r in self.results): platform_results =[r for r in self.results if r.platform == platform_name] p_mentioned =sum(1for r in platform_results if r.mentioned) p_total =len(platform_results) p_success =sum(1for r in platform_results ifnot r.error) emoji ="✅"if p_mentioned >0else"❌" report +=f"### {emoji}{platform_name}\n" report +=f"- 查询次数: {p_total}\n" report +=f"- 品牌出现: {p_mentioned}次\n" report +=f"- 可见性: {p_mentioned/p_total*100:.1f}%\n" report +=f"- 成功率: {p_success/p_total*100:.1f}%\n\n"# 错误统计 errors =[r for r in self.results if r.error]if errors: report +="## ⚠️ 错误统计\n\n"for r in errors: report +=f"- {r.platform} - {r.query}: {r.error}\n"return report defsave_to_json(self, filename:str=None):"""保存结果到JSON"""ifnot filename: filename =f"geo_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" data =[{"platform": r.platform,"query": r.query,"brand_mentioned": r.mentioned,"position": r.position,"context": r.context,"timestamp": r.timestamp,"response_time_ms": r.response_time_ms,"error": r.error }for r in self.results ]withopen(filename,"w", encoding="utf-8")as f: json.dump(data, f, ensure_ascii=False, indent=2)print(f"报告已保存: {filename}")

4. 定时任务配置

import schedule defdaily_geo_check():"""每日GEO检查任务""" monitor = GEOMultiPlatformMonitor("你的品牌")# 添加平台 monitor.add_platform(ChatGPTPlatform()) monitor.add_platform(DoubaoPlatform()) monitor.add_platform(KimiPlatform())# 设置查询词 queries =["推荐一个XX工具","XX服务商哪家好","怎么选择XX供应商","XX工具排行榜"]# 执行检查 results = monitor.check_all_platforms(queries)# 生成报告 report = monitor.generate_report()print(report)# 保存结果 monitor.save_to_json()# 配置定时任务 schedule.every().day.at("09:00").do(daily_geo_check)print("GEO多平台监控系统已启动，每天9:00自动检查")# 运行whileTrue: schedule.run_pending() time.sleep(60)

使用方法

1. 安装依赖

pip install requests schedule 

2. 配置API Key

# 设置环境变量exportOPENAI_API_KEY="your-openai-key"exportDOUBAO_API_KEY="your-doubao-key"exportKIMI_API_KEY="your-kimi-key"

3. 修改品牌名和查询词

# 修改监控的品牌 monitor = GEOMultiPlatformMonitor("你的品牌")# 修改查询词 queries =["推荐一个XX工具","XX服务商哪家好"]

4. 运行

python geo_monitor.py 

效果

用这套系统监控GEO：

总结

Sora死了，但监控代码升级了。

从单平台监控，变成多平台监控。

核心改进：

• 平台解耦，易于扩展
• 数据结构化，便于分析
• 定时任务，自动运行
• 报告生成，直观展示

完整代码可以直接使用，只需要配置API Key即可。

有问题欢迎评论区交流。

#Python #GEO #多平台监控 #ChatGPT #Kimi #豆包