# # -*- coding: utf-8 -*- # """ # @Time : 2025/8/1 10:53 # @Auth : 九月的海 # @File : cookie.py # @IDE : PyCharm # @Motto : Catch as catch can.... # """ # import requests # from datetime import datetime, timedelta # import json # import logging # import http.client # import urllib3 # # # 启用详细日志记录 # urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) # http.client.HTTPConnection.debuglevel = 1 # # # 配置更详细的日志 # logging.basicConfig(level=logging.DEBUG) # logger = logging.getLogger('requests.packages.urllib3') # logger.setLevel(logging.DEBUG) # logger.propagate = True # # # # 创建自定义的 HTTP 适配器用于记录请求/响应细节 # class DebugHTTPAdapter(requests.adapters.HTTPAdapter): # def send(self, request, **kwargs): # """发送请求并记录所有细节""" # # 记录请求详情 # print("\n" + "=" * 40) # print(f"请求方法: {request.method}") # print(f"请求URL: {request.url}") # print("请求头:") # for key, value in request.headers.items(): # print(f" {key}: {value}") # # if request.body: # print("请求体:") # if isinstance(request.body, bytes): # try: # body = request.body.decode('utf-8') # print(body) # except UnicodeDecodeError: # print("<无法解码的二进制数据>") # else: # print(request.body) # else: # print("无请求体") # # # 发送请求 # response = super().send(request, **kwargs) # # # 记录响应详情 # print("\n" + "=" * 40) # print(f"响应状态码: {response.status_code}") # print("响应头:") # for key, value in response.headers.items(): # print(f" {key}: {value}") # # # 尝试解析JSON响应体 # try: # if response.text: # print("响应内容:") # # 尝试格式化为可读JSON # try: # json_data = json.loads(response.text) # print(json.dumps(json_data, indent=2, ensure_ascii=False)) # except json.JSONDecodeError: # print(response.text) # else: # print("无响应内容") # except Exception as e: # print(f"解析响应失败: {e}") # # print("=" * 40 + "\n") # # return response # # # class MobgiAPIClient: # def __init__(self): # self.session = requests.Session() # # # 添加调试适配器 # adapter = DebugHTTPAdapter() # self.session.mount('http://', adapter) # self.session.mount('https://', adapter) # # self.login_time = None # self.expire_time = None # self.api_base = "https://cli1.mobgi.com" # self.is_logged_in = False # self.session_cookie = None # self.user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36" # # def login(self, email, password, expire_hours=24): # """登录并保存会话状态""" # # 清除可能的旧会话 # self.session.cookies.clear() # self.is_logged_in = False # url = f"{self.api_base}/User/AdminUser/login" # payload = {"email": email, "password": password, "product_version": 0} # try: # # 登录请求使用独立的 headers # headers = { # "Content-Type": "application/json", # "User-Agent": self.user_agent # } # # 打印登录请求细节 # print("\n" + "*" * 50) # print("发送登录请求...") # print("*" * 50 + "\n") # response = self.session.post( # url, # json=payload, # headers=headers, # timeout=10 # ) # response.raise_for_status() # # 解析响应 # resp_data = response.json() # if resp_data.get('code') != 0: # print(f"登录失败: {resp_data.get('message', '未知错误')}") # print(f"完整响应: {json.dumps(resp_data, indent=2)}") # return False # # 提取并保存会话 cookie # self.session_cookie = self._extract_session_cookie(response) # # if not self.session_cookie: # print("ERROR: 未能获取会话cookie") # if 'Set-Cookie' in response.headers: # print(f"Set-Cookie 头: {response.headers['Set-Cookie']}") # else: # print("响应中没有 Set-Cookie 头") # return False # # # 更新登录状态 # self.login_time = datetime.now() # self.expire_time = self.login_time + timedelta(hours=expire_hours) # self.is_logged_in = True # # print(f"登录成功! 会话有效期至 {self.expire_time.strftime('%Y-%m-%d %H:%M:%S')}") # print(f"获取的会话Cookie: {self.session_cookie}") # return True # # except requests.exceptions.RequestException as e: # print(f"登录请求异常: {str(e)}") # if e.response is not None: # print(f"响应状态码: {e.response.status_code}") # try: # print(f"响应内容: {e.response.text}") # except: # pass # return False # except json.JSONDecodeError as e: # print(f"登录响应解析失败: 无效的JSON格式, 原始内容: {response.text}") # return False # except Exception as e: # print(f"登录过程中发生意外错误: {str(e)}") # return False # # def _extract_session_cookie(self, response): # """从响应中提取会话cookie""" # # 方法1: 从cookies字典获取 # if 'chuangliang_session' in response.cookies: # cookie_value = response.cookies['chuangliang_session'] # return f"chuangliang_session={cookie_value};" # # # 方法2: 从Set-Cookie头获取 # if 'Set-Cookie' in response.headers: # set_cookie = response.headers['Set-Cookie'] # # 寻找chuangliang_session条目 # for cookie_str in set_cookie.split(','): # if 'chuangliang_session' in cookie_str: # # 提取cookie值 # parts = cookie_str.split(';') # for part in parts: # if 'chuangliang_session' in part: # return part.strip() + ';' # # # 方法3: 如果cookie是多个值，可能在cookie jar中 # for cookie in response.cookies: # if cookie.name == 'chuangliang_session': # return f"{cookie.name}={cookie.value};" # # return None # # def _get_request_headers(self): # """构造请求头，包含必要的cookie""" # headers = { # "Content-Type": "application/json", # "User-Agent": self.user_agent, # # 添加其他可能需要的通用头 # } # # if self.session_cookie: # # 将会话 cookie 添加到 headers # headers["Cookie"] = self.session_cookie # # return headers # # def check_session(self): # """检查会话有效性""" # if not self.is_logged_in or not self.expire_time: # print("会话未初始化，请先登录") # return False # # if datetime.now() > self.expire_time: # print(f"会话已过期，有效期至 {self.expire_time.strftime('%Y-%m-%d %H:%M:%S')}") # return False # # return True # # def get_advertiser_list( # self, # start_date="2025-07-31", # end_date="2025-07-31", # conditions=None, # page=1, # page_size=20 # ): # """获取广告主列表数据""" # if not self.check_session(): # return None # # # 准备请求参数 # params = { # "conditions": json.dumps(conditions) if conditions else json.dumps({ # "companys": [], # "cl_project_id": [], # "optimize_user_id": [], # "keyword": "" # }), # "start_date": start_date, # "end_date": end_date, # "page": page, # "page_size": page_size, # "total_count": 0, # "total_page": 1, # "sort_field": "", # "sort_direction": "", # "data_type": "list" # } # # url = f"{self.api_base}/Baidu/Advertiser/getList" # # try: # # 获取带 cookie 的 headers # headers = self._get_request_headers() # # # 打印API请求细节 # print("\n" + "*" * 50) # print(f"发送广告主列表请求到: {url}") # print("*" * 50 + "\n") # # response = self.session.post( # url, # json=params, # headers=headers, # timeout=15 # ) # response.raise_for_status() # # # 检查是否有新的cookie # if 'Set-Cookie' in response.headers: # new_cookie = self._extract_session_cookie(response) # if new_cookie: # print(f"接收到新cookie: {new_cookie}") # self.session_cookie = new_cookie # else: # print("响应包含Set-Cookie，但未解析出有效值") # # resp_data = response.json() # if resp_data.get('code') != 0: # print(f"广告主列表请求失败: {resp_data.get('message', '未知错误')}") # print(f"完整响应: {json.dumps(resp_data, indent=2)}") # return None # # return resp_data.get('data', {}) # # except requests.exceptions.RequestException as e: # print(f"广告主列表请求异常: {str(e)}") # if e.response is not None: # print(f"响应状态码: {e.response.status_code}") # try: # print(f"响应内容: {e.response.text}") # except: # pass # return None # except json.JSONDecodeError: # print(f"广告主列表响应解析失败: 无效的JSON格式, 原始内容: {response.text}") # return None # except Exception as e: # print(f"广告主列表请求过程中发生意外错误: {str(e)}") # return None # # def __del__(self): # """关闭会话""" # self.session.close() # print("会话已关闭") # # # # 使用示例 # if __name__ == "__main__": # print("=" * 60) # print("启动 API 客户端调试") # print("=" * 60) # # # 初始化客户端 # client = MobgiAPIClient() # # # 登录（使用您提供的凭证） # if client.login( # email="974509022@qq.com", # password="0e3bc01f8f0409f4541015737925ed8e" # ): # print("\n登录成功! 尝试获取广告主列表...") # # # 获取广告主列表 # advertiser_data = client.get_advertiser_list( # start_date="2025-07-01", # end_date="2025-07-31", # conditions=None, # page=1, # page_size=50 # ) # # if advertiser_data: # print("\n成功获取广告主列表数据:") # print(f"总记录数: {advertiser_data.get('total_count', 0)}") # print(f"总页数: {advertiser_data.get('total_page', 1)}") # print(f"返回记录数: {len(advertiser_data.get('list', []))}") # else: # print("\n获取广告主列表失败") # else: # print("\n登录失败，无法继续") # # print("\n调试结束")