"""跨俱乐部资产迁移核心服务（独立模块，未开计划时零影响）。 P0：所有成功过户必须写 Ledger + Item；金额以锁库实读为准；令牌单次消费；占用表防双迁。 """ from __future__ import annotations import hashlib import logging import secrets import uuid from datetime import datetime, timedelta from decimal import Decimal, ROUND_HALF_UP from typing import Dict, List, Optional, Tuple from django.db import transaction from django.db.models import F, Q from django.utils import timezone from jituan.models import ( ClubMigrateConsumption, ClubMigrateLedger, ClubMigrateLedgerItem, ClubMigratePlan, ClubMigrateToken, ClubMigrateUserQuota, ) from jituan.services.club_migrate_assets import ( ASSET_CATALOG, ROLE_PROFILE_ATTR, normalize_whitelist, roles_from_whitelist, ) logger = logging.getLogger(__name__) TOKEN_TTL_HOURS = 24 PASSWORD_MAX_FAIL = 5 SCENE_PREFIX = 'MIG' # 未完结订单：待抢/进行中/退款审/指定中/结算中/未支付 OPEN_ORDER_STATUSES = (1, 2, 4, 7, 8, 9) # 未缴清罚单：待缴纳/申诉中/平台审核中 UNPAID_PENALTY_STATUSES = (1, 3, 5) # 目标店新建打手时绑定的默认管事邀请码（与小程序 PLATFORM 默认码一致） CLUB_MIGRATE_DEFAULT_GUANSHI_INVITE = { 'lxs': '0000008VfJKPu4Spmcjt', 'xq': '0000008RffVgKHMj7kQC', } # 可提现类字段：迁入走目标店冻结网关（押金/积分/已冻结池除外） _FREEZE_CREDIT_FIELDS = frozenset({ ('dashou', 'yue'), ('guanshi', 'yue'), ('zuzhang', 'ketixian_jine'), ('shenheguan', 'yue'), }) MEMBER_JIFEN_CAP = 10 def _money(v) -> Decimal: return Decimal(str(v or 0)).quantize(Decimal('0.01'), rounding=ROUND_HALF_UP) def _now(): return timezone.now() def _hash_password(password: str, salt: str) -> str: raw = f'{salt}:{password}'.encode('utf-8') return hashlib.sha256(raw).hexdigest() def new_plan_id() -> str: return 'mp' + uuid.uuid4().hex[:14] def new_token() -> str: # 短令牌便于塞进微信 scene（总长 32） return secrets.token_urlsafe(12).replace('-', '').replace('_', '')[:20] def scene_for_token(token: str) -> str: return f'{SCENE_PREFIX}{token}' def parse_migrate_scene(scene: str) -> str: s = str(scene or '').strip() if s.upper().startswith(SCENE_PREFIX): return s[len(SCENE_PREFIX):] return '' def plan_in_window(plan: ClubMigratePlan, now=None) -> bool: now = now or _now() if plan.start_at and now < plan.start_at: return False if plan.end_at and now > plan.end_at: return False return True def find_active_plan_for_source(from_club_id: str) -> Optional[ClubMigratePlan]: cid = (from_club_id or '').strip() if not cid: return None qs = ClubMigratePlan.query.filter(from_club_id=cid, enabled=True).order_by('-id') now = _now() for p in qs[:20]: if plan_in_window(p, now): return p return None def find_active_plan_for_target(to_club_id: str) -> Optional[ClubMigratePlan]: cid = (to_club_id or '').strip() if not cid: return None qs = ClubMigratePlan.query.filter(to_club_id=cid, enabled=True).order_by('-id') now = _now() for p in qs[:20]: if plan_in_window(p, now): return p return None def max_allowed_success(plan: ClubMigratePlan, from_uid: str) -> int: base = 1 if plan.default_once_per_user else 999 q = ClubMigrateUserQuota.query.filter(plan_id=plan.plan_id, from_uid=str(from_uid)).first() extra = int(q.max_extra_times or 0) if q else 0 return base + extra def success_count(plan_id: str, from_uid: str) -> int: row = ClubMigrateConsumption.query.filter(plan_id=plan_id, from_uid=str(from_uid)).first() return int(row.success_count or 0) if row else 0 def can_migrate_again(plan: ClubMigratePlan, from_uid: str) -> bool: """有剩余次数，或上次未迁干净（源店仍有可迁资产）时可再迁。""" if success_count(plan.plan_id, from_uid) < max_allowed_success(plan, from_uid): return True # 次数已用尽但仍有资产（如会员曾被 skip 留下）→ 允许补迁，不依赖二次名额 return user_has_migratable_assets(plan, from_uid) def _load_user(uid: str): from users.business_models import User return User.query.filter(UserUID=str(uid)).first() def _resolve_default_guanshi_uid(club_id: str) -> Optional[str]: """按俱乐部硬编码邀请码解析默认管事 UID；解析失败返回 None。""" code = CLUB_MIGRATE_DEFAULT_GUANSHI_INVITE.get((club_id or '').strip()) if not code: return None from users.models import UserGuanshi gs = UserGuanshi.query.filter(yaoqingma=code).select_related('user').first() if not gs or not getattr(gs, 'user', None): logger.warning('migrate default guanshi invite not found club=%s code=%s', club_id, code) return None return str(gs.user.UserUID) def _get_or_create_profile(user, role: str, *, to_club_id: str = ''): """目标店无角色时创建空扩展行（仅迁移模块调用）。 新建打手：绑定目标店默认管事（源店邀请人在新店无效）。 已有打手资料：不改 yaoqingren（曾在新店注册过的情况保留原绑定）。 """ from users.models import UserDashou, UserShangjia, UserGuanshi, UserZuzhang, UserShenheguan mapping = { 'dashou': (UserDashou, 'DashouProfile'), 'shangjia': (UserShangjia, 'ShopProfile'), 'guanshi': (UserGuanshi, 'GuanshiProfile'), 'zuzhang': (UserZuzhang, 'ZuzhangProfile'), 'shenheguan': (UserShenheguan, 'ShenheguanProfile'), } model, attr = mapping[role] try: return getattr(user, attr) except Exception: pass defaults = {'user': user} if role == 'shangjia': defaults['nicheng'] = f'SJ{user.UserUID}' elif role == 'dashou': defaults['nicheng'] = f'DS{user.UserUID}' gs_uid = _resolve_default_guanshi_uid(to_club_id) if gs_uid: defaults['yaoqingren'] = gs_uid obj = model.query.create(**defaults) return obj def _profile_of(user, role: str): attr = ROLE_PROFILE_ATTR.get(role) if not attr: return None try: return getattr(user, attr) except Exception: return None def _lock_profile(user, role: str): """事务内锁扩展行；无行返回 None。""" from users.models import UserDashou, UserShangjia, UserGuanshi, UserZuzhang, UserShenheguan mapping = { 'dashou': UserDashou, 'shangjia': UserShangjia, 'guanshi': UserGuanshi, 'zuzhang': UserZuzhang, 'shenheguan': UserShenheguan, } model = mapping.get(role) if not model: return None return model.objects.select_for_update().filter(user=user).first() def _read_field(profile, field: str, kind: str): if profile is None: return Decimal('0.00') if kind != 'int' else 0 val = getattr(profile, field, None) if kind == 'int': try: return int(val or 0) except (TypeError, ValueError): return 0 return _money(val) def preview_assets(plan: ClubMigratePlan, from_uid: str) -> List[dict]: keys = normalize_whitelist(plan.asset_whitelist) user = _load_user(from_uid) if not user: return [] items = [] for key in keys: meta = ASSET_CATALOG[key] role, field, kind = meta['role'], meta['field'], meta['kind'] if kind == 'huiyuan': items.extend(_preview_huiyuan(from_uid, plan.from_club_id, role)) continue profile = _profile_of(user, role) amount = _read_field(profile, field, kind) if kind == 'money': display = f'¥{amount}' elif kind == 'int': display = str(int(amount) if amount is not None else 0) else: display = str(amount) items.append({ 'key': key, 'role': role, 'field': field, 'label': meta['label'], 'kind': kind, 'amount': str(amount), 'display': display, 'has_profile': profile is not None, }) return items def asset_item_has_value(item: dict) -> bool: """预览项是否有可迁金额/积分/会员。""" kind = (item or {}).get('kind') or '' if kind == 'huiyuan': return True try: if kind == 'int': return int(item.get('amount') or 0) > 0 return _money(item.get('amount')) > 0 except (TypeError, ValueError): return False def user_has_migratable_assets(plan: ClubMigratePlan, from_uid: str) -> bool: """白名单内是否有任意可迁资产；全无则不应强制进迁移页。""" try: items = preview_assets(plan, from_uid) except Exception as e: logger.warning('migrate asset check failed uid=%s: %s', from_uid, e) return True # 检查失败时保守：仍允许进页，避免误放行有资产用户 return any(asset_item_has_value(it) for it in items) def _membership_core_key(huiyuan_id: str) -> str: from jituan.services.huiyuan_bundle import normalize_huiyuan_id hid = normalize_huiyuan_id(huiyuan_id) if not hid: return '' return hid.lstrip('0') or '0' def _club_row_belongs_to_source(rec_cid: str, from_club: str) -> bool: """会员/充值单 club_id 是否归属源店（历史默认 xq/空也算）。""" from jituan.constants import CLUB_ID_DEFAULT from_club = (from_club or '').strip() rec_cid = (rec_cid or '').strip() if not rec_cid or rec_cid in (from_club, CLUB_ID_DEFAULT, ''): return True return bool(from_club) and rec_cid == from_club def _collect_migratable_huiyuan(uid: str, from_club: str, *, for_update: bool = False) -> List[dict]: """ 可迁会员 = 用户端能看到的有效会员（开通表 + 未撤销充值单合成）。 第一次迁移若只弄坏了开通表、充值单还在，第二次仍能迁。 """ from jituan.constants import CLUB_ID_DEFAULT from jituan.services.huiyuan_bundle import normalize_huiyuan_id from jituan.services.member_recharge import ( MEMBER_LEIXING, MEMBER_PAID_STATUS, czjilu_entitlement_active, _czjilu_entitlement_end, _order_is_trial_like, ) from products.models import Czjilu, Huiyuan, Huiyuangoumai now = _now_aware() from_club = (from_club or '').strip() merged = {} # core_key -> entry def upsert(*, huiyuan_id, daoqi, jieshao='', is_trial=False, has_used_trial=False, goumai_row=None, source='goumai'): hid = normalize_huiyuan_id(huiyuan_id) key = _membership_core_key(hid) exp = _aware(daoqi) if not key or not exp or exp <= now: return cur = merged.get(key) if cur and _aware(cur['daoqi_time']) and _aware(cur['daoqi_time']) >= exp: # 已有更晚到期；仍补挂开通行便于失效 if goumai_row is not None and cur.get('goumai_row') is None: cur['goumai_row'] = goumai_row return merged[key] = { 'huiyuan_id': hid, 'daoqi_time': exp, 'jieshao': (jieshao or '').strip(), 'is_trial': bool(is_trial), 'has_used_trial': bool(has_used_trial), 'goumai_row': goumai_row, 'source': source, 'record_club_id': (getattr(goumai_row, 'club_id', None) or '') if goumai_row else '', } qs = Huiyuangoumai.objects.filter(yonghu_id=str(uid)) if for_update: qs = qs.select_for_update() for r in qs: if not _club_row_belongs_to_source(getattr(r, 'club_id', None) or '', from_club): continue # 开通表有效才直接进；失效的留给充值单补 if int(r.huiyuan_zhuangtai or 0) != 1: continue upsert( huiyuan_id=r.huiyuan_id, daoqi=r.daoqi_time, jieshao=r.jieshao or '', is_trial=bool(r.is_trial), has_used_trial=bool(r.has_used_trial), goumai_row=r, source='goumai', ) for order in Czjilu.query.filter( yonghuid=str(uid), leixing=MEMBER_LEIXING, zhuangtai=MEMBER_PAID_STATUS, ).order_by('-CreateTime')[:50]: hid = normalize_huiyuan_id(order.huiyuan_id) if not hid: continue order_cid = (getattr(order, 'club_id', None) or '').strip() if not _club_row_belongs_to_source(order_cid, from_club): continue cid = order_cid or from_club or CLUB_ID_DEFAULT if not czjilu_entitlement_active(order, cid): continue end = _czjilu_entitlement_end(order, cid) name = '' hy = Huiyuan.query.filter(huiyuan_id=hid).first() if hy: name = (hy.jieshao or '').strip() # 若开通表已有同行但状态失效，挂上便于迁完统一置到期 goumai = None key = _membership_core_key(hid) if key not in merged or merged[key].get('goumai_row') is None: goumai = _find_user_huiyuangoumai(uid, hid, for_update=for_update) if goumai and not _club_row_belongs_to_source(getattr(goumai, 'club_id', None) or '', from_club): goumai = None upsert( huiyuan_id=hid, daoqi=end, jieshao=name, is_trial=bool(_order_is_trial_like(order, cid)), has_used_trial=False, goumai_row=goumai, source='czjilu', ) return list(merged.values()) def _list_active_huiyuan_rows(uid: str, from_club: str, *, for_update: bool = False): """兼容旧调用：仅返回开通表有效行。迁移请用 _collect_migratable_huiyuan。""" from products.models import Huiyuangoumai now = _now_aware() from_club = (from_club or '').strip() qs = Huiyuangoumai.objects.filter( yonghu_id=str(uid), huiyuan_zhuangtai=1, ) if for_update: qs = qs.select_for_update() out = [] for r in qs: exp = _aware(r.daoqi_time) if not exp or exp <= now: continue if not _club_row_belongs_to_source(getattr(r, 'club_id', None) or '', from_club): continue out.append(r) return out def _preview_huiyuan(uid: str, club_id: str, role: str) -> List[dict]: from products.models import Huiyuan entries = _collect_migratable_huiyuan(uid, club_id) if not entries: return [] ids = [e['huiyuan_id'] for e in entries] name_map = { h.huiyuan_id: (h.jieshao or '').strip() or '会员' for h in Huiyuan.query.filter(huiyuan_id__in=ids) } out = [] for e in entries: name = e.get('jieshao') or name_map.get(e['huiyuan_id']) or '会员' exp = _aware(e.get('daoqi_time')) if exp is None: exp_text = '' elif timezone.is_aware(exp): exp_text = timezone.localtime(exp).strftime('%Y-%m-%d %H:%M') else: exp_text = exp.strftime('%Y-%m-%d %H:%M') src_tag = '开通' if e.get('source') == 'goumai' else '充值权益' out.append({ 'key': f'{role}.huiyuan:{e["huiyuan_id"]}', 'role': role, 'field': 'huiyuan', 'label': name, 'kind': 'huiyuan', 'amount': exp_text, 'display': f'{name} · 到期 {exp_text}' if exp_text else name, 'huiyuan_id': e['huiyuan_id'], 'daoqi_time': exp.isoformat() if exp else '', 'jieshao': name, 'has_profile': True, 'record_club_id': e.get('record_club_id') or '', 'source': e.get('source') or '', 'source_tag': src_tag, }) return out def _resolve_migrate_target_huiyuan_ids( *, src_huiyuan_id: str, from_club: str = '', to_club: str = '', ) -> List[str]: """ 普通会员：原样迁同 ID。 总会员：按子会员格式迁出（写入各子会员开通记录），不在目标店再挂一条总会员。 未配置子项时才退回迁总会员本身，避免丢权益。 """ from jituan.services.huiyuan_bundle import ( is_bundle_huiyuan, normalize_huiyuan_id, resolve_bundle_included_ids, ) src_id = normalize_huiyuan_id(src_huiyuan_id) if not src_id: return [] if not is_bundle_huiyuan(src_id): return [src_id] kids = [] seen = set() for cid in (from_club, to_club): for hid in resolve_bundle_included_ids( src_id, club_id=(cid or '').strip() or None, record_club_id=(from_club or '').strip() or None, ) or []: kid = normalize_huiyuan_id(hid) if not kid: continue core = kid.lstrip('0') or '0' if core in seen: continue seen.add(core) kids.append(kid) if kids: break if kids: return kids return [src_id] def _revoke_source_member_czjilu(from_uid: str, huiyuan_id: str): """撤销源店充值单权益，避免 Huiyuangoumai 失效后仍被 clumber 从 Czjilu 合成展示。""" from jituan.services.huiyuan_bundle import huiyuan_ids_match from jituan.services.member_recharge import ( ENTITLEMENT_REVOKED_MARKER, MEMBER_LEIXING, MEMBER_PAID_STATUS, _czjilu_entitlement_revoked, _truncate_shuoming, ) from products.models import Czjilu for order in Czjilu.query.filter( yonghuid=str(from_uid), leixing=MEMBER_LEIXING, zhuangtai=MEMBER_PAID_STATUS, ): if not huiyuan_ids_match(order.huiyuan_id, huiyuan_id): continue if _czjilu_entitlement_revoked(order): continue note = (order.shuoming or '').strip() order.shuoming = _truncate_shuoming( f'{note} |跨店迁移|{ENTITLEMENT_REVOKED_MARKER}|' if note else f'跨店迁移|{ENTITLEMENT_REVOKED_MARKER}' ) order.save(update_fields=['shuoming']) def _expire_source_huiyuan(row, now): """源会员置失效。""" row.huiyuan_zhuangtai = 0 row.daoqi_time = now row.save(update_fields=['huiyuan_zhuangtai', 'daoqi_time', 'UpdateTime']) try: _revoke_source_member_czjilu(row.yonghu_id, row.huiyuan_id) except Exception as e: logger.warning( 'migrate revoke czjilu fail uid=%s hy=%s: %s', row.yonghu_id, row.huiyuan_id, e, ) def _expire_source_membership(*, from_uid: str, huiyuan_id: str, goumai_row=None): """迁完后清掉源店开通记录 + 充值单合成权益。""" now = _now_aware() row = goumai_row if row is None: row = _find_user_huiyuangoumai(from_uid, huiyuan_id, for_update=True) if row is not None: row.huiyuan_zhuangtai = 0 row.daoqi_time = now row.save(update_fields=['huiyuan_zhuangtai', 'daoqi_time', 'UpdateTime']) try: _revoke_source_member_czjilu(from_uid, huiyuan_id) except Exception as e: logger.warning( 'migrate revoke czjilu fail uid=%s hy=%s: %s', from_uid, huiyuan_id, e, ) def _find_user_huiyuangoumai(uid: str, huiyuan_id: str, *, for_update: bool = False): """按会员 ID 变体查找购买记录（001 / 1 等同）。""" from jituan.services.huiyuan_bundle import expand_huiyuan_id_variants from products.models import Huiyuangoumai variants = expand_huiyuan_id_variants(huiyuan_id) or [str(huiyuan_id)] qs = Huiyuangoumai.objects.filter( yonghu_id=str(uid), huiyuan_id__in=variants, ) if for_update: qs = qs.select_for_update() return qs.first() def _latest_effective_cover_until(uid: str, huiyuan_id: str, club_id: str, *, goumai_row=None): """ 目标店对该会员类型的有效覆盖到期： - 自己开通的该子会员/总会员未到期 - 或任意未到期总会员包含该子会员 用于迁移累加：总会员覆盖剩 10 天 + 迁入同子会员 10 天 → 子会员落到 20 天。 """ from jituan.services.huiyuan_bundle import ( _included_huiyuan_id_set, _membership_record_active, expand_huiyuan_id_variants, huiyuan_ids_match, is_bundle_huiyuan, normalize_huiyuan_id, resolve_bundle_included_ids, ) from products.models import Huiyuangoumai now = _now_aware() hid = normalize_huiyuan_id(huiyuan_id) if not hid: return None req_set = {hid, hid.lstrip('0') or '0'} for vid in expand_huiyuan_id_variants(hid): req_set.add(vid) req_set.add(vid.lstrip('0') or '0') latest = None def bump(dt): nonlocal latest exp = _aware(dt) if not exp or exp <= now: return if latest is None or exp > latest: latest = exp row = goumai_row if row is None: row = _find_user_huiyuangoumai(uid, hid) if row is not None and int(row.huiyuan_zhuangtai or 0) == 1: bump(row.daoqi_time) for rec in Huiyuangoumai.query.filter(yonghu_id=str(uid)): if not _membership_record_active(rec): continue if huiyuan_ids_match(rec.huiyuan_id, hid): bump(rec.daoqi_time) continue if not is_bundle_huiyuan(rec.huiyuan_id): continue included = resolve_bundle_included_ids( rec.huiyuan_id, club_id=club_id, yonghu_id=uid, record_club_id=getattr(rec, 'club_id', None), ) if req_set & _included_huiyuan_id_set(included): bump(rec.daoqi_time) return latest def _add_huiyuan_duration(*, to_uid: str, to_club: str, huiyuan_id: str, remain, jieshao: str, is_trial: bool, has_used_trial: bool): """给目标用户对应会员加时长。 累加基数 = max(现在, 该会员自身到期, 覆盖它的总会员到期)。 避免：目标已有总会员覆盖子会员 10 天，迁入同子会员 10 天却只得到约 10 天（应为 20）。 """ from products.models import Huiyuangoumai now = _now_aware() remain = remain if remain and remain.total_seconds() > 0 else None if remain is None: return None, {} hid = str(huiyuan_id).strip() dst = _find_user_huiyuangoumai(to_uid, hid, for_update=True) before_dst = {} if dst: before_dst = { 'daoqi_time': dst.daoqi_time.isoformat() if dst.daoqi_time else '', 'zhuangtai': dst.huiyuan_zhuangtai, 'club_id': dst.club_id, } cover = _latest_effective_cover_until(to_uid, hid, to_club, goumai_row=dst) base = cover if (cover and cover > now) else now exp = base + remain if dst: dst.daoqi_time = exp dst.huiyuan_zhuangtai = 1 dst.club_id = to_club if jieshao: dst.jieshao = jieshao dst.save(update_fields=['daoqi_time', 'huiyuan_zhuangtai', 'club_id', 'jieshao', 'UpdateTime']) return exp, before_dst try: Huiyuangoumai.query.create( huiyuan_id=hid, yonghu_id=str(to_uid), huiyuan_zhuangtai=1, jieshao=jieshao or '', daoqi_time=exp, club_id=to_club, is_trial=bool(is_trial), has_used_trial=bool(has_used_trial), ) except Exception: # 并发/ID 变体撞唯一约束：再查一次改成累加 dst2 = _find_user_huiyuangoumai(to_uid, hid, for_update=True) if not dst2: raise before_dst = { 'daoqi_time': dst2.daoqi_time.isoformat() if dst2.daoqi_time else '', 'zhuangtai': dst2.huiyuan_zhuangtai, 'club_id': dst2.club_id, } cover2 = _latest_effective_cover_until(to_uid, hid, to_club, goumai_row=dst2) base2 = cover2 if (cover2 and cover2 > now) else now exp = base2 + remain dst2.daoqi_time = exp dst2.huiyuan_zhuangtai = 1 dst2.club_id = to_club if jieshao: dst2.jieshao = jieshao dst2.save(update_fields=['daoqi_time', 'huiyuan_zhuangtai', 'club_id', 'jieshao', 'UpdateTime']) return exp, before_dst return exp, before_dst def _transfer_huiyuan(*, from_uid, to_uid, from_club, to_club, role) -> List[dict]: """ 未到期会员：目标店加同等剩余时长；源店置已到期并撤销充值权益。 总会员按子会员拆分写入（不用总会员 ID 在目标店再挂一条）。 """ from products.models import Huiyuan from jituan.services.huiyuan_bundle import is_bundle_huiyuan, normalize_huiyuan_id now = _now_aware() entries = _collect_migratable_huiyuan(from_uid, from_club, for_update=True) out = [] # 同一迁移内按子会员去重，避免源店「总会员+子会员」重复加时长 transferred_cores = set() for e in entries: src_exp = _aware(e.get('daoqi_time')) name = (e.get('jieshao') or '').strip() if not name: hy = Huiyuan.query.filter(huiyuan_id=e['huiyuan_id']).first() name = (hy.jieshao if hy else '') or '会员' before = { 'huiyuan_id': e['huiyuan_id'], 'daoqi_time': src_exp.isoformat() if src_exp else '', 'zhuangtai': 1, 'is_trial': bool(e.get('is_trial')), 'jieshao': name, 'record_club_id': e.get('record_club_id') or '', 'source': e.get('source') or '', } remain = (src_exp - now) if src_exp and src_exp > now else None if remain is None: continue target_ids = _resolve_migrate_target_huiyuan_ids( src_huiyuan_id=e['huiyuan_id'], from_club=from_club, to_club=to_club, ) if not target_ids: raise ValueError(f'会员 ID 无效，无法迁移: {e["huiyuan_id"]}') formal_floor = 0 if not bool(e.get('is_trial')): from jituan.services.member_recharge import count_formal_paid_member_orders src_paid = count_formal_paid_member_orders( from_uid, e['huiyuan_id'], from_club, ) formal_floor = max(int(src_paid or 0), 1) wrote_any = False src_is_bundle = is_bundle_huiyuan(e['huiyuan_id']) for tid in target_ids: tid = normalize_huiyuan_id(tid) or tid core = (tid.lstrip('0') or '0') if tid else '' if core and core in transferred_cores: continue child_name = name if src_is_bundle or tid != normalize_huiyuan_id(e['huiyuan_id']): chy = Huiyuan.query.filter(huiyuan_id=tid).first() if not chy: from jituan.services.huiyuan_bundle import expand_huiyuan_id_variants for vid in expand_huiyuan_id_variants(tid): chy = Huiyuan.query.filter(huiyuan_id=vid).first() if chy: break child_name = ((chy.jieshao if chy else '') or '').strip() or tid exp, before_dst = _add_huiyuan_duration( to_uid=to_uid, to_club=to_club, huiyuan_id=tid, remain=remain, jieshao=child_name, is_trial=bool(e.get('is_trial')), has_used_trial=bool(e.get('has_used_trial')), ) if exp is None: raise ValueError(f'会员过户失败（加时长）: {child_name}({tid})') if core: transferred_cores.add(core) wrote_any = True out.append({ 'item_type': ClubMigrateLedgerItem.ITEM_HUIYUAN, 'role': role, 'field': 'huiyuan', 'amount': Decimal('1'), 'before_src': before.get('daoqi_time', ''), 'after_src': 'expired', 'before_dst': before_dst.get('daoqi_time', ''), 'after_dst': exp.isoformat() if exp else '', 'huiyuan_id': tid, 'meta_json': { 'before_src': before, 'before_dst': before_dst, 'src_huiyuan_id': e['huiyuan_id'], 'name': child_name, 'mapped_from_bundle': bool(src_is_bundle and tid != normalize_huiyuan_id(e['huiyuan_id'])), 'formal_purchase_floor': formal_floor, 'is_trial': bool(e.get('is_trial')), 'source': e.get('source') or '', }, }) if not wrote_any: raise ValueError(f'会员过户失败: {name}({e["huiyuan_id"]})') _expire_source_membership( from_uid=from_uid, huiyuan_id=e['huiyuan_id'], goumai_row=e.get('goumai_row'), ) return out def _aware(dt): """与 timezone.now() 对齐 aware/naive，避免混比炸 500。""" if dt is None: return None now_sample = timezone.now() if timezone.is_aware(now_sample): if timezone.is_naive(dt): return timezone.make_aware(dt, timezone.get_current_timezone()) return dt # 项目未开 USE_TZ：一律 naive if timezone.is_aware(dt): return timezone.make_naive(dt, timezone.get_current_timezone()) return dt def _now_aware(): """业务比较用的当前时间（与 _aware 同一套时区规则）。""" return _aware(_now()) or _now() def check_conditions(plan: ClubMigratePlan, from_uid: str) -> Tuple[bool, str]: """返回 (ok, msg)。""" cond = plan.condition_json or {} user = _load_user(from_uid) if not user: return False, '源用户不存在' if cond.get('require_not_banned', True): ds = _profile_of(user, 'dashou') if ds is not None and int(getattr(ds, 'zhanghaozhuangtai', 1) or 1) != 1: return False, '账号已封禁，无法迁移' if cond.get('require_no_unpaid_penalty'): try: from orders.models import Penalty unpaid = Penalty.query.filter( PenalizedUserID=str(from_uid), ClubID=plan.from_club_id, Status__in=list(UNPAID_PENALTY_STATUSES), ).exists() if unpaid: return False, '存在未缴罚单，请先缴清' except Exception as e: logger.warning('migrate penalty check failed uid=%s: %s', from_uid, e) return False, '罚单门槛校验失败，请稍后重试' if cond.get('require_orders_closed'): try: from orders.models import Order open_st = Order.objects.filter( PlayerID=str(from_uid), ClubID=plan.from_club_id, Status__in=list(OPEN_ORDER_STATUSES), ).exists() if open_st: return False, '仍有未完结订单，请先结算或退款完成' except Exception as e: logger.warning('migrate order check failed uid=%s: %s', from_uid, e) return False, '订单门槛校验失败，请稍后重试' return True, '' def gate_for_user(*, club_id: str, user_uid: str, roles: Optional[List[str]] = None) -> dict: """小程序启动/进打手端调用。无计划 → 全 false，现网无感。""" plan = find_active_plan_for_source(club_id) if not plan: target = find_active_plan_for_target(club_id) return { 'active': False, 'force': False, 'show_entry': False, 'completed': False, 'plan_id': '', 'to_club_id': '', 'is_target': bool(target), 'target_from_club_id': (target.from_club_id if target else ''), } done = not can_migrate_again(plan, user_uid) force_roles = plan.force_roles or ['dashou'] role_hit = True if roles: role_hit = any(r in force_roles for r in roles) has_assets = user_has_migratable_assets(plan, user_uid) # 关后台「启用」→ find_active 找不到计划 → 上面已全 false，前端零感知 # 开「启用」：源店展示入口（与有无资产无关）；「强制拦截」只控制自动跳 force = bool(plan.force_intercept and role_hit and not done and has_assets) show_entry = True return { 'active': True, 'force': force, 'show_entry': show_entry, 'completed': done, 'has_assets': has_assets, 'plan_id': plan.plan_id, 'plan_name': plan.name or '', 'from_club_id': plan.from_club_id, 'to_club_id': plan.to_club_id, 'force_roles': force_roles, 'force_intercept': bool(plan.force_intercept), 'success_count': success_count(plan.plan_id, user_uid), 'max_allowed': max_allowed_success(plan, user_uid), 'is_target': False, } @transaction.atomic def set_password_and_issue( *, plan: ClubMigratePlan, from_uid: str, password: str, qrcode_url: str = '', ) -> Tuple[Optional[ClubMigrateToken], str]: pwd = str(password or '').strip() if not (pwd.isdigit() and len(pwd) == 6): return None, '转移密码须为6位数字' if not can_migrate_again(plan, from_uid): return None, '本计划已迁移完成，不可再次发码' ok, msg = check_conditions(plan, from_uid) if not ok: return None, msg ClubMigrateToken.query.filter( plan_id=plan.plan_id, from_uid=str(from_uid), status=ClubMigrateToken.STATUS_ISSUED, ).update(status=ClubMigrateToken.STATUS_CANCELLED, UpdateTime=_now()) salt = secrets.token_hex(8) token = new_token() snap = {'assets': preview_assets(plan, from_uid), 'at': _now().isoformat()} row = ClubMigrateToken.query.create( token=token, plan_id=plan.plan_id, from_club_id=plan.from_club_id, to_club_id=plan.to_club_id, from_uid=str(from_uid), password_hash=_hash_password(pwd, salt), password_salt=salt, status=ClubMigrateToken.STATUS_ISSUED, asset_snapshot=snap, qrcode_url=(qrcode_url or '')[:512], expire_at=_now() + timedelta(hours=TOKEN_TTL_HOURS), ) return row, '' def verify_token_password(token_row: ClubMigrateToken, password: str) -> Tuple[bool, str]: if token_row.status != ClubMigrateToken.STATUS_ISSUED: return False, '令牌已失效' if token_row.expire_at and _now() > token_row.expire_at: token_row.status = ClubMigrateToken.STATUS_EXPIRED token_row.save(update_fields=['status', 'UpdateTime']) return False, '令牌已过期，请回源店重新设密出码' if int(token_row.fail_count or 0) >= PASSWORD_MAX_FAIL: return False, '密码错误次数过多，请回源店重置密码' pwd = str(password or '').strip() expect = _hash_password(pwd, token_row.password_salt) if expect != token_row.password_hash: ClubMigrateToken.query.filter(pk=token_row.pk).update(fail_count=F('fail_count') + 1) left = PASSWORD_MAX_FAIL - int(token_row.fail_count or 0) - 1 return False, f'转移密码错误（还可试{max(left, 0)}次）' return True, '' @transaction.atomic def confirm_migrate( *, token: str, password: str, to_uid: str, client_meta: Optional[dict] = None, ) -> Tuple[Optional[dict], str]: """目标店确认过户。成功返回 ledger 摘要。""" token_row = ( ClubMigrateToken.query.select_for_update() .filter(token=str(token or '').strip()) .first() ) if not token_row: return None, '令牌无效' if token_row.status == ClubMigrateToken.STATUS_CONSUMED: led = ClubMigrateLedger.query.filter(token=token_row.token).order_by('-id').first() if led: return serialize_ledger(led), '' return None, '令牌已使用' ok, msg = verify_token_password(token_row, password) if not ok: return None, msg plan = ClubMigratePlan.query.filter(plan_id=token_row.plan_id).first() if not plan or not plan.enabled or not plan_in_window(plan): return None, '迁移计划未生效' if not can_migrate_again(plan, token_row.from_uid): return None, '本计划已迁移完成' ok, msg = check_conditions(plan, token_row.from_uid) if not ok: return None, msg to_user = _load_user(to_uid) if not to_user: return None, '目标用户不存在，请先登录' from_user = _load_user(token_row.from_uid) if not from_user: return None, '源用户不存在' ds = _lock_profile(to_user, 'dashou') if ds is not None and int(getattr(ds, 'zhanghaozhuangtai', 1) or 1) != 1: return None, '目标账号已封禁' cons, _ = ClubMigrateConsumption.query.get_or_create( plan_id=plan.plan_id, from_uid=str(token_row.from_uid), defaults={'success_count': 0}, ) cons = ClubMigrateConsumption.query.select_for_update().filter(pk=cons.pk).first() if not can_migrate_again(plan, token_row.from_uid): return None, '本计划已迁移完成' keys = normalize_whitelist(plan.asset_whitelist) # 目标店：白名单涉及角色先建空扩展行 for role in roles_from_whitelist(keys): if _profile_of(to_user, role) is None: _get_or_create_profile(to_user, role, to_club_id=plan.to_club_id) items_payload = [] migrate_seq = int(cons.success_count or 0) + 1 freeze_biz_prefix = f'cm:{token_row.token}:{migrate_seq}' try: for key in keys: meta = ASSET_CATALOG[key] if meta['kind'] == 'huiyuan': items_payload.extend( _transfer_huiyuan( from_uid=token_row.from_uid, to_uid=str(to_uid), from_club=plan.from_club_id, to_club=plan.to_club_id, role=meta['role'], ) ) continue items_payload.append( _transfer_balance_field( from_user=from_user, to_user=to_user, role=meta['role'], field=meta['field'], kind=meta['kind'], key=key, label=meta['label'], to_club_id=plan.to_club_id, freeze_biz_id=f'{freeze_biz_prefix}:{key}'[:64], ) ) except ValueError as e: transaction.set_rollback(True) return None, str(e) or '会员过户失败' ledger = ClubMigrateLedger.query.create( plan_id=plan.plan_id, token=token_row.token, from_club_id=plan.from_club_id, to_club_id=plan.to_club_id, from_uid=str(token_row.from_uid), to_uid=str(to_uid), migrate_seq=migrate_seq, status=ClubMigrateLedger.STATUS_SUCCESS, summary_json={'item_count': len(items_payload), 'keys': keys}, client_meta=client_meta or {}, ) for it in items_payload: ClubMigrateLedgerItem.query.create( ledger_id=ledger.id, plan_id=plan.plan_id, item_type=it['item_type'], role=it['role'], field=it.get('field', ''), amount=_money(it.get('amount') or 0), before_src=str(it.get('before_src', '')), after_src=str(it.get('after_src', '')), before_dst=str(it.get('before_dst', '')), after_dst=str(it.get('after_dst', '')), huiyuan_id=str(it.get('huiyuan_id') or ''), meta_json=it.get('meta_json') or {}, ) cons.success_count = migrate_seq cons.last_ledger_id = ledger.id cons.save(update_fields=['success_count', 'last_ledger_id', 'UpdateTime']) token_row.status = ClubMigrateToken.STATUS_CONSUMED token_row.consumed_at = _now() token_row.save(update_fields=['status', 'consumed_at', 'UpdateTime']) logger.info( 'club_migrate ok plan=%s seq=%s from=%s to=%s ledger=%s items=%s', plan.plan_id, migrate_seq, token_row.from_uid, to_uid, ledger.id, len(items_payload), ) return serialize_ledger(ledger), '' def _transfer_balance_field( *, from_user, to_user, role, field, kind, key, label, to_club_id: str = '', freeze_biz_id: str = '', ) -> dict: src_prof = _lock_profile(from_user, role) amount = _read_field(src_prof, field, kind) before_src = amount after_src = Decimal('0.00') if kind != 'int' else 0 if src_prof is not None and amount: if kind == 'int': type(src_prof).objects.filter(pk=src_prof.pk).update(**{field: 0}) else: type(src_prof).objects.filter(pk=src_prof.pk).update(**{field: Decimal('0.00')}) dst_prof = _lock_profile(to_user, role) created = False if dst_prof is None: dst_prof = _get_or_create_profile(to_user, role, to_club_id=to_club_id) created = True dst_prof = _lock_profile(to_user, role) before_dst = _read_field(dst_prof, field, kind) if dst_prof else (0 if kind == 'int' else Decimal('0.00')) after_dst = before_dst meta_extra = {} if dst_prof is not None and amount: # 积分：目标已满 10 保持 10；相加后也不得超过 10 if field == 'jifen' and kind == 'int': if int(before_dst) >= MEMBER_JIFEN_CAP: after_dst = MEMBER_JIFEN_CAP else: after_dst = min(MEMBER_JIFEN_CAP, int(before_dst) + int(amount)) type(dst_prof).objects.filter(pk=dst_prof.pk).update(jifen=int(after_dst)) meta_extra['jifen_capped'] = True meta_extra['jifen_cap'] = MEMBER_JIFEN_CAP elif (role, field) in _FREEZE_CREDIT_FIELDS: # 可提现入账：走目标店冻结网关（开了冻结则按配置拆分） from jituan.services.fund_freeze import credit_role_balance, SOURCE_CLUB_MIGRATE to_uid = str(getattr(to_user, 'UserUID', '') or '') credit_res = credit_role_balance( club_id=to_club_id, user_id=to_uid, role=role, amount=_money(amount), source_type=SOURCE_CLUB_MIGRATE, biz_id=(freeze_biz_id or f'cm:{to_uid}:{key}')[:64], profile=dst_prof, freeze_reason='跨店资产迁移入账', ignore_source_scope=True, ) dst_prof = _lock_profile(to_user, role) after_dst = _read_field(dst_prof, field, kind) if dst_prof else before_dst meta_extra.update({ 'freeze_available': str(credit_res.get('available') or 0), 'freeze_frozen': str(credit_res.get('frozen') or 0), 'freeze_ledger_id': credit_res.get('ledger_id'), 'freeze_skipped': bool(credit_res.get('skipped_freeze')), 'freeze_skip_reason': credit_res.get('skip_reason'), }) elif kind == 'int': type(dst_prof).objects.filter(pk=dst_prof.pk).update(**{field: F(field) + int(amount)}) after_dst = int(before_dst) + int(amount) else: # 押金 / 已冻结池等：原样累加，不再二次冻结 type(dst_prof).objects.filter(pk=dst_prof.pk).update(**{field: F(field) + _money(amount)}) after_dst = _money(before_dst) + _money(amount) return { 'item_type': ClubMigrateLedgerItem.ITEM_BALANCE, 'role': role, 'field': field, 'amount': amount, 'before_src': before_src, 'after_src': after_src if amount else before_src, 'before_dst': before_dst, 'after_dst': after_dst, 'meta_json': {'key': key, 'label': label, 'dst_created': created, **meta_extra}, } def _fmt_ledger_time(v: str) -> str: s = str(v or '').strip() if not s or s in ('skipped', 'expired') or 'T' not in s: return s try: dt = datetime.fromisoformat(s.replace('Z', '+00:00')) if timezone.is_naive(dt): dt = timezone.make_aware(dt, timezone.get_current_timezone()) return timezone.localtime(dt).strftime('%Y-%m-%d %H:%M') except Exception: return s def serialize_ledger(ledger: ClubMigrateLedger) -> dict: items = list(ClubMigrateLedgerItem.query.filter(ledger_id=ledger.id).order_by('id')) return { 'ledger_id': ledger.id, 'plan_id': ledger.plan_id, 'from_club_id': ledger.from_club_id, 'to_club_id': ledger.to_club_id, 'from_uid': ledger.from_uid, 'to_uid': ledger.to_uid, 'migrate_seq': ledger.migrate_seq, 'status': ledger.status, 'created_at': ledger.CreateTime.isoformat() if ledger.CreateTime else '', 'items': [ { 'item_type': it.item_type, 'role': it.role, 'field': it.field, 'amount': str(it.amount), 'before_src': it.before_src, 'after_src': it.after_src, 'before_dst': it.before_dst, 'after_dst': _fmt_ledger_time(it.after_dst) if it.field == 'huiyuan' else it.after_dst, 'huiyuan_id': it.huiyuan_id, 'meta': it.meta_json or {}, 'label': (it.meta_json or {}).get('name') or (it.meta_json or {}).get('label') or f'{it.role}.{it.field}', } for it in items ], } def list_user_records(uid: str, limit: int = 50) -> List[dict]: qs = ( ClubMigrateLedger.query.filter(Q(from_uid=str(uid)) | Q(to_uid=str(uid))) .order_by('-id')[: max(1, min(int(limit or 50), 100))] ) return [serialize_ledger(x) for x in qs] def serialize_plan(plan: ClubMigratePlan) -> dict: return { 'plan_id': plan.plan_id, 'name': plan.name or '', 'from_club_id': plan.from_club_id, 'to_club_id': plan.to_club_id, 'enabled': bool(plan.enabled), 'start_at': plan.start_at.isoformat() if plan.start_at else '', 'end_at': plan.end_at.isoformat() if plan.end_at else '', 'asset_whitelist': normalize_whitelist(plan.asset_whitelist), 'condition_json': plan.condition_json or {}, 'force_intercept': bool(plan.force_intercept), 'force_roles': plan.force_roles or ['dashou'], 'default_once_per_user': bool(plan.default_once_per_user), 'remark': plan.remark or '', 'updated_by': plan.updated_by or '', 'created_at': plan.CreateTime.isoformat() if plan.CreateTime else '', 'updated_at': plan.UpdateTime.isoformat() if plan.UpdateTime else '', } def serialize_token(token_row: ClubMigrateToken) -> dict: qr = (token_row.qrcode_url or '').strip() if qr and not qr.startswith('http'): from django.conf import settings cos_domain = getattr(settings, 'COS_DOMAIN', '').rstrip('/') if cos_domain: qr = f'{cos_domain}/{qr.lstrip("/")}' return { 'token': token_row.token, 'scene': scene_for_token(token_row.token), 'plan_id': token_row.plan_id, 'from_club_id': token_row.from_club_id, 'to_club_id': token_row.to_club_id, 'status': token_row.status, 'qrcode_url': qr, 'expire_at': token_row.expire_at.isoformat() if token_row.expire_at else '', 'asset_snapshot': token_row.asset_snapshot or {}, } def lookup_migrate_user_brief(uid: str) -> dict: """后台二次名额：校验用户并返回简要资料。""" uid = str(uid or '').strip() user = _load_user(uid) if not user: return {'exists': False, 'from_uid': uid, 'nicheng': '', 'phone': ''} nicheng = '' try: ds = getattr(user, 'DashouProfile', None) if ds and getattr(ds, 'nicheng', None): nicheng = str(ds.nicheng) except Exception: pass if not nicheng: nicheng = str(getattr(user, 'UserName', '') or getattr(user, 'Phone', '') or '') return { 'exists': True, 'from_uid': uid, 'nicheng': nicheng, 'phone': str(getattr(user, 'Phone', '') or ''), 'club_id': str(getattr(user, 'ClubID', '') or ''), } def serialize_user_quota(row: ClubMigrateUserQuota, plan: Optional[ClubMigratePlan] = None, user_info: Optional[dict] = None) -> dict: plan = plan or ClubMigratePlan.query.filter(plan_id=row.plan_id).first() info = user_info if user_info is not None else lookup_migrate_user_brief(row.from_uid) used = success_count(row.plan_id, row.from_uid) allowed = max_allowed_success(plan, row.from_uid) if plan else (1 + int(row.max_extra_times or 0)) return { 'plan_id': row.plan_id, 'from_uid': row.from_uid, 'max_extra_times': int(row.max_extra_times or 0), 'reason': row.reason or '', 'updated_by': row.updated_by or '', 'created_at': row.CreateTime.isoformat() if row.CreateTime else '', 'updated_at': row.UpdateTime.isoformat() if row.UpdateTime else '', 'nicheng': info.get('nicheng') or '', 'phone': info.get('phone') or '', 'user_club_id': info.get('club_id') or '', 'success_count': used, 'max_allowed': allowed, 'remaining': max(0, int(allowed) - int(used)), 'can_migrate_again': used < allowed, }