基于 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

初始化步骤：

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

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

python run_all.py

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

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

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

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

说明：

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

python run_all.py

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

1. 输入Excel文件名（默认：keywords_results.xlsx）

   • 如果文件存在，自动加载并继续处理
   • 如果文件不存在，创建新文件

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

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

3. 程序自动处理所有待处理关键字

4. 每完成一个关键字，结果立即保存到Excel

python run_all.py

选择 2，然后：

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

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

自动标记跳过：

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

手动跳过记录：

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

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

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

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

原子写入：

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

异常恢复：

• 如果程序异常关闭，临时文件会被保留
• 下次启动时会加载最新成功的版本
• 不会出现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/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     │
                    │  (电话号码提取器)    │
                    └─────────────────────┘

设计模式：单例模式

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

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"	输出文件编码

职责：管理 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...
    # 退出时自动关闭

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

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() 中添加正则表达式。

设计模式：模板方法模式

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

核心方法：

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() - 自定义导航逻辑

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

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()

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

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

常见选择器模式：

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

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

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

关键原则：

• 使用 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()

重要原则：

• 只返回 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)

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

• 创建平台文件（如 new_platform.py）
• 定义 PLATFORM_NAME 和 PLATFORM_URL
• 实现 search() 方法
• 实现 _get_input_xpath() 方法
• 实现 _wait_for_response() 方法（使用内容稳定性检测）
• 实现 _extract_content() 方法（只提取回答区域）
• 在 platforms/__init__.py 中注册平台
• 测试搜索功能是否正常
• 测试电话号码提取是否准确
• 测试异常处理是否完善

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

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

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

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

1. 浏览器要求：首次运行时会自动打开 Chrome/Edge 浏览器，请确保已安装
2. 等待机制：不同 AI 平台的响应特征不同，建议重写 _wait_for_response() 方法实现精确判断
3. 内容提取：各平台应重写 _extract_content() 方法，只返回 AI 回答区域，避免提取错误的电话号码
4. 网络环境：需要稳定的网络连接访问各 AI 平台
5. 资源清理：使用 platform.cleanup() 或上下文管理器确保浏览器资源正确释放

MIT License