feat: 资金冻结 P1 打手结单收口网关 + 到期解冻定时任务骨架

Co-authored-by: Cursor <cursoragent@cursor.com>

a_long_dianjing/celery.py

@@ -17,7 +17,7 @@ app.config_from_object('django.conf:settings', namespace='CELERY')
 # 自动发现所有已注册app中的tasks.py文件
 #app.autodiscover_tasks()
 # 自动发现任务（不含 orders.tasks：订单超时自动结算已永久关闭）
-app.autodiscover_tasks(['users.tasks', 'users.ranking_tasks', 'config.tasks'])
+app.autodiscover_tasks(['users.tasks', 'users.ranking_tasks', 'config.tasks', 'jituan.tasks'])
 
 # 配置周期性任务（Celery Beat Schedule）
 app.conf.beat_schedule = {
@@ -84,7 +84,17 @@ app.conf.beat_schedule = {
         },
         'args': (),
         'kwargs': {}
-    }
+    },
+
+    # 5. 资金冻结到期解冻（默认无冻结单时几乎空跑）
+    'scan_fund_freeze_due': {
+        'task': 'jituan.tasks.scan_fund_freeze_due',
+        'schedule': crontab(minute='*/15'),
+        'options': {
+            'queue': 'periodic_tasks',
+            'priority': 4,
+        },
+    },
 }
 
 
@@ -97,6 +107,7 @@ app.conf.task_routes = {
     # 'orders.tasks.*': {'queue': 'order_tasks'},
     'users.tasks.*': {'queue': 'periodic_tasks'},
     'config.tasks.*': {'queue': 'periodic_tasks'},
+    'jituan.tasks.*': {'queue': 'periodic_tasks'},
 }
 
 # 任务序列化

jituan/services/fund_freeze.py

@@ -1,9 +1,9 @@
-"""资金冻结入账网关（P0）。
+"""资金冻结入账网关与解冻骨架。
 
 默认：无配置或 enabled=False → 全额进可提现，与现网一致，不写流水。
 开启后：拆「可提现 + 冻结池」并写 FundFreezeLedger。
 
