如何扩展Google Ads Python库自定义服务与插件开发指南【免费下载链接】googleads-python-libThe Python client library for Googles Ads APIs项目地址: https://gitcode.com/gh_mirrors/go/googleads-python-lib想要提升Google Ads API的使用效率本文将为您详细介绍如何扩展Google Ads Python库通过自定义服务和插件开发来满足您的个性化需求。Google Ads Python库是一个强大的客户端库但有时标准功能可能无法完全满足您的特定业务场景。通过掌握扩展技巧您可以构建更灵活、更高效的广告管理解决方案。 Google Ads Python库扩展基础Google Ads Python库基于SOAP协议构建提供了完整的Ad Manager API访问能力。要理解如何扩展它首先需要了解其核心架构客户端类AdManagerClient是主要的入口点负责认证和服务的创建服务代理通过GetService()方法获取各种API服务的代理对象插件系统基于Zeep SOAP库的插件机制支持自定义日志、缓存等配置管理通过YAML配置文件管理认证信息和客户端设置️ 自定义服务开发实战1. 创建自定义服务代理当标准服务无法满足需求时您可以创建自定义服务代理。以下是一个创建自定义报告服务的示例from googleads import ad_manager class CustomReportService: def __init__(self, client, versionv202511): self.client client self.version version self.service client.GetService(ReportService, versionversion) def run_custom_delivery_report(self, report_query): 运行自定义交付报告 # 添加自定义逻辑 report_query[dimensions] [DATE, AD_UNIT_NAME] report_query[columns] [TOTAL_LINE_ITEM_LEVEL_IMPRESSIONS] # 调用标准报告服务 report_job self.service.runReportJob(report_query) return self._process_custom_report(report_job) def _process_custom_report(self, report_job): 自定义报告处理逻辑 # 添加业务特定的处理逻辑 processed_data [] # ... 自定义处理代码 return processed_data2. 扩展现有服务功能您可以通过继承现有服务类来添加新功能from googleads import ad_manager class EnhancedCreativeService(ad_manager.AdManagerClient): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.creative_service self.GetService(CreativeService) def create_creative_with_validation(self, creative_data): 创建创意前进行自定义验证 if not self._validate_creative_data(creative_data): raise ValueError(创意数据验证失败) return self.creative_service.createCreatives([creative_data]) def _validate_creative_data(self, creative_data): 自定义验证逻辑 # 添加您的业务验证规则 required_fields [name, advertiserId, size] return all(field in creative_data for field in required_fields) 插件开发与集成1. 创建自定义Zeep插件Zeep插件系统允许您拦截和修改SOAP请求/响应。以下是一个自定义日志插件的示例import logging from lxml import etree import zeep class CustomZeepPlugin(zeep.Plugin): 自定义Zeep插件添加业务特定的日志记录 def __init__(self, custom_loggerNone): self.logger custom_logger or logging.getLogger(custom_soap) def ingress(self, envelope, http_headers, operation): 处理响应 # 记录响应摘要 self.logger.info(f收到{operation.name}的响应) # 添加自定义响应处理逻辑 if fault in envelope.tag.lower(): self._handle_soap_fault(envelope) return envelope, http_headers def egress(self, envelope, http_headers, operation, binding_options): 处理请求 # 记录请求摘要 self.logger.info(f发送{operation.name}请求到{binding_options[address]}) # 添加自定义请求处理逻辑 self._add_custom_headers(envelope) return envelope, http_headers def _handle_soap_fault(self, envelope): 自定义SOAP错误处理 fault_element envelope.find(.//{http://schemas.xmlsoap.org/soap/envelope/}Fault) if fault_element is not None: fault_string fault_element.find(faultstring) if fault_string is not None: self.logger.error(fSOAP错误: {fault_string.text}) def _add_custom_headers(self, envelope): 添加自定义SOAP头 # 添加业务特定的SOAP头部信息 header etree.SubElement(envelope, {http://schemas.xmlsoap.org/soap/envelope/}Header) custom_header etree.SubElement(header, {urn:custom}TrackingHeader) etree.SubElement(custom_header, RequestId).text custom-request-1232. 集成自定义插件将自定义插件集成到Google Ads客户端from googleads import ad_manager import logging # 配置日志 logging.basicConfig(levellogging.INFO) logger logging.getLogger(custom_soap) # 创建客户端 ad_manager_client ad_manager.AdManagerClient.LoadFromStorage() # 创建自定义插件实例 custom_plugin CustomZeepPlugin(custom_loggerlogger) # 将插件添加到Zeep客户端 ad_manager_client.zeep_client.plugins.append(custom_plugin) # 现在所有SOAP请求都会经过自定义插件处理 自定义缓存策略Google Ads Python库默认使用文件系统缓存WSDL文档。您可以根据需要实现自定义缓存策略import zeep.cache import redis import pickle class RedisCache(zeep.cache.Base): Redis缓存实现 def __init__(self, redis_client, ttl3600): self.redis redis_client self.ttl ttl def get(self, url): 从Redis获取缓存 key fzeep:wsdl:{url} cached self.redis.get(key) if cached: return pickle.loads(cached) return None def add(self, url, content): 添加到Redis缓存 key fzeep:wsdl:{url} self.redis.setex(key, self.ttl, pickle.dumps(content)) def clear(self): 清空缓存可选实现 # Redis缓存通常不需要手动清空 pass # 使用自定义缓存 import redis redis_client redis.Redis(hostlocalhost, port6379, db0) redis_cache RedisCache(redis_client, ttl7200) ad_manager_client ad_manager.AdManagerClient( oauth2_client, application_name, network_codenetwork_code, cacheredis_cache # 使用Redis缓存 ) 自定义错误处理1. 创建自定义异常类class GoogleAdsCustomError(Exception): 自定义Google Ads异常 def __init__(self, message, original_exceptionNone, request_dataNone): super().__init__(message) self.original_exception original_exception self.request_data request_data def __str__(self): base_message super().__str__() if self.original_exception: return f{base_message} (原始异常: {self.original_exception}) return base_message class RateLimitError(GoogleAdsCustomError): API速率限制异常 pass class ValidationError(GoogleAdsCustomError): 数据验证异常 pass2. 实现错误处理装饰器import functools import time def retry_on_failure(max_retries3, delay1): 重试装饰器 def decorator(func): functools.wraps(func) def wrapper(*args, **kwargs): for attempt in range(max_retries): try: return func(*args, **kwargs) except Exception as e: if attempt max_retries - 1: raise time.sleep(delay * (2 ** attempt)) # 指数退避 return None return wrapper return decorator # 使用装饰器 retry_on_failure(max_retries3, delay2) def safe_api_call(service_method, *args): 安全的API调用 return service_method(*args) 自定义认证流程1. 实现自定义OAuth2处理器from googleads import oauth2 import requests class CustomOAuth2Handler: 自定义OAuth2处理器 def __init__(self, client_id, client_secret, refresh_token): self.client_id client_id self.client_secret client_secret self.refresh_token refresh_token self.access_token None self.token_expiry None def get_access_token(self): 获取访问令牌 if self._token_needs_refresh(): self._refresh_access_token() return self.access_token def _token_needs_refresh(self): 检查令牌是否需要刷新 if not self.access_token or not self.token_expiry: return True # 在过期前5分钟刷新 return time.time() (self.token_expiry - 300) def _refresh_access_token(self): 刷新访问令牌 token_url https://oauth2.googleapis.com/token data { client_id: self.client_id, client_secret: self.client_secret, refresh_token: self.refresh_token, grant_type: refresh_token } response requests.post(token_url, datadata) if response.status_code 200: token_data response.json() self.access_token token_data[access_token] self.token_expiry time.time() token_data.get(expires_in, 3600) else: raise Exception(f令牌刷新失败: {response.status_code})2. 集成自定义认证from googleads import ad_manager # 创建自定义认证处理器 custom_auth CustomOAuth2Handler( client_idYOUR_CLIENT_ID, client_secretYOUR_CLIENT_SECRET, refresh_tokenYOUR_REFRESH_TOKEN ) # 创建Google OAuth2客户端 from googleads import oauth2 google_oauth2_client oauth2.GoogleRefreshTokenClient( custom_auth.client_id, custom_auth.client_secret, custom_auth.refresh_token ) # 使用自定义认证创建客户端 ad_manager_client ad_manager.AdManagerClient( google_oauth2_client, YourApplicationName, network_codeYOUR_NETWORK_CODE ) 性能优化扩展1. 批量操作处理器class BatchOperationHandler: 批量操作处理器 def __init__(self, service, batch_size100): self.service service self.batch_size batch_size def batch_create(self, items, create_method): 批量创建项目 results [] for i in range(0, len(items), self.batch_size): batch items[i:i self.batch_size] try: batch_result create_method(batch) results.extend(batch_result) except Exception as e: # 记录失败但继续处理其他批次 print(f批次 {i//self.batch_size 1} 失败: {e}) # 可以在这里添加重试逻辑 return results def batch_update(self, items, update_method): 批量更新项目 # 类似的批量更新逻辑 pass2. 缓存装饰器import hashlib import json from functools import lru_cache def api_cache(ttl300): API响应缓存装饰器 def decorator(func): cache {} functools.wraps(func) def wrapper(*args, **kwargs): # 创建缓存键 cache_key hashlib.md5( json.dumps({ func: func.__name__, args: args, kwargs: kwargs }, defaultstr).encode() ).hexdigest() # 检查缓存 if cache_key in cache: cached_data, timestamp cache[cache_key] if time.time() - timestamp ttl: return cached_data # 调用函数并缓存结果 result func(*args, **kwargs) cache[cache_key] (result, time.time()) return result return wrapper return decorator 测试自定义扩展1. 创建测试用例import unittest from unittest.mock import Mock, patch from googleads import ad_manager class TestCustomService(unittest.TestCase): def setUp(self): self.mock_client Mock(specad_manager.AdManagerClient) self.custom_service CustomReportService(self.mock_client) def test_custom_report_generation(self): 测试自定义报告生成 mock_report_service Mock() self.mock_client.GetService.return_value mock_report_service # 模拟报告作业 mock_report_job {id: test_job_123} mock_report_service.runReportJob.return_value mock_report_job # 测试自定义报告 report_query {dateRange: LAST_7_DAYS} result self.custom_service.run_custom_delivery_report(report_query) # 验证调用 self.mock_client.GetService.assert_called_with(ReportService, versionv202511) mock_report_service.runReportJob.assert_called() def test_validation_logic(self): 测试验证逻辑 enhanced_service EnhancedCreativeService(self.mock_client) # 测试有效数据 valid_creative { name: Test Creative, advertiserId: 12345, size: {width: 300, height: 250} } # 应该通过验证 self.assertTrue(enhanced_service._validate_creative_data(valid_creative)) # 测试无效数据 invalid_creative {name: Test Creative} self.assertFalse(enhanced_service._validate_creative_data(invalid_creative))2. 集成测试配置# test_config.yaml ad_manager: application_name: Test Application network_code: TEST_NETWORK oauth2_client: oauth2 client_id: test_client_id client_secret: test_client_secret refresh_token: test_refresh_token client_customer_id: test_customer_id user_agent: Test User Agent # 在测试中使用配置 import yaml def load_test_config(): with open(test_config.yaml, r) as f: return yaml.safe_load(f) class IntegrationTest(unittest.TestCase): classmethod def setUpClass(cls): config load_test_config() # 使用测试配置创建客户端 cls.client ad_manager.AdManagerClient.LoadFromString( yaml.dump(config), ad_manager ) 部署最佳实践1. 配置管理创建可配置的扩展模块# extensions/config.py import yaml import os class ExtensionConfig: 扩展配置管理 def __init__(self, config_fileextensions_config.yaml): self.config_file config_file self.config self._load_config() def _load_config(self): 加载配置文件 if os.path.exists(self.config_file): with open(self.config_file, r) as f: return yaml.safe_load(f) return {} def get_plugin_config(self, plugin_name): 获取插件配置 return self.config.get(plugins, {}).get(plugin_name, {}) def get_cache_config(self): 获取缓存配置 return self.config.get(cache, {}) def get_retry_config(self): 获取重试配置 return self.config.get(retry, {max_retries: 3, delay: 1})2. 工厂模式创建扩展# extensions/factory.py from extensions.plugins import CustomZeepPlugin from extensions.cache import RedisCache from extensions.services import CustomReportService, EnhancedCreativeService class ExtensionFactory: 扩展工厂 staticmethod def create_plugin(plugin_type, **kwargs): 创建插件 plugins { custom_logger: CustomZeepPlugin, # 添加更多插件类型 } plugin_class plugins.get(plugin_type) if plugin_class: return plugin_class(**kwargs) raise ValueError(f未知插件类型: {plugin_type}) staticmethod def create_service(service_type, client, **kwargs): 创建服务 services { custom_report: lambda: CustomReportService(client, **kwargs), enhanced_creative: lambda: EnhancedCreativeService(client, **kwargs), # 添加更多服务类型 } service_creator services.get(service_type) if service_creator: return service_creator() raise ValueError(f未知服务类型: {service_type}) 总结与建议通过本文的介绍您已经了解了如何扩展Google Ads Python库来满足特定的业务需求。以下是一些关键建议逐步扩展从简单的自定义服务开始逐步添加更复杂的功能保持兼容性确保自定义扩展与原始库的API保持兼容充分测试为所有自定义扩展编写全面的测试用例文档化为您的扩展创建清晰的文档和使用示例性能监控监控自定义扩展的性能确保不会影响系统稳定性Google Ads Python库的扩展性使其成为构建强大广告管理解决方案的理想基础。通过合理利用自定义服务和插件开发您可以创建出完全符合业务需求的广告自动化系统。记住扩展库功能时始终要考虑维护性和可升级性。随着Google Ads API的更新确保您的自定义扩展能够平滑迁移到新版本。开始您的Google Ads Python库扩展之旅吧 通过自定义服务和插件开发您将能够构建更强大、更灵活的广告管理工具提升工作效率并满足特定的业务需求。【免费下载链接】googleads-python-libThe Python client library for Googles Ads APIs项目地址: https://gitcode.com/gh_mirrors/go/googleads-python-lib创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考