shanxia2.0/backend/app/services/wechat_notice_binding.py

"""服务号身份绑定：扫码候选 + 原登录端确认，两步均在短时会话内完成。"""
import json
import re
import secrets
import time
from urllib.parse import urlencode, urlparse

from fastapi import HTTPException
from redis.exceptions import RedisError
from sqlalchemy.exc import IntegrityError

from app.core.config import settings
from app.core.redis_client import get_sync_redis_client
from app.core.utils import generate_uuid
from app.models.wechat_notice import WechatBinding
from app.services.wechat_oa_client import client, WechatAPIError

TTL = 300
_STATE_RE = re.compile(r"^[A-Za-z0-9_-]{20,100}$")
_GETDEL = "local v=redis.call('GET',KEYS[1]); redis.call('DEL',KEYS[1]); return v"
_UNLOCK = "if redis.call('GET',KEYS[1]) == ARGV[1] then return redis.call('DEL',KEYS[1]) end return 0"


def enabled():
    base = urlparse(settings.WECHAT_OA_PUBLIC_BASE_URL)
    return bool(settings.WECHAT_OA_ENABLED and settings.WECHAT_OA_APPID and settings.WECHAT_OA_SECRET
                and base.scheme == "https" and base.netloc and not base.username
                and not base.query and not base.fragment)


def _redis():
    r = get_sync_redis_client()
    if r is None:
        raise HTTPException(503, "服务号绑定需要 Redis，请联系管理员")
    return r


def _call(method, *args, **kwargs):
    try:
        return getattr(_redis(), method)(*args, **kwargs)
    except RedisError:
        raise HTTPException(503, "绑定服务暂不可用，请稍后重试") from None


def _key(session_id):
    return "wechat:oa:binding:" + session_id


def _owner_key(user):
    return "wechat:oa:current:" + settings.WECHAT_OA_APPID + ":" + user["tenant_no"] + ":" + user["user_id"]


def _decode(value):
    return value.decode() if isinstance(value, bytes) else value


def _load(session_id):
    raw = _call("get", _key(session_id))
    if not raw:
        return None
    data = json.loads(raw)
    if data["expiresAt"] <= time.time() or data["appid"] != settings.WECHAT_OA_APPID:
        return None
    return data


def _owned(user, session_id):
    data = _load(session_id)
    if not data:
        return None
    if (data["userId"], data["tenantNo"]) != (user["user_id"], user["tenant_no"]):
        raise HTTPException(403, "不能操作其他用户的绑定会话")
    if _decode(_call("get", _owner_key(user))) != session_id:
        return None
    return data


def _save(session_id, data):
    remaining = int(data["expiresAt"] - time.time())
    if remaining <= 0:
        raise HTTPException(400, "绑定会话已过期，请重新扫码")
    _call("set", _key(session_id), json.dumps(data), ex=remaining)


def binding_status(db, user, *, refresh=True):
    if not enabled():
        return {"enabled": False, "bound": False, "subscribed": False}
    row = db.query(WechatBinding).filter_by(
        user_id=user["user_id"], tenant_no=user["tenant_no"], appid=settings.WECHAT_OA_APPID, del_flag=0,
    ).first()
    if row and row.openid and refresh:
        previous_openid, binding_id = row.openid, row.id
        try:
            subscribed = client.subscribed(previous_openid)
        except WechatAPIError:
            raise HTTPException(502, "微信关注状态校验失败，请稍后刷新") from None
        # 微信请求期间可能已重绑/解绑；旧身份的关注结果不能覆盖新绑定。
        db.query(WechatBinding).filter_by(
            id=binding_id, user_id=user["user_id"], tenant_no=user["tenant_no"],
            appid=settings.WECHAT_OA_APPID, openid=previous_openid, del_flag=0,
        ).update({"subscribed": int(subscribed)}, synchronize_session=False)
        db.commit()
        db.expire_all()
        row = db.query(WechatBinding).filter_by(
            user_id=user["user_id"], tenant_no=user["tenant_no"], appid=settings.WECHAT_OA_APPID, del_flag=0,
        ).first()
    return {"enabled": True, "bound": bool(row and row.openid), "subscribed": bool(row and row.subscribed)}