-注意：P0 仅提供网关；结算调用点在 P1 收口，避免一次改太多接口。
+P1：打手订单结算入口已统一走 credit_dashou_order_settle（未开冻结时行为与改造前一致）。
 """
 from __future__ import annotations
 
@@ -273,3 +273,260 @@ def split_preview(club_id: Optional[str], role: str, amount, source_type: str =
         return credit, Decimal('0.00')
     frozen = calc_freeze_amount(credit, cfg)
     return _money(credit - frozen), frozen
+
+
+def order_club_id(order) -> str:
+    return _club(getattr(order, 'ClubID', None) or getattr(order, 'club_id', None))
+
+
+def credit_dashou_order_settle(
+    *,
+    order,
+    dashou_profile,
+    user_id: str,
+    amount,
+    set_idle_status: bool = False,
+    increment_chengjiao: bool = True,
+) -> Dict[str, Any]:
+    """
+    打手订单结算专用入账（P1）。
+
+    - 可提现走 credit_role_balance（未开冻结=全额进 yue，与改造前一致）
+    - zonge / 今日今月收益仍按「全额」累加（统计口径不变）
+    - 成交单量 +1、可选置空闲状态
+    调用方须已在 transaction.atomic() 内；本函数用 F() 更新，勿再对同一 profile save 覆盖余额字段。
+    """
+    from users.models import UserDashou
+
+    credit = _money(amount)
+    order_id = str(getattr(order, 'OrderID', None) or getattr(order, 'order_id', '') or '')
+    cid = order_club_id(order)
+
+    if credit <= 0:
+        updates = {}
+        if increment_chengjiao:
+            updates['chengjiaozongliang'] = F('chengjiaozongliang') + 1
+        if set_idle_status:
+            updates['zhuangtai'] = 1
+        if updates:
+            UserDashou.query.filter(pk=dashou_profile.pk).update(**updates)
+        return {
+            'available': Decimal('0.00'),
+            'frozen': Decimal('0.00'),
+            'ledger_id': None,
+            'skipped_freeze': True,
+        }
+
+    result = credit_role_balance(
+        club_id=cid,
+        user_id=str(user_id),
+        role=ROLE_DASHOU,
+        amount=credit,
+        source_type=SOURCE_ORDER_SETTLE,
+        biz_id=order_id,
+        profile=dashou_profile,
+        order_id=order_id,
+        freeze_reason=f'订单结算 {order_id}',
+        also_credit_zonge=False,
+    )
+    if result.get('idempotent'):
+        logger.warning('打手订单结算幂等跳过 order=%s user=%s', order_id, user_id)
+        return result
+
+    updates = {
+        'zonge': F('zonge') + credit,
+        'jinrishouyi': F('jinrishouyi') + credit,
+        'jinyueshouyi': F('jinyueshouyi') + credit,
+    }
+    if increment_chengjiao:
+        updates['chengjiaozongliang'] = F('chengjiaozongliang') + 1
+    if set_idle_status:
+        updates['zhuangtai'] = 1
+    UserDashou.query.filter(pk=dashou_profile.pk).update(**updates)
+    return result
+
+
+def _step_unfreeze_amount(ledger: FundFreezeLedger, step: dict) -> Decimal:
+    remain = _money(ledger.amount_total - ledger.amount_unfrozen)
+    if remain <= 0:
+        return Decimal('0.00')
+    typ = (step.get('type') or 'remain').strip().lower()
+    if typ == 'remain':
+        return remain
+    if typ == 'fixed':
+        return min(remain, _money(step.get('value') or 0))
+    # ratio
+    ratio = Decimal(str(step.get('value') or 0))
+    if ratio <= 0:
+        return Decimal('0.00')
+    return min(remain, _money(ledger.amount_total * ratio))
+
+
+def _plan_steps(snapshot: dict) -> list:
+    schedule = (snapshot or {}).get('schedule_mode') or 'all_at_once'
+    if schedule == 'installment':
+        steps = ((snapshot or {}).get('installment_json') or {}).get('steps') or []
+        if steps:
+            return sorted(steps, key=lambda s: int(s.get('seq') or 0))
+    # 一次全解：单步 remain，天数用 freeze_min_days（已在 next_eligible_at）
+    return [{'seq': 1, 'after_days': 0, 'type': 'remain', 'value': 1, 'need_condition': False}]
+
+
+def apply_unfreeze_step(
+    ledger_id: int,
+    *,
+    trigger: str = 'schedule',
+    operator_id: str = '',
+    reason: str = '',
+    force: bool = False,
+) -> Dict[str, Any]:
+    """
+    执行一笔冻结单的当前分期步（骨架：先支持到期/手动；条件解冻 P4 再接指标）。
+    force=True 时忽略 next_eligible_at（后台手动）。
+    """
+    from django.db import transaction
+    from users.models import UserDashou, UserGuanshi, UserZuzhang, UserShenheguan
+    from jituan.models import FundUnfreezeEvent
+
+    with transaction.atomic():
+        ledger = (
+            FundFreezeLedger.objects.select_for_update()
+            .filter(pk=ledger_id)
+            .first()
+        )
+        if not ledger:
+            return {'ok': False, 'msg': '冻结单不存在'}
+        if ledger.status in (FundFreezeLedger.STATUS_DONE, FundFreezeLedger.STATUS_VOID):
+            return {'ok': False, 'msg': '已完结或已作废'}
+
+        now = timezone.now()
+        if not force and ledger.next_eligible_at and ledger.next_eligible_at > now:
+            return {'ok': False, 'msg': '未到解冻时间', 'next_eligible_at': ledger.next_eligible_at}
+
+        snap = ledger.config_snapshot or {}
+        trigger_mode = (snap.get('unfreeze_trigger') or 'time_only')
+        if trigger_mode == 'manual_only' and not force:
+            return {'ok': False, 'msg': '仅允许手动解冻'}
+        # condition_*：自动任务暂不评估指标（P4）；手动 force 可解
+        if trigger_mode == 'condition_only' and not force:
+            return {'ok': False, 'msg': '条件解冻尚未启用自动评估'}
+
+        steps = _plan_steps(snap)
+        step_idx = max(0, int(ledger.current_step or 1) - 1)
+        if step_idx >= len(steps):
+            step = {'seq': ledger.current_step, 'type': 'remain', 'value': 1}
+        else:
+            step = steps[step_idx]
+
+        amount = _step_unfreeze_amount(ledger, step)
+        if amount <= 0:
+            ledger.status = FundFreezeLedger.STATUS_DONE
+            ledger.save(update_fields=['status', 'UpdateTime'])
+            return {'ok': True, 'amount': Decimal('0.00'), 'done': True}
+
+        profile_map = {
+            ROLE_DASHOU: UserDashou,
+            ROLE_GUANSHI: UserGuanshi,
+            ROLE_ZUZHANG: UserZuzhang,
+            ROLE_SHENHEGUAN: UserShenheguan,
+        }
+        model = profile_map.get(ledger.role)
+        if not model:
+            return {'ok': False, 'msg': f'未知角色 {ledger.role}'}
+
+        from users.models import User
+        user = User.query.filter(UserUID=ledger.user_id).first()
+        if not user:
+            return {'ok': False, 'msg': '用户不存在'}
+        profile = model.objects.select_for_update().filter(user=user).first()
+        if not profile:
+            return {'ok': False, 'msg': '角色扩展表不存在'}
+
+        # 冻结池不足则按实际可解
+        cur_frozen = _money(getattr(profile, 'dongjie_yue', 0) or 0)
+        if cur_frozen < amount:
+            amount = cur_frozen
+        if amount <= 0:
+            return {'ok': False, 'msg': '冻结池余额不足'}
+
+        if ledger.role == ROLE_ZUZHANG:
+            model.query.filter(pk=profile.pk).update(
+                dongjie_yue=F('dongjie_yue') - amount,
+                ketixian_jine=F('ketixian_jine') + amount,
+            )
+        else:
+            model.query.filter(pk=profile.pk).update(
+                dongjie_yue=F('dongjie_yue') - amount,
+                yue=F('yue') + amount,
+            )
+
+        new_unfrozen = _money(ledger.amount_unfrozen + amount)
+        remain = _money(ledger.amount_total - new_unfrozen)
+        ledger.amount_unfrozen = new_unfrozen
+        next_step = int(ledger.current_step or 1) + 1
+
+        if remain <= 0:
+            ledger.status = FundFreezeLedger.STATUS_DONE
+            ledger.current_step = next_step
+            ledger.next_eligible_at = None
+        else:
+            ledger.status = FundFreezeLedger.STATUS_PARTIAL
+            ledger.current_step = next_step
+            # 下一步到期时间
+            if next_step - 1 < len(steps):
+                nxt = steps[next_step - 1]
+                days = int(nxt.get('after_days') or 0)
+                # after_days 相对冻结日
+                ledger.next_eligible_at = ledger.frozen_at + timedelta(days=days)
+            else:
+                ledger.next_eligible_at = now
+
+        ledger.save(update_fields=[
+            'amount_unfrozen', 'status', 'current_step', 'next_eligible_at', 'UpdateTime',
+        ])
+
+        FundUnfreezeEvent.query.create(
+            ledger_id=ledger.id,
+            club_id=ledger.club_id,
+            user_id=ledger.user_id,
+            role=ledger.role,
+            step_seq=int(step.get('seq') or ledger.current_step),
+            amount=amount,
+            trigger=trigger,
+            reason=(reason or f'{trigger}解冻')[:255],
+            operator_id=(operator_id or '')[:32],
+            condition_snapshot={},
+        )
+
+        return {
+            'ok': True,
+            'amount': amount,
+            'remain': max(remain, Decimal('0.00')),
+            'done': remain <= 0,
+            'ledger_id': ledger.id,
+        }
+
+
+def scan_due_freeze_ledgers(limit: int = 200) -> Dict[str, Any]:
+    """定时任务入口：扫到期冻结单并尝试解冻一步。"""
+    now = timezone.now()
+    qs = (
+        FundFreezeLedger.query.filter(
+            status__in=[FundFreezeLedger.STATUS_FROZEN, FundFreezeLedger.STATUS_PARTIAL],
+            next_eligible_at__lte=now,
+        )
+        .order_by('next_eligible_at', 'id')[:limit]
+    )
+    ok_n = fail_n = 0
+    for row in qs:
+        try:
+            r = apply_unfreeze_step(row.id, trigger='schedule')
+            if r.get('ok'):
+                ok_n += 1
+            else:
+                fail_n += 1
+                logger.info('unfreeze skip id=%s msg=%s', row.id, r.get('msg'))
+        except Exception as e:
+            fail_n += 1
+            logger.exception('unfreeze error id=%s: %s', row.id, e)
+    return {'ok': ok_n, 'fail': fail_n, 'scanned': ok_n + fail_n}

jituan/tasks.py

@@ -0,0 +1,15 @@
+"""资金冻结相关 Celery 定时任务。"""
+import logging
+
+from celery import shared_task
+
+logger = logging.getLogger(__name__)
+
+
+@shared_task(name='jituan.tasks.scan_fund_freeze_due')
+def scan_fund_freeze_due(limit: int = 200):
+    """扫到期冻结单，按配置执行一步解冻（未开冻结的俱乐部无数据，空跑）。"""
+    from jituan.services.fund_freeze import scan_due_freeze_ledgers
+    result = scan_due_freeze_ledgers(limit=limit)
+    logger.info('scan_fund_freeze_due %s', result)
+    return result

orders/tasks.py

@@ -70,13 +70,12 @@ def process_expired_order(self, dingdan_id):
                         # 获取打手扩展表
                         dashou_profile = getattr(dashou_user, 'DashouProfile', None)
                         if dashou_profile:
-                            # 使用F表达式原子更新
-                            UserDashou.query.filter(pk=dashou_profile.pk).update(
-                                chengjiaozongliang=F('chengjiaozongliang') + 1,
-                                yue=F('yue') + dashou_fencheng,
-                                zonge=F('zonge') + dashou_fencheng,
-                                jinrishouyi=F('jinrishouyi') + dashou_fencheng,
-                                jinyueshouyi=F('jinyueshouyi') + dashou_fencheng
+                            from jituan.services.fund_freeze import credit_dashou_order_settle
+                            credit_dashou_order_settle(
+                                order=order,
+                                dashou_profile=dashou_profile,
+                                user_id=dashou_id,
+                                amount=dashou_fencheng,
                             )
                             updated_dashou = True
                             logger.info(f"更新打手{dashou_id}信息成功")

orders/views/admin.py

@@ -613,23 +613,13 @@ class AdQiangZhiJieDan(APIView):
                         # 获取打手扩展表
                         dashou_profile = dashou_user.DashouProfile
 
-                        # 更新打手扩展表数据
-                        # ✅ 确认字段：chengjiaozongliang 成交单总量
-                        dashou_profile.chengjiaozongliang += 1
-
-                        # ✅ 确认字段：yue 可提现余额
-                        dashou_profile.yue += dashou_fencheng
-
-                        # ✅ 确认字段：zonge 打单赚取总额
-                        dashou_profile.zonge += dashou_fencheng
-
-                        # ✅ 确认字段：jinrishouyi 今日收益金额
-                        dashou_profile.jinrishouyi += dashou_fencheng
-
-                        # ✅ 确认字段：jinyueshouyi 今月收益金额
-                        dashou_profile.jinyueshouyi += dashou_fencheng
-
-                        dashou_profile.save()
+                        from jituan.services.fund_freeze import credit_dashou_order_settle
+                        credit_dashou_order_settle(
+                            order=dingdan_obj,
+                            dashou_profile=dashou_profile,
+                            user_id=jiedan_dashou_id,
+                            amount=dashou_fencheng,
+                        )
                         logger.info(f"强制结单成功，打手{jiedan_dashou_id}结算分成：{dashou_fencheng}")
 
                     except User.DoesNotExist:
@@ -789,23 +779,13 @@ class AdJuJueTuiKuan(APIView):
                         # 获取打手扩展表
                         dashou_profile = dashou_user.DashouProfile
 
-                        # 更新打手扩展表数据
-                        # ✅ 确认字段：chengjiaozongliang 成交单总量
-                        dashou_profile.chengjiaozongliang += 1
-
-                        # ✅ 确认字段：yue 可提现余额
-                        dashou_profile.yue += dashou_fencheng
-
-                        # ✅ 确认字段：zonge 打单赚取总额
-                        dashou_profile.zonge += dashou_fencheng
-
-                        # ✅ 确认字段：jinrishouyi 今日收益金额
-                        dashou_profile.jinrishouyi += dashou_fencheng
-
-                        # ✅ 确认字段：jinyueshouyi 今月收益金额
-                        dashou_profile.jinyueshouyi += dashou_fencheng
-
-                        dashou_profile.save()
+                        from jituan.services.fund_freeze import credit_dashou_order_settle
+                        credit_dashou_order_settle(
+                            order=dingdan_obj,
+                            dashou_profile=dashou_profile,
+                            user_id=jiedan_dashou_id,
+                            amount=dashou_fencheng,
+                        )
                         logger.info(f"拒绝退款成功，打手{jiedan_dashou_id}结算分成：{dashou_fencheng}")
 
                     except User.DoesNotExist:

orders/views/boss_orders.py

@@ -432,21 +432,14 @@ class JiedanView(APIView):
                             dashou_kuozhan = None
 
                         if dashou_kuozhan:
-                            # 更新四个金额字段
-                            dashou_kuozhan.zonge = (dashou_kuozhan.zonge or Decimal('0.00')) + dashou_fencheng
-                            dashou_kuozhan.yue = (dashou_kuozhan.yue or Decimal('0.00')) + dashou_fencheng
-                            dashou_kuozhan.jinrishouyi = (dashou_kuozhan.jinrishouyi or Decimal(
-                                '0.00')) + dashou_fencheng
-                            dashou_kuozhan.jinyueshouyi = (dashou_kuozhan.jinyueshouyi or Decimal(
-                                '0.00')) + dashou_fencheng
-
-                            # 成交总量加1
-                            dashou_kuozhan.chengjiaozongliang = (dashou_kuozhan.chengjiaozongliang or 0) + 1
-
-                            # 状态改为数字1
-                            dashou_kuozhan.zhuangtai = 1
-
-                            dashou_kuozhan.save()
+                            from jituan.services.fund_freeze import credit_dashou_order_settle
+                            credit_dashou_order_settle(
+                                order=dingdan,
+                                dashou_profile=dashou_kuozhan,
+                                user_id=jiedan_dashou_id,
+                                amount=dashou_fencheng,
+                                set_idle_status=True,
+                            )
 
                 # 8. 更新订单状态为已完成（状态3）
                 dingdan.Status = 3
@@ -749,13 +742,14 @@ class JiedanView2(APIView):
                     if dashou_user:
                         try:
                             dashou_ext = dashou_user.DashouProfile
-                            dashou_ext.zonge = (dashou_ext.zonge or Decimal('0.00')) + dashou_fencheng
-                            dashou_ext.yue = (dashou_ext.yue or Decimal('0.00')) + dashou_fencheng
-                            dashou_ext.jinrishouyi = (dashou_ext.jinrishouyi or Decimal('0.00')) + dashou_fencheng
-                            dashou_ext.jinyueshouyi = (dashou_ext.jinyueshouyi or Decimal('0.00')) + dashou_fencheng
-                            dashou_ext.chengjiaozongliang = (dashou_ext.chengjiaozongliang or 0) + 1
-                            dashou_ext.zhuangtai = 1
-                            dashou_ext.save()
+                            from jituan.services.fund_freeze import credit_dashou_order_settle
+                            credit_dashou_order_settle(
+                                order=order,
+                                dashou_profile=dashou_ext,
+                                user_id=jiedan_dashou_id,
+                                amount=dashou_fencheng,
+                                set_idle_status=True,
+                            )
                         except UserDashou.DoesNotExist:
                             pass
 

orders/views/merchant.py

@@ -880,13 +880,13 @@ class ShangjiaJiesuanView(APIView):
                         'data': None
                     }, status=status.HTTP_400_BAD_REQUEST)
 
-                # 原子更新打手数据（使用 F 表达式）
-                UserDashou.query.filter(id=dashou.id).update(
-                    chengjiaozongliang=F('chengjiaozongliang') + 1,
-                    yue=F('yue') + dashou_fencheng,
-                    zonge=F('zonge') + dashou_fencheng,
-                    jinrishouyi=F('jinrishouyi') + dashou_fencheng,
-                    jinyueshouyi=F('jinyueshouyi') + dashou_fencheng,
+                # 原子更新打手数据（可提现走冻结网关；未开启时与原先全额进 yue 一致）
+                from jituan.services.fund_freeze import credit_dashou_order_settle
+                credit_dashou_order_settle(
+                    order=order,
+                    dashou_profile=dashou,
+                    user_id=jiedan_dashou_id,
+                    amount=dashou_fencheng,
                 )
 
                 # 更新商家成交单量

shop/views/shop_refund_views.py

@@ -379,13 +379,14 @@ class ShopForceCompleteView(APIView):
                 try:
                     dashou_user = User.query.select_related('DashouProfile').get(UserUID=dashou_id)
                     dashou_profile = dashou_user.DashouProfile
-                    dashou_profile.chengjiaozongliang += 1
-                    dashou_profile.yue += dashou_fencheng
-                    dashou_profile.zonge += dashou_fencheng
-                    dashou_profile.jinrishouyi += dashou_fencheng
-                    dashou_profile.jinyueshouyi += dashou_fencheng
-                    dashou_profile.zhuangtai = 1       # 空闲
-                    dashou_profile.save()
+                    from jituan.services.fund_freeze import credit_dashou_order_settle
+                    credit_dashou_order_settle(
+                        order=order,
+                        dashou_profile=dashou_profile,
+                        user_id=dashou_id,
+                        amount=dashou_fencheng,
+                        set_idle_status=True,
+                    )
                 except (User.DoesNotExist, ObjectDoesNotExist):
                     logger.warning(f"本地订单结算：打手 {dashou_id} 不存在，跳过打手更新")
 

users/views/kefu_orders.py

@@ -1020,13 +1020,13 @@ class KefuRejectRefundView(APIView):
                     try:
                         dashou_user = User.query.get(UserUID=jiedan_dashou_id)
                         dashou_profile = dashou_user.DashouProfile
-                        # 增加打手相关统计数据
-                        dashou_profile.chengjiaozongliang += 1
-                        dashou_profile.yue += dashou_fencheng
-                        dashou_profile.zonge += dashou_fencheng
-                        dashou_profile.jinrishouyi += dashou_fencheng
-                        dashou_profile.jinyueshouyi += dashou_fencheng
-                        dashou_profile.save()
+                        from jituan.services.fund_freeze import credit_dashou_order_settle
+                        credit_dashou_order_settle(
+                            order=order,
+                            dashou_profile=dashou_profile,
+                            user_id=jiedan_dashou_id,
+                            amount=dashou_fencheng,
+                        )
                         logger.info(f"拒绝退款：打手 {jiedan_dashou_id} 获得分成 {dashou_fencheng}")
                     except (User.DoesNotExist, AttributeError):
                         logger.warning(f"拒绝退款：打手 {jiedan_dashou_id} 不存在或扩展表缺失，跳过结算")
@@ -1739,16 +1739,13 @@ class KefuForceCompleteView(APIView):
                 try:
                     dashou_user = User.query.get(UserUID=dashou_id)
                     dashou_profile = dashou_user.DashouProfile
-                    # 原子更新（避免并发重复加）
-                    dashou_profile.chengjiaozongliang = F('chengjiaozongliang') + 1
-                    dashou_profile.yue = F('yue') + dashou_fencheng
-                    dashou_profile.zonge = F('zonge') + dashou_fencheng
-                    dashou_profile.jinrishouyi = F('jinrishouyi') + dashou_fencheng
-                    dashou_profile.jinyueshouyi = F('jinyueshouyi') + dashou_fencheng
-                    dashou_profile.save(update_fields=[
-                        'chengjiaozongliang', 'yue', 'zonge',
-                        'jinrishouyi', 'jinyueshouyi'
-                    ])
+                    from jituan.services.fund_freeze import credit_dashou_order_settle
+                    credit_dashou_order_settle(
+                        order=order,
+                        dashou_profile=dashou_profile,
+                        user_id=dashou_id,
+                        amount=dashou_fencheng,
+                    )
                 except (User.DoesNotExist, AttributeError):
                     return Response({'code': 400, 'msg': '打手不存在或扩展表缺失'})