zhenxiauto/GEO-Common-AI-Keyword-Results-Collection

Public

WeChat Login

Code Issues

Pull requests Events Packages Insights

master

Branch

Tag

迪迦

fix(kimi): 添加缺失的 import time 修复运行时 NameError

78203d1d

40 commits

.ide
.trae
core
platforms
.cnb.yml
.gitignore
.python-version
CLAUDE.md
README.md
create_test_excel.py
init_tabs.py
pyproject.toml
run_all.py
test_concurrent.py
test_concurrent.xlsx
uv.lock

GEO Common AI 关键词电话号码提取工具

基于 DrissionPage 的多平台 AI 对话工具，可自动在多个 AI 平台上搜索关键词并提取电话号码。

核心特性

Excel批量模式：支持批量处理关键字，结果自动保存到Excel表格
断点续传：程序中断后可从上次位置继续，支持跳过指定记录
数据安全：采用临时文件+原子重命名机制，避免异常关闭导致Excel损坏
面向对象设计：采用抽象基类和模板方法模式，代码复用率高
支持 6 个 AI 平台：DeepSeek、豆包、文心一言、Kimi、通义千问、腾讯元宝
自动提取电话：智能识别多种电话号码格式（400、手机、座机等）
浏览器管理：自动连接已打开的浏览器或创建新实例
可配置：支持等待时间、浏览器端口等参数配置
异常处理：完善的异常体系，错误信息清晰

项目架构

GEO-Common-AI-Keyword-Results-Collection/
├── core/                     # 核心模块
│   ├── __init__.py          # 统一导出接口
│   ├── config.py            # 配置管理（单例模式）
│   ├── excel_manager.py     # Excel表格管理器
│   ├── phone_extractor.py   # 电话号码提取器
│   ├── browser_manager.py   # 浏览器生命周期管理
│   ├── base_platform.py     # 平台抽象基类
│   └── exceptions.py        # 自定义异常类
├── platforms/               # 平台实现
│   ├── __init__.py         # 平台导出 + PLATFORMS 字典
│   ├── deepseek.py         # DeepSeek 平台
│   ├── doubao.py           # 豆包平台
│   ├── wenxin.py           # 文心一言平台
│   ├── kimi.py             # Kimi 平台
│   ├── tongyi.py           # 通义千问平台
│   └── yuanbao.py          # 腾讯元宝平台
├── run_all.py             # 主程序入口（Excel批量模式）
├── pyproject.toml         # 项目配置
└── uv.lock               # 依赖锁定文件

快速开始

环境要求

Python 3.13+
DrissionPage 4.0.0+（浏览器自动化库）

安装依赖

使用 uv（推荐）：

uv sync

使用 pip：

pip install DrissionPage>=4.0.0

基本使用

from platforms import DeepSeekPlatform

# 创建平台实例
platform = DeepSeekPlatform()

# 搜索关键词并提取电话
phones = platform.search("劳力士维修服务中心")

# 清理资源
platform.cleanup()

print(f"找到电话号码：{phones}")

使用便捷函数

from platforms import search_deepseek

# 直接搜索（自动清理资源）
phones = search_deepseek("劳力士维修服务中心", "output.txt")

批量搜索所有平台

from platforms import PLATFORMS

keyword = "劳力士维修服务中心"
all_phones = []

for platform_name, platform_info in PLATFORMS.items():
    print(f"正在搜索 {platform_name}...")
    search_func = platform_info["function"]
    phones = search_func(keyword)
    all_phones.extend(phones)

print(f"共找到 {len(all_phones)} 个电话号码")

首次使用：平台初始化

在首次使用前，需要在各AI平台完成登录。

运行初始化脚本，在多个标签页中同时打开所有AI平台：

python init_tabs.py

初始化步骤：

运行初始化脚本
在打开的浏览器标签页中完成各平台的登录
登录完成后关闭脚本

注意：只需在首次使用时执行，之后浏览器的登录状态会保持。

运行主程序

python run_all.py

Excel 批量模式（推荐）

功能特性

Excel批量模式是本工具的主要使用方式，支持以下特性：

