教程：构建用于自动化报告的 Python SEO 数据管道

使用 Python 构建一个自动化 SEO 数据管道，从多个数据源采集数据、进行处理，并生成自动化报告。

前置条件

第 1 步：搭建项目结构

seo-pipeline/
├── config/
│   ├── credentials.json
│   └── settings.yaml
├── src/
│   ├── extractors/
│   │   ├── gsc_extractor.py
│   │   ├── ga4_extractor.py
│   │   └── ahrefs_extractor.py
│   ├── processors/
│   │   ├── data_processor.py
│   │   └── keyword_processor.py
│   ├── reporters/
│   │   ├── weekly_report.py
│   │   └── monthly_report.py
│   └── main.py
├── data/
│   ├── raw/
│   └── processed/
└── output/
    └── reports/

第 2 步：创建数据提取器

GSC 提取器

import pandas as pd
from google.oauth2 import service_account
from googleapiclient.discovery import build

class GSCExtractor:
    def __init__(self, credentials_path, site_url):
        self.credentials = service_account.Credentials.from_service_account_file(
            credentials_path,
            scopes=['https://www.googleapis.com/auth/webmasters.readonly']
        )
        self.service = build('searchconsole', 'v1', credentials=self.credentials)
        self.site_url = site_url

    def extract_performance(self, start_date, end_date, dimensions=None):
        if dimensions is None:
            dimensions = ['query', 'page']

        request = {
            'startDate': start_date,
            'endDate': end_date,
            'dimensions': dimensions,
            'rowLimit': 25000,
        }

        response = self.service.searchanalytics().query(
            siteUrl=self.site_url, body=request
        ).execute()

        rows = []
        for row in response.get('rows', []):
            data = dict(zip(dimensions, row['keys']))
            data.update({
                'clicks': row['clicks'],
                'impressions': row['impressions'],
                'ctr': round(row['ctr'], 4),
                'position': round(row['position'], 1)
            })
            rows.append(data)

        return pd.DataFrame(rows)

GA4 提取器

from google.analytics.data_v1beta import BetaAnalyticsDataClient
from google.analytics.data_v1beta import RunReportRequest, DateRange, Dimension, Metric

class GA4Extractor:
    def __init__(self, credentials_path, property_id):
        self.client = BetaAnalyticsDataClient.from_service_account_file(credentials_path)
        self.property_id = property_id

    def extract_organic_traffic(self, start_date, end_date):
        request = RunReportRequest(
            property=f'properties/{self.property_id}',
            date_ranges=[DateRange(start_date=start_date, end_date=end_date)],
            dimensions=[
                Dimension(name='pagePath'),
                Dimension(name='deviceCategory'),
            ],
            metrics=[
                Metric(name='sessions'),
                Metric(name='averageSessionDuration'),
                Metric(name='conversions'),
            ],
            dimension_filter=self._organic_filter(),
        )
        response = self.client.run_report(request)

        rows = []
        for row in response.rows:
            rows.append({
                'page_path': row.dimension_values[0].value,
                'device': row.dimension_values[1].value,
                'sessions': int(row.metric_values[0].value),
                'avg_session_duration': float(row.metric_values[1].value),
                'conversions': int(row.metric_values[2].value),
            })

        return pd.DataFrame(rows)

    def _organic_filter(self):
        from google.analytics.data_v1beta import FilterExpression, Filter
        return FilterExpression(
            filter=Filter(
                field_name='medium',
                string_filter=Filter.StringFilter(value='organic')
            )
        )

第 3 步：数据处理

class SEODataProcessor:
    def __init__(self, gsc_data, ga4_data):
        self.gsc_data = gsc_data
        self.ga4_data = ga4_data

    def find_opportunities(self):
        # Keywords with high impressions but low CTR
        opportunities = self.gsc_data[
            (self.gsc_data['impressions'] > 100) &
            (self.gsc_data['ctr'] < 0.05) &
            (self.gsc_data['position'] > 5) &
            (self.gsc_data['position'] < 20)
        ].sort_values('impressions', ascending=False)
        return opportunities

    def find_quick_wins(self):
        # Keywords ranking 11-20 with decent volume
        quick_wins = self.gsc_data[
            (self.gsc_data['position'] >= 11) &
            (self.gsc_data['position'] <= 20) &
            (self.gsc_data['impressions'] > 50)
        ].sort_values('impressions', ascending=False)
        return quick_wins

    def calculate_content_roi(self):
        # Merge GSC and GA4 data on page URL
        merged = pd.merge(
            self.gsc_data.groupby('page').agg({'clicks': 'sum', 'impressions': 'sum'}),
            self.ga4_data.groupby('page_path').agg({'sessions': 'sum', 'conversions': 'sum'}),
            left_index=True, right_index=True, how='outer'
        )
        return merged.fillna(0)

第 4 步：生成报告

class WeeklySEOReport:
    def __init__(self, processor):
        self.processor = processor

    def generate(self, output_path):
        report = {
            'opportunities': self.processor.find_opportunities().head(20),
            'quick_wins': self.processor.find_quick_wins().head(15),
            'content_roi': self.processor.calculate_content_roi(),
        }

        # Export to Excel with multiple sheets
        with pd.ExcelWriter(output_path) as writer:
            report['opportunities'].to_excel(writer, sheet_name='Opportunities')
            report['quick_wins'].to_excel(writer, sheet_name='Quick Wins')
            report['content_roi'].to_excel(writer, sheet_name='Content ROI')

        return report

第 5 步：调度管道

# Cron job for weekly report generation
0 6 * * 1 cd /path/to/seo-pipeline && python src/main.py >> /var/log/seo-pipeline.log 2>&1

验证清单