def start_binding(user, *, mode="desktop"):
    if not enabled():
        raise HTTPException(503, "服务号通知尚未配置启用")
    lock_key, lock_token = _owner_key(user) + ":lock", secrets.token_urlsafe(16)
    if not _call("set", lock_key, lock_token, ex=60, nx=True):
        raise HTTPException(409, "绑定正在处理中，请稍后再试")
    try:
        return _start_binding(user, mode=mode)
    finally:
        _call("eval", _UNLOCK, 1, lock_key, lock_token)


def _start_binding(user, *, mode="desktop"):
    session_id, state = secrets.token_urlsafe(32), secrets.token_urlsafe(32)
    data = {"userId": user["user_id"], "tenantNo": user["tenant_no"], "appid": settings.WECHAT_OA_APPID,
            "mode": mode, "status": "waiting", "expiresAt": time.time() + TTL, "subscribed": False,
            "verificationCode": f"{secrets.randbelow(1000000):06d}"}
    _save(session_id, data)
    _call("set", "wechat:oa:state:" + state, session_id, ex=TTL)
    _call("set", _owner_key(user), session_id, ex=TTL)
    # 小程序 web-view 只能打开已配置的业务域名。open.weixin.qq.com 不能作为 src，
    # 移动端先打开门户中转页，再由该页跳到微信授权；电脑扫码仍直接使用授权链接。
    authorize_url = mobile_launch_url(state) if mode == "mobile" else oauth_authorize_url(state)
    return {"sessionId": session_id, "authorizeUrl": authorize_url, "expiresIn": TTL}


def oauth_authorize_url(state):
    callback = settings.WECHAT_OA_PUBLIC_BASE_URL.rstrip("/") + settings.API_V1_PREFIX + "/public/wechat-notices/oauth/callback"
    params = {"appid": settings.WECHAT_OA_APPID, "redirect_uri": callback,
              "response_type": "code", "scope": "snsapi_base", "state": state}
    return "https://open.weixin.qq.com/connect/oauth2/authorize?" + urlencode(params) + "#wechat_redirect"


def mobile_launch_url(state):
    base = settings.WECHAT_OA_PUBLIC_BASE_URL.rstrip("/") + settings.API_V1_PREFIX
    return base + "/public/wechat-notices/oauth/launch?" + urlencode({"state": state})


def resolve_mobile_authorize_target(state):
    """校验尚未消费的 state，返回官方授权地址。不删除 state，授权回调才消费。"""
    if not _STATE_RE.fullmatch(state or ""):
        raise HTTPException(400, "授权链接无效，请返回小程序重新绑定")
    if not enabled():
        raise HTTPException(503, "服务号通知尚未配置启用")
    raw = _call("get", "wechat:oa:state:" + state)
    session_id = _decode(raw) if raw else ""
    if not session_id or not _load(session_id):
        raise HTTPException(400, "授权已过期，请返回小程序重新绑定")
    return oauth_authorize_url(state)


def oauth_callback(state, code):
    if not enabled():
        raise HTTPException(503, "服务号通知尚未配置启用")
    raw = _call("eval", _GETDEL, 1, "wechat:oa:state:" + state)
    session_id = _decode(raw)
    data = _load(session_id) if session_id else None
    if not data:
        raise HTTPException(400, "绑定二维码已过期或已使用，请重新生成")
    user = {"user_id": data["userId"], "tenant_no": data["tenantNo"]}
    if not _owned(user, session_id):
        raise HTTPException(400, "绑定二维码已失效，请使用最新二维码")
    try:
        openid = client.oauth_openid(code)
        subscribed = client.subscribed(openid)
    except WechatAPIError:
        raise HTTPException(502, "微信授权校验失败，请重新生成二维码") from None
    data.update(status="scanned", openid=openid, subscribed=subscribed)
    _save(session_id, data)
    return {"status": "scanned", "subscribed": subscribed, "verificationCode": data["verificationCode"],
            "mode": data.get("mode", "desktop")}


