X25020501_card_book/core/routers/card/scraper.py

# -*- coding = utf-8 -*-
# @Time : 2025/2/14 上午10:34
# @File : scraper.py
# @Software : PyCharm
# @Author : xingc
# @Desc :
import logging
import re

from curl_cffi import requests
from fake_useragent import UserAgent
from parsel import Selector
from simple_spider_tool import jsonpath, regx_match, format_json

from config import settings
from core.base.exceptions import SearchCardNotFound, SearchCardRetry
from core.utils.helper import async_retry

logger = logging.getLogger(__name__)


class SiteApi:
    def __init__(self):
        self.user_agent = UserAgent(
            browsers=['Google', 'Chrome', 'Firefox', 'Edge', 'Safari'],
            os=['Windows', 'Mac OS X'],
            platforms=['desktop'],
            min_version=127.0
        )
        self.session = requests.AsyncSession()
        self.headers = {
            'User-Agent': "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36 Edg/132.0.0.0",
            'Accept-Encoding': "gzip, deflate, br, zstd",
            'Sec-Fetch-Site': "same-origin",
            'Sec-Fetch-Mode': "cors",
            'Sec-Fetch-Dest': "empty",
            'Accept-Language': "zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6",
        }

    async def request(
            self,
            method,
            url,
            headers=None,
            params=None,
            data=None,
            json=None,
            use_abroad_proxy: bool = False,
            **kwargs
    ):
        headers = {
            **self.headers,
            'User-Agent': self.user_agent.random,
            **(headers or {}),
        }
        # 境外代理
        if use_abroad_proxy:
            kwargs['proxies'] = {
                'http': settings.OVERSEAS_PROXY,
                'https': settings.OVERSEAS_PROXY
            }
        r = await self.session.request(
            method, url, headers=headers, params=params, data=data, json=json, **kwargs
        )
        logger.debug(f'耗时：{url} {r.elapsed}')
        return r

    async def get(self, url, headers=None, params=None, **kwargs):
        return await self.request('GET', url, headers=headers, params=params, **kwargs)

    async def post(self, url, headers=None, params=None, data=None, json=None, **kwargs):
        return await self.request(
            'POST', url, headers=headers, params=params, data=data, json=json, **kwargs
        )

    @async_retry(exceptions=(SearchCardRetry,))
    async def psa(self, cert_number: str | int):
        """PSA"""
        url = f'https://www.psacard.com/cert/{cert_number}'
        headers = {
            # 'X-Requested-With': 'XMLHttpRequest',
            'Origin': 'https://www.psacard.com',
            'Referer': f'https://www.psacard.com/cert/{cert_number}',
        }
        r = await self.get(url, headers=headers, use_abroad_proxy=True)
        chck_string = 'Just a moment...'
        if chck_string in r.text:
            raise SearchCardRetry(chck_string)
        return r

    async def psa_images(self, cert_number: str | int, images_id: str | int):
        """PSA图片"""
        url = 'https://www.psacard.com/GetPSACertImages'
        headers = {
            'X-Requested-With': 'XMLHttpRequest',
            'Origin': 'https://www.psacard.com',
            'Referer': f'https://www.psacard.com/cert/{cert_number}',
        }
        payload = {
            'certID': images_id
        }
        return await self.post(url, headers=headers, params=payload, use_abroad_proxy=True)

    async def bgs(self, cert_number: str | int):
        """BGS"""
        url = 'https://www.beckett.com/api/grading/lookup'
        params = {
            'category': 'BGS',
            'serialNumber': cert_number
        }
        r = await self.get(url, headers=self.headers, params=params, use_abroad_proxy=True)
        return r

    async def cgc(self, cert_number: str | int):
        """CGC"""
        url = f'https://www.cgccards.com/certlookup/{cert_number}/'
        headers = {
            'referer': 'https://www.cgccards.com/'
        }
        return await self.get(url, headers=headers)

    async def gbtc(self, cert_number: str | int):
        """北京公博"""
        url = 'https://wapi.gongbocoins.com/gbca/orderCoin/getWebsiteRatingInfo'
        headers = {
            'Content-Type': "application/json;charset=UTF-8",
            'origin': 'https://www.gongbocoins.com',
            'referer': "https://www.gongbocoins.com"
        }
        payload = {
            'ratingCode': cert_number
        }
        return await self.post(url, headers=headers, json=payload)

    async def bctc(self, cert_number: str | int):
        """南京保粹"""
        url = f'https://web-api.baocuicoin.com/Search/index?keyword={cert_number}&code'
        headers = {
            'origin': 'https://www.baocuicoin.com/',
            'referer': "https://www.baocuicoin.com/"
        }
        r = await self.post(url, headers=headers)
        return r

    async def ccg(self, cert_number: str | int):
        """重庆藏卡"""
        url = 'https://webapi.cangcard.com/webapi/card/num_list'
        payload = {
            'num': cert_number
        }
        headers = {
            'Content-Type': 'application/json',
            'Referer': "https://servicewechat.com/wx0b020dffc7cf7e74/75/page-frame.html",
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36 MicroMessenger/7.0.20.1781(0x6700143B) NetType/WIFI MiniProgramEnv/Windows WindowsWechat/WMPF WindowsWechat(0x63090c1f)XWEB/11581'
        }
        r = await self.post(url, headers=headers, json=payload)
        return r

    async def close(self):
        await self.session.close()


