diff --git a/a_long_dianjing/celery.py b/a_long_dianjing/celery.py index af17945..5ea0dc0 100644 --- a/a_long_dianjing/celery.py +++ b/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'}, } # 任务序列化 diff --git a/jituan/services/fund_freeze.py b/jituan/services/fund_freeze.py index 69d2d7e..874ec45 100644 --- a/jituan/services/fund_freeze.py +++ b/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} diff --git a/jituan/tasks.py b/jituan/tasks.py new file mode 100644 index 0000000..84ceb83 --- /dev/null +++ b/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 diff --git a/orders/tasks.py b/orders/tasks.py index 4424bbb..896c74c 100644 --- a/orders/tasks.py +++ b/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}信息成功") diff --git a/orders/views/admin.py b/orders/views/admin.py index cbe68db..fd70dbf 100644 --- a/orders/views/admin.py +++ b/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: diff --git a/orders/views/boss_orders.py b/orders/views/boss_orders.py index acce73e..77e4d59 100644 --- a/orders/views/boss_orders.py +++ b/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 diff --git a/orders/views/merchant.py b/orders/views/merchant.py index 59aff9e..7bc7d35 100644 --- a/orders/views/merchant.py +++ b/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, ) # 更新商家成交单量 diff --git a/shop/views/shop_refund_views.py b/shop/views/shop_refund_views.py index 92199aa..0699d06 100644 --- a/shop/views/shop_refund_views.py +++ b/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} 不存在，跳过打手更新") diff --git a/users/views/kefu_orders.py b/users/views/kefu_orders.py index 3b9e986..118ce91 100644 --- a/users/views/kefu_orders.py +++ b/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': '打手不存在或扩展表缺失'})