Selenium自动化
GEO效果追踪实战指南

# Selenium环境搭建与浏览器配置
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager
import pickle
import time
import os

# Cookie存储路径
COOKIE_FILE = 'cookies/ai_platform_cookies.pkl'

def create_driver(headless=True):
    '''创建配置好的Chrome WebDriver'''
    options = Options()
    if headless:
        options.add_argument('--headless=new')
    options.add_argument('--disable-blink-features=AutomationControlled')
    options.add_argument('--no-sandbox')
    options.add_argument('--disable-dev-shm-usage')
    options.add_argument('--window-size=1920,1080')
    # 禁用图片加载提升速度
    prefs = {'profile.managed_default_content_settings.images': 2}
    options.add_experimental_option('prefs', prefs)
    # 设置User-Agent
    options.add_argument('--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36')

    service = Service(ChromeDriverManager().install())
    driver = webdriver.Chrome(service=service, options=options)
    driver.set_page_load_timeout(30)
    driver.implicitly_wait(15)
    return driver

def save_cookies(driver, platform_name):
    '''保存当前浏览器Cookie'''
    cookies = driver.get_cookies()
    os.makedirs('cookies', exist_ok=True)
    with open(f'cookies/{platform_name}_cookies.pkl', 'wb') as f:
        pickle.dump(cookies, f)
    print(f'Cookie已保存: {len(cookies)}条')

def load_cookies(driver, platform_name, domain):
    '''加载已保存的Cookie'''
    cookie_file = f'cookies/{platform_name}_cookies.pkl'
    if os.path.exists(cookie_file):
        driver.get(domain)  # 先访问域名才能设置Cookie
        with open(cookie_file, 'rb') as f:
            cookies = pickle.load(f)
        for cookie in cookies:
            driver.add_cookie(cookie)
        print(f'Cookie已加载: {len(cookies)}条')
        return True
    return False

# 多AI平台搜索结果抓取
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import time

class ChatGPTScraper:
    def __init__(self, driver):
        self.driver = driver
        self.url = 'https://chatgpt.com'

    def search(self, keyword, brand_name):
        '''在ChatGPT中搜索关键词并分析品牌提及'''
        self.driver.get(self.url)
        time.sleep(3)  # 等待页面加载

        # 输入关键词
        textarea = WebDriverWait(self.driver, 15).until(
            EC.presence_of_element_located((By.CSS_SELECTOR, 'textarea'))
        )
        textarea.send_keys(keyword)
        textarea.send_keys('\n')

        # 等待回答完成（监测停止生成按钮消失）
        time.sleep(5)  # 初始等待生成开始
        for _ in range(60):  # 最多等待60秒
            try:
                stop_btn = self.driver.find_element(By.CSS_SELECTOR, '[data-testid="stop-button"]')
                time.sleep(1)
            except:
                break  # 停止按钮消失，回答完成

        # 提取回答文本
        try:
            response_el = self.driver.find_element(By.CSS_SELECTOR, '[data-testid="final-message"]')
            response_text = response_el.text
        except:
            # 备选选择器
            response_el = self.driver.find_element(By.CSS_SELECTOR, '.markdown')
            response_text = response_el.text

        # 分析品牌提及
        brand_mentioned = brand_name.lower() in response_text.lower()

        return {
            'platform': 'chatgpt',
            'keyword': keyword,
            'response': response_text,
            'brand_mentioned': brand_mentioned,
            'response_length': len(response_text),
            'timestamp': int(time.time())
        }

class WenxinScraper:
    def __init__(self, driver):
        self.driver = driver
        self.url = 'https://yiyan.baidu.com'

    def search(self, keyword, brand_name):
        '''在文心一言中搜索'''
        self.driver.get(self.url)
        # 加载已保存的登录Cookie
        load_cookies(self.driver, 'wenxin', self.url)
        self.driver.refresh()
        time.sleep(3)

        # 输入并发送
        input_box = WebDriverWait(self.driver, 15).until(
            EC.presence_of_element_located((By.CSS_SELECTOR, '.chat-input textarea'))
        )
        input_box.send_keys(keyword)
        send_btn = self.driver.find_element(By.CSS_SELECTOR, '.send-btn')
        send_btn.click()

        # 等待回答完成（最多等待60秒）
        time.sleep(10)
        for _ in range(50):
            try:
                # 检测是否还在生成
                loading = self.driver.find_element(By.CSS_SELECTOR, '.loading-indicator')
                time.sleep(1)
            except:
                break

        response_text = self.driver.find_element(By.CSS_SELECTOR, '.response-content').text
        brand_mentioned = brand_name in response_text

        return {
            'platform': 'wenxin',
            'keyword': keyword,
            'response': response_text,
            'brand_mentioned': brand_mentioned,
            'response_length': len(response_text),
            'timestamp': int(time.time())
        }

# 效果指标计算与趋势分析
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

