# -*- coding: utf-8 -*- # Author : Charley # Python : 3.10.8 # Date : 2026/2/27 11:22 import time import inspect import requests import schedule import user_agent from loguru import logger from crypto_utils import CryptoHelper from mysql_pool import MySQLConnectionPool # 2026/08/06 停用板块发现：get_board_shop_list 的旧 cateId 已随版本升级失效， # 改用「全部」接口(get_all_shop_list)覆盖全部在售店铺，故移除该导入（文件保留备用）。 from tenacity import retry, stop_after_attempt, wait_fixed logger.remove() logger.add("./logs/{time:YYYYMMDD}.log", encoding='utf-8', rotation="00:00", format="[{time:YYYY-MM-DD HH:mm:ss.SSS}] {level} {message}", level="DEBUG", retention="7 day") # 基础配置 BASE_URL = "https://cashier.yqszpay.com" PAGE_SIZE = 10 headers = { "User-Agent": user_agent.generate_user_agent(os="android"), # 设置为安卓模拟器 "Connection": "Keep-Alive", "Accept-Encoding": "gzip", "Content-Type": "application/json", "channelNo": "88888888", "pageSize": str(PAGE_SIZE), # "pageNum": 1, # 2026/08/06 版本失效修复：服务端升级后按 version 头做灰度校验，旧值 1.9.9.82537 # 会被静默降级为返回空 rows（HTTP 200 不报错、数据悄悄变空）。改为抓包实测的当前 # 真实版本 1.3.09 后 getActList / getActEndList 均恢复正常返回。 "version": "1.3.09" } def after_log(retry_state): """ retry 回调 :param retry_state: RetryCallState 对象 """ # 检查 args 是否存在且不为空 if retry_state.args and len(retry_state.args) > 0: log = retry_state.args[0] # 获取传入的 logger else: log = logger # 使用全局 logger if retry_state.outcome.failed: log.warning( f"Function '{retry_state.fn.__name__}', Attempt {retry_state.attempt_number} Times") else: log.info(f"Function '{retry_state.fn.__name__}', Attempt {retry_state.attempt_number} succeeded") @retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log) def get_proxys(log): """ 获取代理配置 :param log: 日志对象 :return: 代理字典 """ tunnel = "x371.kdltps.com:15818" kdl_username = "t13753103189895" kdl_password = "o0yefv6z" try: proxies = { "http": "http://%(user)s:%(pwd)s@%(proxy)s/" % {"user": kdl_username, "pwd": kdl_password, "proxy": tunnel}, "https": "http://%(user)s:%(pwd)s@%(proxy)s/" % {"user": kdl_username, "pwd": kdl_password, "proxy": tunnel} } return proxies except Exception as e: log.error(f"Error getting proxy: {e}") raise e @retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log) def make_encrypted_post_request(log, url: str, request_data: dict, extra_headers: dict = None): """ 通用加密POST请求函数（带重试机制） :param log: 日志对象 :param url: 请求URL :param request_data: 请求数据字典（会被加密） :param extra_headers: 额外的请求头 :return: 解密后的响应数据，失败返回None """ request_headers = headers.copy() if extra_headers: request_headers.update(extra_headers) log.debug(f"Request URL: {url}, Data: {request_data}") encrypted_body = CryptoHelper.encrypt_request_data(request_data) # print(request_headers) # response = requests.post(url, headers=request_headers, json=encrypted_body, timeout=22, proxies=get_proxys(log)) response = requests.post(url, headers=request_headers, json=encrypted_body, timeout=(5, 30), proxies=get_proxys(log)) # response = requests.post(url, headers=request_headers, json=encrypted_body, timeout=(5, 30)) # response.raise_for_status() if response.status_code == 200: response_json = response.json() # log.debug(f"Raw response: {response_json}") if 'data' in response_json: decrypted = CryptoHelper.decrypt_response_data(response_json) # log.debug(f"Decrypted response: {decrypted}") return decrypted return response_json else: log.error(f"请求失败: {response.status_code}, Response: {response.text}") return None def get_shop_single_page(log, page_num, page_size=PAGE_SIZE): """ 获取商户列表（支持翻页） :param log: 日志对象 :param page_num: 页码 :param page_size: 每页条数 """ log.debug(f"Getting shop list, page: {page_num}") url = f"{BASE_URL}/zc-api/merchant/getMerMyList" request_data = {'pageNum': page_num, 'pageSize': page_size} try: resp = make_encrypted_post_request(log, url, request_data, extra_headers={"pageNum": str(page_num)}) except Exception as e: log.error(f"Error getting shop list: {e}") resp = None return resp # 「全部」板块的分类ID。来源：抓包数据.txt 里「全部」两条 getActList 请求体（AES 解密后得到）， # 2026/08/06 版本升级后该值为 "0,6,3"。此接口为公开接口，不校验 Authorization，无需 token。 ALL_SHOP_CATE_ID = "0,6,3" def get_all_shop_single_page(log, page_num, page_size=PAGE_SIZE): """获取「全部」板块的活动列表单页（用于发现全部在售店铺）。 走 getActList「全部」板块（cateId=ALL_SHOP_CATE_ID），从每条活动里解析出 merNo / merName，覆盖当前所有有进行中活动的店铺。为公开接口，无需 token。 Args: log: 日志对象，从调用方透传。 page_num (int): 页码，从 1 开始。 page_size (int, optional): 每页条数。Defaults to PAGE_SIZE。 Returns: dict | None: 解密后的响应字典（含 rows）；请求失败时返回 None。 """ log.debug(f"Getting all-shop list, page: {page_num}") url = f"{BASE_URL}/zc-api/act/actProduct/getActList" # 请求体字段与抓包解密结果一一对应，仅 cateId 固定为「全部」板块： # finish / loading -> 前端分页状态位，抓包固定 false / true，照搬即可 # actStatus=1 -> 只取进行中的活动 request_data = { 'cateId': ALL_SHOP_CATE_ID, 'pageSize': page_size, 'pageNum': page_num, 'finish': False, 'loading': True, 'actStatus': 1 } try: resp = make_encrypted_post_request(log, url, request_data, extra_headers={"pageNum": str(page_num)}) except Exception as e: log.error(f"Error getting all-shop list, page {page_num}: {e}") resp = None return resp def parse_all_shop_data(log, items, sql_pool, seen): """解析「全部」接口活动里的店铺并写库（INSERT IGNORE，不覆盖存量）。 「全部」接口只返回 merNo / merName，拿不到 sold_number(销量) / fans(粉丝)。因此对 新发现的店铺 sold_number 写死哨兵值 1，让其能被每日任务「WHERE sold_number != 0」 选中进入商品采集管道；用 INSERT IGNORE 跳过已存在的店铺，避免覆盖 get_shop_list （getMerMyList）回填的真实 sold_number / fans。 Args: log: 日志对象，从调用方透传。 items (list[dict]): getActList「全部」板块返回的活动列表（每条含 merNo / merName）。 sql_pool (MySQLConnectionPool): MySQL 连接池。 seen (set): 跨页去重的 shop_id 集合，避免同一店铺重复写库。 Returns: int: 本次实际写入的店铺数量（已去重）。 """ info_list = [] for item in items: shop_id = item.get('merNo') shop_name = item.get('merName') # 同一店铺会出现在多条活动里，跨页去重以减少无谓写库 if shop_id in seen: continue seen.add(shop_id) # sold_number 写死哨兵值 1：仅作「需采集」标记（板块拿不到真实销量），配合 INSERT IGNORE # 只对新插入店铺生效，已存在店铺整行被跳过；后续 get_shop_list 会回填真实值 info_list.append({'shop_id': shop_id, 'shop_name': shop_name, 'sold_number': 1}) if not info_list: return 0 sql_pool.insert_many(table='zc_shop_record', data_list=info_list, ignore=True) return len(info_list) def get_all_shop_list(log, sql_pool): """「全部」接口店铺发现入口：翻页遍历全部在售店铺并写库。 店铺主来源，替代原来覆盖不全的 getMerMyList「我的商户列表」。翻页采用 「空页或不足整页即停」的可靠判断，受 MAX 页数保护。 Args: log: 日志对象，从调用方透传。 sql_pool (MySQLConnectionPool): MySQL 连接池。 Returns: int: 去重后合计写入的店铺数量。 """ log.info("开始「全部」接口店铺发现......") page_num = 1 max_pages = 100 seen = set() while page_num <= max_pages: result = get_all_shop_single_page(log, page_num, PAGE_SIZE) if result is None: log.error(f"「全部」第 {page_num} 页请求失败，停止翻页") break data_list = result.get('rows', []) if len(data_list) == 0: log.info(f"「全部」第 {page_num} 页无数据，停止翻页") break parse_all_shop_data(log, data_list, sql_pool, seen) log.info(f"「全部」第 {page_num} 页完成，本页活动数: {len(data_list)}，累计去重店铺: {len(seen)}") # 不足整页说明已到末页，停止翻页 if len(data_list) < PAGE_SIZE: log.info(f"「全部」第 {page_num} 页不足整页，已到末页，停止翻页") break page_num += 1 log.info(f"「全部」接口店铺发现结束，去重店铺数: {len(seen)}") return len(seen) def get_sold_single_page(log, mer_no, page_num): """ 获取商品列表（支持翻页） :param log: 日志对象 :param mer_no: 商户编号 :param page_num: 页码 """ log.info(f"Getting sold items for mer_no: {mer_no}, page: {page_num}") # url = f"{BASE_URL}/zc-api/act/actProduct/getActList" # url = "https://cashier.yqszpay.com/zc-api/act/actProduct/getActEndList" 20260407接口变化 修改 url = f"{BASE_URL}/zc-api/act/actProduct/getActEndList" request_data = { 'merNo': mer_no, 'pageNum': page_num, 'pageSize': PAGE_SIZE, 'queryType': 1 } return make_encrypted_post_request(log, url, request_data, extra_headers={"pageNum": str(page_num)}) def get_player_single_page(log, act_id, token, page_num, page_size=PAGE_SIZE): """ 获取玩家列表（支持翻页） :param log: 日志对象 :param act_id: 活动ID :param token: Authorization token :param page_num: 页码 :param page_size: 每页条数 """ log.debug(f"Getting player list for act_id: {act_id}, page: {page_num}") url = f"{BASE_URL}/zc-api/act/actOrder/getActOrderPublicDetails" request_data = {'actId': act_id, 'pageNum': page_num, 'pageSize': page_size} return make_encrypted_post_request( log, url, request_data, extra_headers={"Authorization": token, "pageNum": str(page_num)} ) def parse_shop_data(log, items, sql_pool): """ 解析商户数据 :param log: 日志对象 :param items: 商户列表 :param sql_pool: MySQL连接池 :return: 解析后的数据列表 """ log.debug(f"Parsing shop data...........") info_list = [] for item in items: # log.debug(f"Processing shop item: {item}") shop_id = item.get('merNo') shop_name = item.get('merName') sold_number = item.get('spell_number') # link_man = item.get('linkMan') # user_id = item.get('userId') fans = item.get('attentionNumber') data_dict = { 'shop_id': shop_id, 'shop_name': shop_name, 'sold_number': sold_number, 'fans': fans } log.debug(f"Parsed shop data: {data_dict}") info_list.append(data_dict) # 保存/更新 根据shop_id判断 是否存在，存在则更新，不存在则插入 sql = "INSERT INTO zc_shop_record (shop_id, shop_name, sold_number, fans) VALUES (%s, %s, %s, %s) ON DUPLICATE KEY UPDATE shop_name=VALUES(shop_name), sold_number=VALUES(sold_number), fans=VALUES(fans)" # 将字典列表转换为元组列表 args_list = [tuple(d.values()) for d in info_list] sql_pool.insert_many(query=sql, args_list=args_list) @retry(stop=stop_after_attempt(3), wait=wait_fixed(1), after=after_log) def get_video(log, token, pid): """ 获取活动视频信息 :param log: 日志对象 :param token: Authorization token :param pid: 活动ID :return: (live_id, open_time, close_time, video_url) """ url = "https://cashier.yqszpay.com/zc-api/live/actLive/getMerLiveInfo" request_data = {'actId': pid} log.debug(f"获取视频信息，actId: {pid}") resp_data = make_encrypted_post_request( log, url, request_data, extra_headers={"Authorization": token} ) # log.debug(f"视频响应: {resp_data}") live_id = resp_data.get('live', {}).get('liveId') live_open_time = resp_data.get('live', {}).get('openTime') live_close_time = resp_data.get('live', {}).get('closeTime') video_url = resp_data.get('live', {}).get('videoUrl') return live_id, live_open_time, live_close_time, video_url def parse_sold_data(log, token, items, sql_pool, shop_name): """ 解析商品数据 :param log: 日志对象 :param token: Authorization token :param items: 商品列表 :param sql_pool: MySQL连接池 :param shop_name: 商户名称 :return: 解析后的数据列表 """ info_list = [] for item in items: # log.debug(f"Processing sold item: {item}") shop_id = item.get('merNo') # 商户编号 pid = item.get('id') act_day = item.get('actDay') # 活动天数 act_logo = item.get('actLogo') act_name = item.get('actName') # 活动名称 act_no = item.get('actNo') # 活动编号 act_status = item.get('actStatus') # 活动状态 startDate = item.get('startDate') # 开始时间 endDate = item.get('complete_date') # 结束时间 storageId = item.get('storageId') # 存储ID storageName = item.get('storageName') # 存储名称 unitPrice = item.get('unitPrice') # 单价 sumPrice = item.get('sumPrice') # 总价 reality_price = item.get('realityPrice') # 实际价格 packageNumber = item.get('packageNumber') # 包配置 schedule_ = item.get('schedule') # 库存 try: live_id, live_open_time, live_close_time, video_url = get_video(log, token, pid) except Exception as e: log.error(f"Error getting video info for pid {pid}: {e}") live_id, live_open_time, live_close_time, video_url = None, None, None, None data_dict = { 'shop_id': shop_id, 'shop_name': shop_name, 'pid': pid, 'act_day': act_day, 'act_img': act_logo, 'act_name': act_name, 'act_no': act_no, 'act_status': act_status, 'start_date': startDate, 'end_date': endDate, 'storage_id': storageId, 'storage_name': storageName, 'unit_price': unitPrice, 'sum_price': sumPrice, 'reality_price': reality_price, 'package_number': packageNumber, 'schedule': schedule_, 'live_id': live_id, 'live_open_time': live_open_time, 'live_close_time': live_close_time, 'video_url': video_url } # log.debug(f"Parsed sold data: {data_dict}") # { 'live_close_time': None, 'video_url': None} info_list.append(data_dict) # 保存数据 sql_pool.insert_many(table='zc_product_record', data_list=info_list, ignore=True) def parse_player_data(log, items, sql_pool): """ 解析玩家数据 :param log: 日志对象 :param items: 玩家列表 :param sql_pool: MySQL连接池 :return: 解析后的数据列表 """ log.debug(f"Parsing player data...........") info_list = [] for item in items: # log.debug(f"Processing player item: {item}") pid = item.get('actId') give_number = item.get('giveNumber') # 份数 user_id = item.get('userId') user_name = item.get('userName') data_dict = { 'pid': pid, 'give_number': give_number, 'user_id': user_id, 'user_name': user_name } # log.debug(f"Parsed player data: {data_dict}") info_list.append(data_dict) # 保存数据 sql_pool.insert_many(table='zc_player_record', data_list=info_list, ignore=True) def get_shop_list(log, sql_pool): """ 商户列表翻页生成器 :param log: 日志对象 :param sql_pool: MySQL连接池 """ page_num = 1 max_pages = 100 # 2026/08/06 翻页 bug 修复：原逻辑 total 初值为 0，却用「total is None」判断是否取总数， # 导致 total 永远停在 0，停止条件「(page_num-1)*PAGE_SIZE >= total」在第 1 页跑完后即 # 0 >= 0 成立、立刻 break，只能采到第 1 页 10 个店铺。改为「空页 / 不足整页即停」的可靠判断。 while page_num <= max_pages: result = get_shop_single_page(log, page_num, PAGE_SIZE) if result is None: log.error(f"第 {page_num} 页请求失败，停止翻页") break data_list = result.get('rows', []) # 空页直接停止，避免用空列表调用 insert_many 触发 ValueError if len(data_list) == 0: log.info(f"第 {page_num} 页无数据，停止翻页") break # getMerMyList 含真实 sold_number / fans，走 ON DUPLICATE KEY UPDATE 回填覆盖哨兵值 parse_shop_data(log, data_list, sql_pool) log.info(f"第 {page_num} 页查询完成，本页条数: {len(data_list)}") # 不足整页说明已到末页，停止翻页 if len(data_list) < PAGE_SIZE: log.info(f"第 {page_num} 页不足整页，已到末页，停止翻页") break page_num += 1 def get_sold_list(log, shop_id, token, sql_pool, shop_name): """ 商品列表翻页生成器 :param log: 日志对象 :param shop_id: shop_id :param token: Authorization token :param sql_pool: MySQL连接池 :param shop_name: 商户名称 """ page_num = 1 max_pages = 10 while page_num <= max_pages: result = get_sold_single_page(log, shop_id, page_num) # time.sleep(random.uniform(0.5, 1)) # 添加随机延迟，防止对目标服务器造成过大负载 # print(result) if result is None: log.error(f"第 {page_num} 页请求失败，停止翻页") break data_list = result.get('rows', []) if not data_list: log.info(f"第 {page_num} 页无数据，停止翻页") # log.info(f'该店铺{shop_name}无数据, 修改为店铺注销状态..........') # # 更新店铺状态为注销状态 # sql_pool.update_one("UPDATE zc_shop_record SET is_deleted = 1 WHERE shop_id = %s", (shop_id,)) break parse_sold_data(log, token, data_list, sql_pool, shop_name) # 检查是否有数据 if len(data_list) < 10: log.info(f"第 {page_num} 页无数据，停止翻页") break log.info(f"第 {page_num} 页查询完成，本页条数: {len(data_list)}") page_num += 1 def get_player_list(log, act_id, token, sql_pool): """ 玩家列表翻页生成器 :param log: 日志对象 :param act_id: 活动ID :param token: Authorization token :param sql_pool: MySQL连接池 :return: has_data (True: 有数据, False: 无数据) """ page_num = 1 max_pages = 100 has_data = False while page_num <= max_pages: result = get_player_single_page(log, act_id, token, page_num) if result is None: log.error(f"第 {page_num} 页请求失败，停止翻页") break data_list = result.get('rows', []) # 如果有数据才解析 if len(data_list) > 0: has_data = True parse_player_data(log, data_list, sql_pool) # 检查是否有数据 if len(data_list) < 10: log.info(f"第 {page_num} 页无数据，停止翻页") break log.info(f"第 {page_num} 页查询完成，本页条数: {len(data_list)}") page_num += 1 return has_data @retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log) def zc_main(log): """ 主函数 :param log: logger对象 """ log.info( f'开始运行 {inspect.currentframe().f_code.co_name} 爬虫任务....................................................') # 配置 MySQL 连接池 sql_pool = MySQLConnectionPool(log=log) if not sql_pool.check_pool_health(): log.error("数据库连接池异常") raise RuntimeError("数据库连接池异常") try: # 获取 token token_row = sql_pool.select_one("SELECT token FROM zc_token WHERE id = 1") if not token_row: log.error("未查询到 token") return token = token_row[0] # player test # has_data = get_player_list(log, 1800, token, sql_pool) # 店铺主来源：「全部」接口发现全部在售店铺（替代覆盖不全的 getMerMyList） # 2026/08/06 停用板块发现 get_board_shop_list：其 5 个旧 cateId 已随版本升级失效， # 且「全部」接口(cateId=0,6,3)已覆盖所有在售店铺，板块发现冗余。 try: get_all_shop_list(log, sql_pool) except Exception as e: log.error(f'get_all_shop_list error: {e}') # 补充：getMerMyList 回填「我的店铺」真实 sold_number / fans（「全部」接口拿不到这两个字段） try: get_shop_list(log, sql_pool) except Exception as e: log.error(f'iterate_shop_list error: {e}') time.sleep(5) # 获取sold data - 遍历所有商户 try: # 从 shop 表查询所有 merNo 2026/5/9增加is_deleted判断是否注销店铺 mer_no_rows = sql_pool.select_all( # "SELECT shop_id, shop_name FROM zc_shop_record WHERE is_deleted = 0 AND sold_number != 0" "SELECT shop_id, shop_name FROM zc_shop_record WHERE sold_number != 0" ) log.info(f"查询到 {len(mer_no_rows)} 个商户编号: {mer_no_rows}") for shop_id, shop_name in mer_no_rows: log.info(f"开始爬取商户 {shop_id}, {shop_name} 的商品数据") # get_sold_list(log, shop_id, shop_name, token, sql_pool) get_sold_list(log, shop_id, token, sql_pool, shop_name) except Exception as e: log.error(f'get_sold_list error: {e}') time.sleep(5) # 获取player data - 遍历所有活动 try: # 从 sold 表查询所有 actId act_id_rows = sql_pool.select_all("SELECT pid FROM zc_product_record WHERE player_state != 1") act_id_list = [row[0] for row in act_id_rows] if act_id_rows else [] log.info(f"查询到 {len(act_id_list)} 个活动ID") for act_id in act_id_list: try: # 先将当前 pid 的状态改为 1，表示开始查询 sql_pool.update_one("UPDATE zc_product_record SET player_state = 1 WHERE pid = %s", (act_id,)) log.info(f"将 pid: {act_id} 的状态更新为 1（开始查询）") log.info(f"开始爬取pid: {act_id} 的玩家数据") has_data = get_player_list(log, act_id, token, sql_pool) # 根据是否有数据更新状态 if has_data: log.info(f"pid: {act_id} 查询到数据，状态保持为 1") else: log.info(f"pid: {act_id} 没有数据，状态更新为 2") sql_pool.update_one("UPDATE zc_product_record SET player_state = 2 WHERE pid = %s", (act_id,)) except Exception as pid_error: # 如果查询失败，将状态改为 3 log.error(f"pid: {act_id} 查询失败，错误: {pid_error}") try: sql_pool.update_one("UPDATE zc_product_record SET player_state = 3 WHERE pid = %s", (act_id,)) log.info(f"已将 pid: {act_id} 的状态更新为 3（查询异常）") except Exception as update_error: log.error(f"更新 pid: {act_id} 状态失败: {update_error}") except Exception as e: log.error(f'iterate_player_list error: {e}') except Exception as e: log.error(f'{inspect.currentframe().f_code.co_name} error: {e}') finally: log.info(f'爬虫程序 {inspect.currentframe().f_code.co_name} 运行结束,等待下一轮的采集任务............') def schedule_task(): """ 爬虫模块 定时任务 的启动文件 """ # 立即运行一次任务 # zc_main(log=logger) # 设置定时任务 schedule.every().day.at("00:01").do(zc_main, log=logger) while True: schedule.run_pending() time.sleep(1) if __name__ == '__main__': # zc_main(logger) schedule_task()