# -*- coding: utf-8 -*- # Author : Charley # Python : 3.10.8 # Date : 2025/12/2 14:08 import shutil import threading import time import inspect import requests import schedule import user_agent from loguru import logger from parsel import Selector from datetime import datetime from mysql_pool import MySQLConnectionPool from DrissionPage import ChromiumPage, ChromiumOptions 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 = { "accept": "application/json", "referer": "https://courtyard.io/", "user-agent": user_agent.generate_user_agent() } # 全局变量标识首次运行是否完成 detail_first_run_completed = False # 详情页浏览器需拦截的资源模式。目标: 该页只需 DOM 结构做 xpath 解析(Activity history), # 故把「不影响 DOM 数据」的重资源全部拦掉——省请求、提速、降内存, 且不会 OOM。 # 注意: 拦 CSS/字体不影响 xpath(xpath 只认 DOM 不认样式), 图片已由 no_imgs 关闭。 # 视频/大媒体: 最占带宽与内存 VIDEO_PATTERNS = [ '*.mp4', '*.webm', '*.avi', '*.mov', '*.flv', '*.m3u8', '*.ts', '*googlevideo.com*', '*videoplayback*', '*video.twimg.com*', '*videocdn*', '*.mpd', # DASH视频流 ] # 字体: 纯展示资源, 拦掉可省请求且绝不影响 DOM/JS。 # 注意: 不要拦 *.css! courtyard.io 是 Next.js 应用, CSS 与 JS 同属路由 chunk 依赖图, # 拦掉 CSS 会导致 chunk-loading Promise 被 reject → 触发 React 错误边界(client-side exception)白屏。 STYLE_FONT_PATTERNS = [ '*.woff', '*.woff2', '*.ttf', '*.otf', '*.eot', ] # 第三方埋点/统计/广告/监控: 与业务数据无关, 纯拖慢加载 TRACKER_PATTERNS = [ '*google-analytics.com*', '*googletagmanager.com*', '*doubleclick.net*', '*facebook.net*', '*connect.facebook*', '*segment.com*', '*segment.io*', '*sentry.io*', '*mixpanel.com*', '*hotjar.com*', '*intercom.io*', '*fullstory.com*', '*amplitude.com*', ] # 汇总: 实际下发给浏览器的拦截清单 BLOCKED_URL_PATTERNS = VIDEO_PATTERNS + STYLE_FONT_PATTERNS + TRACKER_PATTERNS # 连续失败达到该阈值即判定浏览器卡死, 触发重启 MAX_CONSECUTIVE_FAILURES = 3 # 每连续处理该条数就主动重启一次浏览器, 释放 Chromium 累积内存(V8 堆/渲染进程/缓存随页面数线性增长) BROWSER_RESTART_INTERVAL = 100 # 每处理该条数就清理一次缓存/cookies, 减缓单实例内存增长 BROWSER_CLEAR_CACHE_INTERVAL = 10 # auto_port 临时用户目录的基路径(放 D 盘专属子目录)。auto_port 每次用随机端口, 会在此 # 目录下不断新建 userData/{port} 且退出不自动删除, 故每次新建浏览器前会清空整个此目录 BROWSER_TMP_PATH = r'D:\Drissionpage_temp\courtyard' 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_goods_list(log, sql_pool): """ 获取商品列表 :param log: logger对象 :param sql_pool: MySQL连接池对象 :return: """ log.info(f"========================== 开始获取商品列表 ==========================") url = "https://api.courtyard.io/vending-machines" response = requests.get(url, headers=headers, timeout=22) # print(response.text) response.raise_for_status() # 同样用 `or []` 兜底: 防止 vendingMachines 为 null 时下方 for 迭代抛 TypeError vendingMachines = response.json().get("vendingMachines") or [] for item in vendingMachines: bag_id = item.get("id") bag_title = item.get("title") # sealed_pack_animation = item.get("sealedPackAnimation") # sealed_pack_image = item.get("sealedPackImage") category_title = item.get("category", {}).get("title") price = item.get("saleDetails", {}).get("salePriceUsd") data_dict = { "bag_id": bag_id, "bag_title": bag_title, "category": category_title, "price":price } # log.info(f'get_goods_list: {data_dict}') try: get_goods_detail(log, data_dict, sql_pool) except Exception as e: log.error(f"Error processing item: {e}") # 保存数据 # if info_list: # log.info(f"获取商品列表成功, 共 {len(info_list)} 条数据") # sql_pool.insert_many(table="courtyard_vending_machines_record", data_list=info_list, ignore= True) @retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log) def get_goods_detail(log, query_dict: dict, sql_pool=None): """ 获取商品详情 :param log: logger对象 :param query_dict: query_dict :param sql_pool: MySQL连接池对象 :return: """ log.info(f"========================== 获取商品详情 ==========================") url = "https://api.courtyard.io/index/query/recent-pulls" params = { "limit": "250", # "vendingMachineIds": "pkmn-basic-pack" "vendingMachineIds": query_dict["bag_id"] } response = requests.get(url, headers=headers, params=params, timeout=22) # print(response.text) response.raise_for_status() resp_json = response.json() # 接口偶发返回 {"error": ...}（无 assets 键，多为限流/暂无数据），显式跳过并告警，避免静默吞掉 if isinstance(resp_json, dict) and "assets" not in resp_json: log.warning(f"recent-pulls 返回无 assets, bag_id: {query_dict['bag_id']}, resp: {str(resp_json)[:150]}") return # 注意: dict.get(key, default) 的 default 仅在 key 缺失时生效; # 若 key 存在但值为 null(None) 仍返回 None, 故统一用 `or []` 兜底, 防止后续 len()/迭代抛 TypeError pulls = resp_json.get("assets") or [] info_list = [] for item in pulls: detail_title = item.get("title") detail_id = item.get("proof_of_integrity") if not detail_id: log.error(f"信息异常, detail_id: {detail_id}") continue # asset_pictures 实测存在为 null 的情况, 原写法 len(None) 是本次 TypeError 崩溃主因, 用 `or []` 兜底 asset_pictures = item.get("asset_pictures") or [] img_front = asset_pictures[0] if len(asset_pictures) > 0 else None img_back = asset_pictures[1] if len(asset_pictures) > 1 else None # crawl_date = time.strftime("%Y-%m-%d", time.localtime()) data_dict = { "bag_id": query_dict["bag_id"], "bag_title": query_dict["bag_title"], "category": query_dict["category"], "price": query_dict["price"], "detail_id": detail_id, "detail_title": detail_title, "img_front": img_front, "img_back": img_back, # "crawler_date": crawl_date } # log.info(f'data_dict:{data_dict}') info_list.append(data_dict) # 保存数据 if info_list: log.info(f"获取商品详情成功, 共 {len(info_list)} 条数据") sql_pool.insert_many(table="courtyard_list_record", data_list=info_list, ignore=True) def convert_time_format(time_str): """ 将时间字符串转换为标准格式 :param time_str: 原始时间字符串，如 "December 3, 2025 at 4:29 PM" :return: 标准时间格式字符串，如 "2025-12-03 16:29:00" """ if not time_str: return None try: dt_obj = datetime.strptime(time_str, "%B %d, %Y at %I:%M %p") return dt_obj.strftime("%Y-%m-%d %H:%M:%S") except ValueError as e: logger.warning(f"时间转换失败: {time_str}, 错误: {e}") return None @retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log) def get_sale_detail_single_page(log, page, sql_id, detail_id, sql_pool=None): """ 获取商品详情 :param log: logger对象 :param page: page对象 :param sql_id: 数据库id :param detail_id: 商品详情id :param sql_pool: MySQL连接池对象 :return: """ log.info(f"========================== 获取商品 详情, sql_id: {sql_id} ==========================") # page_url = "https://courtyard.io/asset/a4f0bbebd858370567f1779fddf0f55630810116d80965e33940fc8ff5ac94b4" page_url = f"https://courtyard.io/asset/{detail_id}" page.get(page_url) # 加载策略为 none, get() 立即返回; 这里只等目标节点 "Activity history" 渲染出现即可开始解析, # 不等整页 load 完成——目标节点在, 就代表要抓的数据已就绪, 大幅缩短单条耗时 target = page.ele('xpath://h6[text()="Activity history"]', timeout=25) if not target: log.error(f'{inspect.currentframe().f_code.co_name} -> 目标节点未出现(可能加载失败/被限流), 重试..........') raise Exception('目标节点未出现, 重新加载........') # 抛出异常以便重试 log.debug(f'{inspect.currentframe().f_code.co_name} -> 目标节点已就绪, url: {page_url}') html = page.html if not html: log.error(f'{inspect.currentframe().f_code.co_name} -> 页面加载失败...........') raise Exception('页面加载失败, 重新加载........') # 抛出异常以便重试 selector = Selector(text=html) # 方法一：通过文本内容匹配（优先） correlation_spans = selector.xpath('//span[contains(text(), ":")]/text()') correlation_id = None for text_selector in correlation_spans: correlation_id = text_selector.get() # ✅ 获取字符串 # match = re.search(r'[\w\s]+:\s*(\d+)', text) # if match: # correlation_id = match.group(1) # break # 获取第一个有效 ID # correlation_spans = selector.xpath('//span[contains(text(), ":")]') # correlation_text = None # # for span in correlation_spans: # text = span.get() # if text and ":" in text: # correlation_text = text # break # 如果方法一失败，使用方法二：通过结构定位（备用） if not correlation_id: correlation_span = selector.xpath('//a[contains(@href, "cgccards.com")]/preceding-sibling::span[1]/text()') correlation_id = correlation_span.get() # 初始化所有可能的字段为None data_dict = {"detail_id": detail_id, "correlation_id": correlation_id, "burn_from": None, "burn_to": None, "burn_time": None, "sale_price": None, "sale_from": None, "sale_to": None, "sale_time": None, "mint_price": None, "mint_from": None, "mint_to": None, "mint_time": None} # 获取 "Activity history" 后面的 div activity_div = selector.xpath('//h6[text()="Activity history"]/following-sibling::div[1]/div') for tag_div in activity_div: tag_name = tag_div.xpath('./div[1]/div/span/text()').get() if not tag_name: continue if tag_name == "Burn": data_dict["burn_from"] = tag_div.xpath('./div[2]/div[1]//h6/text()').get() data_dict["burn_to"] = tag_div.xpath('./div[2]/div[2]//h6/text()').get() data_dict["burn_time"] = tag_div.xpath('./div[2]/div[3]/span/@aria-label').get() # December 3, 2025 at 4:29 PM 转换时间格式 data_dict["burn_time"] = convert_time_format(data_dict["burn_time"]) elif tag_name == "Sale": sale_price = tag_div.xpath('./div[2]/span/text()').get() if sale_price: sale_price = sale_price.replace("$", "").replace(",", "") data_dict["sale_price"] = sale_price data_dict["sale_from"] = tag_div.xpath('./div[3]/div[1]//h6/text()').get() data_dict["sale_to"] = tag_div.xpath('./div[3]/div[2]//h6/text()').get() data_dict["sale_time"] = tag_div.xpath('./div[3]/div[3]/span/@aria-label').get() # December 3, 2025 at 4:29 PM 转换时间格式 data_dict["sale_time"] = convert_time_format(data_dict["sale_time"]) elif tag_name == "Mint": mint_price = tag_div.xpath('./div[2]/span/text()').get() if mint_price: mint_price = mint_price.replace("$", "").replace(",", "") data_dict["mint_price"] = mint_price data_dict["mint_from"] = tag_div.xpath('./div[3]/div[1]//h6/text()').get() data_dict["mint_to"] = tag_div.xpath('./div[3]/div[2]//h6/text()').get() data_dict["mint_time"] = tag_div.xpath('./div[3]/div[3]/span/@aria-label').get() # December 3, 2025 at 4:29 PM 转换时间格式 data_dict["mint_time"] = convert_time_format(data_dict["mint_time"]) # log.info(f'Sale detail data: {data_dict}') # 保存数据 sql_pool.insert_one_or_dict(table="courtyard_detail_record", data=data_dict, ignore=True) sql_pool.update_one("UPDATE courtyard_list_record SET state = 1 WHERE id = %s", (sql_id,)) def _create_detail_browser(log): """创建并配置用于详情采集的浏览器实例。 集中管理浏览器启动参数、视频拦截规则与超时, 供首次启动和卡死重启复用。 Args: log: loguru logger 对象, 从调用方透传。 Returns: ChromiumPage: 已配置拦截规则与超时的浏览器页面对象。 """ # auto_port 的临时用户目录退出不会自动清理, 每次随机端口都会在 BROWSER_TMP_PATH 下留一份; # 故新建实例前先清空整个目录, 保证不残留缓存、不累积占磁盘(重启时旧实例已 quit, 无占用) shutil.rmtree(BROWSER_TMP_PATH, ignore_errors=True) options = ChromiumOptions() # 无需登录态, 不做任何持久化: auto_port 自动分配空闲端口(每次全新进程, 确保内存被彻底释放), # 相比固定端口: 分批重启不会因端口被残留进程占用而失败/误连到旧浏览器 options.auto_port(True) # 临时用户目录基路径放到 D 盘专属子目录(默认在 C 盘 %TEMP%/DrissionPage) options.set_tmp_path(BROWSER_TMP_PATH) # options.set_proxy("http://" + tunnel) options.no_imgs(True) # 加载策略设为 none: page.get() 不等整页 load 完成立即返回, 之后由业务代码只等 # 目标元素(Activity history)出现即开始解析——这是提速最关键的一招(SPA 整页 load # 往往还挂着一堆无关请求/长连接, 傻等会白白拖慢每一条) options.set_load_mode('none') # 禁止媒体相关设置 options.set_argument('--autoplay-policy=user-gesture-required') options.set_argument('--disable-features=PreloadMediaEngagementData') # ---- 省资源(真正有效且无副作用的项): 减少下载与缓存占用, 不掐内存上限 ---- # 说明: 不再使用 --max-old-space-size / --renderer-process-limit=1 / --process-per-site。 # 前者会在重 SPA 需要更多 V8 堆时直接触发渲染进程 OOM 崩溃(即"喔唷崩溃啦 Out of Memory"); # 后两者强制单渲染进程复用, 反而关闭了浏览器靠"进程销毁"回收内存的机制, 让内存只增不减。 # 真正的内存回收依赖每 BROWSER_RESTART_INTERVAL 条整浏览器重启(见 get_sale_detail_list)。 options.set_argument('--disk-cache-size=1') # 几乎禁用磁盘缓存 options.set_argument('--media-cache-size=1') # 几乎禁用媒体缓存 options.set_argument('--disable-application-cache') # 禁用应用缓存 options.set_argument('--disable-dev-shm-usage') # 不占用共享内存, 避免共享内存耗尽 options.set_argument('--disable-extensions') # 禁用扩展 options.set_argument('--disable-background-networking') # 关闭后台网络活动 # 最大化 options.set_argument("--start-maximized") options.set_argument("--disable-gpu") options.set_argument("-accept-lang=en-US") page = ChromiumPage(options) # 使用 set.blocked_urls() 拦截视频/样式/字体/埋点等无关资源(不影响 DOM 取数) page.set.blocked_urls(BLOCKED_URL_PATTERNS) # 设置超时: 页面加载最多等 30s, 基础操作 20s, 避免浏览器卡死时无限等待 page.set.timeouts(base=20, page_load=30) return page def _restart_detail_browser(log, page, reason='卡死'): """关闭旧浏览器并重建一个新实例。 先尽力 quit 旧实例(失败忽略), 等待端口释放后重新创建。既用于连续采集失败时的 自愈重启, 也用于每处理 BROWSER_RESTART_INTERVAL 条后的主动内存释放重启。 Args: log: loguru logger 对象, 从调用方透传。 page (ChromiumPage): 待关闭的旧浏览器实例。 reason (str, optional): 重启原因, 仅用于日志区分。Defaults to '卡死'。 Returns: ChromiumPage: 全新的浏览器页面对象。 """ try: page.quit() except Exception as e: # 卡死的浏览器 quit 本身也可能超时/报错, 忽略即可 log.warning(f'关闭旧浏览器失败(忽略): {e}') # 等待端口(9138)与用户数据目录释放, 再重建, 避免端口占用导致启动失败 time.sleep(3) log.warning(f'浏览器{reason}, 正在重启新实例..........') return _create_detail_browser(log) @retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log) def get_sale_detail_list(log, sql_pool=None): """ 获取商品详情 :param log: logger对象 # :param detail_id_list: 详情id列表 :param sql_pool: MySQL连接池对象 :return: """ log.info(f"========================== 获取商品 详情 LIST ==========================") page = _create_detail_browser(log) # 连续失败计数: 用于区分"单条数据问题"与"浏览器整体卡死", 后者需重启浏览器 consecutive_failures = 0 try: sql_detail_id_list = sql_pool.select_all("SELECT id, detail_id FROM courtyard_list_record WHERE state = 0") for idx, sql_detail_id in enumerate(sql_detail_id_list): sql_id = sql_detail_id[0] detail_id = sql_detail_id[1] # 每处理 BROWSER_RESTART_INTERVAL 条主动重启浏览器, 彻底释放 Chromium 累积内存; # 单实例连续跑几百页时 V8 堆/渲染进程/缓存会线性增长到数 GB, 这是内存/CPU 高的主因 if idx > 0 and idx % BROWSER_RESTART_INTERVAL == 0: log.info(f'已连续处理 {idx} 条, 达到批次阈值, 主动重启浏览器释放内存..........') page = _restart_detail_browser(log, page, reason='达到批次重启阈值') consecutive_failures = 0 try: get_sale_detail_single_page(log, page, sql_id, detail_id, sql_pool) consecutive_failures = 0 # 成功即清零, 只累计"连续"失败 # 定期清理缓存/cookies, 减缓单实例在两次批次重启之间的内存增长 if idx > 0 and idx % BROWSER_CLEAR_CACHE_INTERVAL == 0: try: page.clear_cache() except Exception as ce: log.warning(f'清理浏览器缓存失败(忽略): {ce}') except Exception as e: consecutive_failures += 1 log.error(f'get_sale_detail_single_page error ' f'({consecutive_failures}/{MAX_CONSECUTIVE_FAILURES}), sql_id: {sql_id}: {e}') if consecutive_failures >= MAX_CONSECUTIVE_FAILURES: # 连续多条失败, 判定浏览器已卡死: 不标记 state=2(避免误伤), # 保留 state=0 待下轮重试, 并重启浏览器后继续处理后续条目 page = _restart_detail_browser(log, page) consecutive_failures = 0 else: # 未达阈值, 视为该条数据自身问题, 标记 state=2 跳过 sql_pool.update_one("UPDATE courtyard_list_record SET state = 2 WHERE id = %s", (sql_id,)) except Exception as e: log.error(f'get_response error: {e}') raise # 直接透传原始异常(原写法 raise 字符串会抛 TypeError 掩盖真因) finally: try: page.quit() except Exception as e: log.warning(f'最终关闭浏览器失败(忽略): {e}') @retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log) def list_main(log): """ 主函数 自动售货机 :param log: logger对象 """ log.info( f'开始运行 {inspect.currentframe().f_code.co_name} 爬虫任务....................................................') start = time.time() # 配置 MySQL 连接池 sql_pool = MySQLConnectionPool(log=log) if not sql_pool.check_pool_health(): log.error("数据库连接池异常") raise RuntimeError("数据库连接池异常") try: try: log.debug('------------------- 开始获取商品列表 -------------------') get_goods_list(log, sql_pool) except Exception as e: log.error(f'get_goods_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} 运行结束,等待下一轮的采集任务............') end = time.time() elapsed_time = end - start log.info(f'============================== 本次爬虫运行时间:{elapsed_time:.2f} 秒 ===============================') return elapsed_time @retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log) def detail_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("数据库连接池异常") global detail_first_run_completed try: # 获取详情页信息 try: log.debug('------------------- 获取商品 detail 数据 -------------------') get_sale_detail_list(log, sql_pool) except Exception as e: log.error(f'get_sale_detail_list error: {e}') except Exception as e: log.error(f'{inspect.currentframe().f_code.co_name} error: {e}') finally: detail_first_run_completed = True log.info(f'爬虫程序 {inspect.currentframe().f_code.co_name} 运行结束,等待下一轮的采集任务............') def control_list_mask(log): """ 控制列表爬虫任务 每10分钟运行 :param log: logger对象 """ while True: log.info( f'--------------------- 开始运行 {inspect.currentframe().f_code.co_name} 新一轮的爬虫任务 ---------------------') elapsed_time = list_main(log) # 计算剩余时间 wait_time = max(0, 300 - int(elapsed_time)) if wait_time > 0: log.info(f"程序运行时间{elapsed_time:.2f}秒, 小于 5 分钟，等待 {wait_time:.2f} 秒后再开始下一轮任务") time.sleep(wait_time) else: log.info("程序运行时间大于等于5分钟，直接开始下一轮任务") def scheduled_detail_main(log): """定时任务调用的包装函数""" global detail_first_run_completed if detail_first_run_completed: detail_main(log) else: log.info("Skipping scheduled task as first run is not completed yet") def run_threaded(job_func, *args, **kwargs): """ 在新线程中运行给定的函数，并传递参数。 :param job_func: 要运行的目标函数 :param args: 位置参数 :param kwargs: 关键字参数 """ job_thread = threading.Thread(target=job_func, args=args, kwargs=kwargs) job_thread.start() def schedule_task(): """ 设置定时任务 """ # 启动 control_list_mask 任务线程 list_thread = threading.Thread(target=control_list_mask, args=(logger,)) list_thread.daemon = True # 设置为守护线程，主程序退出时自动结束 list_thread.start() # 启动 detail_main 任务线程（首次运行） detail_thread = threading.Thread(target=detail_main, args=(logger,)) detail_thread.daemon = True detail_thread.start() # 设置定时任务 每天 # schedule.every().day.at("00:01").do(run_threaded, detail_main, logger) schedule.every().day.at("00:01").do(run_threaded, scheduled_detail_main, logger) while True: schedule.run_pending() time.sleep(1) if __name__ == '__main__': schedule_task() # detail_main(log=logger) # get_sale_detail_list(log, ((1, 'a4f0bbebd858370567f1779fddf0f55630810116d80965e33940fc8ff5ac94b4'), # (2, 'a4f0bbebd858370567f1779fddf0f55630810116d80965e33940fc8ff5ac94b4')), sql_pool)