auctions90s_spider.py 5.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166
  1. # -*- coding: utf-8 -*-
  2. # Author : Charley
  3. # Python : 3.12.10
  4. # Date : 2026/5/28
  5. """
  6. 90sAuctions 增量爬虫(半月调度)
  7. 逻辑:
  8. 1. GET 首页解析当前网站全部 auction id
  9. 2. 查库 select distinct auction_id from auctions90s_record,得到已爬过的 auction
  10. 3. 差集 = 新增 auction
  11. 4. 没有新增 → 本轮无数据可抓,结束
  12. 5. 对每个新增 auction:postback 切换 → 翻页 → 写库
  13. 6. 补抓 state != 1 的详情页
  14. 7. 补采 status='Open' 的 lot:拍卖结束后回写最终 bids/status/price
  15. """
  16. import time
  17. import random
  18. import inspect
  19. import schedule
  20. from curl_cffi import requests
  21. from loguru import logger
  22. from tenacity import retry, stop_after_attempt, wait_fixed
  23. from mysql_pool import MySQLConnectionPool
  24. from auctions90s_core import (
  25. TABLE_NAME,
  26. client_identifier_list,
  27. crawl_one_auction,
  28. get_auction_list,
  29. update_details_for_pending,
  30. update_open_lots,
  31. after_log,
  32. )
  33. logger.remove()
  34. logger.add("./logs/{time:YYYYMMDD}.log", encoding='utf-8', rotation="00:00",
  35. format="[{time:YYYY-MM-DD HH:mm:ss.SSS}] {level} {message}",
  36. level="DEBUG", retention="7 day")
  37. def get_existing_auction_ids(log, sql_pool):
  38. """
  39. 查库返回已爬过的 auction_id 集合。
  40. :param log: (loguru.Logger) 日志对象
  41. :param sql_pool: (MySQLConnectionPool) 数据库连接池
  42. :return: (set[str]) 已存在的 auction_id 集合(字符串形式,与首页解析值对齐)
  43. """
  44. rows = sql_pool.select_all(
  45. f"select distinct auction_id from {TABLE_NAME} where auction_id is not null"
  46. )
  47. ids = {str(r[0]) for r in rows} if rows else set()
  48. log.info(f"库中已存在 {len(ids)} 个 auction_id: {sorted(ids)}")
  49. return ids
  50. def diff_new_auctions(log, all_auctions, existing_ids):
  51. """
  52. 从首页解析的全部 auctions 中筛出库里没有的。
  53. :param log: (loguru.Logger) 日志对象
  54. :param all_auctions: (list[dict]) get_auction_list 返回的全部拍卖会列表
  55. :param existing_ids: (set[str]) 已存在的 auction_id 集合
  56. :return: (list[dict]) 新增待抓的 auction 列表
  57. """
  58. new_list = [a for a in all_auctions if a["id"] not in existing_ids]
  59. log.info(f"新增待抓取 auction 数: {len(new_list)} -> {[(a['id'], a['name']) for a in new_list]}")
  60. return new_list
  61. def run_incremental(log, sql_pool):
  62. """
  63. 增量抓取主流程:拉首页 → 差集 → 逐个抓新增 auction。
  64. :param log: (loguru.Logger) 日志对象
  65. :param sql_pool: (MySQLConnectionPool) 数据库连接池
  66. :return: None
  67. """
  68. impersonate = random.choice(client_identifier_list)
  69. with requests.Session() as session:
  70. try:
  71. all_auctions = get_auction_list(log, session, impersonate)
  72. except Exception as e:
  73. log.error(f"获取拍卖会列表失败: {e}")
  74. return
  75. existing_ids = get_existing_auction_ids(log, sql_pool)
  76. new_auctions = diff_new_auctions(log, all_auctions, existing_ids)
  77. if not new_auctions:
  78. log.info("本轮无新增 auction,跳过 list 抓取")
  79. return
  80. for idx, auc in enumerate(new_auctions, 1):
  81. aid, name = auc["id"], auc["name"]
  82. log.info(f"========== [{idx}/{len(new_auctions)}] 开始抓 auction={aid} ({name}) ==========")
  83. try:
  84. crawl_one_auction(log, sql_pool, session, impersonate,
  85. auction_id=aid, auction_name=name)
  86. except Exception as e:
  87. log.error(f"auction={aid} 抓取异常: {e}")
  88. continue
  89. @retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
  90. def nineties_main(log):
  91. """
  92. 日调度主函数:增量 list + 补详情。失败时按小时级重试(最多 100 次)。
  93. :param log: (loguru.Logger) 日志对象
  94. :return: None
  95. :raises Exception: MySQL 连接失败时抛出,由 tenacity 触发整轮重试
  96. """
  97. log.info(f'开始运行 {inspect.currentframe().f_code.co_name} 增量爬虫任务 ...')
  98. sql_pool = MySQLConnectionPool(log=log)
  99. if not sql_pool:
  100. log.error("MySQL数据库连接失败")
  101. raise Exception("MySQL数据库连接失败")
  102. try:
  103. try:
  104. run_incremental(log, sql_pool)
  105. except Exception as e:
  106. log.error(f'增量抓取失败: {e}')
  107. try:
  108. update_details_for_pending(log, sql_pool)
  109. except Exception as e:
  110. log.error(f'详情补抓失败: {e}')
  111. try:
  112. # 补采:重访 status='Open' 的 lot,拍卖结束后回写最终 bids/status/price
  113. update_open_lots(log, sql_pool)
  114. except Exception as e:
  115. log.error(f'拍卖结果补采失败: {e}')
  116. except Exception as e:
  117. log.error(f'{inspect.currentframe().f_code.co_name} error: {e}')
  118. finally:
  119. log.info(f'爬虫程序 {inspect.currentframe().f_code.co_name} 运行结束,等待下一轮采集 ...')
  120. def schedule_task():
  121. """
  122. 启动半月调度:脚本启动时先跑一次,之后每月 1 号和 15 号 05:00 各跑一次。
  123. :return: None(永不返回,内部死循环)
  124. """
  125. # nineties_main(log=logger)
  126. def run_semimonthly():
  127. # 每月 1 号和 15 号执行(半月一次)
  128. from datetime import date
  129. if date.today().day in (1, 15):
  130. nineties_main(log=logger)
  131. schedule.every().day.at("05:00").do(run_semimonthly)
  132. while True:
  133. schedule.run_pending()
  134. time.sleep(1)
  135. if __name__ == '__main__':
  136. # nineties_main(log=logger)
  137. schedule_task()