along_django/utils/wechat_mch_service.py

"""
微信商家转账商户配置服务。
配置来源：peizhi.WechatPayMchConfig；无启用记录时回退 settings.WECHAT_PAY_V3_CONFIG。
"""
from __future__ import annotations

import logging
import os
from typing import Dict, List, Optional

from django.conf import settings

logger = logging.getLogger('utils.wechat_mch')


def _settings_fallback_cfg() -> Dict:
    raw = getattr(settings, 'WECHAT_PAY_V3_CONFIG', {}) or {}
    return {
        'MCHID': str(raw.get('MCHID') or '').strip(),
        'APPID': str(raw.get('APPID') or '').strip(),
        'API_V3_KEY': str(raw.get('API_V3_KEY') or '').strip(),
        'PRIVATE_KEY_PATH': str(raw.get('PRIVATE_KEY_PATH') or '').strip(),
        'CERT_SERIAL_NO': str(raw.get('CERT_SERIAL_NO') or '').strip(),
        'PLATFORM_PUB_KEY_PATH': str(raw.get('PLATFORM_CERT_DIR') or '').strip(),
        'NOTIFY_URL': str(raw.get('NOTIFY_URL') or '').strip(),
        'TRANSFER_SCENE_ID': str(raw.get('TRANSFER_SCENE_ID') or '1005').strip(),
        '_source': 'settings',
    }


def row_to_wx_cfg(row) -> Dict:
    return {
        'MCHID': str(row.mch_id or '').strip(),
        'APPID': str(row.appid or '').strip(),
        'API_V3_KEY': str(row.api_v3_key or '').strip(),
        'PRIVATE_KEY_PATH': str(row.private_key_path or '').strip(),
        'CERT_SERIAL_NO': str(row.cert_serial_no or '').strip(),
        'PLATFORM_PUB_KEY_PATH': str(row.platform_pub_key_path or '').strip(),
        'NOTIFY_URL': str(row.notify_url or '').strip(),
        'TRANSFER_SCENE_ID': str(row.transfer_scene_id or '1005').strip(),
        '_source': 'db',
        '_db_id': row.id,
        '_name': row.name or '',
        '_priority': row.priority,
    }


def wx_cfg_is_complete(cfg: Optional[Dict]) -> 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)


def list_enabled_wx_cfgs() -> List[Dict]:
    """启用商户按 priority 升序；库空时回退 settings 单户。"""
    from peizhi.models import WechatPayMchConfig

    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)
    if result:
        return result

    fb = _settings_fallback_cfg()
    if wx_cfg_is_complete(fb):
        logger.warning('无启用商户配置，回退 settings.WECHAT_PAY_V3_CONFIG mch=%s', fb.get('MCHID'))
        return [fb]
    return []


def get_wx_cfg_by_mch_id(mch_id: str) -> Optional[Dict]:
    mch_id = (mch_id or '').strip()
    if not mch_id:
        return None
    from peizhi.models import WechatPayMchConfig

    row = WechatPayMchConfig.objects.filter(mch_id=mch_id).first()
    if row:
        cfg = row_to_wx_cfg(row)
        return cfg if wx_cfg_is_complete(cfg) else None

    fb = _settings_fallback_cfg()
    if fb.get('MCHID') == mch_id and wx_cfg_is_complete(fb):
        return fb
    return None


def get_default_wx_cfg() -> Optional[Dict]:
    cfgs = list_enabled_wx_cfgs()
    return cfgs[0] if cfgs else None


def resolve_wx_cfg_for_bill(out_bill_no: str) -> Optional[Dict]:
    """查单/对账：必须优先用打款记录上的 mch_id。"""
    from yonghu.models import TixianAutoRecord

    rec = (
        TixianAutoRecord.objects.filter(tixian_id=out_bill_no)
        .only('mch_id')
        .first()
    )
    if rec and rec.mch_id:
        cfg = get_wx_cfg_by_mch_id(rec.mch_id)
        if cfg:
            return cfg
        logger.error('打款单 mch_id 无对应配置 bill=%s mch_id=%s', out_bill_no, rec.mch_id)
    return get_default_wx_cfg()


def cert_base_dir() -> str:
    raw = getattr(settings, 'WECHAT_MCH_CERT_BASE_DIR', '') or ''
    if raw:
        return raw
    fb = _settings_fallback_cfg().get('PRIVATE_KEY_PATH') or ''
    if fb:
        return os.path.dirname(fb)
    return os.path.join(settings.BASE_DIR, 'wechat_mch_certs')


def seed_from_settings(force: bool = False) -> Optional[Dict]:
    """将 settings.WECHAT_PAY_V3_CONFIG 写入表；已存在同 mch_id 时默认跳过。"""
    from peizhi.models import WechatPayMchConfig

    fb = _settings_fallback_cfg()
    if not wx_cfg_is_complete(fb):
        return None
    exists = WechatPayMchConfig.objects.filter(mch_id=fb['MCHID']).first()
    if exists and not force:
        return row_to_wx_cfg(exists)

    row, _ = WechatPayMchConfig.objects.update_or_create(
        mch_id=fb['MCHID'],
        defaults={
            'name': '默认商户(自settings迁移)',
            'appid': fb['APPID'],
            'api_v3_key': fb['API_V3_KEY'],
            'private_key_path': fb['PRIVATE_KEY_PATH'],
            'cert_serial_no': fb['CERT_SERIAL_NO'],
            'platform_pub_key_path': fb['PLATFORM_PUB_KEY_PATH'],
            'notify_url': fb['NOTIFY_URL'],
            'transfer_scene_id': fb['TRANSFER_SCENE_ID'] or '1005',
            'priority': 1,
            'enabled': True,
        },
    )
    return row_to_wx_cfg(row)