GEO数据管道
从采集到可视化全流程

# 数据采集层：统一采集器框架
import abc
import json
from datetime import datetime

class BaseCollector(abc.ABC):
    '''数据采集器基类'''
    def __init__(self, source_name):
        self.source_name = source_name

    @abc.abstractmethod
    def collect(self, **kwargs):
        '''执行数据采集，返回标准格式数据列表'''
        pass

    def _format_record(self, entity_type, payload, metadata=None):
        '''格式化为标准数据结构'''
        return {
            'data_source': self.source_name,
            'entity_type': entity_type,
            'timestamp': datetime.now().isoformat(),
            'payload': payload,
            'metadata': metadata or {}
        }

class GoogleAnalyticsCollector(BaseCollector):
    '''Google Analytics数据采集器'''
    def __init__(self, ga_credentials):
        super().__init__('google_analytics')
        self.credentials = ga_credentials

    def collect(self, start_date, end_date, metrics=['sessions', 'pageviews']):
        '''从GA API采集流量数据'''
        # 模拟GA API调用
        # 实际使用时替换为google-analytics-data API调用
        data = []
        for date in self._date_range(start_date, end_date):
            record = self._format_record(
                entity_type='traffic_daily',
                payload={
                    'date': date,
                    'sessions': 1200,  # 实际从API获取
                    'pageviews': 3500,
                    'ai_search_visits': 85  # AI搜索来源流量
                }
            )
            data.append(record)
        return data

    def _date_range(self, start, end):
        from datetime import datetime, timedelta
        d1 = datetime.strptime(start, '%Y-%m-%d')
        d2 = datetime.strptime(end, '%Y-%m-%d')
        dates = []
        while d1 <= d2:
            dates.append(d1.strftime('%Y-%m-%d'))
            d1 += timedelta(days=1)
        return dates

class CrawlerLogCollector(BaseCollector):
    '''爬虫日志采集器'''
    def __init__(self, log_path):
        super().__init__('crawler_log')
        self.log_path = log_path

    def collect(self, date=None):
        '''解析Nginx日志，提取AI爬虫活动'''
        data = []
        log_file = f'{self.log_path}/access_{date or datetime.now().strftime("%Y%m%d")}.log'
        try:
            with open(log_file, 'r', encoding='utf-8') as f:
                for line in f:
                    # 使用第75章的日志解析逻辑
                    parsed = parse_nginx_log(line)
                    if parsed and parsed.get('crawler'):
                        record = self._format_record(
                            entity_type='crawler_request',
                            payload=parsed
                        )
                        data.append(record)
        except FileNotFoundError:
            print(f'日志文件不存在: {log_file}')
        return data

# ETL流程：数据清洗与聚合
import psycopg2
from psycopg2.extras import execute_batch
import pandas as pd
from datetime import datetime, timedelta

