6c96eb3bcb
后台条件真正按罚款率/单量等评估；选项式配置所需字段规范化；Celery 改为每5分钟；小程序说明文案接口补齐。 Co-authored-by: Cursor <cursoragent@cursor.com>
645 lines
22 KiB
Python
645 lines
22 KiB
Python
"""资金冻结入账网关与解冻骨架。
|
||
|
||
默认：无配置或 enabled=False → 全额进可提现，与现网一致，不写流水。
|
||
开启后：拆「可提现 + 冻结池」并写 FundFreezeLedger。
|
||
|
||
P1：打手订单结算入口已统一走 credit_dashou_order_settle（未开冻结时行为与改造前一致）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from datetime import timedelta
|
||
from decimal import Decimal, ROUND_HALF_UP
|
||
from typing import Any, Dict, Optional, Tuple
|
||
|
||
from django.core.cache import cache
|
||
from django.db.models import F
|
||
from django.utils import timezone
|
||
|
||
from jituan.constants import CLUB_ID_DEFAULT
|
||
from jituan.models import ClubFundFreezeConfig, FundFreezeLedger
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
ROLE_DASHOU = 'dashou'
|
||
ROLE_GUANSHI = 'guanshi'
|
||
ROLE_ZUZHANG = 'zuzhang'
|
||
ROLE_SHENHEGUAN = 'shenheguan'
|
||
|
||
SOURCE_ORDER_SETTLE = 'order_settle'
|
||
SOURCE_FENHONG = 'fenhong'
|
||
SOURCE_KAOHE_FEE = 'kaohe_fee'
|
||
|
||
_CFG_CACHE_TTL = 60
|
||
|
||
|
||
def _money(v) -> Decimal:
|
||
return Decimal(str(v or 0)).quantize(Decimal('0.01'), rounding=ROUND_HALF_UP)
|
||
|
||
|
||
def _club(club_id: Optional[str]) -> str:
|
||
return (club_id or '').strip() or CLUB_ID_DEFAULT
|
||
|
||
|
||
def get_freeze_config(club_id: Optional[str], role: str) -> Optional[ClubFundFreezeConfig]:
|
||
cid = _club(club_id)
|
||
role = (role or '').strip().lower()
|
||
key = f'fund_freeze_cfg:{cid}:{role}'
|
||
cached = cache.get(key)
|
||
if cached is not None:
|
||
return cached or None
|
||
row = ClubFundFreezeConfig.query.filter(club_id=cid, role=role).first()
|
||
cache.set(key, row or False, _CFG_CACHE_TTL)
|
||
return row
|
||
|
||
|
||
def invalidate_freeze_config_cache(club_id: Optional[str] = None, role: Optional[str] = None):
|
||
if club_id and role:
|
||
cache.delete(f'fund_freeze_cfg:{_club(club_id)}:{(role or "").strip().lower()}')
|
||
return
|
||
# 粗粒度：调用方改配置后建议带 club+role 清；此处留空避免误扫
|
||
|
||
|
||
def calc_freeze_amount(credit: Decimal, cfg: ClubFundFreezeConfig) -> Decimal:
|
||
credit = _money(credit)
|
||
if credit <= 0:
|
||
return Decimal('0.00')
|
||
min_c = _money(cfg.min_credit_to_freeze)
|
||
if min_c > 0 and credit < min_c:
|
||
return Decimal('0.00')
|
||
|
||
mode = int(cfg.freeze_mode or 1)
|
||
frozen = Decimal('0.00')
|
||
if mode == 2:
|
||
frozen = _money(cfg.freeze_fixed)
|
||
elif mode == 3:
|
||
frozen = _money(credit * Decimal(str(cfg.freeze_rate or 0)))
|
||
cap = _money(cfg.freeze_cap)
|
||
if cap > 0:
|
||
frozen = min(frozen, cap)
|
||
else:
|
||
frozen = _money(credit * Decimal(str(cfg.freeze_rate or 0)))
|
||
cap = _money(cfg.freeze_cap)
|
||
if cap > 0:
|
||
frozen = min(frozen, cap)
|
||
|
||
if frozen < 0:
|
||
frozen = Decimal('0.00')
|
||
if frozen > credit:
|
||
frozen = credit
|
||
return frozen
|
||
|
||
|
||
def _source_allowed(cfg: ClubFundFreezeConfig, source_type: str) -> bool:
|
||
scope = cfg.source_scope
|
||
if not scope:
|
||
return True
|
||
if isinstance(scope, str):
|
||
return source_type == scope
|
||
try:
|
||
return source_type in list(scope)
|
||
except TypeError:
|
||
return True
|
||
|
||
|
||
def _config_snapshot(cfg: ClubFundFreezeConfig) -> Dict[str, Any]:
|
||
return {
|
||
'club_id': cfg.club_id,
|
||
'role': cfg.role,
|
||
'enabled': cfg.enabled,
|
||
'freeze_mode': cfg.freeze_mode,
|
||
'freeze_rate': str(cfg.freeze_rate),
|
||
'freeze_fixed': str(cfg.freeze_fixed),
|
||
'freeze_cap': str(cfg.freeze_cap),
|
||
'min_credit_to_freeze': str(cfg.min_credit_to_freeze),
|
||
'source_scope': cfg.source_scope,
|
||
'unfreeze_trigger': cfg.unfreeze_trigger,
|
||
'schedule_mode': cfg.schedule_mode,
|
||
'installment_json': cfg.installment_json,
|
||
'condition_json': cfg.condition_json,
|
||
'freeze_min_days': cfg.freeze_min_days,
|
||
'display_reason': (cfg.remark or '')[:255],
|
||
'remark': (cfg.remark or '')[:255],
|
||
}
|
||
|
||
|
||
def _first_step_delay_days(cfg: ClubFundFreezeConfig) -> int:
|
||
if (cfg.schedule_mode or '') == 'installment':
|
||
steps = (cfg.installment_json or {}).get('steps') or []
|
||
if steps:
|
||
try:
|
||
return max(0, int(steps[0].get('after_days') or 0))
|
||
except (TypeError, ValueError):
|
||
pass
|
||
return max(0, int(cfg.freeze_min_days or 0))
|
||
|
||
|
||
def _add_withdrawable(role: str, profile, available: Decimal, also_zonge: bool = False):
|
||
"""原子增加可提现；组长字段为 ketixian_jine。"""
|
||
available = _money(available)
|
||
if available == 0:
|
||
return
|
||
if role == ROLE_ZUZHANG:
|
||
type(profile).query.filter(pk=profile.pk).update(
|
||
ketixian_jine=F('ketixian_jine') + available,
|
||
)
|
||
else:
|
||
updates = {'yue': F('yue') + available}
|
||
if also_zonge and hasattr(profile, 'zonge'):
|
||
updates['zonge'] = F('zonge') + available
|
||
type(profile).query.filter(pk=profile.pk).update(**updates)
|
||
|
||
|
||
def _add_dongjie(profile, frozen: Decimal):
|
||
frozen = _money(frozen)
|
||
if frozen == 0:
|
||
return
|
||
type(profile).query.filter(pk=profile.pk).update(
|
||
dongjie_yue=F('dongjie_yue') + frozen,
|
||
)
|
||
|
||
|
||
def credit_role_balance(
|
||
*,
|
||
club_id: Optional[str],
|
||
user_id: str,
|
||
role: str,
|
||
amount,
|
||
source_type: str,
|
||
biz_id: str,
|
||
profile=None,
|
||
order_id: str = '',
|
||
czjilu_id: str = '',
|
||
freeze_reason: str = '',
|
||
also_credit_zonge: bool = False,
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
统一入账。返回 {available, frozen, ledger_id, skipped_freeze}。
|
||
|
||
调用方须在外层 transaction.atomic() 内，并已持有/即将更新的业务一致性。
|
||
profile: 已加载的角色扩展实例（UserDashou 等）；若空则按 role+user_id 查。
|
||
"""
|
||
role = (role or '').strip().lower()
|
||
cid = _club(club_id)
|
||
credit = _money(amount)
|
||
biz_id = (biz_id or '').strip()
|
||
source_type = (source_type or '').strip() or 'unknown'
|
||
|
||
if credit <= 0:
|
||
return {'available': Decimal('0.00'), 'frozen': Decimal('0.00'), 'ledger_id': None, 'skipped_freeze': True}
|
||
|
||
if profile is None:
|
||
profile = _load_profile(role, user_id)
|
||
if profile is None:
|
||
raise ValueError(f'找不到角色资料 role={role} user_id={user_id}')
|
||
|
||
cfg = get_freeze_config(cid, role)
|
||
freeze_on = bool(
|
||
cfg and cfg.enabled and _source_allowed(cfg, source_type) and biz_id
|
||
)
|
||
frozen = calc_freeze_amount(credit, cfg) if freeze_on else Decimal('0.00')
|
||
available = _money(credit - frozen)
|
||
|
||
# 幂等：已存在同 biz 冻结单则只补可提现差额不重复冻（极端重入）
|
||
if freeze_on and frozen > 0:
|
||
existed = FundFreezeLedger.query.filter(
|
||
club_id=cid, role=role, source_type=source_type, biz_id=biz_id,
|
||
).first()
|
||
if existed:
|
||
logger.warning(
|
||
'fund_freeze idempotent hit club=%s role=%s source=%s biz=%s — 不再入账',
|
||
cid, role, source_type, biz_id,
|
||
)
|
||
return {
|
||
'available': Decimal('0.00'),
|
||
'frozen': Decimal('0.00'),
|
||
'ledger_id': existed.id,
|
||
'skipped_freeze': True,
|
||
'idempotent': True,
|
||
}
|
||
|
||
_add_withdrawable(role, profile, available, also_zonge=also_credit_zonge)
|
||
ledger_id = None
|
||
if frozen > 0:
|
||
_add_dongjie(profile, frozen)
|
||
delay = _first_step_delay_days(cfg)
|
||
next_at = timezone.now() + timedelta(days=delay) if delay else timezone.now()
|
||
ledger = FundFreezeLedger.query.create(
|
||
club_id=cid,
|
||
user_id=str(user_id),
|
||
role=role,
|
||
amount_total=frozen,
|
||
amount_unfrozen=Decimal('0.00'),
|
||
status=FundFreezeLedger.STATUS_FROZEN,
|
||
source_type=source_type,
|
||
order_id=(order_id or '')[:64],
|
||
czjilu_id=(czjilu_id or '')[:64],
|
||
biz_id=biz_id[:64],
|
||
freeze_reason=(
|
||
freeze_reason
|
||
or (getattr(cfg, 'remark', None) if cfg else None)
|
||
or f'{source_type}冻结'
|
||
)[:255],
|
||
config_snapshot=_config_snapshot(cfg),
|
||
current_step=1,
|
||
next_eligible_at=next_at,
|
||
)
|
||
ledger_id = ledger.id
|
||
|
||
return {
|
||
'available': available,
|
||
'frozen': frozen,
|
||
'ledger_id': ledger_id,
|
||
'skipped_freeze': frozen <= 0,
|
||
}
|
||
|
||
|
||
def _load_profile(role: str, user_id: str):
|
||
from users.models import User, UserDashou, UserGuanshi, UserZuzhang, UserShenheguan
|
||
user = User.query.filter(UserUID=user_id).first()
|
||
if not user:
|
||
return None
|
||
mapping = {
|
||
ROLE_DASHOU: UserDashou,
|
||
ROLE_GUANSHI: UserGuanshi,
|
||
ROLE_ZUZHANG: UserZuzhang,
|
||
ROLE_SHENHEGUAN: UserShenheguan,
|
||
}
|
||
model = mapping.get(role)
|
||
if not model:
|
||
return None
|
||
return model.query.filter(user=user).first()
|
||
|
||
|
||
def split_preview(club_id: Optional[str], role: str, amount, source_type: str = SOURCE_ORDER_SETTLE) -> Tuple[Decimal, Decimal]:
|
||
"""只算不写库：后台试算用。"""
|
||
credit = _money(amount)
|
||
cfg = get_freeze_config(club_id, role)
|
||
if not cfg or not cfg.enabled or not _source_allowed(cfg, source_type):
|
||
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]:
|
||
"""
|
||
执行一笔冻结单的当前分期步。
|
||
force=True：后台手动，跳过时间/条件门槛。
|
||
条件未达标时不推进步数、不改 next_eligible_at，供定时任务下次重试。
|
||
"""
|
||
from django.db import transaction
|
||
from users.models import UserDashou, UserGuanshi, UserZuzhang, UserShenheguan
|
||
from jituan.models import FundUnfreezeEvent
|
||
from jituan.services.fund_freeze_conditions import evaluate_unfreeze_conditions
|
||
|
||
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()
|
||
snap = ledger.config_snapshot or {}
|
||
trigger_mode = (snap.get('unfreeze_trigger') or 'time_only')
|
||
min_days = max(0, int(snap.get('freeze_min_days') or 0))
|
||
frozen_at = ledger.frozen_at or now
|
||
min_days_ok = force or (now >= frozen_at + timedelta(days=min_days))
|
||
time_ok = force or (
|
||
not ledger.next_eligible_at or ledger.next_eligible_at <= now
|
||
)
|
||
|
||
if trigger_mode == 'manual_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, 'need_condition': False}
|
||
else:
|
||
step = steps[step_idx]
|
||
need_step_cond = bool(step.get('need_condition'))
|
||
|
||
cond_ok = True
|
||
cond_snap: Dict[str, Any] = {}
|
||
need_eval = (
|
||
not force
|
||
and (
|
||
need_step_cond
|
||
or trigger_mode in (
|
||
'condition_only', 'time_and_condition', 'time_or_condition',
|
||
)
|
||
)
|
||
)
|
||
|
||
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': '角色扩展表不存在'}
|
||
|
||
if need_eval:
|
||
cond_ok, cond_snap = evaluate_unfreeze_conditions(
|
||
ledger.club_id,
|
||
ledger.user_id,
|
||
ledger.role,
|
||
snap.get('condition_json') or {},
|
||
profile=profile,
|
||
)
|
||
|
||
if not force:
|
||
if trigger_mode == 'time_only':
|
||
if not time_ok:
|
||
return {'ok': False, 'msg': '未到解冻时间', 'next_eligible_at': ledger.next_eligible_at}
|
||
elif trigger_mode == 'condition_only':
|
||
if not min_days_ok:
|
||
return {'ok': False, 'msg': f'未满最短冻结天数({min_days}天)'}
|
||
if not cond_ok:
|
||
return {
|
||
'ok': False,
|
||
'msg': '条件未达标',
|
||
'condition': cond_snap,
|
||
'retry': True,
|
||
}
|
||
elif trigger_mode == 'time_and_condition':
|
||
if not time_ok:
|
||
return {'ok': False, 'msg': '未到解冻时间', 'next_eligible_at': ledger.next_eligible_at}
|
||
if not cond_ok:
|
||
return {
|
||
'ok': False,
|
||
'msg': '已到期但条件未达标，等待达标后自动解',
|
||
'condition': cond_snap,
|
||
'retry': True,
|
||
}
|
||
elif trigger_mode == 'time_or_condition':
|
||
if not (time_ok or cond_ok):
|
||
return {
|
||
'ok': False,
|
||
'msg': '未到期且条件未达标',
|
||
'condition': cond_snap,
|
||
'retry': True,
|
||
}
|
||
|
||
# 分期单步要求达标时，即使全局是 time_only / time_or 已到期，也仍需条件
|
||
if need_step_cond and not cond_ok:
|
||
return {
|
||
'ok': False,
|
||
'msg': '本分期步要求条件达标',
|
||
'condition': cond_snap,
|
||
'retry': True,
|
||
}
|
||
|
||
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}
|
||
|
||
# 冻结池不足则按实际可解
|
||
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)
|
||
ledger.next_eligible_at = frozen_at + timedelta(days=days)
|
||
else:
|
||
# 无更多配置步：剩余立即进入可扫状态，下一步按 remain 吃完
|
||
ledger.next_eligible_at = now
|
||
|
||
ledger.save(update_fields=[
|
||
'amount_unfrozen', 'status', 'current_step', 'next_eligible_at', 'UpdateTime',
|
||
])
|
||
|
||
event_trigger = trigger
|
||
if not force and trigger_mode in ('condition_only', 'time_or_condition') and cond_ok and not time_ok:
|
||
event_trigger = 'condition'
|
||
|
||
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=event_trigger,
|
||
reason=(reason or f'{event_trigger}解冻')[:255],
|
||
operator_id=(operator_id or '')[:32],
|
||
condition_snapshot=cond_snap or {},
|
||
)
|
||
|
||
return {
|
||
'ok': True,
|
||
'amount': amount,
|
||
'remain': max(remain, Decimal('0.00')),
|
||
'done': remain <= 0,
|
||
'ledger_id': ledger.id,
|
||
'condition': cond_snap,
|
||
}
|
||
|
||
|
||
def scan_due_freeze_ledgers(limit: int = 200) -> Dict[str, Any]:
|
||
"""
|
||
定时任务入口。
|
||
1) 到期（next_eligible_at<=now）的冻结中/部分解冻单
|
||
2) 条件触发类：condition_only / time_or_condition —— 即使未到期也扫描，避免「条件先达标却要干等到期」
|
||
临界：条件未达标只记 fail+retry，不改账本；下次 beat 再试。
|
||
"""
|
||
from django.db.models import Q
|
||
|
||
now = timezone.now()
|
||
base = FundFreezeLedger.query.filter(
|
||
status__in=[FundFreezeLedger.STATUS_FROZEN, FundFreezeLedger.STATUS_PARTIAL],
|
||
)
|
||
due_ids = list(
|
||
base.filter(next_eligible_at__lte=now)
|
||
.order_by('next_eligible_at', 'id')
|
||
.values_list('id', flat=True)[:limit]
|
||
)
|
||
# JSON 字段路径：MySQL/Postgres Django JSONField 支持 key 查询
|
||
cond_ids = []
|
||
remain = max(0, limit - len(due_ids))
|
||
if remain > 0:
|
||
try:
|
||
cond_ids = list(
|
||
base.filter(
|
||
Q(config_snapshot__unfreeze_trigger='condition_only')
|
||
| Q(config_snapshot__unfreeze_trigger='time_or_condition')
|
||
)
|
||
.exclude(id__in=due_ids)
|
||
.order_by('id')
|
||
.values_list('id', flat=True)[:remain]
|
||
)
|
||
except Exception as e:
|
||
logger.warning('scan condition branch skipped: %s', e)
|
||
|
||
ids = due_ids + cond_ids
|
||
ok_n = fail_n = retry_n = 0
|
||
for lid in ids:
|
||
try:
|
||
r = apply_unfreeze_step(lid, trigger='schedule')
|
||
if r.get('ok'):
|
||
ok_n += 1
|
||
else:
|
||
fail_n += 1
|
||
if r.get('retry'):
|
||
retry_n += 1
|
||
logger.info('unfreeze skip id=%s msg=%s', lid, r.get('msg'))
|
||
except Exception as e:
|
||
fail_n += 1
|
||
logger.exception('unfreeze error id=%s: %s', lid, e)
|
||
return {
|
||
'ok': ok_n,
|
||
'fail': fail_n,
|
||
'retry': retry_n,
|
||
'scanned': len(ids),
|
||
'due': len(due_ids),
|
||
'cond_scan': len(cond_ids),
|
||
}
|