fix(card): 修复cic搜卡失败，增加响应data解密处理

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

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')
         # 卡片编号

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')