# -*- coding: utf-8 -*- # Author : Charley # Python : 3.10.8 # Date : 2025/11/11 11:54 import time import random import inspect import requests from loguru import logger from datetime import datetime from mysql_pool import MySQLConnectionPool 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") headers = { "authority": "kurabu.feishezhang.com", "accept": "*/*", "accept-language": "zh-CN,zh;q=0.9", "content-type": "application/json", "referer": "https://servicewechat.com/wxa5880b2d8e8a0f37/17/page-frame.html", "sec-fetch-dest": "empty", "sec-fetch-mode": "cors", "sec-fetch-site": "cross-site", "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.0.0 Safari/537.36 MicroMessenger/7.0.20.1781(0x6700143B) NetType/WIFI MiniProgramEnv/Windows WindowsWechat/WMPF WindowsWechat(0x63090c33) XWEB/9129", "xweb_xhr": "1" } token = "26eed48c80bad053717687daef3a8668" crawl_category = "一番赏" 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): """ 获取代理 :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 get_reward_single_page(log, page, sql_pool, sql_reward_list): """ 获取 赏池 单个页面数据 :param log: 日志对象 :param page: 页码 :param sql_pool: 数据库连接池对象 :param sql_reward_list: sql_reward_list :return: len(data_list) -> 获取数据长度 """ log.debug(f"{inspect.currentframe().f_code.co_name} 开始获取第{page}页数据") url = "https://kurabu.feishezhang.com/api.php" params = { "s": "reward/index", "system_type": "default", "application": "app", "application_client_type": "weixin", "token": token, "uuid": "227de1d4-c429-4e73-b972-42c26560ed83", "ajax": "ajax" } data = { # "page": 1, "page": page, "type": 0 } # data = json.dumps(data, separators=(',', ':')) response = requests.post(url, headers=headers, params=params, json=data) response.raise_for_status() resp_json = response.json() if resp_json['code'] == 0: data_list = resp_json.get('data', {}).get('data', []) info_list = [] for cdata in data_list: reward_pool_id = cdata.get('id') # 赏池id title = cdata.get('title') simple_desc = cdata.get('simple_desc') show_image = cdata.get('show_image') total_sessions = cdata.get('total_sessions') # 总箱数 price = cdata.get('price_per_draw') status = cdata.get('status') reward_type = cdata.get('type') sales_count = cdata.get('sales_count') # 判断是否在库中 if reward_pool_id in sql_reward_list: log.debug(f"{inspect.currentframe().f_code.co_name} 赏池ID: {reward_pool_id} 已存在") sql_pool.update_one_or_dict(table="feishezhang_reward_list_record", data={"total_sessions": total_sessions, "price": price, "sales_count": sales_count}, condition={"reward_pool_id": reward_pool_id}) continue sale_time_int = cdata.get('sale_time') # 开售时间 sale_time = datetime.fromtimestamp(sale_time_int).strftime('%Y-%m-%d %H:%M:%S') if sale_time_int != 0 else 0 order_time_int = cdata.get('order_time') # 下单时间 order_time = datetime.fromtimestamp(order_time_int).strftime( '%Y-%m-%d %H:%M:%S') if order_time_int != 0 else 0 data_dict = { "reward_pool_id": reward_pool_id, "title": title, "simple_desc": simple_desc, "show_image": show_image, "total_sessions": total_sessions, "price": price, "status": status, "reward_type": reward_type, "sales_count": sales_count, "sale_time": sale_time, "order_time": order_time, "crawl_category": crawl_category } # print(data_dict) info_list.append(data_dict) # 保存数据 if info_list: sql_pool.insert_many(table="feishezhang_reward_list_record", data_list=info_list, ignore=True) return len(data_list) else: log.warning(resp_json['msg']) return 0 def get_reward_list(log, sql_pool, sql_reward_list): """ 获取 赏池 列表 :param log: 日志对象 :param sql_pool: 数据库连接池对象 :param sql_reward_list: sql_reward_list """ log.debug(f"{inspect.currentframe().f_code.co_name} 开始获取列表数据") page = 1 max_page = 1000 while page <= max_page: len_data_list = get_reward_single_page(log, page, sql_pool, sql_reward_list) if len_data_list < 15: log.debug(f"当前页数据不足15条，已结束获取数据，当前页数：{page}") break page += 1 time.sleep(random.uniform(1, 2)) @retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log) def get_reward_detail(log, reward_pool_id, sql_pool): """ 获取 赏池 详情 :param log: 日志对象 :param reward_pool_id: 赏池id :param sql_pool: 数据库连接池对象 """ log.debug(f"{inspect.currentframe().f_code.co_name} 开始获取详情数据") url = "https://kurabu.feishezhang.com/api.php" params = { "s": "reward/GetRandomAvailableSession", "system_type": "default", "application": "app", "application_client_type": "weixin", "token": token, "uuid": "227de1d4-c429-4e73-b972-42c26560ed83", "ajax": "ajax" } data = { # "reward_pool_id": "3cd3fa34-b0b7-4b1b-a593-411191fc745f", "reward_pool_id": reward_pool_id, "session_id": "" } # data = json.dumps(data, separators=(',', ':')) response = requests.post(url, headers=headers, params=params, json=data) response.raise_for_status() resp_json = response.json() # print(resp_json) if resp_json['code'] == 0: session_data = resp_json.get('data', {}).get('session', {}) session_id = session_data.get('id') prize_items = resp_json.get('data', {}).get('prize_items', []) info_list = [] for item in prize_items: prize_id = item.get('id') prize_name = item.get('name') level_nickname = item.get('level_nickname') parameter = item.get('parameter') prize_image = item.get('image') prize_num = item.get('num') goods_ids = item.get('goods_ids', []) goods_ids = ','.join(map(str, goods_ids)) if goods_ids else None prize_price = item.get('reference_price') prize_inventory = item.get('inventory') # 剩余库存 data_dict = { "reward_pool_id": reward_pool_id, "session_id": session_id, # int "prize_id": prize_id, "prize_name": prize_name, "level_nickname": level_nickname, "parameter": parameter, "prize_image": prize_image, "prize_num": prize_num, "goods_ids": goods_ids, "prize_price": prize_price, "prize_inventory": prize_inventory, "crawl_category": crawl_category } # print(data_dict) info_list.append(data_dict) sql_pool.insert_many(table="feishezhang_reward_detail_record", data_list=info_list, ignore=True) else: log.warning(resp_json['msg']) @retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log) def get_reward_order_single_page(log, reward_pool_id, session_id, page, sql_pool): """ 获取 赏池 订单列表 :param log: 日志对象 :param reward_pool_id: 赏池id :param session_id: 订单id :param page: 页码 :param sql_pool: 数据库连接池对象 :return: len(data_list) -> 获取数据长度 """ log.debug(f"{inspect.currentframe().f_code.co_name} 开始获取第{page}页数据, session_id:{session_id}") url = "https://kurabu.feishezhang.com/api.php" params = { "s": "reward/MiniappGetCurrentSessionDrawRecords", "system_type": "default", "application": "app", "application_client_type": "weixin", "token": token, "uuid": "227de1d4-c429-4e73-b972-42c26560ed83", "ajax": "ajax", # "session_id": "3065", "session_id": session_id, "record_page": str(page) # "record_page": "1" } response = requests.get(url, headers=headers, params=params) response.raise_for_status() resp_json = response.json() if resp_json['code'] == 0: data_list = resp_json.get('data', []) if not data_list: log.debug(f"当前页数据为空，已结束获取数据，当前页数：{page}, session_id:{session_id}") return 0 info_list = [] for cdata in data_list: order_id = cdata.get('id') goods_id = cdata.get('goods_id') draw_time = cdata.get('draw_time') level_nickname = cdata.get('level_nickname') user_id = cdata.get('user_id') images = cdata.get('images') title = cdata.get('title') nickname = cdata.get('nickname') data_dict = { "reward_pool_id": reward_pool_id, "session_id": session_id, "order_id": order_id, "goods_id": goods_id, "draw_time": draw_time, "level_nickname": level_nickname, "user_id": user_id, "images": images, "title": title, "nickname": nickname, "crawl_category": crawl_category } # print(data_dict) info_list.append(data_dict) sql_pool.insert_many(table="feishezhang_reward_order_record", data_list=info_list, ignore=True) return len(data_list) else: log.warning(resp_json['msg']) return 0 def get_reward_order_list(log, reward_pool_id, total_sessions, sql_pool): """ 获取 赏池 订单列表 :param log: 日志对象 :param reward_pool_id: 赏池id :param total_sessions: 订单总箱数 :param sql_pool: 数据库连接池对象 """ log.debug(f"{inspect.currentframe().f_code.co_name} 开始获取数据") session_id_list = get_all_boxes(log, reward_pool_id, total_sessions) for session_id in session_id_list: log.debug(f"{inspect.currentframe().f_code.co_name} 获取第{session_id}的订单数据") page = 1 max_page = 1000 while page <= max_page: try: len_data_list = get_reward_order_single_page(log, reward_pool_id, session_id, page, sql_pool) except Exception as e: log.error(e) len_data_list = 0 if len_data_list < 20: log.debug(f"当前页数据不足20条，已结束获取数据，当前页数：{page}") break page += 1 # time.sleep(random.uniform(1, 2)) def get_all_boxes(log, reward_pool_id, total_sessions): url = "https://kurabu.feishezhang.com/api.php" params = { "s": "reward/MiniappGetPreviousAvailableSession", "system_type": "default", "application": "app", "application_client_type": "weixin", # "token": "26eed48c80bad053717687daef3a8668", "token": token, "uuid": "0b2f5448-a8ca-4dbe-9bac-0c49a10d9873", "ajax": "ajax" } session_id_list = [] for i in range(1, total_sessions + 1): log.debug(f"{inspect.currentframe().f_code.co_name} 获取reward_pool_id:{reward_pool_id}, 第{i}箱赏") data = { "reward_pool_id": reward_pool_id, # "reward_pool_id": "65dc0347-63b5-41de-8bb8-811958a3b6c8", # "current_session_num": 2 "current_session_num": i } response = requests.post(url, headers=headers, params=params, json=data) response.raise_for_status() resp_json = response.json() if resp_json['code'] == 0: session_id = resp_json.get('data', {}).get('id') session_id_list.append(session_id) else: log.warning(f"{resp_json['msg']}") session_id_list = list(set(session_id_list)) # print(session_id_list) return session_id_list @retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log) def fs_yifan_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: # 获取赏池列表数据 try: sql_reward_list = sql_pool.select_all( f"select distinct reward_pool_id from feishezhang_reward_list_record where crawl_category = '{crawl_category}'") sql_reward_list = [item[0] for item in sql_reward_list] get_reward_list(log, sql_pool, sql_reward_list) except Exception as e: log.error(f'get_reward_list -> error: {e}') # # 获取赏列表数据 # sql_detail_list = sql_pool.select_all( # f"select id, reward_pool_id from feishezhang_reward_list_record where detail_state = 0 and crawl_category = '{crawl_category}'") # for sql_detail in sql_detail_list: # sql_id = sql_detail[0] # reward_pool_id = sql_detail[1] # log.info(f"开始处理数据：{reward_pool_id}") # try: # get_reward_detail(log, reward_pool_id, sql_pool) # sql_pool.update_one( # "update feishezhang_reward_list_record set detail_state = 1 where id = %s", (sql_id,)) # except Exception as e: # log.error(f'get_reward_detail -> error: {e}') # sql_pool.update_one( # "update feishezhang_reward_list_record set detail_state = 2 where id = %s", (sql_id,)) # # time.sleep(random.uniform(0.1, 1)) # 获取订单列表数据 sql_order_list = sql_pool.select_all( f"select reward_pool_id, total_sessions from feishezhang_reward_list_record where yi_order_state = 0 and crawl_category = '{crawl_category}'") for sql_order in sql_order_list: reward_pool_id = sql_order[0] total_sessions = sql_order[1] log.info(f"开始处理数据, reward_pool_id：{reward_pool_id}, total_sessions:{total_sessions}") try: get_reward_order_list(log, reward_pool_id, total_sessions, sql_pool) sql_pool.update_one( "update feishezhang_reward_list_record set yi_order_state = 1 where reward_pool_id = %s", (reward_pool_id,) ) except Exception as e: log.error(f'get_reward_order_list -> error: {e}') sql_pool.update_one( "update feishezhang_reward_list_record set yi_order_state = 2 where reward_pool_id = %s", (reward_pool_id,) ) 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} 运行结束,等待下一轮的采集任务............') if __name__ == '__main__': # get_login() # get_reward_list(logger) # get_reward_detail(logger, "3cd3fa34-b0b7-4b1b-a593-411191fc745f", None) # get_reward_order_list(logger, "3cd3fa34-b0b7-4b1b-a593-411191fc745f", "3065", None) # get_all_boxes(logger, "c058b102-8ea6-48c2-9195-1216b010f579", 3) fs_yifan_main(logger)