From 537e5385283af2c25b8c4ea84e6ea99cf167ce45 Mon Sep 17 00:00:00 2001 From: XingQue Date: Tue, 14 Jul 2026 03:28:42 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E5=A4=9A=E5=95=86=E6=88=B7?= =?UTF-8?q?=E8=BD=AE=E6=8D=A2=E6=AD=BB=E9=94=81=EF=BC=8C=E5=B9=B6=E5=8A=A0?= =?UTF-8?q?=E5=9B=BA=E5=9B=9E=E8=B0=83=E8=A7=A3=E5=AF=86=E4=B8=8E=E8=AF=81?= =?UTF-8?q?=E4=B9=A6=E6=A0=A1=E9=AA=8C=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 余额不足等明确错误在查单宽限 PENDING 时会卡死不换号；现确认无单后切换，停用商户仍可解回调、请求带 notify_url。 Co-authored-by: Cursor --- utils/wechat_mch_service.py | 52 +++++++++++++++++++---- utils/wechat_v3.py | 8 ++-- yonghu/tixian_shenhe_views.py | 78 +++++++++++++++++++++++++++++++++-- 3 files changed, 121 insertions(+), 17 deletions(-) diff --git a/utils/wechat_mch_service.py b/utils/wechat_mch_service.py index 423762b..a0643d4 100644 --- a/utils/wechat_mch_service.py +++ b/utils/wechat_mch_service.py @@ -45,11 +45,29 @@ def row_to_wx_cfg(row) -> Dict: } -def wx_cfg_is_complete(cfg: Optional[Dict]) -> bool: +def wx_cfg_is_complete(cfg: Optional[Dict], *, require_key_file: bool = True) -> bool: if not cfg: return False required = ('MCHID', 'APPID', 'API_V3_KEY', 'PRIVATE_KEY_PATH', 'CERT_SERIAL_NO', 'TRANSFER_SCENE_ID') - return all(str(cfg.get(k) or '').strip() for k in required) + if not all(str(cfg.get(k) or '').strip() for k in required): + return False + if require_key_file: + pk = str(cfg.get('PRIVATE_KEY_PATH') or '').strip() + if not os.path.isfile(pk): + logger.warning('商户私钥文件不存在已跳过 mch_id=%s path=%s', cfg.get('MCHID'), pk) + return False + return True + + +def _rows_to_complete_cfgs(rows) -> List[Dict]: + result = [] + for row in rows: + cfg = row_to_wx_cfg(row) + if wx_cfg_is_complete(cfg): + result.append(cfg) + else: + logger.warning('商户配置不完整已跳过 mch_id=%s id=%s', row.mch_id, row.id) + return result def list_enabled_wx_cfgs() -> List[Dict]: @@ -59,13 +77,7 @@ def list_enabled_wx_cfgs() -> List[Dict]: rows = list( WechatPayMchConfig.objects.filter(enabled=True).order_by('priority', 'id') ) - result = [] - for row in rows: - cfg = row_to_wx_cfg(row) - if wx_cfg_is_complete(cfg): - result.append(cfg) - else: - logger.warning('商户配置不完整已跳过 mch_id=%s id=%s', row.mch_id, row.id) + result = _rows_to_complete_cfgs(rows) if result: return result @@ -76,7 +88,29 @@ def list_enabled_wx_cfgs() -> List[Dict]: return [] +def list_wx_cfgs_for_callback() -> List[Dict]: + """ + 回调验签/解密：包含已停用商户。 + 解密只需 API_V3_KEY；验签缺平台公钥时会自动跳过该户。 + """ + from peizhi.models import WechatPayMchConfig + + rows = list(WechatPayMchConfig.objects.all().order_by('priority', 'id')) + result = [] + for row in rows: + cfg = row_to_wx_cfg(row) + if str(cfg.get('MCHID') or '').strip() and str(cfg.get('API_V3_KEY') or '').strip(): + result.append(cfg) + if result: + return result + fb = _settings_fallback_cfg() + if str(fb.get('MCHID') or '').strip() and str(fb.get('API_V3_KEY') or '').strip(): + return [fb] + return [] + + def get_wx_cfg_by_mch_id(mch_id: str) -> Optional[Dict]: + """按 mch_id 取配置（含停用）；查单需可签名，故要求私钥文件存在。""" mch_id = (mch_id or '').strip() if not mch_id: return None diff --git a/utils/wechat_v3.py b/utils/wechat_v3.py index a8e1fe9..6211e6e 100644 --- a/utils/wechat_v3.py +++ b/utils/wechat_v3.py @@ -102,8 +102,8 @@ def verify_wechat_sign(headers, body, wx_cfg=None): if wx_cfg: cfgs = [wx_cfg] else: - from utils.wechat_mch_service import list_enabled_wx_cfgs - cfgs = list_enabled_wx_cfgs() + from utils.wechat_mch_service import list_wx_cfgs_for_callback + cfgs = list_wx_cfgs_for_callback() for cfg in cfgs: cert_path = _resolve_platform_pub_path(cfg, serial=serial) @@ -139,10 +139,10 @@ def decrypt_callback_with_merchants(associated_data, nonce, ciphertext): 仅对启用商户密钥做 AES 尝试（通常很少），成功后返回 (plaintext, wx_cfg)。 业务侧再用 out_bill_no 反查记录上的 mch_id，对齐一致性。 """ - from utils.wechat_mch_service import list_enabled_wx_cfgs + from utils.wechat_mch_service import list_wx_cfgs_for_callback last_err = None - for cfg in list_enabled_wx_cfgs(): + for cfg in list_wx_cfgs_for_callback(): try: text = decrypt_callback_ciphertext(associated_data, nonce, ciphertext, wx_cfg=cfg) return text, cfg diff --git a/yonghu/tixian_shenhe_views.py b/yonghu/tixian_shenhe_views.py index 94288d8..7f23bb4 100644 --- a/yonghu/tixian_shenhe_views.py +++ b/yonghu/tixian_shenhe_views.py @@ -29,6 +29,7 @@ from .tixian_shenhe_services import ( ensure_shenhe_collect_completed, get_audit_for_collect, handle_post_transfer_failure, + query_wechat_transfer_bill, reconcile_shenhe_wechat_bills, reject_audit_and_refund, release_collect_quota_for_record, @@ -37,6 +38,7 @@ from .tixian_shenhe_services import ( shijidaozhang_exceeds_wechat_limit, validate_collect_eligibility, wechat_limit_collect_message, + _sync_record_from_wx_query, ) logger = logging.getLogger('yonghu.tixian_shenhe') @@ -65,15 +67,16 @@ def _is_wx_balance_insufficient(err_code='', err_msg=''): def _is_wx_switchable_error(err_code='', err_msg=''): """ 可立刻换下一商户重试的错误（本笔尚未进入 WAIT_USER_CONFIRM）。 - 仅白名单，避免误伤导致重复开单。 + 仅白名单，避免误伤导致重复开单。不含裸词 LIMIT，防误匹配。 """ if _is_wx_balance_insufficient(err_code, err_msg): return True blob = f'{err_code} {err_msg}'.upper() keywords = ( - 'FREQUENCY_LIMITED', 'RATE_LIMIT', 'QUOTA', 'LIMIT', + 'FREQUENCY_LIMITED', 'RATE_LIMIT', 'QUOTA_EXCEEDED', 'QUOTA', '日限额', '超出限额', '超过限额', '限额', '次数超限', - 'ACCOUNT_ERROR', 'NO_AUTH', 'PERM', '商户号异常', '账户异常', + 'ACCOUNT_ERROR', 'NO_AUTH', 'PERMISSION', '商户号异常', '账户异常', + 'CERT_ERROR', 'PRIVATE_KEY', 'SIGN_ERROR', '商户证书', '私钥', ) return any(k in blob for k in keywords) @@ -120,7 +123,7 @@ def _build_wx_transfer_body(wx_cfg, tixian_id, openid, yonghuid, leixing, shijid transfer_scene_report_infos = [ {'info_type': '采购商品名称', 'info_content': '平台服务'}, ] - return { + body = { 'appid': wx_cfg['APPID'], 'out_bill_no': tixian_id, 'transfer_scene_id': transfer_scene_id, @@ -129,6 +132,10 @@ def _build_wx_transfer_body(wx_cfg, tixian_id, openid, yonghuid, leixing, shijid 'transfer_remark': f'兄弟们 收款前记得录屏发快手', 'transfer_scene_report_infos': transfer_scene_report_infos, } + notify_url = str(wx_cfg.get('NOTIFY_URL') or '').strip() + if notify_url: + body['notify_url'] = notify_url + return body def _call_wechat_transfer(body, wx_cfg): @@ -363,6 +370,28 @@ def process_audit_collect(request): 'code': 12, 'msg': '收款请求已提交，系统正在核对状态，请稍后再试，切勿重复点击', }) + except (OSError, IOError, ValueError) as e: + # 证书/私钥本地问题：未真正发到微信，可换下一商户 + logger.error( + '微信打款本地配置异常: bill=%s mch=%s err=%s', + tixian_id, mch_id, e, exc_info=True, + ) + err_code, err_msg = 'CERT_ERROR', f'商户证书/私钥异常: {e}' + last_user_msg = err_msg + last_was_balance = False + _mark_auto_record_failed(tixian_id, f'[{mch_id}] {err_msg}') + if mch_idx < len(mch_cfgs) - 1: + logger.warning( + '商户本地配置失败，切换下一商户 shenhe=%s from=%s idx=%s/%s', + shenhe_danhao, mch_id, mch_idx + 1, len(mch_cfgs), + ) + continue + if quota_reserved: + release_collect_quota_for_record( + TixianAutoRecord.objects.get(tixian_id=tixian_id), + ) + quota_reserved = False + return Response({'code': 99, 'msg': '微信商户配置异常，请联系管理员'}) if resp.status_code == 200 and wx_result.get('state') == 'WAIT_USER_CONFIRM': package_info = wx_result.get('package_info') @@ -408,6 +437,47 @@ def process_audit_collect(request): 'data': {'already_completed': True, 'tixianjilu_id': tixianjilu_id_val}, }) if post_action == RECONCILE_PENDING: + # 关键漏洞修复：微信已返回明确可切换业务错误时，查单宽限会得到 PENDING(not_found)， + # 若不处理会死卡「请稍后再试」永不换号。确认查单仍 not_found 后才换；查单不可用则仍禁换。 + can_try_switch = ( + _is_wx_switchable_error(err_code, err_msg) + and mch_idx < len(mch_cfgs) - 1 + ) + if can_try_switch: + wx_q = query_wechat_transfer_bill(tixian_id, wx_cfg=wx_cfg) + if wx_q is None: + return Response({ + 'code': 12, + 'msg': '收款请求已提交，系统正在核对状态，请稍后再试，切勿重复点击', + }) + if wx_q.get('_not_found'): + user_msg = err_msg or '微信收款发起失败，请稍后重试' + last_user_msg = user_msg + last_was_balance = _is_wx_balance_insufficient(err_code, err_msg) + _mark_auto_record_failed(tixian_id, f'[{mch_id}] {user_msg}') + logger.warning( + '商户转账失败(查单无单)，切换下一商户 shenhe=%s from=%s idx=%s/%s', + shenhe_danhao, mch_id, mch_idx + 1, len(mch_cfgs), + ) + continue + # 微信侧其实已有单：按查单结果走，禁止盲换 + ar = TixianAutoRecord.objects.get(tixian_id=tixian_id) + sync_action, sync_payload = _sync_record_from_wx_query(ar, wx_q) + if sync_action == RECONCILE_WAIT_CONFIRM: + if isinstance(sync_payload, dict): + sync_payload.setdefault('mch_id', mch_id) + sync_payload.setdefault('tixianjilu_id', tixianjilu_id_val) + quota_resp = _verify_collect_quota_or_response(user_main, audit, shenhe_danhao) + if quota_resp: + return quota_resp + return _build_collect_success_response(sync_payload) + if sync_action == RECONCILE_COMPLETED: + ensure_shenhe_collect_completed(shenhe_danhao) + return Response({ + 'code': 0, + 'msg': '提现已成功到账', + 'data': {'already_completed': True, 'tixianjilu_id': tixianjilu_id_val}, + }) pending_msg = ( post_payload.get('msg', post_payload) if isinstance(post_payload, dict) else post_payload