批量处理：一次输入多个关键字，自动在所有AI平台搜索
断点续传：程序中断后可从上次位置继续，自动跳过已完成的记录
数据安全：使用临时文件+原子重命名机制，确保数据不会因异常关闭而损坏
实时保存：每个关键字处理完成后立即保存到Excel
进度跟踪：实时显示处理进度和完成状态
灵活控制：可手动标记某些记录为"跳过"

Excel表格结构

所有搜索结果保存在一个Excel表格中，结构如下：

关键字	DeepSeek	豆包	文心一言	Kimi	通义千问	腾讯元宝	是否跳过
上海劳力士维修电话	13812345678, 13987654321	-	13987654321	-	400-123-4567	-
北京百达翡丽服务中心	-	-	-	-	-	-	是
广东欧米茄手表维修电话	13800138000	-	-	13800138000	-	-

说明：

第1行是表头
从第2行开始是数据
多个电话号码用逗号分隔
未找到电话号码显示 -
是否跳过 列标记为 是 的记录不会处理

使用方式

模式1：Excel批量模式（默认）

python run_all.py

选择 1（或直接回车），然后：

输入Excel文件名（默认：keywords_results.xlsx）
- 如果文件存在，自动加载并继续处理
- 如果文件不存在，创建新文件

逐个输入关键字（空行结束）：

关键字 1: 上海劳力士维修电话
关键字 2: 北京百达翡丽服务中心
关键字 3: （直接回车结束）

程序自动处理所有待处理关键字
每完成一个关键字，结果立即保存到Excel

模式2：仅创建Excel模板

python run_all.py

选择 2，然后：

输入Excel文件名
输入关键字列表
程序创建Excel模板，包含所有关键字

之后可以手动编辑Excel（如标记某些记录为"跳过"），再使用模式1进行处理。

断点续传机制

自动标记跳过：

每个关键字处理完成后，自动在"是否跳过"列标记为"是"
下次运行程序时，已标记的记录会被自动跳过
支持随时中断（Ctrl+C），进度已保存，下次继续处理

手动跳过记录：

打开Excel文件
在 是否跳过 列填写 是
保存文件
重新运行程序，这些记录会被自动跳过

Excel文件配置

默认配置在 core/config.py 中：

配置项	默认值	说明
`DEFAULT_EXCEL_FILE`	`keywords_results.xlsx`	默认Excel文件名
`SHEET_NAME`	`搜索结果`	工作表名称
`DATA_START_ROW`	`2`	数据起始行（第1行是表头）

可以在运行时指定其他文件名。

数据安全机制

原子写入：

数据先写入临时文件 ~keywords_results.xlsx.tmp
写入完成后，原文件备份为 .bak
临时文件重命名为目标文件
删除备份文件

异常恢复：

如果程序异常关闭，临时文件会被保留
下次启动时会加载最新成功的版本
不会出现Excel文件损坏的情况

使用示例

# 启动程序
python run_all.py

# 选择模式1（Excel批量模式）
请输入选择 (1-2, 默认1): 1

# 指定文件（回车使用默认文件名）
请输入Excel文件名 (默认: keywords_results.xlsx, 回车使用现有或创建新文件): 

# 输入关键字
请输入关键字列表（每行一个关键字，空行结束）:
关键字 1: 上海劳力士维修中心电话
关键字 2: 北京百达翡丽维修服务中心电话
关键字 3: 广东欧米茄手表维修电话
关键字 4: （回车结束）

# 程序开始处理
=== Excel 批量模式 ===
支持断点续传，自动跳过已完成或标记为跳过的记录

进度统计:
  总计: 3 个关键字
  已完成: 0 个
  待处理: 3 个

开始处理 3 个待处理关键字...

[上海劳力士维修中心电话] (第 2 行)
==================================================

[DeepSeek]
✓ 找到 2 个电话号码
--------------------------------------------------

[豆包]
→ 未找到电话号码
--------------------------------------------------

...

Excel文件管理

查看结果：

直接用Excel/WPS打开 keywords_results.xlsx
每个关键字的搜索结果一目了然
可以手动编辑、添加备注等

继续处理：

如果需要处理新关键字，直接追加到Excel
或重新运行程序创建新文件

备份建议：

定期备份Excel文件
处理大量关键字时建议分批处理

根据提示输入关键词和输出文件名即可在所有平台搜索。

架构设计

核心类关系图

                    ┌─────────────────────┐
                    │     Config          │
                    │  (配置管理-单例)     │
                    └─────────────────────┘
                            │
                            │ 使用
                            ▼
┌───────────────────────┐       ┌─────────────────────┐
│   PlatformBase        │◄──────│  BrowserManager     │
│   (抽象基类)           │       │  (浏览器管理)        │
├───────────────────────┤       └─────────────────────┘
│ + PLATFORM_NAME       │
│ + PLATFORM_URL        │
│ + search()            │◄──────┐
│ + _execute_search()   │       │
│ + _wait_for_response()│       │ 继承
│ + _extract_content()  │       │
│ + _get_input_xpath()  │       │
└───────────────────────┘       │
                                │
        ┌───────────────────────┼───────────────────────┐
        │                       │                       │
        ▼                       ▼                       ▼
┌───────────────┐      ┌───────────────┐      ┌───────────────┐
│ DeepSeekPlatform│     │ DoubaoPlatform│    │ WenxinPlatform│
├───────────────┤      ├───────────────┤      ├───────────────┤
│ (实现特定逻辑) │      │ (实现特定逻辑) │      │ (实现特定逻辑) │
└───────────────┘      └───────────────┘      └───────────────┘
        │                       │                       │
        └───────────────────────┴───────────────────────┘
                                │
                                │ 使用
                                ▼
                    ┌─────────────────────┐
                    │  PhoneExtractor     │
                    │  (电话号码提取器)    │
                    └─────────────────────┘

核心模块说明

1. Config（配置管理）

设计模式：单例模式

集中管理所有配置参数，支持动态更新。

from core import Config

# 获取所有配置
config = Config.get_all()

# 更新配置
Config.update(BROWSER_PORT=9222, PAGE_LOAD_WAIT=5)

# 重置为默认值
Config.reset()

默认配置：

参数	默认值	说明
BROWSER_PORT	9222	浏览器调试端口
BROWSER_HEADLESS	False	是否无头模式
PAGE_LOAD_WAIT	3	页面加载等待时间（秒）
INPUT_WAIT	1	输入后等待时间（秒）
DEFAULT_OUTPUT_FILE	"phones.txt"	默认输出文件名
FILE_ENCODING	"utf-8"	输出文件编码

2. BrowserManager（浏览器管理器）

职责：管理 ChromiumPage 实例的创建、获取和关闭

from core import BrowserManager

# 方式1：手动管理
manager = BrowserManager(port=9222)
page = manager.get_page()
# 使用 page...
manager.close()

# 方式2：上下文管理器（推荐）
with BrowserManager() as manager:
    page = manager.get_page()
    # 使用 page...
    # 退出时自动关闭

3. PhoneExtractor（电话提取器）

职责：从文本中提取并保存电话号码

from core import PhoneExtractor

extractor = PhoneExtractor()

# 提取电话
phones = extractor.extract("客服热线：400-123-4567")

# 保存到文件
extractor.save_to_file(phones, "output.txt", "DeepSeek")

重要：只在 AI 回答区域提取

PhoneExtractor 本身是通用的文本提取工具，不负责识别页面内容区域。各平台需要通过重写 _extract_content() 方法，只返回 AI 回答的内容区域，避免从导航栏、页脚等位置提取错误电话。

正确示例（使用 DrissionPage）：

def _extract_content(self) -> str:
    """只提取 AI 回答内容区域"""
    # 方式1：直接定位回答元素（推荐）
    answer_element = self._page.ele('css:.ai-answer-content', timeout=5)
    if answer_element:
        return answer_element.text

    # 方式2：遍历查找符合条件的元素
    all_divs = self._page.eles('xpath://div')
    for div in all_divs:
        text = div.text
        # 只提取包含"电话"且长度足够的内容块
        if '电话' in text and '用户' not in text and '搜索' not in text and len(text) > 500:
            return text[:5000]

    # 未找到特定区域，返回整个页面
    return self._page.html

DrissionPage 元素操作方法：

