# -*- coding: utf-8 -*- # Author : Charley # Python : 3.10.8 # Date : 2025/6/30 19:35 import time from datetime import datetime import utils import random import inspect import schedule from loguru import logger from mysql_pool import MySQLConnectionPool from tenacity import retry, stop_after_attempt, wait_fixed logger.remove() logger.add("./logs/chouka_add_{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") 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") def get_draw_card_list(log, sql_pool): """ 获取 抽卡机 列表页信息 :param log: :param sql_pool: """ page = 1 max_page = 500 total_count = 0 sql_pid_list = sql_pool.select_all("SELECT pid FROM qiandao_mall_draw_card_list_record") sql_pid_list = [item[0] for item in sql_pid_list] # sql_pid_list = [] while page <= max_page: len_item, totalCount = draw_card_one_page(log, page, sql_pool, sql_pid_list) if len_item < 20: log.debug( f"--------------- page {page}, len_item: {len_item}, totalCount: {totalCount}, break !!! ---------------") break total_count += len_item if total_count >= int(totalCount): log.debug(f"total_count: {total_count} >= totalCount: {totalCount}, break...........") break page += 1 time.sleep(random.uniform(0.1, 1)) sql_pid_list.clear() @retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log) def draw_card_one_page(log, page, sql_pool, sql_pid_list): """ 获取 抽卡机 单页信息 :param log: :param page: :param sql_pool: :param sql_pid_list: """ log.debug(f"--------------- {inspect.currentframe().f_code.co_name}, page: {page} ---------------") url = "https://api.qiandao.com/shelf-web/list-home-recommender" params = { # "offset": "0", "offset": str((page - 1) * 20), "limit": "20", "cmd": "b2c_homepage_feed", "name": "cgmarket", "project": "channel", "type": "LIVE_GROUPON", "typeIds": [1443323], "withLive": "true", "tagIds": [1701855], "quickCouponTemplateId": "0" } resp_json = utils.request_get_data(log, url, params) # print(resp_json) rows = resp_json["data"]["rows"] parse_draw_card_data(log, rows, sql_pool, sql_pid_list) len_rows = len(rows) total_count = resp_json["data"]["count"] return len_rows, total_count def parse_draw_card_data(log, resp_json, sql_pool, sql_pid_list): log.debug(f"--------------- {inspect.currentframe().f_code.co_name} ---------------") info_list = [] for item in resp_json: # print(item) pid = item.get("id") if pid in sql_pid_list: log.debug(f"{inspect.currentframe().f_code.co_name}, pid: {pid} already exists") continue p_type = item.get("type") orgId = item.get("orgId") title = item.get("name") # description = item.get("description") isSoldOut = item.get("isSoldOut") price = item.get("price", {}).get("unitPriceOfCash") soldAmountText = item.get("soldAmountText") nickname = item.get("org", {}).get("nickname", '') info_dict = { "pid": pid, "p_type": p_type, "title": title, "price": price, "sold_amount_text": soldAmountText, "org_id": orgId, "nickname": nickname, # "description": description, "is_sold_out": isSoldOut } # print(info_dict) sql_pid_list.append(pid) info_list.append(info_dict) if info_list: # sql_pool.insert_many(table="qiandao_mall_draw_card_list_record", data_list=info_list) query = """ INSERT INTO qiandao_mall_draw_card_list_record (pid, p_type, title, price, sold_amount_text, org_id, nickname, is_sold_out) VALUES (%s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE p_type = VALUES(p_type), title = VALUES(title), price = VALUES(price), sold_amount_text = VALUES(sold_amount_text), org_id = VALUES(org_id), nickname = VALUES(nickname), is_sold_out = VALUES(is_sold_out) """ args_list = [(item['pid'], item['p_type'], item['title'], item['price'], item['sold_amount_text'], item['org_id'], item['nickname'], item['is_sold_out']) for item in info_list] sql_pool.insert_many(query=query, args_list=args_list) @retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log) def get_draw_card_details(log, sql_pool, pid): """ 获取 抽卡机 详情页信息 :param log: :param sql_pool: :param pid: :return: {'pid': '867827029033121899', 'p_type': 'GACHA', 'org_id': '710150753377039896', 'title': '0.01/包 灵焰包第2弹 可合成 累计抽卡领好礼', 'price': 1.4, 'sold_amount_text': '已售2000+', 'nickname': '哇咔咔', 'is_sold_out': False} """ url = "https://api.qiandao.com/b2c-web/v1/gacha/query/detail" params = { "shelfId": "867827029033121899" } resp_json = utils.request_get_data(log, url, params) # print(resp_json) if resp_json.get("code") == 0: parse_draw_card_details_data(log, resp_json, pid, sql_pool) else: log.error(f"{inspect.currentframe().f_code.co_name} error: {resp_json.get('message')}") def parse_draw_card_details_data(log, resp_json, pid, sql_pool): log.debug(f"--------------- {inspect.currentframe().f_code.co_name} ---------------") data = resp_json.get("data") specification = data.get("packageConfigRamark") # 规格说明 introImages = data.get('images', {}).get("introImages", []) introImages = '|'.join(introImages) # 多图链接, |分割 categoryId = data.get("categoryId") vipPrice = data.get("vipPrice") unitPriceOfCash = data.get("unitPriceOfCash") detail_dict = { "specification": specification, "images": introImages, "category_id": categoryId, "unit_price_of_cash": unitPriceOfCash, "vip_price": vipPrice, } # print(detail_dict) sql_pool.update_one_or_dict(table="qiandao_mall_draw_card_list_record", data=detail_dict, condition={"pid": pid}) @retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log) def get_draw_ids(log, pid, sql_pool,max_trading_time): """ 获取 抽卡记录 id列表 信息 :param log: :param pid: :param sql_pool: :param max_trading_time: """ log.debug(f"--------------- {inspect.currentframe().f_code.co_name}, pid: {pid} ---------------") url = "https://api.qiandao.cn/b2c-web/v1/box/gacha/query/digest-draw-records" params = { # "shelfId": "855250935293687891" "shelfId": pid } resp_json = utils.request_get_data(log, url, params) # print(resp_json) if resp_json.get("code") == 0: rows = resp_json.get("data", {}).get("rows", []) draw_id_list = [item.get("id") for item in rows] # 每隔10个抽奖，获取一次 for i in range(0, len(draw_id_list), 10): log.debug( f"{inspect.currentframe().f_code.co_name}, get_draw_ids, i: {i}, len draw_id_list: {len(draw_id_list)}") try: get_draw_buy_data(log, pid, sql_pool, draw_id_list[i:i + 10],max_trading_time) except Exception as e: log.error(f"{inspect.currentframe().f_code.co_name} call draw_id_list error: {e}") else: log.error(f"{inspect.currentframe().f_code.co_name} error: {resp_json.get('message')}") @retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log) def get_draw_buy_data(log, pid, sql_pool, ids: list,max_trading_time): """ 获取 抽卡记录 信息 :param log: :param pid: :param sql_pool: :param ids: :param max_trading_time: """ url = "https://api.qiandao.cn/b2c-web/v1/box/gacha/query/ids-draw-records" # data = { # "ids": [ # "878724165295676667", # "878724165295676666", # "878724165295676665", # "878724165295676664", # "878724165295676663", # "878723922630058286", # "878723922630058285", # "878723922630058284", # "878723922630058283", # "878723922630058282" # ], # "shelfId": "855250935293687891" # } data = { "ids": ids, # "shelfId": "855250935293687891" "shelfId": pid } resp_json = utils.request_post_data(log, url, data) # print(resp_json) if resp_json.get("code") == 0: try: parse_draw_buy_data(log, resp_json, pid, sql_pool,max_trading_time) except Exception as e: log.error(f"{inspect.currentframe().f_code.co_name} call parse_draw_buy_data error: {e}") else: log.error(f"{inspect.currentframe().f_code.co_name} error: {resp_json.get('message')}") def is_draw_id_exists(log, sql_pool, draw_id): """ 判断 draw_id 是否存在于表中 :param log: 日志记录器 :param sql_pool: 数据库连接池 :param draw_id: 要检查的 draw_id :return: True 如果存在，否则 False """ try: query = """ SELECT EXISTS ( SELECT 1 FROM qiandao_mall_draw_card_buy_record WHERE draw_id = %s ) AS found """ result = sql_pool.select_one(query=query, args=(draw_id,)) return bool(result[0]) if result else False except Exception as e: log.error(f"{inspect.currentframe().f_code.co_name} error: {e}") return False def parse_draw_buy_data(log, resp_json, pid, sql_pool,max_trading_time): log.debug(f"--------------- {inspect.currentframe().f_code.co_name} ---------------") rows = resp_json.get('data', {}).get('rows', []) info_list = [] for item in rows: draw_id = item.get("id") # # 判断 draw_id 是否在 qiandao_mall_draw_card_buy_record 表中 # if is_draw_id_exists(log, sql_pool, draw_id): # log.debug(f"draw_id: {draw_id} 已存在，跳过插入") # continue productInfo = item.get("productInfo", {}) spu_id = productInfo.get("id", {}) spuName = productInfo.get("spuName") spuImage = productInfo.get("spuImage") rarity = item.get("rarity") userInfo = item.get("userInfo", {}) buyer_id = userInfo.get("id") buyer_name = userInfo.get("name") draw_time = item.get("time") trading_time_str = utils.transform_ms(log, draw_time) if not trading_time_str: log.debug(f"Invalid trading time: {trading_time_str}") continue # 跳过无效时间 # 字符串 -> datetime trading_time = datetime.strptime(trading_time_str, "%Y-%m-%d %H:%M:%S") if max_trading_time and trading_time <= max_trading_time: log.debug(f"trading_time: {trading_time_str} <= max_trading_time: {max_trading_time}, 跳过旧数据") break draw_dict = { "pid": pid, "draw_id": draw_id, "spu_id": spu_id, "spu_name": spuName, "spu_image": spuImage, "rarity": rarity, "buyer_id": buyer_id, "buyer_name": buyer_name, "draw_time": trading_time_str } # print(draw_dict) info_list.append(draw_dict) if info_list: try: sql_pool.insert_many(table="qiandao_mall_draw_card_buy_record", data_list=info_list) except Exception as e: log.error(f"{inspect.currentframe().f_code.co_name} call insert_many error: {e}") @retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log) def qd_draw_card_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: # 请求抽卡记录 # sql_pid2_list = sql_pool.select_all("SELECT pid FROM qiandao_mall_draw_card_list_record WHERE is_sold_out = 1") sql_pid2_list = sql_pool.select_all("SELECT pid FROM qiandao_mall_draw_card_list_record") sql_pid2_list = [item[0] for item in sql_pid2_list] # 测试 if not sql_pid2_list: # sql_pid2_list = ["855250935293687891"] log.debug("No sql_pid2_list") return for pid2 in sql_pid2_list: try: # 获取 该pid 购买数据的最大时间。 max_trading_time = sql_pool.select_one( query="SELECT MAX(draw_time) FROM qiandao_mall_draw_card_buy_record WHERE pid = %s", args=(pid2,)) max_trading_time = max_trading_time[0] if max_trading_time else None get_draw_ids(log, pid2, sql_pool,max_trading_time) except Exception as e: log.error(f"{inspect.currentframe().f_code.co_name} for sql_pid2_list: {pid2}, 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} 运行结束,等待下一轮的采集任务............') if __name__ == '__main__': qd_draw_card_main(log=logger)