fix: 收入恢复命令修正

config/management/commands/recover_wechat_income_stats.py

@@ -1,38 +1,39 @@
 """
 恢复被 repair_wechat_income 搞乱的收入统计。
 
-repair 会对已入账订单重复加钱（200 变 900 就是这个）。
+repair 会对已入账订单重复加钱（200 变 900）。
-本命令：删掉当日错误日志 → 按真实微信支付重建 → 重算日统计和 szjilu。
+本命令：删当日错误日志 → 按真实微信支付重建 → 扣回 szjilu 多记部分。
 
-先不加 --yes 可预览重建后金额。
+用法：
+  1. 先预览：python manage.py recover_wechat_income_stats
+  2. 看「重建后」金额是否对（应接近真实收入，如 200 左右）
+  3. 确认后：python manage.py recover_wechat_income_stats --yes
+  4. 若多日错乱：python manage.py recover_wechat_income_stats --days 7 --yes
+  5. 最后全量对齐日统计表：python manage.py recover_wechat_income_stats --yes --rebuild-all-daily
 """
 from datetime import date, timedelta
 
 from django.core.management.base import BaseCommand
+from django.db.models import Sum
 
-from jituan.services.szjilu_accounting import recover_income_stats_for_date
+from config.models import DailyIncomeStat
+from jituan.services.szjilu_accounting import (
+    rebuild_all_daily_income_from_logs,
+    recover_income_stats_for_date,
+)
 
 
 class Command(BaseCommand):
-    help = '恢复 repair_wechat_income 造成的重复记账（先预览，再加 --yes 执行）'
+    help = '恢复 repair_wechat_income 重复记账（先预览，--yes 执行）'
 
     def add_arguments(self, parser):
+        parser.add_argument('--date', type=str, default='', help='日期 YYYY-MM-DD，默认今天')
+        parser.add_argument('--days', type=int, default=1, help='向前恢复几天')
+        parser.add_argument('--yes', action='store_true', help='确认执行')
         parser.add_argument(
-            '--date',
+            '--rebuild-all-daily',
-            type=str,
-            default='',
-            help='恢复日期 YYYY-MM-DD，默认今天',
-        )
-        parser.add_argument(
-            '--days',
-            type=int,
-            default=1,
-            help='从指定日期起向前恢复几天，默认 1',
-        )
-        parser.add_argument(
-            '--yes',
             action='store_true',
-            help='确认执行（不加则仅预览，不改数据库）',
+            help='恢复后按 platform_income_log 全量重写 daily_income_stat（纠正累计总收入）',
         )
 
     def handle(self, *args, **options):
@@ -46,40 +47,46 @@ class Command(BaseCommand):
 
         if dry_run:
             self.stdout.write(self.style.WARNING(
-                '【预览模式】未改数据库。确认金额无误后加 --yes 执行。\n'
+                '【预览】未改数据库。下面「重建后」应是正确金额。\n'
             ))
         else:
-            self.stdout.write(self.style.WARNING('【执行模式】正在恢复数据...\n'))
+            self.stdout.write(self.style.WARNING('【执行】正在恢复...\n'))
 
         for i in range(days):
             d = start_day + timedelta(days=i)
             r = recover_income_stats_for_date(d, dry_run=dry_run)
-            self.stdout.write(f"=== {r['date']} ===")
+            self.stdout.write(f'=== {r["date"]} ===')
             self.stdout.write(
-                f"  当前日统计（错乱）: {r['before_daily_amount']:.2f} 元 / {r['before_daily_count']} 笔"
+                f'  错乱前日统计: {r["before_daily_amount"]:.2f} 元 / {r["before_daily_count"]} 笔'
             )
             self.stdout.write(
-                f"  当前幂等日志:       {r['before_log_amount']:.2f} 元 / {r['before_log_count']} 笔"
+                f'  当日幂等日志: {r["before_log_amount"]:.2f} 元 / {r["before_log_count"]} 笔'
             )
             self.stdout.write(self.style.SUCCESS(
-                f"  重建后（真实支付）: {r['after_amount']:.2f} 元 / {r['after_count']} 笔"
+                f'  重建后(正确值): {r["after_amount"]:.2f} 元 / {r["after_count"]} 笔'
             ))
-            if r['items'] and dry_run:
+            if r['items']:
-                self.stdout.write('  明细（前20笔）:')
+                self.stdout.write('  明细:')
-                for ref, cid, amt, kind in r['items'][:20]:
+                for ref, cid, amt, kind in r['items']:
                     self.stdout.write(f'    {ref}  {amt}元  club={cid}  ({kind})')
-                if len(r['items']) > 20:
-                    self.stdout.write(f'    ... 共 {len(r["items"])} 笔')
 
-        if dry_run:
+        if not dry_run and options['rebuild_all_daily']:
-            self.stdout.write(self.style.WARNING(
+            n = rebuild_all_daily_income_from_logs()
-                '\n预览完成。若重建后金额正确，执行：\n'
+            self.stdout.write(self.style.SUCCESS(f'已按日志全量重写 daily_income_stat: {n} 行'))
-                '  python manage.py recover_wechat_income_stats --yes\n'
+
-                '指定日期：\n'
+        if not dry_run:
-                '  python manage.py recover_wechat_income_stats --date 2026-06-24 --yes'
+            agg = DailyIncomeStat.objects.filter(date=end_day).aggregate(
-            ))
+                amt=Sum('total_amount'),
-        else:
+                cnt=Sum('total_count'),
+            )
             self.stdout.write(self.style.SUCCESS(
-                '\n恢复完成。请刷新后台财务页核对今日收入。'
+                f'\n数据库今日日统计现为: {float(agg["amt"] or 0):.2f} 元 / {int(agg["cnt"] or 0)} 笔'
-                '\n勿再执行 repair_wechat_income（已禁用）。'
+            ))
+            self.stdout.write(self.style.SUCCESS('恢复完成，请刷新后台财务页。'))
+        else:
+            self.stdout.write(self.style.WARNING(
+                '\n预览完成。重建后金额对的话执行：\n'
+                '  python manage.py recover_wechat_income_stats --yes\n'
+                '累计总收入仍不对再加：\n'
+                '  python manage.py recover_wechat_income_stats --yes --rebuild-all-daily'
             ))