方法	说明	示例
`.ele()`	定位单个元素	`page.ele('xpath://div[@class="answer"]')`
`.eles()`	定位多个元素	`page.eles('css:.message')`
`.text`	获取元素文本	`element.text`
`.html`	获取元素HTML	`element.html`
`.click()`	点击元素	`element.click()`
`.input()`	输入文本	`element.input('内容')`
`.run_js()`	执行 JavaScript	`page.run_js('return document.title')`

支持格式（统一在基类维护）：

400/800 电话：400-123-4567
手机号：13812345678、138-1234-5678
座机号：021-12345678、010-12345678
国际号码：+86 21 2319 3688

如需支持新格式，只需在 PhoneExtractor._get_patterns() 中添加正则表达式。

4. PlatformBase（平台基类）

设计模式：模板方法模式

所有平台的抽象基类，定义统一接口和搜索流程骨架。

核心方法：

class PlatformBase(ABC):
    # 子类必须定义的类属性
    PLATFORM_NAME: str = ""      # 平台名称
    PLATFORM_URL: str = ""       # 平台URL
    
    # 子类必须实现的抽象方法
    @abstractmethod
    def search(self, keyword: str, output_file: str = "") -> List[str]:
        """搜索关键词并提取电话号码"""
        pass
    
    @abstractmethod
    def _wait_for_response(self) -> None:
        """等待AI响应完成"""
        pass
    
    @abstractmethod
    def _extract_content(self) -> str:
        """提取AI回答内容"""
        pass
    
    @abstractmethod
    def _get_input_xpath(self) -> str:
        """获取输入框的XPath"""
        pass
    
    # 模板方法（定义算法骨架）
    def _execute_search(self, keyword: str, output_file: str = "") -> List[str]:
        """执行搜索流程"""
        # 1. 导航到平台
        self._navigate_to_platform()
        # 2. 输入关键词
        self._input_keyword(keyword)
        # 3. 等待响应
        self._wait_for_response()
        # 4. 提取内容
        content = self._extract_content()
        # 5. 提取电话号码
        phones = self.phone_extractor.extract(content)
        # 6. 保存结果
        if phones:
            self.phone_extractor.save_to_file(phones, output_file, self.PLATFORM_NAME)
        return phones

可重写的方法：

子类可通过重写以下方法实现平台特定逻辑：

_get_input_xpath() - 自定义输入框定位符
_wait_for_response() - 自定义等待逻辑
_extract_content() - 自定义内容提取逻辑
_navigate_to_platform() - 自定义导航逻辑

5. 异常类

完善的异常体系便于错误处理。

from core.exceptions import (
    PlatformError,
    BrowserConnectionError,
    InputBoxNotFoundError,
    ResponseTimeoutError
)

try:
    platform = DeepSeekPlatform()
    phones = platform.search("关键词")
except InputBoxNotFoundError as e:
    print(f"输入框未找到：{e}")
except ResponseTimeoutError as e:
    print(f"响应超时：{e}")
except PlatformError as e:
    print(f"平台错误：{e}")

新平台接入指南

接入步骤

第一步：创建平台文件

在 platforms/ 目录下创建新文件，例如 new_platform.py：

"""
新平台搜索模块

提供新平台的关键词搜索和电话号码提取功能。
"""
import time
from typing import List
from core.base_platform import PlatformBase


