财务恢复：只读日统计表，废弃szjilu，新增restore_finance_stats

jituan/services/caiwu_stats.py

@@ -1,7 +1,5 @@
 """俱乐部维度财务统计（供 /jituan/houtai/caiwu，旧 /houtai/caiwu 不变）。"""
-from calendar import monthrange
 from datetime import date, datetime, timedelta
-from decimal import Decimal
 
 from django.db.models import Sum
 
@@ -12,13 +10,8 @@ from jituan.services.club_context import (
     resolve_club_scope,
 )
 from jituan.constants import DATA_SCOPE_ALL
-from jituan.services.live_income_stats import (
-    income_daily_stats_last_days,
-    income_for_date,
-    income_total_all_time,
-)
 
-from config.models import DailyPayoutStat, Szjilu
+from config.models import DailyIncomeStat, DailyPayoutStat
 from orders.models import Order
 from products.models import Czjilu, Huiyuangoumai
 from users.models import UserDashou, UserGuanshi, UserShangjia, UserZuzhang
@@ -38,27 +31,6 @@ def _sum_role_balance(profile_model, balance_field, request):
     )
 
 
-def _build_szjilu_payload(request):
-    """全局收支流水表 szjilu（按俱乐部或集团汇总）。"""
-    if resolve_club_scope(request) == DATA_SCOPE_ALL:
-        agg = Szjilu.query.aggregate(
-            zongshouyi=Sum('TotalIncome'),
-            zongliushui=Sum('TotalFlow'),
-            zongzhichu=Sum('TotalExpense'),
-            jrls=Sum('DailyFlow'),
-            jrzc=Sum('DailyExpense'),
-        )
-        return {
-            'zongliushui': float(agg['zongliushui'] or 0),
-            'zongshouyi': float(agg['zongshouyi'] or 0),
-            'zongzhichu': float(agg['zongzhichu'] or 0),
-            'jrls': float(agg['jrls'] or 0),
-            'jrzc': float(agg['jrzc'] or 0),
-        }
-    from jituan.services.szjilu_accounting import get_szjilu_snapshot
-    return get_szjilu_snapshot(resolve_club_id_from_request(request))
-
-
 def _order_qs(request):
     return filter_queryset_by_club(Order.query.all(), request, club_field='ClubID')
 
@@ -71,12 +43,16 @@ def _huiyuangoumai_qs(request):
     return filter_club_char_field(Huiyuangoumai.query.all(), request, field='club_id')
 
 
+def _daily_income_qs(request):
+    return filter_club_char_field(DailyIncomeStat.query.all(), request, field='club_id')
+
+
 def _daily_payout_qs(request):
     return filter_club_char_field(DailyPayoutStat.query.all(), request, field='club_id')
 
 
 def build_caiwu_payload(request):
-    """与 CaiwuView 返回结构一致，但按俱乐部过滤订单/充值流水。"""
+    """与 CaiwuView 返回结构一致；收入/支出只读 daily_income_stat、daily_payout_stat。"""
     today = date.today()
     today_start = datetime.combine(today, datetime.min.time())
     today_end = today_start + timedelta(days=1)
@@ -85,11 +61,19 @@ def build_caiwu_payload(request):
     order_qs = _order_qs(request)
     cz_qs = _czjilu_qs(request)
     hy_qs = _huiyuangoumai_qs(request)
+    income_qs = _daily_income_qs(request)
     payout_qs = _daily_payout_qs(request)
 
+    scope = resolve_club_scope(request)
+    today_income_qs = income_qs.filter(date=today)
     today_payout_qs = payout_qs.filter(date=today)
 
