# -*- coding: utf-8 -*- # Author : Charley # Python : 3.10.8 # Date : 2025/8/19 15:58 import re import json import inspect import requests import user_agent from loguru import logger from parsel import Selector from mysql_pool import MySQLConnectionPool 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") crawler_language = "jp" pokemon_products = { "M-P": "ポケモンカードゲーム MEGAプロモカード", "SV-P": "ポケモンカードゲームスカーレット＆バイオレット プロモカード", "950": "ハイクラスパック 「MEGAドリームex」", "949": "拡張パック「インフェルノX」", "947": "スターターセットMEGA メガゲンガーex", "948": "スターターセットMEGA メガディアンシーex", "946": "プレミアムトレーナーボックスMEGA", "944": "拡張パック「メガブレイブ」", "945": "拡張パック「メガシンフォニア」", "942": "拡張パック「ブラックボルト」", "943": "拡張パック「ホワイトフレア」", "941": "拡張パック「ロケット団の栄光」", "940": "強化拡張パック「熱風のアリーナ」", "939": "スターターセットex ダイゴのダンバル＆メタグロスex", "938": "スターターセットex マリィのモルペコ＆オーロンゲex", "935": "拡張パック「バトルパートナーズ」", "936": "デッキビルドBOX「バトルパートナーズ」", "934": "ハイクラスパック「テラスタルフェスex」", "925": "スタートデッキGenerations ピカチュウex・カビゴンex", "926": "スタートデッキGenerations ルギアex・バンギラスex", "927": "スタートデッキGenerations カイオーガex・バシャーモex", "928": "スタートデッキGenerations ディアルガex・ルカリオex", "929": "スタートデッキGenerations レシラムex・モロバレルex", "930": "スタートデッキGenerations ゼルネアスex・オンバーンex", "931": "スタートデッキGenerations カプ・コケコex・ミミッキュex", "932": "スタートデッキGenerations ザシアンex・マホイップex", "933": "スタートデッキGenerations コライドンex・パルデアドオーex", "923": "拡張パック「超電ブレイカー」", "922": "強化拡張パック「楽園ドラゴーナ」", "921": "スターターセット テラスタイプ：ステラ ニンフィアex", "920": "スターターセット テラスタイプ：ステラ ソウブレイズex", "918": "拡張パック「ステラミラクル」", "919": "デッキビルドBOX ステラミラクル", "917": "強化拡張パック「ナイトワンダラー」", "10917": "スペシャルジャンボカードセット オーガポン", "915": "バトルマスターデッキテラスタルリザードンex", "916": "バトルマスターデッキパオジアンex", "914": "拡張パック「変幻の仮面」", "913": "強化拡張パック「クリムゾンヘイズ」", "911": "バトルアカデミー", "912": "いつでもどこでも バトルアカデミー", "906": "拡張パック「ワイルドフォース」", "907": "拡張パック「サイバージャッジ」", "908": "スターターデッキ＆ビルドセット「古代のコライドンex」", "909": "スターターデッキ＆ビルドセット「未来のミライドンex」", "905": "ハイクラスパック「シャイニートレジャーex」", "904": "スペシャルデッキセットex フシギバナ・リザードン・カメックス", "901": "拡張パック「古代の咆哮」", "902": "拡張パック「未来の一閃」", "897": "強化拡張パック「レイジングサーフ」", "898": "スターターセット テラスタル ミュウツーex", "899": "スターターセット テラスタル ラウドボーンex", "894": "拡張パック「黒炎の支配者」", "895": "デッキビルドBOX 黒炎の支配者", "896": "ポケモンワールドチャンピオンシップス2023横浜 記念デッキ「ピカチュウ」", "884": "exスタートデッキ 草 ジュナイパー", "885": "exスタートデッキ 炎 ビクティニ", "886": "exスタートデッキ 水 ゲッコウガ", "887": "exスタートデッキ 雷 ミライドン", "888": "exスタートデッキ 超 ピクシー", "889": "exスタートデッキ 闘 コライドン", "890": "exスタートデッキ 悪 ヘルガー", "891": "exスタートデッキ 鋼 メルメタル", "892": "exスタートデッキ テラスタル カイリュー", "893": "exスタートデッキ テラスタル ヨクバリス", "882": "強化拡張パック「ポケモンカード151（イチゴーイチ）」", "881": "exスペシャルセット", "879": "拡張パック「スノーハザード」", "880": "拡張パック「クレイバースト」", "878": "スターターセットex ピカチュウex&パーモット", "877": "強化拡張パック「トリプレットビート」", "870": "拡張パック「スカーレットex」", "871": "拡張パック「バイオレットex」", "872": "スターターセットex ニャオハ＆ルカリオex", "873": "スターターセットex ホゲータ＆デンリュウex", "874": "スターターセットex クワッス＆ミミッキュex", "875": "プレミアムトレーナーボックスex" } class JPPokemonCardSpider: # 类变量 BASE_URL = "https://www.pokemon-card.com" def __init__(self, log=None): self.log = log or logger self.headers = { "accept": "application/json, text/javascript, */*; q=0.01", "user-agent": user_agent.generate_user_agent() } @staticmethod 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_pokemon_card_single_page(self, pg_value, page=1): """ 获取指定分类和页码的卡片列表 """ self.log.debug(f"{inspect.currentframe().f_code.co_name} start.....................") url = f"{self.BASE_URL}/card-search/resultAPI.php" params = { "keyword": "", "se_ta": "", "regulation_sidebar_form": "all", "pg": pg_value, "illust": "", "sm_and_keyword": "true", "page": str(page) } response = requests.get(url, headers=self.headers, params=params, timeout=10) response.raise_for_status() resp_json = response.json() return resp_json def _parse_pokemon_card_list(self, card_list, pg_value, pg_label, sql_pool): """ 解析卡片列表，获取卡片信息 """ self.log.debug(f"{inspect.currentframe().f_code.co_name} start.....................") card_info_list = [] for card in card_list: card_id = card.get("cardID") card_name = card.get("cardNameViewText") card_thumb = card.get("cardThumbFile") card_thumb = f'{self.BASE_URL}{card_thumb}' data_dict = { "card_id": card_id, "card_name": card_name, "img": card_thumb, "pg_value": pg_value, "pg_label": pg_label, "crawler_language": crawler_language } card_info_list.append(data_dict) if card_info_list: sql_pool.insert_many(table="pokemon_card_record_copy1", data_list=card_info_list, ignore=True) def get_pokemon_card_list(self, pg_value='', pg_label='', sql_pool=None): """ 获取指定分类下的所有卡片列表 翻页 """ self.log.debug(f"{inspect.currentframe().f_code.co_name} start.....................") page = 1 max_page = 1 while page <= max_page: self.log.debug(f"正在获取第 {page} 页数据, pg_label: {pg_label} .........") page_data = self._get_pokemon_card_single_page(pg_value, page) if page_data.get("result") == 1: # 更新max_page（仅在第一页时需要更新） if page == 1: max_page = page_data.get("maxPage", 1) self.log.info(f"分类 {pg_label} 共有 {max_page} 页数据") cardList = page_data.get("cardList", []) if not cardList: self.log.warning(f"{inspect.currentframe().f_code.co_name} NOt found cardList !!!") break try: self._parse_pokemon_card_list(cardList, pg_value, pg_label, sql_pool) except Exception as e: self.log.error(f"{inspect.currentframe().f_code.co_name} parse_pokemon_card_list error: {e}") if len(cardList) < 39: self.log.debug(f"{inspect.currentframe().f_code.co_name} 获取的卡片数量小于39 !!! 停止翻页") break if page >= max_page: self.log.debug( f"{inspect.currentframe().f_code.co_name} -> page: {page}, max_page: {max_page}, 停止翻页") break if page >= 10: self.log.debug( f"{inspect.currentframe().f_code.co_name} -> page: {page}, page >= 10, 停止翻页.......") break page += 1 else: self.log.warning(f"获取第 {page} 页数据失败: {page_data.get('errMsg')}") break def get_details(self, item_id, sql_pool): """ 获取商品详情 """ self.log.debug(f"{inspect.currentframe().f_code.co_name} start, item_id: {item_id}.....................") url = f'{self.BASE_URL}/card-search/details.php/card/{item_id}' response = requests.get(url, headers=self.headers, timeout=10) response.raise_for_status() selector = Selector(response.text) card_no_list = selector.xpath('//div[@class="subtext Text-fjalla"]/text()').getall() card_no = ''.join(card_no_list) card_no = card_no.strip().replace('\xa0', '') if card_no else None tag_ic_rare = selector.xpath( '//div[@class="subtext Text-fjalla"]/img[not(contains(@class, "img-regulation"))]/@src').get() ic_rare_sp = tag_ic_rare.split('/')[-1].split('.')[0] if tag_ic_rare else None if ic_rare_sp and '_' in ic_rare_sp: ic_rare = ic_rare_sp.split('_')[-1] else: ic_rare = ic_rare_sp data_dict = { "card_no": card_no, "rarity": ic_rare } sql_pool.update_one_or_dict( table="pokemon_card_record_copy1", data=data_dict, condition={"card_id": item_id} ) @retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=_after_log) def run(self): """ 主函数 """ self.log.info( f'开始运行 {inspect.currentframe().f_code.co_name} 爬虫任务....................................................') # 配置 MySQL 连接池 sql_pool = MySQLConnectionPool(log=self.log) if not sql_pool.check_pool_health(): self.log.error("数据库连接池异常") raise RuntimeError("数据库连接池异常") try: self.log.debug(f"........... 开始获取已售出商品列表 按系列获取 ..........") for pg_value, pg_label in pokemon_products.items(): # print(pg_value, pg_label) try: self.get_pokemon_card_list(pg_value, pg_label, sql_pool) except Exception as e: self.log.error(f"Request get_pokemon_card_list error: {e}") # 获取商品详情 # self.log.debug(f"........... 获取商品详情 ..........") # sql_ietm_id_list = sql_pool.select_all( # f"SELECT card_id FROM pokemon_card_record WHERE card_no IS NULL AND crawler_language='{crawler_language}'") # sql_ietm_id_list = [item_id[0] for item_id in sql_ietm_id_list] # for item_id in sql_ietm_id_list: # try: # self.get_details(item_id, sql_pool) # except Exception as e: # self.log.error(f"Request get_details error: {e}") except Exception as e: self.log.error(f'{inspect.currentframe().f_code.co_name} error: {e}') finally: self.log.info(f'爬虫程序 {inspect.currentframe().f_code.co_name} 运行结束,等待下一轮的采集任务............') # def schedule_task(): # """ # 爬虫模块 定时任务 的启动文件 # """ # # 创建爬虫实例 # spider = JPPokemonCardSpider() # # # 立即运行一次任务 # # spider.run() # # # 设置定时任务 # schedule.every().day.at("01:06").do(spider.run) # # while True: # schedule.run_pending() # time.sleep(1) if __name__ == '__main__': # schedule_task() spider = JPPokemonCardSpider() spider.run()