Playwright自动化
AI搜索检测脚本开发指南

# 环境安装命令
# pip install playwright
# playwright install chromium

# 验证安装
from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    browser = p.chromium.launch(headless=True)
    page = browser.new_page()
    page.goto('https://example.com')
    print(f'Page title: {page.title()}')
    browser.close()
    print('Playwright environment ready!')

# 多平台检测脚本架构
class PlatformDetector:
    '''AI平台检测器基类'''
    def __init__(self, headless=True):
        self.headless = headless
        self.platform_name = 'base'

    def detect(self, page, keyword, brand_name):
        '''执行检测，返回结果字典'''
        raise NotImplementedError

    def _check_mention(self, text, brand_name):
        '''检查品牌是否被提及'''
        return brand_name.lower() in text.lower()


class ChatGPTDetector(PlatformDetector):
    def __init__(self, headless=True):
        super().__init__(headless)
        self.platform_name = 'chatgpt'
        self.url = 'https://chatgpt.com'

    def detect(self, page, keyword, brand_name):
        page.goto(self.url)
        page.wait_for_selector('textarea', timeout=15000)
        page.fill('textarea', keyword)
        page.press('textarea', 'Enter')
        # 等待AI回答完成
        page.wait_for_selector('[data-testid="final-message"]', timeout=60000)
        response_text = page.text_content('[data-testid="final-message"]')
        return {
            'platform': self.platform_name,
            'keyword': keyword,
            'response': response_text,
            'brand_mentioned': self._check_mention(response_text, brand_name),
            'timestamp': page.evaluate('Date.now()')
        }


class KimiDetector(PlatformDetector):
    def __init__(self, headless=True):
        super().__init__(headless)
        self.platform_name = 'kimi'
        self.url = 'https://kimi.moonshot.cn'

    def detect(self, page, keyword, brand_name):
        page.goto(self.url)
        page.wait_for_selector('.chat-input', timeout=15000)
        page.fill('.chat-input', keyword)
        page.click('.send-button')
        page.wait_for_selector('.response-content', timeout=60000)
        response_text = page.text_content('.response-content')
        return {
            'platform': self.platform_name,
            'keyword': keyword,
            'response': response_text,
            'brand_mentioned': self._check_mention(response_text, brand_name),
            'timestamp': page.evaluate('Date.now()')
        }

# 反检测与稳定性保障实现
import random
import time
from playwright.sync_api import sync_playwright

# Stealth模式配置
STEALTH_JS = '''
// 隐藏webdriver标志
Object.defineProperty(navigator, 'webdriver', {get: () => undefined});
// 模拟真实浏览器插件
Object.defineProperty(navigator, 'plugins', {
    get: () => [1, 2, 3, 4, 5]
});
// 模拟真实语言
Object.defineProperty(navigator, 'languages', {
    get: () => ['zh-CN', 'zh', 'en']
});
'''

def create_stealth_browser(playwright, headless=True):
    browser = playwright.chromium.launch(
        headless=headless,
        args=[
            '--disable-blink-features=AutomationControlled',
            '--no-sandbox',
            '--disable-dev-shm-usage'
        ]
    )
    context = browser.new_context(
        user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
        viewport={'width': 1920, 'height': 1080},
        locale='zh-CN'
    )
    context.add_init_script(STEALTH_JS)
    return browser, context

def human_delay(min_sec=3, max_sec=8):
    '''模拟人类操作延迟'''
    time.sleep(random.uniform(min_sec, max_sec))

def retry_with_backoff(func, max_retries=3):
    '''带退避的重试机制'''
    for attempt in range(max_retries):
        try:
            return func()
        except Exception as e:
            wait = (attempt + 1) * 10
            print(f'Attempt {attempt+1} failed: {e}, retry in {wait}s')
            time.sleep(wait)
    raise Exception(f'Failed after {max_retries} retries')

# 定时调度配置 (Linux crontab)
# 每天早上8点执行核心关键词检测
# 0 8 * * * cd /opt/geo-detector && python -X utf8 run_detection.py --mode=daily

# 每周一早上8点执行全量关键词检测
# 0 8 * * 1 cd /opt/geo-detector && python -X utf8 run_detection.py --mode=weekly

# run_detection.py 核心调度逻辑
import argparse
import json
from datetime import datetime

def run_detection(mode='daily'):
    config = json.load(open('config.json', 'r', encoding='utf-8'))
    if mode == 'daily':
        keywords = config['core_keywords']  # 10-20个核心词
    else:
        keywords = config['all_keywords']   # 50-100个全量词

    results = []
    for kw in keywords:
        for platform in config['platforms']:
            try:
                result = detect_keyword(kw, platform, config['brand_name'])
                results.append(result)
                print(f'OK: {platform} | {kw}')
            except Exception as e:
                print(f'FAIL: {platform} | {kw} | {e}')
                results.append({'keyword': kw, 'platform': platform, 'error': str(e)})

    # 保存结果
    output_file = f'results/detection_{datetime.now().strftime("%Y%m%d_%H%M%S")}.jsonl'
    with open(output_file, 'w', encoding='utf-8') as f:
        for r in results:
            f.write(json.dumps(r, ensure_ascii=False) + '\n')

    # 生成摘要
    total = len(results)
    mentioned = sum(1 for r in results if r.get('brand_mentioned'))
    print(f'\nDetection complete: {mentioned}/{total} mentioned ({mentioned/total:.1%})')
    print(f'Results saved to: {output_file}')

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--mode', default='daily', choices=['daily', 'weekly'])
    args = parser.parse_args()
    run_detection(args.mode)