class NewPlatform(PlatformBase):
    """新平台类
    
    继承自 PlatformBase，实现新平台特定的配置和逻辑。
    """
    
    PLATFORM_NAME = "新平台名称"
    PLATFORM_URL = "https://example.com"
    
    def search(self, keyword: str, output_file: str = "") -> List[str]:
        """搜索关键词并提取电话号码
        
        Args:
            keyword: 搜索关键词
            output_file: 输出文件名
        
        Returns:
            提取到的电话号码列表
        """
        return self._execute_search(keyword, output_file)
    
    def _get_input_xpath(self) -> str:
        """获取输入框的XPath表达式
        
        Returns:
            XPath表达式字符串
        """
        # 示例：查找 textarea 元素
        return 'xpath://textarea'
        
        # 或者使用更具体的选择器
        # return 'xpath://textarea[@placeholder="请输入问题"]'
        # return 'css:textarea.input-box'
    
    def _wait_for_response(self) -> None:
        """等待AI响应完成
        
        通过监听页面元素内容变化判断输出是否完成。
        """
        print("等待AI响应...")
        
        # 方式1：检测特定文本是否稳定（推荐）
        stable_count = 0
        stable_threshold = 3
        check_interval = 1
        last_content = ""
        
        while True:
            # 获取页面文本
            body_text = self._page.run_js('return document.body.innerText;')
            
            # 检测是否有完成标记（如"生成完成"）
            if '生成完成' in body_text:
                # 获取"生成完成"周围内容片段
                index = body_text.index('生成完成')
                start = max(0, index - 30)
                end = min(len(body_text), index + 30)
                current_content = body_text[start:end]
                
                if current_content == last_content:
                    stable_count += 1
                    print(f"完成标记稳定 ({stable_count}/{stable_threshold})")
                    if stable_count >= stable_threshold:
                        print("AI响应完成")
                        return
                else:
                    stable_count = 0
                    last_content = current_content
                    print("检测到内容更新")
            
            time.sleep(check_interval)
    
    def _extract_content(self) -> str:
        """提取AI回答内容
        
        只返回AI回答的内容区域，避免提取错误的电话号码。
        
        Returns:
            AI回答的文本内容
        """
        # 方式1：直接定位回答元素（推荐）
        answer_element = self._page.ele('css:.ai-answer-content', timeout=5)
        if answer_element:
            return answer_element.text
        
        # 方式2：遍历查找符合条件的元素
        all_divs = self._page.eles('xpath://div')
        for div in all_divs:
            text = div.text
            # 只提取包含"电话"且长度足够的内容块
            if '电话' in text and '用户' not in text and '搜索' not in text and len(text) > 500:
                return text[:5000]
        
        # 未找到特定区域，返回整个页面
        return self._page.html


# 便捷函数（可选，提供向后兼容）
def search_new_platform(keyword: str, output_file: str = "") -> List[str]:
    """便捷搜索函数
    
    Args:
        keyword: 搜索关键词
        output_file: 输出文件名
    
    Returns:
        提取到的电话号码列表
    """
    platform = NewPlatform()
    try:
        return platform.search(keyword, output_file)
    finally:
        platform.cleanup()

第二步：注册平台

在 platforms/__init__.py 中添加导入和注册：

# 1. 导入新平台
from .new_platform import NewPlatform, search_new_platform

# 2. 添加到 __all__
__all__ = [
    # ... 其他平台
    "NewPlatform",
    "search_new_platform",
]

# 3. 添加到 PLATFORMS 字典
PLATFORMS = {
    # ... 其他平台
    "新平台名称": {
        "class": NewPlatform,
        "function": search_new_platform
    },
}

第三步：测试新平台

创建测试脚本验证功能：

from platforms import NewPlatform

# 测试搜索
platform = NewPlatform()
phones = platform.search("劳力士维修服务中心", "test_output.txt")
print(f"找到电话号码：{phones}")
platform.cleanup()

接入最佳实践

1. 获取输入框 XPath

使用浏览器开发者工具获取输入框的 XPath：

打开平台网站
按 F12 打开开发者工具
点击元素选择器（左上角箭头图标）
点击输入框
在 Elements 标签中右键点击选中的元素
选择 Copy → Copy XPath

常见选择器模式：

# 根据属性选择
'xpath://textarea[@placeholder="请输入问题"]'
'xpath://input[@id="search-input"]'

# 根据 class 选择
'css:textarea.input-box'
'css:#search-input'

# 根据文本内容选择
'xpath://button[contains(text(), "发送")]'

2. 实现 `_wait_for_response()`

关键原则：

使用 while True 持续检测，不使用超时
通过内容是否变化来判断完成状态
可以检测特定文本（如"生成完成"）周围内容是否稳定
也可以检测回答区域内容是否不再变化

示例1：检测完成标记文本

