fix: undo_repair_wechat_income 反向撤销重复记账

jituan/services/szjilu_accounting.py

@@ -58,6 +58,150 @@ def apply_szjilu_income(amount, club_id=None):
         szjilu.save()
 
 
+def subtract_szjilu_income(amount, club_id=None, subtract_daily_flow=True):
+    """扣减 szjilu 收入（repair 反向操作用）。"""
+    jine = Decimal(str(amount))
+    with transaction.atomic():
+        szjilu = _get_szjilu_for_update(club_id)
+        szjilu.TotalIncome = max(_ZERO, (szjilu.TotalIncome or _ZERO) - jine)
+        szjilu.TotalFlow = max(_ZERO, (szjilu.TotalFlow or _ZERO) - jine)
+        if subtract_daily_flow:
+            szjilu.DailyFlow = max(_ZERO, (szjilu.DailyFlow or _ZERO) - jine)
+        szjilu.save()
+
+
+# repair_wechat_income 原始扫描逻辑（用于反向撤销，一字不改）
+_REPAIR_ORDER_STATUSES = [1, 2, 3, 4, 5, 6, 7, 8]
+
+
+def collect_repair_wechat_income_scan(stat_date):
+    """
+    与已禁用的 repair_wechat_income 命令完全相同的扫描列表。
+    返回 [(biz_ref, club_id, amount, kind), ...]
+    """
+    from datetime import datetime, timedelta
+
+    from django.db.models import Q
+
+    from orders.models import Order
+    from products.models import Czjilu
+
+    day_start = datetime.combine(stat_date, datetime.min.time())
+    day_end = day_start + timedelta(days=1)
+    items = []
+
+    orders = Order.query.filter(
+        CreateTime__gte=day_start,
+        CreateTime__lt=day_end,
+        Status__in=_REPAIR_ORDER_STATUSES,
+    )
+    for od in orders:
+        cid = normalize_szjilu_club_id(getattr(od, 'ClubID', None))
+        items.append((f'order:{od.OrderID}', cid, Decimal(str(od.Amount or 0)), 'order'))
+
+    cz_orders = Czjilu.query.filter(
+        CreateTime__gte=day_start,
+        CreateTime__lt=day_end,
+        zhuangtai=3,
+    ).filter(
+        Q(leixing__in=[1, 2, 3, 4]) | Q(leixing__isnull=False),
+    )
+    for cz in cz_orders:
+        cid = normalize_szjilu_club_id(getattr(cz, 'club_id', None))
+        items.append((f'cz:{cz.dingdan_id}', cid, Decimal(str(cz.jine or 0)), f'cz_{cz.leixing}'))
+
+    return items
+
+
+def undo_repair_wechat_income_one(amount, club_id, biz_ref, inflation_date):
+    """
+    反向执行 repair 的 _repair_one：
+    - 扣减当日 daily_income_stat（repair 当时 update_daily_income 用的是执行日=inflation_date）
+    - 扣减 szjilu
+    - 删除 platform_income_log（repair 若写过）
+    若 repair 曾走「删日志再记」分支，会多记一次；用 had_log 判断扣两次。
+    """
+    from orders.utils import subtract_daily_income
+
+    ref = (biz_ref or '').strip()
+    jine = Decimal(str(amount))
+    cid = normalize_szjilu_club_id(club_id)
+
+    had_log = income_log_exists(ref)
+    subtract_daily_income(jine, cid, inflation_date)
+    subtract_szjilu_income(jine, cid, subtract_daily_flow=(inflation_date == date.today()))
+
+    if had_log:
+        subtract_daily_income(jine, cid, inflation_date)
+        subtract_szjilu_income(jine, cid, subtract_daily_flow=(inflation_date == date.today()))
+
+    try:
+        PlatformIncomeLog.objects.filter(biz_ref=ref).delete()
+    except _PLATFORM_LOG_ERRORS:
+        pass
+
+
+def undo_repair_wechat_income_period(start_date, end_date, inflation_date, dry_run=True):
+    """
+    撤销 repair_wechat_income 在 [start_date, end_date] 扫描范围内造成的入账。
+    inflation_date = 你执行 repair 那天（repair 把所有金额都记到执行日的 daily 里）。
+    """
+    from datetime import timedelta
+
+    from django.db.models import Sum
+
+    all_items = []
+    d = start_date
+    while d <= end_date:
+        all_items.extend(collect_repair_wechat_income_scan(d))
+        d += timedelta(days=1)
+
+    total_amt = sum((x[2] for x in all_items), _ZERO)
+    before_daily = DailyIncomeStat.objects.filter(date=inflation_date).aggregate(
+        amt=Sum('total_amount'), cnt=Sum('total_count'),
+    )
+
+    result = {
+        'start': str(start_date),
+        'end': str(end_date),
+        'inflation_date': str(inflation_date),
+        'undo_count': len(all_items),
+        'undo_amount': float(total_amt),
+        'before_daily_on_inflation_date': float(before_daily['amt'] or 0),
+        'items': all_items,
+        'dry_run': dry_run,
+    }
+
+    if dry_run:
+        return result
+
+    with transaction.atomic():
+        for ref, cid, jine, _kind in all_items:
+            undo_repair_wechat_income_one(jine, cid, ref, inflation_date)
+
+    after_daily = DailyIncomeStat.objects.filter(date=inflation_date).aggregate(
+        amt=Sum('total_amount'), cnt=Sum('total_count'),
+    )
+    result['after_daily_on_inflation_date'] = float(after_daily['amt'] or 0)
+    return result
+
+
+def zero_all_income_statistics():
+    """核弹：收支统计全部清零（日收入表、幂等日志、szjilu 收入/流水）。"""
+    with transaction.atomic():
+        try:
+            PlatformIncomeLog.objects.all().delete()
+        except _PLATFORM_LOG_ERRORS:
+            pass
+        DailyIncomeStat.objects.all().delete()
+        for szjilu in Szjilu.objects.select_for_update().all():
+            szjilu.TotalIncome = _ZERO
+            szjilu.TotalFlow = _ZERO
+            szjilu.DailyFlow = _ZERO
+            szjilu.save(update_fields=['TotalIncome', 'TotalFlow', 'DailyFlow', 'UpdateTime'])
+    return True
+
+
 def _apply_income_stats(amount, club_id):
     """写入 daily_income_stat + szjilu（仅在新 platform_income_log 落库后调用）。"""
     from orders.utils import update_daily_income
