559 lines
20 KiB
Python
559 lines
20 KiB
Python
# -*- coding = utf-8 -*-
|
|
# @Time : 2025/2/14 上午10:34
|
|
# @File : scraper.py
|
|
# @Software : PyCharm
|
|
# @Author : xingc
|
|
# @Desc :
|
|
import logging
|
|
import re
|
|
from typing import Any
|
|
|
|
import ujson
|
|
from curl_cffi import requests, CurlError
|
|
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.db.engine import get_cache
|
|
from core.base.exceptions import SearchCardRetry, SearchCardRetryFail, SearchCardHandlerFail
|
|
from core.utils.helper import async_retry, decrypt_aes_cbc, generate_keyname
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
async def get_cf_clearance(platform: str) -> dict:
|
|
redis_conn = await get_cache()
|
|
async with redis_conn:
|
|
cache_session_str = await redis_conn.get(generate_keyname(f'session:{platform}'))
|
|
if isinstance(cache_session_str, str):
|
|
headers = ujson.decode(cache_session_str)
|
|
else:
|
|
headers = {}
|
|
return headers
|
|
|
|
|
|
class SiteApi:
|
|
def __init__(self):
|
|
self.user_agent = UserAgent(
|
|
browsers=['Google', 'Chrome', 'Edge'],
|
|
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,
|
|
use_random_ua: bool = True,
|
|
retry_num: int = 3,
|
|
**kwargs
|
|
):
|
|
headers = {
|
|
**self.headers,
|
|
**(headers or {}),
|
|
}
|
|
if use_random_ua:
|
|
headers['User-Agent'] = self.user_agent.random
|
|
|
|
# 境外代理
|
|
if use_abroad_proxy:
|
|
kwargs['proxies'] = {
|
|
'http': settings.OVERSEAS_PROXY,
|
|
'https': settings.OVERSEAS_PROXY
|
|
}
|
|
|
|
for i in range(retry_num):
|
|
try:
|
|
r = await self.session.request(
|
|
method, url, headers=headers, params=params, data=data, json=json, **kwargs
|
|
)
|
|
|
|
chck_string = 'Just a moment...'
|
|
if chck_string in r.text:
|
|
raise SearchCardRetry(chck_string)
|
|
|
|
return r
|
|
|
|
except CurlError as e:
|
|
logger.error(e)
|
|
if i == retry_num - 1:
|
|
raise SearchCardRetryFail('请重试，如多次无效反馈管理员')
|
|
|
|
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}/psa'
|
|
headers = {
|
|
# 'X-Requested-With': 'XMLHttpRequest',
|
|
'Origin': 'https://www.psacard.com',
|
|
'Referer': f'https://www.psacard.com/cert/{cert_number}/psa',
|
|
}
|
|
|
|
cf_clearance_headers = await get_cf_clearance(platform='psa')
|
|
headers.update(cf_clearance_headers)
|
|
r = await self.get(url, headers=headers, use_abroad_proxy=True, use_random_ua=False, http_version=2, impersonate="chrome110")
|
|
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 = await get_cf_clearance(platform='cgc')
|
|
return await self.get(url, headers=headers, use_random_ua=False, use_abroad_proxy=True)
|
|
|
|
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 cic(self, cert_number: str | int):
|
|
"""中检检通"""
|
|
url = 'https://www.zhongjianjiantong.com/Api/OrderRatingGoods/detail'
|
|
payload = {
|
|
'cert_no': cert_number
|
|
}
|
|
headers = {
|
|
'Content-Type': 'application/json;charset=UTF-8',
|
|
'Referer': "https://www.zhongjianjiantong.com/web/index.html",
|
|
}
|
|
r = await self.post(url, headers=headers, json=payload)
|
|
return r
|
|
|
|
async def close(self):
|
|
await self.session.close()
|
|
|
|
|
|
class SiteDecrypt:
|
|
@staticmethod
|
|
def cic(ciphertext_b64: str, iv_str):
|
|
key_str = '3bd48ea5e910b195843941351be7cbae'
|
|
decrypted_text = decrypt_aes_cbc(ciphertext_b64, key_str, iv_str)
|
|
return decrypted_text
|
|
|
|
|
|
class CardScraper:
|
|
def __init__(self):
|
|
self.api = SiteApi()
|
|
self.decrypt = SiteDecrypt()
|
|
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,
|
|
'cic': self.cic,
|
|
}
|
|
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, "wrapper")]//h1/text()').get() is not None:
|
|
return
|
|
|
|
info_elements = html.xpath('//dl[@class="text-body1"]/div[contains(@class, "py-3")]')
|
|
info = dict()
|
|
for info_element in info_elements:
|
|
name = info_element.xpath('./dt[1]').xpath('string()').get() or info_element.xpath(
|
|
'./dt/template/@id').get() or ''
|
|
name = re.sub(r'[()]', '', name)
|
|
name = name.lower().replace(' ', '_').replace('/', '_')
|
|
value = info_element.xpath('./dd[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[str, Any] = dict(cert_number=None)
|
|
# 评级卡号
|
|
item['cert_number'] = info['cert_number']
|
|
# 年份
|
|
item['card_year'] = info.get('year')
|
|
# 卡片品牌
|
|
item['card_brand'] = info.get('brand_title')
|
|
# 卡片评分
|
|
card_grade = info.get('item_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('subject') or info.get('p:16')
|
|
# 卡片编号
|
|
item['card_number'] = info.get('card_number')
|
|
# 卡片类型
|
|
item['card_type'] = info.get('category')
|
|
# 卡片正、反图片（远程）
|
|
item['card_spare_picture'] = {
|
|
'front_picture': html.xpath('//div[contains(@class, "flex")]//span//img[@alt="left"]/@src').get(),
|
|
'reverse_picture': html.xpath('//div[contains(@class, "flex")]//span//img[@alt="right"]/@src').get()
|
|
}
|
|
|
|
# 来源
|
|
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[str, Any] = 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) -> dict | None:
|
|
"""CGC"""
|
|
r = await self.api.cgc(cert_number)
|
|
html = Selector(r.text)
|
|
if html.xpath('//div[@class="certlookup-search-box"]').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[str, Any] = 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[str, Any] = 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' or not result['data']:
|
|
return
|
|
|
|
item: dict[str, Any] = 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[str, Any] = 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': details,
|
|
}
|
|
|
|
async def cic(self, cert_number: str | int) -> dict | None:
|
|
"""中检检通"""
|
|
r = await self.api.cic(cert_number)
|
|
result = r.json()
|
|
if result.get('code') != 200:
|
|
return
|
|
|
|
ciphertext = result.get('data')
|
|
iv_str = result.get('iv')
|
|
try:
|
|
plaintext_str = self.decrypt.cic(ciphertext, iv_str)
|
|
|
|
plaintext = ujson.loads(plaintext_str)
|
|
except Exception as e:
|
|
logger.error(e)
|
|
raise SearchCardHandlerFail('请联系客服反馈搜卡异常')
|
|
|
|
details = jsonpath(plaintext, '$.obj_order_rating_goods', first=True)
|
|
|
|
item: dict[str, Any] = dict(cert_number=None)
|
|
# 评级卡号
|
|
item['cert_number'] = details.get('cert_no')
|
|
# 年份
|
|
item['card_year'] = jsonpath(details, '$.obj_detail.fxnf', first=True)
|
|
# 卡片品牌
|
|
item['card_brand'] = jsonpath(details, '$.obj_brand.title', first=True)
|
|
# 卡片评分
|
|
item['card_score'] = details.get('score')
|
|
# 卡片名称
|
|
item['card_name'] = details.get('goods_name')
|
|
# 卡片编号
|
|
item['card_number'] = details.get('tag_no')
|
|
# 卡片类型 无
|
|
# 卡片正、反图片（远程）
|
|
item['card_spare_picture'] = {
|
|
'front_picture': jsonpath(details, '$.lst_images[0]', first=True),
|
|
'reverse_picture': jsonpath(details, '$.lst_images[1]', first=True)
|
|
}
|
|
# 来源
|
|
item['source'] = 'cic'
|
|
return {
|
|
'cert_number': item['cert_number'],
|
|
'data': item,
|
|
'raw_data': details,
|
|
}
|
|
|
|
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())
|