feat(wechat): 小程序服务端api

core/wechat/api.py

@@ -4,33 +4,66 @@
 # @Software : PyCharm
 # @Author : xingc
 # @Desc :
+import base64
+import logging
+import math
 import time
 
 import httpx
 import ujson
+from Crypto.Cipher import AES
+from Crypto.Util.Padding import unpad
+
+from config import settings
+from core.base.exceptions import WechatApiError
 
 from core.db.engine import get_cache
 
+logger = logging.getLogger(__name__)
+
 
 class WechatApi:
     def __init__(self, appid, secret):
         self.appid = appid
         self.secret = secret
         self._access_token = None
-        self._access_token_expire_time = None
+        self._access_token_expire_time = -1
         self.session = httpx.AsyncClient()
 
-    @property
     async def access_token(self):
         if any([
             self._access_token is None,
-            self._access_token_expire_time < time.time(),
+            self._access_token_expire_time < math.floor(time.time()),
         ]):
             await self.get_access_token()
         return self._access_token
 
+    async def request(self, method, url, headers=None, params=None, data=None, json=None, **kwargs):
+        r = await self.session.request(
+            method, url, headers=headers, params=params, data=data, json=json, **kwargs
+        )
+        result = r.json()
+        if result.get('errcode'):
+            logger.debug('\n'.join([
+                f'method : {method}'
+                f'url : {url}'
+                f'params : {params}'
+                f'body : {data or json}'
+                f'response: {r.text}'
+            ]))
+            raise WechatApiError('used result: {}'.format(result.get('errcode')))
+        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 def get_access_token(self):
-        async with get_cache() as redis_conn:
+        redis_conn = await get_cache()
         result_str = await redis_conn.get('access_token')
 
         if result_str:
@@ -44,11 +77,11 @@ class WechatApi:
                 'secret': self.secret,
                 'grant_type': 'client_credential'
             }
-                r = await self.session.get(url, params=params)
+            r = await self.get(url, params=params)
             result = r.json()
-                if 'errcode' not in result_str:
+            if 'errcode' not in result:
                 expires_in = result['expires_in']
-                    self._access_token_expire_time = time.time() + expires_in
+                self._access_token_expire_time = math.floor(time.time()) + expires_in
                 self._access_token = result['access_token']
                 await redis_conn.set(
                     'access_token', f'{self._access_token}|{self._access_token_expire_time}', ex=expires_in
@@ -64,5 +97,61 @@ class WechatApi:
             'js_code': code,
             'grant_type': 'authorization_code'
         }
-        r = await self.session.get(url, params=params)
+        r = await self.get(url, params=params)
         return r.json()
+
+    async def get_phone_number(self, code):
+        url = 'https://api.weixin.qq.com/wxa/business/getuserphonenumber'
+        params = {
+            'access_token': await self.access_token()
+        }
+        payload = {'code': code}
+        r = await self.post(url, params=params, json=payload)
+        return r.json()
+
+
+wechat_api = WechatApi(appid=settings.APPID, secret=settings.APP_SECRET)
+
+
+async def decrypt_sensitive_data(session_key: str, encrypted_data: str, iv: str) -> dict:
+    """解密敏感数据"""
+    session_key_bytes = base64.b64decode(session_key)
+    iv_bytes = base64.b64decode(iv)
+    encrypted_data_bytes = base64.b64decode(encrypted_data)
+    cipher = AES.new(session_key_bytes, AES.MODE_CBC, iv_bytes)
+    # decrypted = cipher.decrypt(encrypted_data_bytes)
+    # pad = decrypted[-1]
+    # decrypted = decrypted[:-pad]
+    decrypted = unpad(cipher.decrypt(encrypted_data_bytes), AES.block_size)
+    return ujson.decode(decrypted.decode('utf-8'))
+
+
+def decrypt_phone_number(encrypted_data: str, session_key: str, iv: str) -> dict:
+    try:
+        # Base64 解码
+        session_key_bytes = base64.b64decode(session_key)
+        encrypted_bytes = base64.b64decode(encrypted_data)
+        iv_bytes = base64.b64decode(iv)
+
+        # AES 解密
+        cipher = AES.new(session_key_bytes, AES.MODE_CBC, iv_bytes)
+        decrypted = unpad(cipher.decrypt(encrypted_bytes), AES.block_size)
+        decrypted_data = ujson.loads(decrypted.decode('utf-8'))
+
+        # 验证数据
+        if decrypted_data.get("watermark", {}).get("appid") != "your-appid":
+            raise ValueError("Invalid appid in decrypted data")
+        return decrypted_data
+    except Exception as e:
+        logger.info(e)
+        # raise HTTPException(status_code=400, detail=f"解密失败: {str(e)}")
+
+
+if __name__ == '__main__':
+    import asyncio
+
+    print(asyncio.run(decrypt_sensitive_data(
+        session_key='uuLloEnWUpnOWkpB4Oyopw==',
+        encrypted_data='Bzdx4BOZEJ10xf4fN2643OJaxHF/Wcj/Qv1q6M6L4MB0BBvKJRG3rlrGUYQd2nimCvDbDVPu+S3zUQidaOgyvBhDG4W0Oy8hgfeLALhqrzUo1AhTQvvxwQrntMHMRvqtdk1AzsiviaOEVGALYmuXneLD/XnGH8VzXaD2eiauzDQNCMtXVUUgB3QzIeqczr9xXSg2t6JB91m0kLs8a7FSsJAXhHIBk+58jKZjgxMUpzuJeJ/fu1Gez7q9o0iXYSVBO//KGpEVgWO0EPe4pINnwduZ3bbXF21UdRTzQ+GG2eYslXA2lespMlQWomwk5RbpVh+GWx/D29Tkprq17gkzeiiCHk9hIEHuY0+ausTbT+gOk/RG11qzYJL0nRT3uJr4GtAMKcqDV4Kmo9SThglwIg==',
+        iv='Nr7pQgCBMextSzroXPNsIw=='
+    )))