diff --git a/app/routers/nomadvip.py b/app/routers/nomadvip.py index 94ece12..0471f58 100644 --- a/app/routers/nomadvip.py +++ b/app/routers/nomadvip.py @@ -333,13 +333,52 @@ async def nomadvip_check_vip(user_id: str = ""): from ..services.pb import get_pb_client pb = get_pb_client() + linked_user_id = "" if not _uid_has_vip(pb, user_id): - logging.info("[payjsapi] check_vip miss user_id=%s vip=False", user_id) - result = {"vip": False} - _cache_set(cache_key, result) - return result + # 关键补链：真实 wechat userid 可通过 vip_email_link.anonymous_user_id 关联到会员 user_id + try: + link_rec = pb.collection("vip_email_link").get_list( + 1, 1, {"filter": f'anonymous_user_id = "{user_id}"'} + ) + if link_rec.items: + linked_user_id = ( + _record_get(link_rec.items[0], "user_id", "") or "" + ).strip() + except Exception: + linked_user_id = "" + if not linked_user_id: + try: + admin_token = _get_pb_admin_token() + wx_record = ( + _pb_get_wechat_private_user(admin_token, user_id) + if admin_token + else None + ) + wx_link = _extract_wechat_vip_link(wx_record) + linked_user_id = ( + str(wx_link.get("pb_user_id", "")).strip() + or str(wx_link.get("anonymous_user_id", "")).strip() + ) + except Exception: + linked_user_id = "" + if not linked_user_id or not _uid_has_vip(pb, linked_user_id): + logging.info("[payjsapi] check_vip miss user_id=%s vip=False", user_id) + result = {"vip": False} + _cache_set(cache_key, result) + return result email = _get_linked_email_for_user(pb, user_id) + if not email and linked_user_id: + try: + link_rec = pb.collection("vip_email_link").get_list( + 1, 1, {"filter": f'user_id = "{linked_user_id}"'} + ) + if link_rec.items: + email = ( + _record_get(link_rec.items[0], "email", "") or "" + ).strip().lower() + except Exception: + pass if not email: admin_token = _get_pb_admin_token() if admin_token and user_id.startswith("user") and user_id[4:].isdigit(): @@ -355,6 +394,8 @@ async def nomadvip_check_vip(user_id: str = ""): email or "(none)", ) result = {"vip": True, "email": email} + if linked_user_id: + result["linked_user_id"] = linked_user_id _cache_set(cache_key, result) return result except Exception as error: @@ -398,6 +439,76 @@ def _pb_get_wechat_private_user(admin_token: str, wechat_userid: str) -> dict | return items[0] if items else None +def _extract_wechat_vip_link(record: dict | None) -> dict: + if not isinstance(record, dict): + return {} + raw_contact = record.get("raw_contact") + if isinstance(raw_contact, str): + try: + import json as _json + + raw_contact = _json.loads(raw_contact) + except Exception: + raw_contact = {} + if not isinstance(raw_contact, dict): + return {} + link = raw_contact.get("_vip_link") + return link if isinstance(link, dict) else {} + + +def _pb_bind_wechat_vip_link( + admin_token: str, + wechat_userid: str, + *, + email: str | None = None, + pb_user_id: str | None = None, + anonymous_user_id: str | None = None, +): + """ + 在 wechat_private_users.raw_contact 中记录 _vip_link 关系。 + 注意：不改变历史 user_id 语义，anonymous_user_id 仍是历史业务 user_id。 + """ + record = _pb_get_wechat_private_user(admin_token, wechat_userid) + if not isinstance(record, dict): + return + record_id = str(record.get("id", "")).strip() + if not record_id: + return + + raw_contact = record.get("raw_contact") + if isinstance(raw_contact, str): + try: + import json as _json + + raw_contact = _json.loads(raw_contact) + except Exception: + raw_contact = {} + if not isinstance(raw_contact, dict): + raw_contact = {} + + current = _extract_wechat_vip_link(record) + raw_contact["_vip_link"] = { + "email": (email or current.get("email") or "").strip().lower() or None, + "pb_user_id": (pb_user_id or current.get("pb_user_id") or "").strip() or None, + "anonymous_user_id": ( + anonymous_user_id or current.get("anonymous_user_id") or "" + ).strip() + or None, + "updated_at": int(time.time()), + } + + base = (PB_URL or "").rstrip("/") + _http_session.patch( + f"{base}/api/collections/wechat_private_users/records/{record_id}", + json={"raw_contact": raw_contact}, + headers={ + "Authorization": f"Bearer {admin_token}", + "Content-Type": "application/json", + }, + timeout=DEFAULT_HTTP_TIMEOUT, + ) + + def _pb_get_user_by_id(admin_token: str, user_id: str) -> dict | None: base = (PB_URL or "").rstrip("/") response = _http_session.get( @@ -493,12 +604,22 @@ async def nomadvip_wechat_profile(item: dict): ) if email and effective_member_user_id: try: + # 保持历史语义：vip_email_link.anonymous_user_id 仍仅承载历史业务 user_id + link_anonymous_id = anonymous_user_id or None _pb_upsert_email_link( admin_token, email, effective_member_user_id, - anonymous_user_id or None, + link_anonymous_id, ) + if wechat_userid: + _pb_bind_wechat_vip_link( + admin_token, + wechat_userid, + email=email, + pb_user_id=effective_member_user_id, + anonymous_user_id=link_anonymous_id, + ) except Exception: pass @@ -1124,6 +1245,7 @@ async def nomadvip_check_vip_by_email(item: dict): anonymous_user_id = ( item.get("user_id") or item.get("anonymousUserId") or "" ).strip() + wechat_userid = (item.get("wechat_userid") or item.get("userid") or "").strip() logging.info( "[payjsapi] check_vip_by_email email=%s user_id=%s", email or "(empty)", @@ -1132,7 +1254,7 @@ async def nomadvip_check_vip_by_email(item: dict): if not email or not password: return {"vip": False, "error": "missing email or password"} - cache_key = f"check_vip_by_email:{email}:{anonymous_user_id}" + cache_key = f"check_vip_by_email:{email}:{anonymous_user_id}:{wechat_userid}" cached = _cache_get(cache_key) if cached is not None: return cached @@ -1175,11 +1297,8 @@ async def nomadvip_check_vip_by_email(item: dict): pb = get_pb_client() matched_user_id = None vip_check_cache: dict[str, bool] = {} - recovered_anonymous_user_id = ( - anonymous_user_id - if anonymous_user_id.startswith("user") and anonymous_user_id[4:].isdigit() - else None - ) + # 保持历史语义：匿名ID只来自历史 user_id，不使用真实 wechat userid 覆盖 + recovered_anonymous_user_id = (anonymous_user_id or None) recovery_source = None def has_vip_cached(uid: str) -> bool: @@ -1216,9 +1335,9 @@ async def nomadvip_check_vip_by_email(item: dict): if not has_vip_cached(linked_uid): continue matched_user_id = linked_uid - if linked_anonymous_user_id.startswith("user") and linked_anonymous_user_id[4:].isdigit(): + if linked_anonymous_user_id: recovered_anonymous_user_id = linked_anonymous_user_id - elif linked_uid.startswith("user") and linked_uid[4:].isdigit(): + elif linked_uid: recovered_anonymous_user_id = linked_uid recovery_source = "vip_email_link" break @@ -1250,6 +1369,14 @@ async def nomadvip_check_vip_by_email(item: dict): ) else: _pb_upsert_email_link(admin_token, email, pb_user_id) + if wechat_userid: + _pb_bind_wechat_vip_link( + admin_token, + wechat_userid, + email=email, + pb_user_id=pb_user_id, + anonymous_user_id=recovered_anonymous_user_id, + ) logging.info( "[payjsapi] check_vip_by_email recovered by %s matched_user_id=%s effectiveUserId=%s anonymousUserId=%s",