def binding_session(user, session_id):
    data = _owned(user, session_id)
    if not data:
        return {"status": "expired", "subscribed": False}
    result = {"status": data["status"], "subscribed": data["subscribed"]}
    if data["status"] == "scanned":
        result["verificationCode"] = data["verificationCode"]
    return result


def confirm_binding(db, user, session_id):
    if not enabled():
        raise HTTPException(503, "服务号通知尚未启用")
    data = _owned(user, session_id)
    if not data or data["status"] not in ("scanned", "confirmed"):
        raise HTTPException(400, "请先扫描有效二维码完成微信授权")
    if data["status"] == "confirmed":
        return binding_status(db, user, refresh=False)
    # 所有确认/解绑使用同用户互斥锁，防止双确认或解绑后的旧会话复活。
    lock_key = _owner_key(user) + ":lock"
    if not _call("set", lock_key, session_id, ex=60, nx=True):
        raise HTTPException(409, "绑定正在处理中，请稍后再试")
    try:
        data = _owned(user, session_id)
        if not data or data["status"] != "scanned":
            raise HTTPException(400, "绑定会话已失效")
        try:
            subscribed = client.subscribed(data["openid"])
        except WechatAPIError:
            raise HTTPException(502, "微信关注状态校验失败，请稍后重试") from None
        if not subscribed:
            raise HTTPException(400, "请先关注山峡化工服务号，再重新扫码绑定")
        existing = db.query(WechatBinding).filter_by(
            tenant_no=user["tenant_no"], appid=settings.WECHAT_OA_APPID, openid=data["openid"], del_flag=0,
        ).first()
        if existing and existing.user_id != user["user_id"]:
            raise HTTPException(409, "该微信已绑定本租户其他账号，请先解绑")
        row = db.query(WechatBinding).filter_by(
            tenant_no=user["tenant_no"], appid=settings.WECHAT_OA_APPID, user_id=user["user_id"],
        ).with_for_update().first()
        if not row:
            row = WechatBinding(id=generate_uuid(), tenant_no=user["tenant_no"], appid=settings.WECHAT_OA_APPID,
                                user_id=user["user_id"], create_by=user["user_id"], create_dept_id=user.get("region_id") or "")
            db.add(row)
        row.openid, row.subscribed, row.del_flag, row.update_by = data["openid"], 1, 0, user["user_id"]
        try:
            db.commit()
        except IntegrityError:
            db.rollback()
            raise HTTPException(409, "该微信已被绑定，请刷新后重试") from None
        data["status"] = "confirmed"
        _save(session_id, data)
        return binding_status(db, user, refresh=False)
    finally:
        _call("eval", _UNLOCK, 1, lock_key, session_id)


def unbind(db, user):
    if not enabled():
        raise HTTPException(503, "服务号通知尚未启用")
    lock_key = _owner_key(user) + ":lock"
    if not _call("set", lock_key, "unbind", ex=60, nx=True):
        raise HTTPException(409, "绑定正在处理中，请稍后再试")
    try:
        _call("delete", _owner_key(user))
        row = db.query(WechatBinding).filter_by(
            tenant_no=user["tenant_no"], appid=settings.WECHAT_OA_APPID, user_id=user["user_id"], del_flag=0,
        ).with_for_update().first()
        if row:
            row.openid, row.subscribed, row.del_flag, row.update_by = None, 0, 1, user["user_id"]
        db.commit()
        return {"enabled": True, "bound": False, "subscribed": False}
    finally:
        _call("eval", _UNLOCK, 1, lock_key, "unbind")