def _wait_for_response(self) -> None:
    """等待豆包思考完成，通过检测内容稳定性判断"""
    stable_count = 0
    stable_threshold = 3
    check_interval = 1
    last_content = ""

    print("等待思考完成...")

    while True:
        body_text = self._page.run_js('return document.body.innerText;')

        if '已完成思考' in body_text:
            # 找到"已完成思考"周围的内容片段
            think_index = body_text.index('已完成思考')
            start = max(0, think_index - 50)
            end = min(len(body_text), think_index + 50)
            current_content = body_text[start:end]

            if current_content == last_content:
                stable_count += 1
                print(f"思考内容稳定 ({stable_count}/{stable_threshold})")
                if stable_count >= stable_threshold:
                    print("思考完成")
                    return
            else:
                stable_count = 0
                last_content = current_content
                print("检测到思考内容更新")
        else:
            stable_count = 0
            last_content = ""
            print("等待'已完成思考'提示...")

        time.sleep(check_interval)

示例2：检测回答区域内容

def _wait_for_response(self) -> None:
    """等待豆包回答完成，通过检测内容稳定性判断"""
    stable_count = 0
    stable_threshold = 3
    check_interval = 1
    last_answer = ""

    print("等待回答完成...")

    while True:
        # 定位回答元素
        answer = self._page.ele('css:.answer-content', timeout=2)
        if answer:
            current_answer = answer.text
            if current_answer == last_answer:
                stable_count += 1
                print(f"回答内容稳定 ({stable_count}/{stable_threshold})")
                if stable_count >= stable_threshold:
                    print("回答完成")
                    return
            else:
                stable_count = 0
                last_answer = current_answer
                print(f"回答内容更新，当前长度: {len(current_answer)}")
        time.sleep(check_interval)

示例3：综合检测（思考+回答）

def _wait_for_response(self) -> None:
    """等待AI响应完成（思考+回答）"""
    # 先等待思考完成
    self._wait_for_thinking_complete()
    # 再等待回答完成
    self._wait_for_answer_complete()

3. 实现 `_extract_content()`

重要原则：

只返回 AI 回答的内容区域
避免从导航栏、页脚、侧边栏等位置提取
优先使用直接定位元素的方式
如果无法定位，可以使用过滤策略

示例1：直接定位回答元素

def _extract_content(self) -> str:
    """只提取 AI 回答内容区域"""
    answer_element = self._page.ele('css:.ai-answer-content', timeout=5)
    if answer_element:
        return answer_element.text
    
    # 降级方案：返回整个页面
    return self._page.html

示例2：遍历查找符合条件的元素

def _extract_content(self) -> str:
    """通过过滤策略提取回答内容"""
    all_divs = self._page.eles('xpath://div')
    for div in all_divs:
        text = div.text
        # 只提取包含"电话"且长度足够的内容块
        if '电话' in text and '用户' not in text and '搜索' not in text and len(text) > 500:
            return text[:5000]
    
    # 未找到特定区域，返回整个页面
    return self._page.html

示例3：使用 JavaScript 提取

def _extract_content(self) -> str:
    """使用 JavaScript 提取内容"""
    js_code = '''
    // 查找包含"电话"的最长文本块
    const divs = document.querySelectorAll('div');
    let longestText = '';
    for (const div of divs) {
        const text = div.innerText || div.textContent;
        if (text.includes('电话') && text.length > longestText.length) {
            longestText = text;
        }
    }
    return longestText || document.body.innerText;
    '''
    return self._page.run_js(js_code)

平台接入检查清单

接入新平台时，请确保完成以下步骤：

参考示例

可以参考已实现的平台作为模板：

deepseek.py - 简单实现
doubao.py - 完整的等待机制实现
wenxin.py - 复杂的导航和输入处理

设计模式

模板方法模式：PlatformBase._execute_search() 定义算法骨架，子类实现细节
单例模式：Config 使用类变量管理全局配置
策略模式：各平台通过重写方法实现不同的搜索策略
工厂模式：通过 PLATFORMS 字典动态创建平台实例

支持的 AI 平台

平台	类名	函数名	状态
DeepSeek	`DeepSeekPlatform`	`search_deepseek()`	✅ 已测试
豆包	`DoubaoPlatform`	`search_doubao()`	✅ 已测试
文心一言	`WenxinPlatform`	`search_wenxin()`	✅ 已测试
Kimi	`KimiPlatform`	`search_kimi()`	✅ 已测试
通义千问	`TongyiPlatform`	`search_tongyi()`	✅ 已测试
腾讯元宝	`YuanbaoPlatform`	`search_yuanbao()`	✅ 已测试