# -*- coding: utf-8 -*- # Author : Charley # Python : 3.12.10 # Date : 2026/07/10 """ REA (collectrea.com) 增量爬虫(日调度) 逻辑(两阶段,与 wheatland 一致): 1. GET /search 只解析「最近」板块(RECENT AUCTIONS tab,约 16 场) 2. 查库 select distinct auction_key from rea_record,得到已爬过的场次 3. 差集 = 新增场次 4. 没有新增 → 本轮无数据可抓,结束 5. 阶段一 列表:对每个新增场次翻页抓 lot 列表入库(state 默认 0) 6. 阶段二 详情:扫库 state != 1 的记录 → 逐条进详情页抓「标题 + 5 字段 + 多图」写回 目标网站: https://collectrea.com/search 说明:REA 的场次一旦结束即定型,新场次会出现在「最近」板块顶部;因此增量只需盯「最近」, 用 auction_key 差集识别新增场次即可(老场次早已在库)。 """ import sys import time import random import schedule from curl_cffi import requests from loguru import logger from tenacity import retry, stop_after_attempt, wait_fixed from mysql_pool import MySQLConnectionPool from rea_core import ( client_identifier_list, crawl_one_auction, get_auction_list, update_details_for_pending, after_log, ) 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") def get_existing_auction_keys(log, sql_pool): """查库返回已爬过的 auction_key 集合。 Args: log: logger 对象。 sql_pool: MySQL 连接池;为 None 时返回空集合(视作库内无任何场次)。 Returns: set[str]: 已存在的 auction_key 字符串集合。 """ if sql_pool is None: log.warning("sql_pool 为 None,视为库内无任何场次(将全量重抓最近板块)") return set() rows = sql_pool.select_all( "select distinct auction_key from rea_record" ) keys = {str(r[0]) for r in rows} if rows else set() log.info(f"库中已存在 {len(keys)} 个 auction_key") return keys def diff_new_auctions(log, recent_auctions, existing_keys): """从「最近」场次中筛出库里没有的新增场次。 Args: log: logger 对象。 recent_auctions (list[dict]): get_auction_list(only_recent=True) 的返回。 existing_keys (set[str]): 已存在的 auction_key 集合。 Returns: list[dict]: 待抓取的新增场次列表。 """ new_list = [a for a in recent_auctions if a["auction_key"] not in existing_keys] log.info(f"新增待抓取场次数: {len(new_list)} -> {[a['auction_key'] for a in new_list]}") return new_list def run_incremental(log, sql_pool): """增量抓取主流程。 Args: log: logger 对象。 sql_pool: MySQL 连接池;为 None 时不入库,仅在内存收集并 print 样本。 """ impersonate = random.choice(client_identifier_list) with requests.Session() as session: try: recent = get_auction_list(log, session, impersonate, only_recent=True) except Exception as e: log.error(f"获取最近场次列表失败: {e}") return existing_keys = get_existing_auction_keys(log, sql_pool) new_auctions = diff_new_auctions(log, recent, existing_keys) if not new_auctions: log.info("本轮无新增场次,跳过 list 抓取") return collected = [] for idx, auc in enumerate(new_auctions, 1): log.info(f"========== [{idx}/{len(new_auctions)}] 开始抓场次 {auc['auction_key']} ({auc['auction_name']}) ==========") try: lots = crawl_one_auction(log, sql_pool, session, impersonate, auc) if sql_pool is None: collected.extend(lots) except Exception as e: log.error(f"场次 {auc['auction_key']} 抓取异常: {e}") continue if sql_pool is None: log.info(f"增量抓取结束,共 {len(collected)} 条 lot(未入库)") for row in collected[:3]: print(row) @retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log) def rea_main(log): """日调度主函数:增量 list + 补详情。 Args: log: logger 对象。 """ log.info(f"开始运行 {sys._getframe().f_code.co_name} 增量爬虫任务 ...") sql_pool = MySQLConnectionPool(log=log) if not sql_pool: log.error("MySQL数据库连接失败") raise Exception("MySQL数据库连接失败") try: # 阶段一:抓新增场次的列表入库 try: run_incremental(log, sql_pool) except Exception as e: log.error(f"增量抓取失败: {e}") # 阶段二:扫库 state != 1 的记录补抓详情 if sql_pool is not None: try: update_details_for_pending(log, sql_pool) except Exception as e: log.error(f"详情补抓失败: {e}") except Exception as e: log.error(f"{sys._getframe().f_code.co_name} error: {e}") finally: log.info(f"爬虫程序 {sys._getframe().f_code.co_name} 运行结束,等待下一轮采集 ...") def schedule_task(): """启动调度器:先即时跑一次,之后每天 05:00 跑一次增量。""" rea_main(log=logger) schedule.every().day.at("05:00").do(rea_main, log=logger) while True: schedule.run_pending() time.sleep(1) if __name__ == "__main__": # 测试时直接跑一次 # rea_main(log=logger) # 上生产再切回 schedule schedule_task()