diff --git a/core/base/exceptions.py b/core/base/exceptions.py index 64fc03a..623fe9f 100644 --- a/core/base/exceptions.py +++ b/core/base/exceptions.py @@ -23,10 +23,17 @@ class SearchCardRetry(ScraperError): """搜索重试""" pass + class SearchCardRetryFail(ScraperError): """搜索重试失败""" pass + +class SearchCardHandlerFail(ScraperError): + """搜索处理失败""" + pass + + class WechatApiError(CardBookException): """微信开放服务调用api错误""" pass diff --git a/core/routers/card/scraper.py b/core/routers/card/scraper.py index da13fa0..8062616 100644 --- a/core/routers/card/scraper.py +++ b/core/routers/card/scraper.py @@ -6,15 +6,18 @@ # @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.base.exceptions import SearchCardNotFound, SearchCardRetry, SearchCardRetryFail +from core.base.exceptions import SearchCardRetry, SearchCardRetryFail, SearchCardHandlerFail from core.utils.helper import async_retry +from core.utils.helper import decrypt_aes_cbc logger = logging.getLogger(__name__) @@ -182,9 +185,18 @@ class SiteApi: 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 @@ -213,7 +225,7 @@ class CardScraper: 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 = re.sub(r'[()]', '', name) name = name.lower().replace(' ', '_').replace('/', '_') value = info_element.xpath('./dd[1]').xpath('string()').get() info[name.strip()] = value.strip() @@ -225,7 +237,7 @@ class CardScraper: } for picture_element in picture_elements ] - item = dict(cert_number=None) + item: dict[str, Any] = dict(cert_number=None) # 评级卡号 item['cert_number'] = info['cert_number'] # 年份 @@ -264,7 +276,7 @@ class CardScraper: return result = r.json() - item = dict(cert_number=None) + item: dict[str, Any] = dict(cert_number=None) # 评级卡号 item['cert_number'] = result.get('item_id') # 年份 @@ -290,7 +302,7 @@ class CardScraper: 'raw_data': result, } - async def cgc(self, cert_number: str | int): + async def cgc(self, cert_number: str | int) -> dict | None: """CGC""" r = await self.api.cgc(cert_number) html = Selector(r.text) @@ -305,7 +317,7 @@ class CardScraper: value = info_element.xpath('./dd[1]').xpath('string()').get() info[name.strip()] = value.strip() - item = dict(cert_number=None) + item: dict[str, Any] = dict(cert_number=None) # 评级卡号 item['cert_number'] = info.get('cert') # 年份 @@ -344,7 +356,7 @@ class CardScraper: attrs = jsonpath(result, '$.data.attr', first=True) attrs_string = ';'.join(attrs) - item = dict(cert_number=None) + item: dict[str, Any] = dict(cert_number=None) # 评级卡号 item['cert_number'] = jsonpath(result, '$.data.ratingCode', first=True) # 年份 @@ -380,7 +392,7 @@ class CardScraper: if result.get('code') != '200' or not result['data']: return - item = dict(cert_number=None) + item: dict[str, Any] = dict(cert_number=None) # 评级卡号 item['cert_number'] = jsonpath(result, '$.data.num', first=True) # 年份 @@ -418,7 +430,7 @@ class CardScraper: details = jsonpath(result, '$.info.details', first=True) - item = dict(cert_number=None) + item: dict[str, Any] = dict(cert_number=None) # 评级卡号 item['cert_number'] = jsonpath(details, '$.grade[?(@.title =="证书编号")].val', first=True) # 年份 @@ -455,9 +467,19 @@ class CardScraper: if result.get('code') != 200: return - details = jsonpath(result, '$.data.obj_order_rating_goods', first=True) + ciphertext = result.get('data') + iv_str = result.get('iv') + try: + plaintext_str = self.decrypt.cic(ciphertext, iv_str) - item = dict(cert_number=None) + 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') # 年份 @@ -465,7 +487,7 @@ class CardScraper: # 卡片品牌 item['card_brand'] = jsonpath(details, '$.obj_brand.title', first=True) # 卡片评分 - item['card_score'] = details.get('score') + item['card_score'] = details.get('score') # 卡片名称 item['card_name'] = details.get('goods_name') # 卡片编号 diff --git a/core/utils/helper.py b/core/utils/helper.py index 33b2356..1542896 100644 --- a/core/utils/helper.py +++ b/core/utils/helper.py @@ -4,10 +4,12 @@ # @Software : PyCharm # @Author : xingc # @Desc : -import asyncio +import base64 import functools import random import string +from Crypto.Cipher import AES +from Crypto.Util.Padding import unpad from datetime import datetime from typing import Callable @@ -73,3 +75,38 @@ def format_date(date: datetime | None) -> str | None: return date else: return date.strftime('%Y-%m-%d %H:%M:%S') + + +def decrypt_aes_cbc(ciphertext_b64, key_str, iv_str): + """ + 等效于 Node.js CryptoJS 的解密实现 + + 参数: + ciphertext_b64: Base64编码的密文 + key_str: 密钥字符串(UTF-8编码) + iv_str: 初始化向量字符串(UTF-8编码) + + 返回: + 解密后的明文(UTF-8字符串) + """ + # 将密钥和IV从UTF-8字符串转换为bytes + key = key_str.encode('utf-8') + iv = iv_str.encode('utf-8') + + # 确保密钥和IV长度正确 + if len(key) not in [16, 24, 32]: + raise ValueError("密钥长度必须是16(AES-128), 24(AES-192)或32(AES-256)字节") + + if len(iv) != 16: + raise ValueError("IV长度必须是16字节") + + # 解码Base64密文 + ciphertext = base64.b64decode(ciphertext_b64) + + # 创建解密器 + cipher = AES.new(key, AES.MODE_CBC, iv) + + # 解密并去除PKCS7填充 + decrypted = unpad(cipher.decrypt(ciphertext), AES.block_size) + + return decrypted.decode('utf-8')