class CardScraper:
    def __init__(self):
        self.api = SiteApi()
        self._handlers = None

    @property
    def handlers(self):
        if self._handlers is None:
            self._handlers = {
                'psa': self.psa,
                'bgs': self.bgs,
                'cgc': self.cgc,
                'gbtc': self.gbtc,
                'ccg': self.ccg,
                'bctc': self.bctc,
            }
        return self._handlers

    async def psa(self, cert_number: str | int) -> dict | None:
        """PSA"""
        r = await self.api.psa(cert_number)
        html = Selector(r.text)
        if html.xpath('//div[contains(@class, "alert-danger")]/p[contains(string(), "not found")]').get() is not None:
            return

        info_elements = html.xpath('//main[@id="mainContent"]/div/table[contains(@class, "text-medium")]//tr')
        info = dict()
        for info_element in info_elements:
            name = info_element.xpath('./th[1]').xpath('string()').get()
            name = re.sub(r'\(|\)', '', name)
            name = name.lower().replace(' ', '_').replace('/', '_')
            value = info_element.xpath('./td[1]').xpath('string()').get()
            info[name.strip()] = value.strip()
        picture_elements = info_elements.xpath('.//img')
        info['pictures'] = [
            {
                'name': picture_element.xpath('./@alt').get(),
                'value': picture_element.xpath('./@src').get(),
            } for picture_element in picture_elements
        ]

        item = dict(cert_number=None)
        # 评级卡号
        item['cert_number'] = info['certification_number']
        # 年份
        item['card_year'] = info.get('year')
        # 卡片品牌
        item['card_brand'] = info.get('brand')
        # 卡片评分
        card_grade = info.get('card_grade') or info.get('grade') or ''
        item['card_score'] = regx_match(r'(\d+)', card_grade, first=True)
        # 签字评分
        auto_score = info.get('autograph_grade') or ''
        item['auto_score'] = regx_match(r'(\d+)', auto_score, first=True)
        # 卡片名称
        item['card_name'] = info.get('player')
        # 卡片编号
        item['card_number'] = info.get('card_number')
        # 卡片类型
        item['card_type'] = info.get('sport')
        images_id = html.xpath(
            '//script[contains(text(), "getPSACertImages")]/text()'
        ).re_first(r'getPSACertImages\(\'(\d+)\'\);')

        if images_id:
            r = await self.api.psa_images(cert_number, images_id)
            result = r.json()
            # 卡片正、反图片（远程）
            item['card_spare_picture'] = {
                'front_picture': jsonpath(result, '$.PSACertImages[0].OriginalImageUrl', first=True),
                'reverse_picture': jsonpath(result, '$.PSACertImages[1].OriginalImageUrl', first=True)
            }
        # 来源
        item['source'] = 'psa'
        return {
            'cert_number': item['cert_number'],
            'data': item,
            'raw_data': info,
        }

    async def bgs(self, cert_number: str | int) -> dict | None:
        r = await self.api.bgs(cert_number)
        if r.status_code != 200:
            return
        result = r.json()

        item = dict(cert_number=None)
        # 评级卡号
        item['cert_number'] = result.get('item_id')
        # 年份
        set_name = result.get('set_name') or ''
        item['card_year'] = regx_match(r'(\d{,4}).*', set_name, first=True)
        # 卡片品牌
        item['card_brand'] = result.get('sport_name')
        # 卡片评分
        item['card_score'] = result.get('final_grade')
        # 签字评分
        auto_score = result.get('autograph_grade')
        item['auto_score'] = auto_score if auto_score != '0.0' else None
        # 卡片名称
        item['card_name'] = result.get('player_name')
        # 卡片编号 无
        # 卡片类型 无
        # 卡片正、反图片（远程） 无
        # 来源
        item['source'] = 'bgs'
        return {
            'cert_number': item['cert_number'],
            'data': item,
            'raw_data': result,
        }

    async def cgc(self, cert_number: str | int):
        """CGC"""
        r = await self.api.cgc(cert_number)
        html = Selector(r.text)
        if html.xpath('//div[@class="error" and @ng-show]').get() is None:
            return

        info_elements = html.xpath('//div[@class="content-wrapper"]//dl')
        info = dict()
        for info_element in info_elements:
            name = info_element.xpath('./dt[1]').xpath('string()').get()
            name = name.lower().replace('#', '').strip().replace(' ', '_')
            value = info_element.xpath('./dd[1]').xpath('string()').get()
            info[name.strip()] = value.strip()

        item = dict(cert_number=None)
        # 评级卡号
        item['cert_number'] = info.get('cert')
        # 年份
        item['card_year'] = info.get('year')
        # 卡片品牌
        item['card_brand'] = info.get('game')
        # 卡片评分
        item['card_score'] = regx_match(r'(\d+)', info.get('grade') or '', first=True)
        # 签字评分 无
        # 卡片名称
        item['card_name'] = info.get('card_name')
        # 卡片编号
        item['card_number'] = info.get('card_number')
        # 卡片类型 无
        # 卡片正、反图片（远程）
        picture_query = '//div[@class="certlookup-images"]/div[@class="certlookup-images-item"][{index}]//img/@src'
        item['card_spare_picture'] = {
            'front_picture': html.xpath(picture_query.format(index=1)).get(),
            'reverse_picture': html.xpath(picture_query.format(index=2)).get()
        }
        # 来源
        item['source'] = 'cgc'
        return {
            'cert_number': item['cert_number'],
            'data': item,
            'raw_data': info,
        }

    async def gbtc(self, cert_number: str | int) -> dict | None:
        """北京公博"""
        r = await self.api.gbtc(cert_number)
        result = r.json()
        if result.get('errorCode') != 0:
            return

        attrs = jsonpath(result, '$.data.attr', first=True)
        attrs_string = ';'.join(attrs)

        item = dict(cert_number=None)
        # 评级卡号
        item['cert_number'] = jsonpath(result, '$.data.ratingCode', first=True)
        # 年份
        item['card_year'] = regx_match(r'年份：(\d{,4})', attrs_string, first=True)
        # 卡片品牌
        item['card_brand'] = regx_match(r'发行商：(.*?);', attrs_string, first=True)
        # 卡片评分
        item['card_score'] = jsonpath(result, '$.data.goodsScore', first=True)
        # 签字评分
        item['auto_score'] = regx_match(r'签字分数：(\d+)', attrs_string, first=True)
        # 卡片名称
        item['card_name'] = jsonpath(result, '$.data.goodsName', first=True)
        # 卡片编号
        item['card_number'] = regx_match(r'卡片编码：(.*?);', attrs_string, first=True)
        # 卡片类型 无
        # 卡片正、反图片（远程）
        item['card_spare_picture'] = {
            'front_picture': jsonpath(result, '$.data.imgList[0].self', first=True),
            'reverse_picture': jsonpath(result, '$.data.imgList[1].self', first=True)
        }
        # 来源
        item['source'] = 'gbtc'
        return {
            'cert_number': item['cert_number'],
            'data': item,
            'raw_data': result.get('data'),
        }

    async def ccg(self, cert_number: str | int) -> dict | None:
        """重庆藏卡"""
        r = await self.api.ccg(cert_number)
        result = r.json()
        if result.get('code') != '200':
            return

        item = dict(cert_number=None)
        # 评级卡号
        item['cert_number'] = jsonpath(result, '$.data.num', first=True)
        # 年份
        item['card_year'] = jsonpath(result, '$.data.year', first=True)
        # 卡片品牌
        content = jsonpath(result, '$.data.content', first=True)
        item['card_brand'] = regx_match(r'\d{4}(.*?)\.', content, first=True)
        # 卡片评分
        item['card_score'] = jsonpath(result, '$.data.zm_num', first=True)
        # 签字评分 无
        # 卡片名称
        item['card_name'] = jsonpath(result, '$.data.name', first=True)
        # 卡片类型 无
        # 卡片编号
        item['card_number'] = content
        # 卡片正、反图片（远程）
        item['card_spare_picture'] = {
            'front_picture': jsonpath(result, '$.data.img_z', first=True),
            'reverse_picture': jsonpath(result, '$.data.img_f', first=True)
        }
        # 来源
        item['source'] = 'ccg'
        return {
            'cert_number': item['cert_number'],
            'data': item,
            'raw_data': result.get('data'),
        }

    async def bctc(self, cert_number: str | int) -> dict | None:
        """南京保粹"""
        r = await self.api.bctc(cert_number)
        result = r.json()
        if result.get('code') != '200':
            return

        details = jsonpath(result, '$.info.details', first=True)

        item = dict(cert_number=None)
        # 评级卡号
        item['cert_number'] = jsonpath(details, '$.grade[?(@.title =="证书编号")].val', first=True)
        # 年份
        year_string = jsonpath(details, '$.grade[?(@.title =="年份")].val', first=True)
        item['card_year'] = year = regx_match(r'(\d{,4})', year_string, first=True)
        # 卡片品牌
        item['card_brand'] = year_string.replace(f'{year}', '').strip() if year_string else None
        # 卡片评分
        item['card_score'] = jsonpath(details, '$.grade[?(@.title =="分值")].val', first=True)
        # 签字评分
        item['auto_score'] = jsonpath(details, '$.grade[?(@.title =="签字分值")].val', first=True)
        # 卡片名称
        item['card_name'] = jsonpath(details, '$.grade[?(@.title =="名称")].val', first=True)
        # 卡片编号
        item['card_number'] = jsonpath(details, '$.grade[?(@.title =="编号")].val', first=True)
        # 卡片类型 无
        # 卡片正、反图片（远程）
        item['card_spare_picture'] = {
            'front_picture': jsonpath(details, '$.info.picurl', first=True),
            'reverse_picture': jsonpath(details, '$.info.dt_pics', first=True)
        }
        # 来源
        item['source'] = 'bctc'
        return {
            'cert_number': item['cert_number'],
            'data': item,
            'raw_data': result.get('info'),
        }

    async def choice(self, platform: str, cert_number: str | int) -> dict | None:
        handler = self.handlers.get(platform.lower())
        if handler is None:
            raise Exception('平台不支持')
        return await handler(cert_number)

    async def close(self):
        await self.api.close()


card_scraper = CardScraper()


async def main():
    import time
    started = time.time()
    print(format_json(await card_scraper.choice('psa', '1')))
    print(time.time() - started)
    # print(format_json(await card_scraper.choice('psa', '83610272')))
    # print(format_json(await card_scraper.psa('83610272')))
    # print(format_json(await card_scraper.cgc('6011456021')))
    # print(format_json(await card_scraper.bgs('16770585')))
    # print(format_json(await card_scraper.gbtc('8110071983')))
    # print(format_json(await card_scraper.bctc('E000033353')))
    # print(format_json(await card_scraper.ccg('230810139152')))


if __name__ == '__main__':
    import asyncio
    import sys

    if sys.platform == 'win32':
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    asyncio.run(main())