jituan/services/szjilu_accounting.py

@@ -204,24 +204,29 @@ _ORDER_INCOME_STATUSES = [1, 2, 3, 4, 6, 7, 8]
 _CZ_WECHAT_LEIXING = [1, 2, 3, 4, 5]
 
 
-def collect_canonical_wechat_income_for_date(stat_date):
+def collect_canonical_wechat_income_for_date(stat_date, existing_refs=None):
-    """
-    按真实支付时间（UpdateTime）收集当日微信收入，每笔 biz_ref 唯一。
-    返回 [(biz_ref, club_id, amount, kind), ...]
     """
+    收集当日真实微信收入（每笔 biz_ref 全局唯一，不重复计已入账订单）。
+    - 订单：当日创建且已付款，或当日完成微信支付（有微信交易号）
+    - 充值：当日创建且 zhuangtai=3
+    - 不含退款单（Status=5）
+  """
     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)
+    skip_refs = existing_refs or set()
     seen = {}
     items = []
 
     def _add(ref, club_id, amount, kind):
         ref = (ref or '').strip()
-        if not ref or ref in seen:
+        if not ref or ref in seen or ref in skip_refs:
             return
         jine = Decimal(str(amount or 0))
         if jine <= 0:
@@ -230,16 +235,25 @@ def collect_canonical_wechat_income_for_date(stat_date):
         seen[ref] = True
         items.append((ref, cid, jine, kind))
 
-    for od in Order.query.filter(
+    order_qs = Order.query.filter(
-        UpdateTime__gte=day_start,
-        UpdateTime__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
         _add(f'order:{od.OrderID}', getattr(od, 'ClubID', None), od.Amount, 'order')
 
     for cz in Czjilu.query.filter(
-        UpdateTime__gte=day_start,
+        CreateTime__gte=day_start,
-        UpdateTime__lt=day_end,
+        CreateTime__lt=day_end,
         zhuangtai=3,
         leixing__in=_CZ_WECHAT_LEIXING,
     ):
@@ -248,6 +262,53 @@ def collect_canonical_wechat_income_for_date(stat_date):
     return items
 
 
+def _existing_refs_outside_date(stat_date):
+    """该日之外已入账的 biz_ref（删当日错误日志后仍保留的）。"""
+    try:
+        return set(
+            PlatformIncomeLog.objects.exclude(
+                CreateTime__date=stat_date,
+            ).values_list('biz_ref', flat=True)
+        )
+    except _PLATFORM_LOG_ERRORS:
+        return set()
+
+
+def rebuild_all_daily_income_from_logs():
+    """按 platform_income_log 全量重写 daily_income_stat（日志为唯一准绳）。"""
+    from django.db.models.functions import TruncDate
+
+    try:
+        grouped = (
+            PlatformIncomeLog.objects.annotate(log_date=TruncDate('CreateTime'))
+            .values('log_date', 'club_id')
+            .annotate(total_amount=Sum('amount'), total_count=Count('id'))
+        )
+    except _PLATFORM_LOG_ERRORS as exc:
+        logger.error('platform_income_log 不可用: %s', exc)
+        return 0
+
+    with transaction.atomic():
+        DailyIncomeStat.objects.all().delete()
+        n = 0
+        for row in grouped:
+            d = row['log_date']
+            if not d:
+                continue
+            cid = normalize_szjilu_club_id(row['club_id'])
+            DailyIncomeStat.objects.create(
+                date=d,
+                club_id=cid,
+                year=d.year,
+                month=d.month,
+                day=d.day,
+                total_amount=row['total_amount'] or _ZERO,
+                total_count=int(row['total_count'] or 0),
+            )
+            n += 1
+    return n
+
+
 def recover_income_stats_for_date(stat_date, dry_run=True):
     """
     撤销 repair_wechat_income 造成的重复记账：
@@ -283,7 +344,8 @@ def recover_income_stats_for_date(stat_date, dry_run=True):
     except _PLATFORM_LOG_ERRORS:
         pass
 
-    items = collect_canonical_wechat_income_for_date(stat_date)
+    existing_refs = _existing_refs_outside_date(stat_date)
+    items = collect_canonical_wechat_income_for_date(stat_date, existing_refs=existing_refs)
     rebuild_amt = sum((x[2] for x in items), _ZERO)
     rebuild_cnt = len(items)
 
@@ -316,6 +378,9 @@ def recover_income_stats_for_date(stat_date, dry_run=True):
 
         DailyIncomeStat.objects.filter(date=stat_date).delete()
 
+        existing_refs = _existing_refs_outside_date(stat_date)
+        items = collect_canonical_wechat_income_for_date(stat_date, existing_refs=existing_refs)
+
         for ref, cid, jine, _kind in items:
             log = PlatformIncomeLog.objects.create(
                 biz_ref=ref,