spider_work/pokemon_tcg_spider/jp_pokemon_card_spider222.py

# -*- 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()