class GEOEffectAnalyzer:
    def __init__(self, results_file):
        '''加载检测结果数据'''
        self.df = pd.read_json(results_file, lines=True)
        self.df['date'] = pd.to_datetime(self.df['timestamp'], unit='s').dt.date

    def calculate_metrics(self):
        '''计算核心GEO效果指标'''
        metrics = {}

        # 品牌提及率
        metrics['mention_rate'] = self.df['brand_mentioned'].mean()

        # 分平台提及率
        platform_rates = self.df.groupby('platform')['brand_mentioned'].mean()
        metrics['by_platform'] = platform_rates.to_dict()

        # 平台覆盖率（品牌在多少个平台上被提及）
        platforms_with_mention = self.df[self.df['brand_mentioned']]['platform'].nunique()
        total_platforms = self.df['platform'].nunique()
        metrics['platform_coverage'] = platforms_with_mention / total_platforms

        # 平均回答长度
        metrics['avg_response_length'] = self.df['response_length'].mean()

        return metrics

    def trend_analysis(self, window=7):
        '''趋势分析（7日移动平均）'''
        daily = self.df.groupby('date')['brand_mentioned'].mean().reset_index()
        daily['ma'] = daily['brand_mentioned'].rolling(window=window, min_periods=1).mean()
        daily['trend'] = np.where(daily['ma'] > daily['ma'].shift(1), 'up',
                            np.where(daily['ma'] < daily['ma'].shift(1), 'down', 'flat'))
        return daily

    def before_after_comparison(self, optimization_date):
        '''优化前后效果对比'''
        opt_date = pd.to_datetime(optimization_date).date()
        before = self.df[self.df['date'] < opt_date]
        after = self.df[self.df['date'] >= opt_date]

        if len(before) == 0 or len(after) == 0:
            return None

        return {
            'before_mention_rate': before['brand_mentioned'].mean(),
            'after_mention_rate': after['brand_mentioned'].mean(),
            'improvement': after['brand_mentioned'].mean() - before['brand_mentioned'].mean(),
            'improvement_pct': (after['brand_mentioned'].mean() - before['brand_mentioned'].mean()) / before['brand_mentioned'].mean() * 100 if before['brand_mentioned'].mean() > 0 else 0
        }

# 使用示例
# analyzer = GEOEffectAnalyzer('results/detection_20260811.jsonl')
# metrics = analyzer.calculate_metrics()
# print(f'品牌提及率: {metrics["mention_rate"]:.1%}')
# trend = analyzer.trend_analysis()
# print(f'最近趋势: {trend.iloc[-1]["trend"]}')
# comparison = analyzer.before_after_comparison('2026-08-01')
# print(f'优化效果: 提及率提升 {comparison["improvement_pct"]:.1f}%')

# 分布式Selenium任务调度（Redis队列）
import redis
import json
import time
from selenium import webdriver

# 任务队列管理
class TaskQueue:
    def __init__(self, redis_host='localhost', redis_port=6379):
        self.redis = redis.Redis(host=redis_host, port=redis_port)
        self.queue_name = 'geo_detection_tasks'
        self.result_queue = 'geo_detection_results'

    def push_task(self, keyword, platform, brand_name):
        '''推送检测任务到队列'''
        task = json.dumps({
            'keyword': keyword,
            'platform': platform,
            'brand_name': brand_name,
            'created_at': time.time()
        })
        self.redis.lpush(self.queue_name, task)

    def pop_task(self, timeout=30):
        '''从队列拉取检测任务（阻塞模式）'''
        result = self.redis.brpop(self.queue_name, timeout=timeout)
        if result:
            return json.loads(result[1])
        return None

    def push_result(self, result):
        '''推送检测结果'''
        self.redis.lpush(self.result_queue, json.dumps(result, ensure_ascii=False))

# 工作节点
class WorkerNode:
    def __init__(self, worker_id):
        self.worker_id = worker_id
        self.queue = TaskQueue()
        self.driver = create_driver(headless=True)
        self.scrapers = {
            'chatgpt': ChatGPTScraper(self.driver),
            'wenxin': WenxinScraper(self.driver)
        }

    def run(self, max_tasks=50):
        '''运行工作节点，处理最多max_tasks个任务'''
        processed = 0
        while processed < max_tasks:
            task = self.queue.pop_task(timeout=30)
            if not task:
                print(f'Worker {self.worker_id}: 队列为空，退出')
                break

            scraper = self.scrapers.get(task['platform'])
            if not scraper:
                print(f'Worker {self.worker_id}: 不支持平台 {task["platform"]}')
                continue

            try:
                result = scraper.search(task['keyword'], task['brand_name'])
                result['worker_id'] = self.worker_id
                self.queue.push_result(result)
                processed += 1
                print(f'Worker {self.worker_id}: [{processed}/{max_tasks}] {task["platform"]} | {task["keyword"]} | 提及: {result["brand_mentioned"]}')
            except Exception as e:
                print(f'Worker {self.worker_id}: 失败 {task["keyword"]} - {e}')

        self.driver.quit()
        print(f'Worker {self.worker_id}: 完成，共处理 {processed} 个任务')