实时数据看板
驱动GEO策略调整场景GEO实战

# GEO实时数据管道核心架构

import asyncio
import schedule
import time
import redis
import json
from datetime import datetime
from playwright.async_api import async_playwright

redis_client = redis.Redis(host="localhost", port=6379, db=0)

class GEODataPipeline:
    def __init__(self):
        self.priority_keywords = {
            "P0": {"keywords": [], "interval_hours": 4, "platforms_count": 5},
            "P1": {"keywords": [], "interval_hours": 12, "platforms_count": 3},
            "P2": {"keywords": [], "interval_hours": 24, "platforms_count": 2}
        }

    async def collect_data(self, priority_level):
        """采集指定优先级的关键词数据"""
        config = self.priority_keywords[priority_level]
        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=True)
            for keyword in config["keywords"]:
                try:
                    page = await browser.new_page()
                    # 搜索逻辑...
                    response_text = "..."  # AI回答文本
                    result = {
                        "timestamp": datetime.now().isoformat(),
                        "keyword": keyword,
                        "priority": priority_level,
                        "brand_found": "YourBrand" in response_text
                    }
                    redis_client.xadd("geo:data_stream", result)
                    await page.close()
                    await asyncio.sleep(5)
                except Exception as e:
                    print(f"Error: {keyword}: {e}")
            await browser.close()

def schedule_pipeline():
    pipeline = GEODataPipeline()
    schedule.every(4).hours.do(lambda: asyncio.run(pipeline.collect_data("P0")))
    schedule.every(12).hours.do(lambda: asyncio.run(pipeline.collect_data("P1")))
    schedule.every(24).hours.do(lambda: asyncio.run(pipeline.collect_data("P2")))
    while True:
        schedule.run_pending()
        time.sleep(60)