@@ -206,15 +350,11 @@ _CZ_WECHAT_LEIXING = [1, 2, 3, 4, 5]
 
 def collect_canonical_wechat_income_for_date(stat_date, existing_refs=None):
     """
-    收集当日真实微信收入（每笔 biz_ref 全局唯一，不重复计已入账订单）。
-    - 订单：当日创建且已付款，或当日完成微信支付（有微信交易号）
-    - 充值：当日创建且 zhuangtai=3
-    - 不含退款单（Status=5）
-  """
+    收集当日真实微信收入（严格：仅统计当日下单且已支付成功的单据，每笔唯一）。
+    不含退款；不按 UpdateTime 计老单（repair 虚高的主要原因之一）。
+    """
     from datetime import datetime, timedelta
 
-    from django.db.models import Q
-
     from orders.models import Order
     from products.models import Czjilu
 
@@ -235,17 +375,11 @@ def collect_canonical_wechat_income_for_date(stat_date, existing_refs=None):
         seen[ref] = True
         items.append((ref, cid, jine, kind))
 
-    order_qs = Order.query.filter(
+    for od in Order.query.filter(
+        CreateTime__gte=day_start,
+        CreateTime__lt=day_end,
         Status__in=_ORDER_INCOME_STATUSES,
-    ).filter(
-        Q(CreateTime__gte=day_start, CreateTime__lt=day_end)
-        | Q(
-            UpdateTime__gte=day_start,
-            UpdateTime__lt=day_end,
-            WechatTransactionID__isnull=False,
-        ),
-    )
-    for od in order_qs:
+    ):
         txn = (getattr(od, 'WechatTransactionID', None) or '').strip()
         if not txn:
             continue
@@ -262,6 +396,141 @@ def collect_canonical_wechat_income_for_date(stat_date, existing_refs=None):
     return items
 
 
+def dedupe_platform_income_logs():
+    """删除 biz_ref 重复的幂等日志（保留最早一条）。"""
+    try:
+        seen = {}
+        dup_ids = []
+        for log in PlatformIncomeLog.objects.order_by('id'):
+            key = (log.biz_ref or '').strip()
+            if not key:
+                dup_ids.append(log.id)
+                continue
+            if key in seen:
+                dup_ids.append(log.id)
+            else:
+                seen[key] = log.id
+        if dup_ids:
+            PlatformIncomeLog.objects.filter(id__in=dup_ids).delete()
+        return len(dup_ids)
+    except _PLATFORM_LOG_ERRORS:
+        return 0
+
+
+def fix_income_period(start_date, end_date, dry_run=True):
+    """
+    核弹级修复：删掉区间内错误日志与日统计，按严格规则重建，再对齐 szjilu。
+    daily_income_stat 只信 platform_income_log，不再累加脏数据。
+    """
+    from datetime import datetime, time as dt_time, timedelta
+
+    from django.db.models import Sum
+    from django.utils import timezone
+
+    before_daily = DailyIncomeStat.objects.filter(
+        date__gte=start_date, date__lte=end_date,
+    ).aggregate(amt=Sum('total_amount'), cnt=Sum('total_count'))
+
+    before_log = _ZERO
+    before_log_cnt = 0
+    try:
+        before_log_cnt = PlatformIncomeLog.objects.filter(
+            CreateTime__date__gte=start_date,
+            CreateTime__date__lte=end_date,
+        ).count()
+        agg = PlatformIncomeLog.objects.filter(
+            CreateTime__date__gte=start_date,
+            CreateTime__date__lte=end_date,
+        ).aggregate(t=Sum('amount'))
+        before_log = agg['t'] or _ZERO
+    except _PLATFORM_LOG_ERRORS:
+        pass
+
+    all_items = _collect_items_for_period(start_date, end_date)
+    rebuild_amt = sum((x[3] for x in all_items), _ZERO)
+    result = {
+        'start': str(start_date),
+        'end': str(end_date),
+        'before_daily_amount': float(before_daily['amt'] or 0),
+        'before_daily_count': int(before_daily['cnt'] or 0),
+        'before_log_amount': float(before_log),
+        'before_log_count': before_log_cnt,
+        'after_amount': float(rebuild_amt),
+        'after_count': len(all_items),
+        'items': all_items,
+        'dry_run': dry_run,
+    }
+
+    if dry_run:
+        return result
+
+    execute_fix_income_period(start_date, end_date)
+    return result
+
+
+def _collect_items_for_period(start_date, end_date):
+    from datetime import timedelta
+
+    items = []
+    d = start_date
+    while d <= end_date:
+        for ref, cid, jine, kind in collect_canonical_wechat_income_for_date(d, existing_refs=set()):
+            items.append((d, ref, cid, jine, kind))
+        d += timedelta(days=1)
+    return items
+
+
+def execute_fix_income_period(start_date, end_date):
+    from datetime import datetime, time as dt_time, timedelta
+
+    from django.utils import timezone
+
+    dedupe_platform_income_logs()
+    items = _collect_items_for_period(start_date, end_date)
+
+    with transaction.atomic():
+        PlatformIncomeLog.objects.filter(
+            CreateTime__date__gte=start_date,
+            CreateTime__date__lte=end_date,
+        ).delete()
+        DailyIncomeStat.objects.filter(
+            date__gte=start_date, date__lte=end_date,
+        ).delete()
+
+        by_day_club = {}
+        for stat_date, ref, cid, jine, _kind in items:
+            pay_ts = timezone.make_aware(datetime.combine(stat_date, dt_time(12, 0, 0)))
+            log = PlatformIncomeLog.objects.create(biz_ref=ref, club_id=cid, amount=jine)
+            PlatformIncomeLog.objects.filter(pk=log.pk).update(CreateTime=pay_ts)
+            key = (stat_date, cid)
+            if key not in by_day_club:
+                by_day_club[key] = {'amt': _ZERO, 'cnt': 0}
+            by_day_club[key]['amt'] += jine
+            by_day_club[key]['cnt'] += 1
+
+        for (stat_date, cid), agg in by_day_club.items():
+            DailyIncomeStat.objects.create(
+                date=stat_date,
+                club_id=cid,
+                year=stat_date.year,
+                month=stat_date.month,
+                day=stat_date.day,
+                total_amount=agg['amt'],
+                total_count=agg['cnt'],
+            )
+
+    sync_szjilu_income_from_logs()
+    return items
+
+
+def align_daily_income_from_logs_only():
+    """不重扫订单：仅按现有 platform_income_log 重写 daily_income_stat + szjilu（日志对、日统计错时用）。"""
+    dedupe_platform_income_logs()
+    n = rebuild_all_daily_income_from_logs()
+    sync_szjilu_income_from_logs()
+    return n
+
+
 def _existing_refs_outside_date(stat_date):
     """该日之外已入账的 biz_ref（删当日错误日志后仍保留的）。"""
     try: