# -*- coding: utf-8 -*- # Author : Charley # Python : 3.8.10 # Date: 2024-11-12 16:40 import time import requests import schedule import user_agent from loguru import logger import concurrent.futures from tenacity import stop_after_attempt, wait_fixed, retry from mysq_pool import MySQLConnectionPool 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="1 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") @retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log) def get_proxys(log): 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 def save_data(mysql_pool, info): """ :param mysql_pool: :param info: :return: """ sql = "INSERT INTO beckett_bgs_record(set_name, player_name, date_graded, centering_grade, corner_grade, edges_grade, surfaces_grade, auto_grade, final_grade, total_grade, cards_grade, number) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)" mysql_pool.insert_one(sql, info) @retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log) def get_data(log, bgs_id, mysql_pool): """ :param log: :param bgs_id: :param mysql_pool: :return: """ headers = { "accept": "application/json, text/plain, */*", "user-agent": user_agent.generate_user_agent() } url = "https://www.beckett.com/api/grading/lookup" params = { "category": "BGS", "serialNumber": str(bgs_id) } response = requests.get(url, headers=headers, params=params, proxies=get_proxys(log), timeout=5) if response.status_code == 404: # 没有数据 No Record Found 将状态改为3 log.warning(f"No Record Found for {bgs_id}") mysql_pool.update_one("UPDATE bgs_task SET state=3 WHERE auth_code=%s", (bgs_id,)) return if response.status_code != 200: # 查询失败 将状态改为2 log.warning(f"Error getting data for {bgs_id}, {response.status_code}") mysql_pool.update_one("UPDATE bgs_task SET state=2 WHERE auth_code=%s", (bgs_id,)) return # print(response.json()) result_dict = response.json() if result_dict: """ "label": "silver", "non_bccg_card_total": 0, "item_id": "17932864", "set_name": "2024 Magic the Gathering Secret Lair Dungeons & Dragons 50th Anniversary Bonus Card Foil", "sport_name": "Magic", "card_key": "0879", "player_name": "Minsc & Boo, Timeless Heroes M", "date_graded": "Thursday, April 17, 2025", "center_grade": "9.5", "corners_grade": "8.5", "edges_grade": "8.5", "surfaces_grade": "10.0", "autograph_grade": "0.0", "final_grade": "8.5", """ set_name = result_dict.get('set_name') player_name = result_dict.get('player_name') date_graded = result_dict.get('date_graded') centering_grade = result_dict.get('center_grade') corner_grade = result_dict.get('corners_grade') edges_grade = result_dict.get('edges_grade') surfaces_grade = result_dict.get('surface_grade') auto_grade = result_dict.get('autograph_grade') final_grade = result_dict.get('final_grade') total_grade = result_dict.get('pop_report') cards_grade = result_dict.get('pop_higher') info = (set_name, player_name, date_graded, centering_grade, corner_grade, edges_grade, surfaces_grade, auto_grade, final_grade, total_grade, cards_grade, int(bgs_id)) # 检查所有值是否都为 None或空字符串, 不包含bgs_id all_none_or_empty = all(x is None or x == '' for x in info[:-1]) if all_none_or_empty: log.debug("All values are empty") else: # print(info) save_data(mysql_pool, info) # 查询成功 将状态改为1 mysql_pool.update_one("UPDATE bgs_task SET state=1 WHERE auth_code=%s", (bgs_id,)) def process_urls(log, ids, mysql_pool, batch_size=1000, max_workers=5): with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: for i in range(0, len(ids), batch_size): # print(i) batch = ids[i:i + batch_size] # print(batch) try: futures_to_urls = {executor.submit(get_data, log, url, mysql_pool): url for url in batch} for future in concurrent.futures.as_completed(futures_to_urls): url = futures_to_urls[future] try: future.result() log.debug(f"处理 {url} 成功") except Exception as exc: log.debug(f"处理 {url} 出错: {exc}") except Exception as e: log.error(f"提交任务失败: {e}") @retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log) def bgs_main(log): try: log.info( "开始运行 bgs_main 爬虫任务............................................................") sql_pool = MySQLConnectionPool(log=log) if not sql_pool: log.error("数据库连接失败") raise Exception("数据库连接失败") # max_bgs_id = sql_pool.select_one("SELECT MAX(number) AS max_number FROM beckett_bgs_record") # # print(max_bgs_id_list) # max_bgs_id = max_bgs_id[0] # log.info(f"max_bgs_id 从 {max_bgs_id} 开始爬取.........................") # bgs_id_list = [i for i in range(max_bgs_id, max_bgs_id + 3001)] # # sql_pool.insert_all("INSERT INTO bgs_task(auth_code) VALUES (%s)", bgs_id_list) # 20250923 查询到Friday, May 16, 2025 往后进行 while True: # 倒序查 5000个 sql_bgs_id_list = sql_pool.select_all( "SELECT auth_code FROM bgs_task WHERE state!=1 AND auth_code > 17990000 LIMIT 5000") sql_bgs_id_list = [bid[0] for bid in sql_bgs_id_list] # for bid in sql_bgs_id_list: try: process_urls(log, sql_bgs_id_list, sql_pool, batch_size=1000, max_workers=10) # 根据需要调整batch_size和max_workers # get_data(bid, mysql_pool) except Exception as e: log.error('process urls: ', e) time.sleep(5) except Exception as e: log.error(e) finally: log.info("爬虫程序运行结束,等待下一轮的采集任务.....................") def schedule_task(): """ 设置定时任务 """ # 立即运行一次任务 bgs_main(logger) # 设置定时任务 schedule.every().day.at("03:01").do(bgs_main, logger) while True: schedule.run_pending() time.sleep(1) if __name__ == '__main__': schedule_task() # get_data('1000743')