# -*- coding = utf-8 -*- # @Time : 2025/2/17 下午4:49 # @File : api.py # @Software : PyCharm # @Author : xingc # @Desc : import base64 import logging import math import time import httpx import ujson import xmltodict from Crypto.Cipher import AES from Crypto.Util.Padding import unpad from fastapi import HTTPException from simple_spider_tool import md5, jsonpath from config import settings from core.base.exceptions import WechatApiError from core.db.engine import get_cache logger = logging.getLogger(__name__) class BaseClient: def __init__(self): self.session = httpx.AsyncClient() 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 ) class WechatApi(BaseClient): def __init__(self, appid, secret): self.appid = appid self.secret = secret self._access_token = None self._access_token_expire_time = -1 super().__init__() async def access_token(self): if any([ self._access_token is None, self._access_token_expire_time < math.floor(time.time()), ]): await self.get_access_token() return self._access_token async def get_access_token(self): redis_conn = await get_cache() result_str = await redis_conn.get('access_token') if result_str: access_token, expire_time = result_str.split('|') self._access_token = access_token self._access_token_expire_time = int(expire_time) else: url = 'https://api.weixin.qq.com/cgi-bin/token' params = { 'appid': self.appid, 'secret': self.secret, 'grant_type': 'client_credential' } r = await self.get(url, params=params) result = r.json() if 'errcode' not in result: expires_in = result['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 ) return self._access_token async def code_to_session(self, code: str): url = 'https://api.weixin.qq.com/sns/jscode2session' params = { 'appid': self.appid, 'secret': self.secret, 'js_code': code, 'grant_type': 'authorization_code' } 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) 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 = unpad(cipher.decrypt(encrypted_data_bytes), AES.block_size) return ujson.decode(decrypted.decode('utf-8')) class WechatPayApi(BaseClient): def __init__(self, appid, mch_id, app_key, notify_url, debug: bool = False): """ :param appid: 小程序id :param mch_id: 商户id :param app_key: 接口密钥 :param debug: """ self.appid = appid self.mch_id = mch_id self.appi_key = app_key self.notify_url = notify_url self.base_url = 'https://api.mch.weixin.qq.com' if debug: self.base_url = f'{self.base_url}/sandboxnew' super().__init__() @staticmethod def generate_nonce_str() -> str: """生成随机字符串""" return md5(f'{time.time()}') def generate_sign(self, params: dict) -> str: """生成签名""" plaintext = '&'.join([f'{k}={params[k]}' for k in sorted(params.keys())]) return md5(f'{plaintext}&key={self.appi_key}').upper() def generate_payment_params(self, prepay_id) -> dict: """生成小程序支付参数""" params = { "appId": self.appid, "timeStamp": f'{int(time.time())}', "nonceStr": self.generate_nonce_str(), "package": f"prepay_id={prepay_id}", "signType": "MD5", } params["paySign"] = self.generate_sign(params) return params async def create_unified_order( self, openid: str, body: str, out_trade_no: str, total_fee: int, create_ip: str, ): url = self.base_url + '/pay/unifiedorder' params = { 'appid': self.appid, 'mch_id': self.mch_id, 'nonce_str': self.generate_nonce_str(), 'body': body, 'out_trade_no': out_trade_no, 'total_fee': total_fee, 'spbill_create_ip': create_ip, 'notify_url': self.notify_url, 'trade_type': 'JSAPI', 'openid': openid, } params['sign'] = self.generate_sign(params) headers = { 'Content-Type': 'application/xml', } data = xmltodict.unparse({'xml': params}, full_document=False) r = await self.post(url, headers=headers, data=data) if r.status_code != 200: raise HTTPException(status_code=500, detail='微信支付接口请求失败') # 解析返回的 XML result = xmltodict.parse(r.text) return_code = jsonpath(result, '$.xml.return_code', first=True) if return_code != 'SUCCESS': return_msg = jsonpath(result, '$.xml.return_msg', first=True) raise HTTPException(status_code=500, detail=return_msg) return result['xml']['prepay_id'] wechat_pay_api = WechatPayApi( appid=settings.APPID, mch_id=settings.PAY_MCH_ID, app_key=settings.PAY_API_KEY, notify_url=settings.PAY_NOTIFY_URL, debug=settings.DEBUG )