-    today_income_amount, today_income_count = income_for_date(today, request)
+    inc_agg = today_income_qs.aggregate(
+        total_amount=Sum('total_amount'),
+        total_count=Sum('total_count'),
+    )
+    today_income_amount = float(inc_agg['total_amount'] or 0.00)
+    today_income_count = int(inc_agg['total_count'] or 0)
 
     pay_agg = today_payout_qs.aggregate(
         total_amount=Sum('total_amount'),
@@ -145,50 +129,69 @@ def build_caiwu_payload(request):
     all_zuzhang_yue = _sum_role_balance(UserZuzhang, 'ketixian_jine', request)
     all_shangjia_yue = _sum_role_balance(UserShangjia, 'yue', request)
 
-    szjilu_payload = _build_szjilu_payload(request)
-
-    total_income, _ = income_total_all_time()
+    total_income = float(
+        DailyIncomeStat.query.aggregate(total=Sum('total_amount'))['total'] or 0.00
+    )
     total_payout = float(
         DailyPayoutStat.query.aggregate(total=Sum('total_amount'))['total'] or 0.00
     )
-    scoped_total_income, _ = income_total_all_time(request)
+    scoped_total_income = float(income_qs.aggregate(total=Sum('total_amount'))['total'] or 0.00)
     scoped_total_payout = float(payout_qs.aggregate(total=Sum('total_amount'))['total'] or 0.00)
     platform_profit = round(scoped_total_income - scoped_total_payout, 2)
 
     daily_stats = []
     since = today - timedelta(days=30)
-    income_by_date = {
-        row['date']: {
-            'income': row['income'],
-            'income_count': row['income_count'],
+    if scope == DATA_SCOPE_ALL:
+        income_by_date = {
+            str(row['date']): {
+                'income': float(row['total_amount'] or 0),
+                'income_count': int(row['total_count'] or 0),
+            }
+            for row in income_qs.filter(date__gte=since).values('date').annotate(
+                total_amount=Sum('total_amount'),
+                total_count=Sum('total_count'),
+            )
         }
-        for row in income_daily_stats_last_days(30, request)
-    }
-    payout_by_date = {
-        str(row['date']): float(row['total_amount'] or 0)
-        for row in payout_qs.filter(date__gte=since).values('date').annotate(
-            total_amount=Sum('total_amount'),
-        )
-    }
-    all_dates = sorted(
-        set(income_by_date.keys()) | set(payout_by_date.keys()),
-        reverse=True,
-    )[:30]
-    for d in all_dates:
-        inc = income_by_date.get(d, {'income': 0.0, 'income_count': 0})
-        payout_amt = payout_by_date.get(d, 0.00)
-        daily_stats.append({
-            'date': d,
-            'income': inc['income'],
-            'income_count': inc['income_count'],
-            'payout': payout_amt,
-            'profit': round(inc['income'] - payout_amt, 2),
-        })
+        payout_by_date = {
+            str(row['date']): float(row['total_amount'] or 0)
+            for row in payout_qs.filter(date__gte=since).values('date').annotate(
+                total_amount=Sum('total_amount'),
+            )
+        }
+        all_dates = sorted(
+            set(income_by_date.keys()) | set(payout_by_date.keys()),
+            reverse=True,
+        )[:30]
+        for d in all_dates:
+            inc = income_by_date.get(d, {'income': 0.0, 'income_count': 0})
+            payout_amt = payout_by_date.get(d, 0.00)
+            daily_stats.append({
+                'date': d,
+                'income': inc['income'],
+                'income_count': inc['income_count'],
+                'payout': payout_amt,
+                'profit': round(inc['income'] - payout_amt, 2),
+            })
+    else:
+        income_list = income_qs.filter(date__gte=since).order_by('-date')[:30]
+        payout_dict = {
+            str(p.date): float(p.total_amount)
+            for p in payout_qs.filter(date__gte=since)
+        }
+        for inc in income_list:
+            d = str(inc.date)
+            payout_amt = payout_dict.get(d, 0.00)
+            daily_stats.append({
+                'date': d,
+                'income': float(inc.total_amount),
+                'income_count': inc.total_count,
+                'payout': payout_amt,
+                'profit': round(float(inc.total_amount) - payout_amt, 2),
+            })
 
     return {
         'club_id': club_id,
         'scope': resolve_club_scope(request),
-        'szjilu': szjilu_payload,
         'today_income': round(today_income_amount, 2),
         'today_income_count': today_income_count,
         'today_payout': round(today_payout_amount, 2),

jituan/services/live_income_stats.py

@@ -1,264 +0,0 @@
-"""
-每日收入：从真实微信支付订单/充值单实时汇总。
-不读 daily_income_stat（该表可被 repair 搞乱，与展示解耦）。
-
-每日支出：仍读 daily_payout_stat（repair 未动支出）。
-收支记录 szjilu：独立表，本模块不负责展示。
-"""
-from calendar import monthrange
-from datetime import date, datetime, timedelta
-from decimal import Decimal
-
-from django.db.models import Count, Q, Sum
-from django.db.models.functions import ExtractYear, TruncDate, TruncMonth
-
-from jituan.constants import CLUB_ID_DEFAULT
-from jituan.services.club_context import (
-    filter_club_char_field,
-    filter_queryset_by_club,
-)
-
-_ORDER_PAID = [1, 2, 3, 4, 6, 7, 8]
-_CZ_PAID_LEIXING = [1, 2, 3, 4, 5]
-_ZERO = Decimal('0.00')
-
-
-def _orders_qs(request=None):
-    from orders.models import Order
-    qs = Order.query.filter(Status__in=_ORDER_PAID)
-    if request is not None:
-        qs = filter_queryset_by_club(qs, request, club_field='ClubID')
-    return qs
-
-
-def _cz_qs(request=None):
-    from products.models import Czjilu
-    qs = Czjilu.query.filter(zhuangtai=3, leixing__in=_CZ_PAID_LEIXING)
-    if request is not None:
-        qs = filter_club_char_field(qs, request, field='club_id')
-    return qs
-
-
-def _merge_amount_count(order_rows, cz_rows, key_field='d'):
-    out = {}
-    for row in order_rows:
-        k = row[key_field]
-        if k is None:
-            continue
-        if k not in out:
-            out[k] = {'amount': _ZERO, 'count': 0}
-        out[k]['amount'] += Decimal(str(row['total'] or 0))
-        out[k]['count'] += int(row['cnt'] or 0)
-    for row in cz_rows:
-        k = row[key_field]
-        if k is None:
-            continue
-        if k not in out:
-            out[k] = {'amount': _ZERO, 'count': 0}
-        out[k]['amount'] += Decimal(str(row['total'] or 0))
-        out[k]['count'] += int(row['cnt'] or 0)
-    return out
-
-
-def income_for_date(stat_date, request=None):
-    """指定日期的微信收入金额与笔数。"""
-    day_start = datetime.combine(stat_date, datetime.min.time())
-    day_end = day_start + timedelta(days=1)
-    o = _orders_qs(request).filter(
-        UpdateTime__gte=day_start, UpdateTime__lt=day_end,
-    ).aggregate(total=Sum('Amount'), cnt=Count('OrderID'))
-    c = _cz_qs(request).filter(
-        UpdateTime__gte=day_start, UpdateTime__lt=day_end,
-    ).aggregate(total=Sum('jine'), cnt=Count('dingdan_id'))
-    amt = Decimal(str(o['total'] or 0)) + Decimal(str(c['total'] or 0))
-    cnt = int(o['cnt'] or 0) + int(c['cnt'] or 0)
-    return float(amt), cnt
-
-
-def income_total_all_time(request=None):
-    """历史累计微信收入。"""
-    o = _orders_qs(request).aggregate(total=Sum('Amount'), cnt=Count('OrderID'))
-    c = _cz_qs(request).aggregate(total=Sum('jine'), cnt=Count('dingdan_id'))
-    amt = Decimal(str(o['total'] or 0)) + Decimal(str(c['total'] or 0))
-    cnt = int(o['cnt'] or 0) + int(c['cnt'] or 0)
-    return float(amt), cnt
-
-
-def income_daily_stats_last_days(days=30, request=None):
-    """近 N 天每日收入（用于财务看板 daily_stats）。"""
-    since = date.today() - timedelta(days=days)
-    since_dt = datetime.combine(since, datetime.min.time())
-
-    o_rows = (
-        _orders_qs(request)
-        .filter(UpdateTime__gte=since_dt)
-        .annotate(d=TruncDate('UpdateTime'))
-        .values('d')
-        .annotate(total=Sum('Amount'), cnt=Count('OrderID'))
-    )
-    c_rows = (
-        _cz_qs(request)
-        .filter(UpdateTime__gte=since_dt)
-        .annotate(d=TruncDate('UpdateTime'))
-        .values('d')
-        .annotate(total=Sum('jine'), cnt=Count('dingdan_id'))
-    )
-    merged = _merge_amount_count(list(o_rows), list(c_rows), 'd')
-    result = []
-    for d, v in sorted(merged.items(), reverse=True)[:days]:
-        result.append({
-            'date': str(d),
-            'income': float(v['amount']),
-            'income_count': v['count'],
-        })
-    return result
-
-
-def income_time_series(granularity, year, month=None, request=None):
-    """
-    收支详细页时间序列（仅收入侧，支出由调用方合并 daily_payout_stat）。
-    granularity: day | month | year
-    """
-    if granularity == 'day':
-        if not year or not month:
-            return {}
-        _, last = monthrange(year, month)
-        start = datetime(year, month, 1)
-        end = datetime(year, month, last) + timedelta(days=1)
-        o_rows = (
-            _orders_qs(request)
-            .filter(UpdateTime__gte=start, UpdateTime__lt=end)
-            .annotate(d=TruncDate('UpdateTime'))
-            .values('d')
-            .annotate(total=Sum('Amount'), cnt=Count('OrderID'))
-        )
-        c_rows = (
-            _cz_qs(request)
-            .filter(UpdateTime__gte=start, UpdateTime__lt=end)
-            .annotate(d=TruncDate('UpdateTime'))
-            .values('d')
-            .annotate(total=Sum('jine'), cnt=Count('dingdan_id'))
-        )
-        merged = _merge_amount_count(list(o_rows), list(c_rows), 'd')
-        out = {}
-        for d, v in merged.items():
-            out[d.day] = {'income': float(v['amount']), 'income_count': v['count']}
-        return out
-
-    if granularity == 'month':
-        start = datetime(year, 1, 1)
-        end = datetime(year + 1, 1, 1)
-        o_rows = (
-            _orders_qs(request)
-            .filter(UpdateTime__gte=start, UpdateTime__lt=end)
-            .annotate(m=TruncMonth('UpdateTime'))
-            .values('m')
-            .annotate(total=Sum('Amount'), cnt=Count('OrderID'))
-        )
-        c_rows = (
-            _cz_qs(request)
-            .filter(UpdateTime__gte=start, UpdateTime__lt=end)
-            .annotate(m=TruncMonth('UpdateTime'))
-            .values('m')
-            .annotate(total=Sum('jine'), cnt=Count('dingdan_id'))
-        )
-        merged = _merge_amount_count(list(o_rows), list(c_rows), 'm')
-        out = {}
-        for m, v in merged.items():
-            out[m.month] = {'income': float(v['amount']), 'income_count': v['count']}
-        return out
-
-    o_rows = (
-        _orders_qs(request)
-        .annotate(y=ExtractYear('UpdateTime'))
-        .values('y')
-        .annotate(total=Sum('Amount'), cnt=Count('OrderID'))
-    )
-    c_rows = (
-        _cz_qs(request)
-        .annotate(y=ExtractYear('UpdateTime'))
-        .values('y')
-        .annotate(total=Sum('jine'), cnt=Count('dingdan_id'))
-    )
-    merged = _merge_amount_count(list(o_rows), list(c_rows), 'y')
-    return {
-        int(k): {'income': float(v['amount']), 'income_count': v['count']}
-        for k, v in merged.items()
-    }
-
-
-def income_summary_filter(granularity, year, month=None, day=None, request=None):
-    """详细页顶部汇总区间的收入。"""
-    if granularity == 'day' and year and month and day:
-        return income_for_date(date(year, month, day), request)
-    if granularity == 'month' and year and month:
-        start = datetime(year, month, 1)
-        if month == 12:
-            end = datetime(year + 1, 1, 1)
-        else:
-            end = datetime(year, month + 1, 1)
-    elif granularity == 'year' and year:
-        start = datetime(year, 1, 1)
-        end = datetime(year + 1, 1, 1)
-    else:
-        return income_total_all_time(request)
-
-    o = _orders_qs(request).filter(
-        UpdateTime__gte=start, UpdateTime__lt=end,
-    ).aggregate(total=Sum('Amount'), cnt=Count('OrderID'))
-    c = _cz_qs(request).filter(
-        UpdateTime__gte=start, UpdateTime__lt=end,
-    ).aggregate(total=Sum('jine'), cnt=Count('dingdan_id'))
-    amt = Decimal(str(o['total'] or 0)) + Decimal(str(c['total'] or 0))
-    cnt = int(o['cnt'] or 0) + int(c['cnt'] or 0)
-    return float(amt), cnt
-
-
-def sync_szjilu_from_live_payments():
-    """按真实支付重算 szjilu 收支记录表（与每日收入展示无关，可单独执行）。"""
-    from config.models import Szjilu
-
-    today = date.today()
-    today_start = datetime.combine(today, datetime.min.time())
-    today_end = today_start + timedelta(days=1)
-
-    clubs = set()
-    for row in _orders_qs().values('ClubID').distinct():
-        clubs.add((row.get('ClubID') or '').strip() or CLUB_ID_DEFAULT)
-    for row in _cz_qs().values('club_id').distinct():
-        clubs.add((row.get('club_id') or '').strip() or CLUB_ID_DEFAULT)
-
-    def _club_orders(cid):
-        qs = _orders_qs()
-        if cid == CLUB_ID_DEFAULT:
-            return qs.filter(Q(ClubID=cid) | Q(ClubID__isnull=True) | Q(ClubID=''))
-        return qs.filter(ClubID=cid)
-
-    updated = 0
-    for cid in clubs:
-        o_total = _club_orders(cid).aggregate(t=Sum('Amount'))['t'] or 0
-        c_total = _cz_qs().filter(club_id=cid).aggregate(t=Sum('jine'))['t'] or 0
-        total = Decimal(str(o_total)) + Decimal(str(c_total))
-
-        o_day = _club_orders(cid).filter(
-            UpdateTime__gte=today_start, UpdateTime__lt=today_end,
-        )
-        o_day_amt = o_day.aggregate(t=Sum('Amount'))['t'] or 0
-        c_day = _cz_qs().filter(
-            club_id=cid, UpdateTime__gte=today_start, UpdateTime__lt=today_end,
-        ).aggregate(t=Sum('jine'))['t'] or 0
-        day_flow = Decimal(str(o_day_amt)) + Decimal(str(c_day))
-
-        szjilu, _ = Szjilu.objects.get_or_create(
-            club_id=cid,
-            defaults={
-                'TotalIncome': _ZERO, 'TotalFlow': _ZERO, 'TotalExpense': _ZERO,
-                'DailyExpense': _ZERO, 'DailyFlow': _ZERO,
-            },
-        )
-        szjilu.TotalIncome = total
-        szjilu.TotalFlow = total
-        szjilu.DailyFlow = day_flow
-        szjilu.save(update_fields=['TotalIncome', 'TotalFlow', 'DailyFlow', 'UpdateTime'])
-        updated += 1
-    return updated

jituan/services/szjilu_accounting.py

@@ -1,4 +1,4 @@
-"""各俱乐部全局收支流水（szjilu）统一记账。"""
+"""【szjilu 已废弃】财务入账只写 daily_income_stat，本模块保留幂等日志与恢复命令。"""
 import logging
 from datetime import date
 from decimal import Decimal
@@ -203,13 +203,12 @@ def zero_all_income_statistics():
 
 
 def _apply_income_stats(amount, club_id):
-    """写入 daily_income_stat + szjilu（仅在新 platform_income_log 落库后调用）。"""
+    """只写 daily_income_stat（财务看板唯一收入数据源），不写 szjilu。"""
     from orders.utils import update_daily_income
 
     cid = normalize_szjilu_club_id(club_id)
     jine = Decimal(str(amount))
     update_daily_income(jine, cid)
-    apply_szjilu_income(jine, cid)
 
 
 def record_wechat_income_once(amount, club_id=None, biz_ref=None):
@@ -783,7 +782,7 @@ def collect_legitimate_wechat_payments(since_year, until_year):
             f'order:{od.OrderID}',
             getattr(od, 'ClubID', None),
             od.Amount,
-            od.CreateTime,
+            od.UpdateTime or od.CreateTime,
             'order',
         )
 
@@ -797,7 +796,7 @@ def collect_legitimate_wechat_payments(since_year, until_year):
             f'cz:{cz.dingdan_id}',
             getattr(cz, 'club_id', None),
             cz.jine,
-            cz.CreateTime,
+            cz.UpdateTime or cz.CreateTime,
             f'cz_{cz.leixing}',
         )
 
@@ -882,33 +881,20 @@ def rebuild_income_stats_from_scratch(since_year=None, until_year=None, dry_run=
                 total_count=agg['cnt'],
             )
 
-        today_d = date.today()
-        club_ids = set(by_day_club.keys())
-        club_ids |= {(today_d, row.club_id) for row in Szjilu.objects.all()}
-
-        all_clubs = {cid for _, cid in club_ids}
-        for cid in all_clubs:
-            total = PlatformIncomeLog.objects.filter(club_id=cid).aggregate(
-                t=Sum('amount'),
-            )['t'] or _ZERO
-            today_flow = PlatformIncomeLog.objects.filter(
-                club_id=cid, CreateTime__date=today_d,
-            ).aggregate(t=Sum('amount'))['t'] or _ZERO
-            szjilu, _ = Szjilu.objects.select_for_update().get_or_create(
-                club_id=cid,
-                defaults=dict(_SZJILU_DEFAULTS),
-            )
-            szjilu.TotalIncome = total
-            szjilu.TotalFlow = total
-            szjilu.DailyFlow = today_flow
-            szjilu.save(update_fields=['TotalIncome', 'TotalFlow', 'DailyFlow', 'UpdateTime'])
-
     result['after_daily_total'] = float(
         DailyIncomeStat.objects.aggregate(t=Sum('total_amount'))['t'] or 0
     )
     return result
 
 
+def restore_finance_daily_stats(since_year=None, until_year=None, dry_run=True):
+    """
+    一键恢复财务每日收入：从订单/充值重建 daily_income_stat + platform_income_log。
+    不碰 daily_payout_stat，不碰 szjilu（收支记录表已废弃）。
+    """
+    return rebuild_income_stats_from_scratch(since_year, until_year, dry_run=dry_run)
+
+
 def collect_today_live_wechat_payments(stat_date=None):
     """今日实时微信支付（含当日下单或当日完成支付的订单/充值）。"""
     from datetime import datetime, timedelta
@@ -998,7 +984,6 @@ def restore_daily_income_from_orders(since_year=None, until_year=None, dry_run=T
                 total_amount=agg['amt'],
                 total_count=agg['cnt'],
             )
-        sync_szjilu_income_from_logs()
 
     result['after_daily_total'] = float(
         DailyIncomeStat.objects.aggregate(t=Sum('total_amount'))['t'] or 0