class GEODataPipeline:
    def __init__(self, db_config):
        self.conn = psycopg2.connect(**db_config)

    def run_daily_etl(self, target_date=None):
        '''执行每日ETL流程'''
        date = target_date or (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
        print(f'开始ETL流程: {date}')

        # Step 1: Extract - 提取当日原始数据
        raw_data = self._extract(date)
        print(f'  提取数据: {len(raw_data)} 条')

        # Step 2: Transform - 数据清洗与聚合
        metrics = self._transform(raw_data, date)
        print(f'  聚合指标: {len(metrics)} 条')

        # Step 3: Load - 写入事实表
        self._load(metrics)
        print(f'  写入完成: {date} ETL完成')

    def _extract(self, date):
        '''提取当日原始数据'''
        query = '''
        SELECT payload->>'keyword' as keyword,
               payload->>'platform' as platform,
               (payload->>'brand_mentioned')::boolean as brand_mentioned,
               payload->>'response' as response,
               timestamp
        FROM raw_data
        WHERE entity_type = 'ai_search_result'
          AND DATE(timestamp) = %s
        '''
        df = pd.read_sql(query, self.conn, params=(date,))
        return df

    def _transform(self, df, date):
        '''数据清洗与聚合计算'''
        if df.empty:
            return []

        # 按关键词×平台聚合
        agg = df.groupby(['keyword', 'platform']).agg(
            total_checks=('brand_mentioned', 'count'),
            brand_mentions=('brand_mentioned', 'sum')
        ).reset_index()
        agg['mention_rate'] = agg['brand_mentions'] / agg['total_checks']
        agg['date'] = date

        return agg.to_dict('records')

    def _load(self, metrics):
        '''写入事实表'''
        query = '''
        INSERT INTO fact_daily_metrics
            (date, keyword, platform, total_checks, brand_mentions, mention_rate)
        VALUES (%(date)s, %(keyword)s, %(platform)s, %(total_checks)s, %(brand_mentions)s, %(mention_rate)s)
        ON CONFLICT (date, keyword, platform) DO UPDATE
        SET total_checks = EXCLUDED.total_checks,
            brand_mentions = EXCLUDED.brand_mentions,
            mention_rate = EXCLUDED.mention_rate
        '''
        with self.conn.cursor() as cur:
            execute_batch(cur, query, metrics)
        self.conn.commit()

# 分析层：指标计算与异常检测
import numpy as np
from scipy import stats

class GEOAnalyzer:
    def __init__(self, db_config):
        self.db_config = db_config

    def calculate_core_metrics(self, date):
        '''计算核心GEO效果指标'''
        conn = psycopg2.connect(**self.db_config)

        # 整体提及率
        query = 'SELECT AVG(mention_rate) as overall_rate FROM fact_daily_metrics WHERE date = %s'
        overall = pd.read_sql(query, conn, params=(date,)).iloc[0]['overall_rate']

        # 分平台提及率
        query = 'SELECT platform, AVG(mention_rate) as rate FROM fact_daily_metrics WHERE date = %s GROUP BY platform'
        by_platform = pd.read_sql(query, conn, params=(date,))

        # 7日移动平均
        query = '''SELECT date, AVG(mention_rate) as daily_rate
                   FROM fact_daily_metrics
                   WHERE date >= %s::date - 7
                   GROUP BY date ORDER BY date'''
        trend = pd.read_sql(query, conn, params=(date,))
        trend['ma7'] = trend['daily_rate'].rolling(7, min_periods=1).mean()

        conn.close()

        return {
            'date': date,
            'overall_mention_rate': float(overall) if overall else 0,
            'by_platform': by_platform.to_dict('records'),
            'trend': trend.to_dict('records')
        }

    def detect_anomalies(self, date, lookback_days=30):
        '''异常检测：Z-Score方法'''
        conn = psycopg2.connect(**self.db_config)
        query = '''
        SELECT date, AVG(mention_rate) as daily_rate
        FROM fact_daily_metrics
        WHERE date >= %s::date - %s AND date < %s
        GROUP BY date ORDER BY date
        '''
        historical = pd.read_sql(query, conn, params=(date, lookback_days, date))

        # 获取当日数据
        query_today = 'SELECT AVG(mention_rate) as rate FROM fact_daily_metrics WHERE date = %s'
        today_rate = pd.read_sql(query_today, conn, params=(date,)).iloc[0]['rate']
        conn.close()

        if today_rate is None or len(historical) < 7:
            return {'anomaly': False, 'reason': '数据不足'}

        mean = historical['daily_rate'].mean()
        std = historical['daily_rate'].std()
        z_score = (today_rate - mean) / std if std > 0 else 0

        return {
            'anomaly': abs(z_score) > 2,
            'z_score': float(z_score),
            'today_rate': float(today_rate),
            'historical_mean': float(mean),
            'historical_std': float(std),
            'direction': '下降' if z_score < -2 else '上升' if z_score > 2 else '正常'
        }