如何复用GA Reporting API V4 Python脚本认证GA4 API？遇授权格式错误

错误信息 #

Authorized user info was not in the expected format, missing fields refresh_token, client_secret

问题根源 #

你使用的是服务账号密钥文件，但错误调用了Credentials.from_authorized_user_file()——这个方法仅适用于OAuth2个人授权用户的token文件，完全不兼容服务账号的JSON格式，因此会报缺少个人授权所需的refresh_token和client_secret字段。

此前GA Reporting API V4使用的是oauth2client.service_account.ServiceAccountCredentials进行服务账号认证，但当前代码错误替换为个人用户的google.oauth2.credentials.Credentials，导致认证逻辑完全不匹配。

修复步骤 #

1. 保留现有服务账号密钥文件（client_secrets_services.json），改用服务账号专用的认证方法：google.oauth2.service_account.Credentials.from_service_account_file()
2. 确保服务账号已被添加到GA4媒体资源的查看者/编辑者权限（在GA4后台「管理」-「访问权限管理」中配置）
3. 修正initialize_analyticsreporting()函数的认证逻辑

修改后的完整代码 #

from googleapiclient.discovery import build
from google.oauth2.service_account import Credentials  # 替换为服务账号专用凭证类
import pandas as pd
from datetime import date
from datetime import timedelta

SCOPES = ['https://www.googleapis.com/auth/analytics.readonly']
KEY_FILE_LOCATION = 'client_secrets_services.json'


def initialize_analyticsreporting():
    """初始化GA4 Data API服务对象"""
    # 加载服务账号凭证
    credentials = Credentials.from_service_account_file(
        KEY_FILE_LOCATION, scopes=SCOPES)
  
    # 构建GA4 Data API服务实例
    analytics = build('analyticsdata', 'v1beta', credentials=credentials)

    return analytics

def get_report(analytics, start_date, end_date_delta):
    """调用GA4 Data API获取报告数据"""
    return analytics.properties().batchRunReports(
        property='properties/3220*****',
        body={
            'requests': [
                {
                    'limit': 100000,
                    'dateRanges': [
                        {'startDate': start_date.strftime("%Y-%m-%d"), 
                         'endDate': (start_date + end_date_delta).strftime("%Y-%m-%d")}
                    ],           
                    'metrics': [ {'name': 'screenPageViews'} ],
                    'dimensions': [
                        {'name':'date'}, 
                        {'name':'sessionSourceMedium'},  
                        {'name':'pagePath'} 
                    ],
                    'dimensionFilter': {  # GA4 API要求该字段为对象，原代码错误使用数组格式
                        "filters": [ 
                            {
                                "dimensionName": "pagePath",
                                "operator": "REGEXP",
                                "expressions": "/pharmacysuccess"
                            }
                        ]
                    }
                }
            ]
        }
    ).execute()

def print_response(response):
    """解析并打印GA4 API响应结果"""
    for report in response.get('reports', []):
        columnHeader = report.get('columnHeader', {})
        dimensionHeaders = columnHeader.get('dimensions', [])
        metricHeaders = columnHeader.get('metricHeader', {}).get('metricHeaderEntries', [])

        for row in report.get('data', {}).get('rows', []):
            dimensions = row.get('dimensions', [])
            dateRangeValues = row.get('metrics', [])

            for header, dimension in zip(dimensionHeaders, dimensions):
                print(f"{header}: {dimension}")

            for i, values in enumerate(dateRangeValues):
                print(f'Date range: {i}')
                for metricHeader, value in zip(metricHeaders, values.get('values')):
                    print(f"{metricHeader.get('name')}: {value}")

def main():
    final_data = pd.DataFrame()
    start_date = date(2023, 1, 1)
    end_date = date(2023, 2, 1)
    delta = timedelta(days=2)
    end_date_delta = timedelta(days=1)
    analytics = initialize_analyticsreporting()
    
    while start_date < end_date:
        response = get_report(analytics, start_date, end_date_delta)
        report = response['reports'][0]
        
        # 构建结果列名
        columns = report['columnHeader']['dimensions'].copy()
        columns.extend([m['name'] for m in report['columnHeader']['metricHeader']['metricHeaderEntries']])
        
        # 处理数据行
        if report['data'].get("rows") is not None:
            # 合并维度与指标数据
            rows = [row['dimensions'] + row['metrics'][0]['values'] for row in report['data']['rows']]
            response_final = pd.DataFrame(rows, columns=columns)
            final_data = pd.concat([final_data, response_final], ignore_index=True)
        
        start_date += delta
    
    print(final_data)
  
if __name__ == '__main__':
    main()

额外修正说明 #

• 修复了dimensionFilter字段格式错误（GA4 API要求为对象，原代码使用数组）
• 移除了无用的导入模块（如openpyxl、csv等），精简代码
• 用pd.concat替代已弃用的pd.DataFrame.append()方法

内容的提问来源于stack exchange，提问作者Akash Agrawal