"""【szjilu 已废弃】财务入账只写 daily_income_stat,本模块保留幂等日志与恢复命令。""" import logging from datetime import date from decimal import Decimal from django.db import IntegrityError, transaction from django.db.models import Count, Sum from django.db.utils import OperationalError, ProgrammingError from config.models import DailyIncomeStat, PlatformIncomeLog, Szjilu from jituan.constants import CLUB_ID_DEFAULT logger = logging.getLogger(__name__) _ZERO = Decimal('0.00') _SZJILU_DEFAULTS = { 'TotalIncome': _ZERO, 'TotalFlow': _ZERO, 'TotalExpense': _ZERO, 'DailyExpense': _ZERO, 'DailyFlow': _ZERO, } _PLATFORM_LOG_ERRORS = (ProgrammingError, OperationalError) def normalize_szjilu_club_id(club_id): return (club_id or '').strip() or CLUB_ID_DEFAULT def income_log_exists(biz_ref): ref = (biz_ref or '').strip() if not ref: return False try: return PlatformIncomeLog.objects.filter(biz_ref=ref).exists() except _PLATFORM_LOG_ERRORS: return False def _get_szjilu_for_update(club_id): cid = normalize_szjilu_club_id(club_id) szjilu, _ = Szjilu.objects.select_for_update().get_or_create( club_id=cid, defaults=dict(_SZJILU_DEFAULTS), ) return szjilu def apply_szjilu_income(amount, club_id=None): """【已废弃】szjilu 禁止写入,财务只维护 daily_income_stat。""" logger.debug('apply_szjilu_income 已禁用 amount=%s club=%s', amount, club_id) 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。""" from orders.utils import update_daily_income cid = normalize_szjilu_club_id(club_id) jine = Decimal(str(amount)) update_daily_income(jine, cid) def record_wechat_income_once(amount, club_id=None, biz_ref=None): """ 微信入账:同一 biz_ref 只记一次。 返回 True=本次新入账;False=已记过(重复回调,绝不重复加钱)。 """ ref = (biz_ref or '').strip() if not ref: raise ValueError('biz_ref 不能为空') if income_log_exists(ref): return False cid = normalize_szjilu_club_id(club_id) jine = Decimal(str(amount)) try: with transaction.atomic(): PlatformIncomeLog.objects.create( biz_ref=ref, club_id=cid, amount=jine, ) _apply_income_stats(jine, cid) except IntegrityError: return False except _PLATFORM_LOG_ERRORS as exc: logger.warning( 'platform_income_log 不可用(%s),无法幂等入账 biz_ref=%s,请 migrate config 后执行 sync_daily_income_from_log', exc, ref, ) return False except Exception as exc: logger.error( 'platform_income_log 入账失败 biz_ref=%s err=%s', ref, exc, exc_info=True, ) return False logger.info('平台入账记账成功 biz_ref=%s club=%s amount=%s', ref, cid, jine) return True def repair_wechat_income_if_missing(amount, club_id=None, biz_ref=None): """已支付单重复回调时补记;日志已存在则绝不重复加钱。""" return record_wechat_income_once(amount, club_id, biz_ref) def sync_daily_income_stat_for_date(stat_date, reset_day=False): """ 用 platform_income_log 重写指定日期的 daily_income_stat(金额/笔数与日志完全一致)。 reset_day=True 时先删掉该日所有日统计行,再按日志重建(纠正重复记账)。 返回更新的俱乐部行数。 """ if reset_day: DailyIncomeStat.objects.filter(date=stat_date).delete() try: log_qs = PlatformIncomeLog.objects.filter(CreateTime__date=stat_date) except _PLATFORM_LOG_ERRORS as exc: logger.error('platform_income_log 不可用: %s', exc) return 0 club_ids = list( log_qs.values_list('club_id', flat=True).distinct() ) updated = 0 y, m, d = stat_date.year, stat_date.month, stat_date.day with transaction.atomic(): for cid in club_ids: agg = log_qs.filter(club_id=cid).aggregate( total_amount=Sum('amount'), total_count=Count('id'), ) amt = agg['total_amount'] or _ZERO cnt = int(agg['total_count'] or 0) stat, _ = DailyIncomeStat.objects.select_for_update().get_or_create( date=stat_date, club_id=cid, defaults={ 'year': y, 'month': m, 'day': d, 'total_amount': amt, 'total_count': cnt, }, ) stat.year = y stat.month = m stat.day = d stat.total_amount = amt stat.total_count = cnt stat.save(update_fields=['year', 'month', 'day', 'total_amount', 'total_count']) updated += 1 return updated def sync_szjilu_income_from_logs(): """【已废弃】szjilu 禁止维护。""" return 0 # 订单已付款且非退款(repair 错误地把 Status=5 退款单也计入了收入) _ORDER_INCOME_STATUSES = [1, 2, 3, 4, 6, 7, 8] # 微信充值类 czjilu:会员/押金/积分/商家/罚单/考核 _CZ_WECHAT_LEIXING = [1, 2, 3, 4, 5] def collect_canonical_wechat_income_for_date(stat_date, existing_refs=None): """ 收集当日真实微信收入(严格:仅统计当日下单且已支付成功的单据,每笔唯一)。 不含退款;不按 UpdateTime 计老单(repair 虚高的主要原因之一)。 """ from datetime import datetime, timedelta 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 or ref in skip_refs: return jine = Decimal(str(amount or 0)) if jine <= 0: return cid = normalize_szjilu_club_id(club_id) seen[ref] = True items.append((ref, cid, jine, kind)) for od in Order.query.filter( CreateTime__gte=day_start, CreateTime__lt=day_end, Status__in=_ORDER_INCOME_STATUSES, ): 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( CreateTime__gte=day_start, CreateTime__lt=day_end, zhuangtai=3, leixing__in=_CZ_WECHAT_LEIXING, ): _add(f'cz:{cz.dingdan_id}', getattr(cz, 'club_id', None), cz.jine, f'cz_{cz.leixing}') 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: 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 造成的重复记账: 1) 删除该日 platform_income_log 2) 按真实支付重建日志(每笔唯一) 3) 重写 daily_income_stat 4) 重算 szjilu 累计收入 返回 dict 供命令行展示。 """ from datetime import datetime from django.db.models import Sum before_daily = DailyIncomeStat.objects.filter(date=stat_date).aggregate( amt=Sum('total_amount'), cnt=Sum('total_count'), ) before_daily_by_club = { row.club_id: { 'amt': Decimal(str(row.total_amount or 0)), 'cnt': int(row.total_count or 0), } for row in DailyIncomeStat.objects.filter(date=stat_date) } before_log_cnt = 0 before_log_amt = _ZERO try: before_log_cnt = PlatformIncomeLog.objects.filter(CreateTime__date=stat_date).count() agg = PlatformIncomeLog.objects.filter(CreateTime__date=stat_date).aggregate( t=Sum('amount'), ) before_log_amt = agg['t'] or _ZERO except _PLATFORM_LOG_ERRORS: pass 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) result = { 'date': str(stat_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_amt), 'before_log_count': before_log_cnt, 'after_amount': float(rebuild_amt), 'after_count': rebuild_cnt, 'items': items, 'dry_run': dry_run, } if dry_run: return result from datetime import time as dt_time from django.utils import timezone pay_ts = timezone.make_aware(datetime.combine(stat_date, dt_time(12, 0, 0))) with transaction.atomic(): try: PlatformIncomeLog.objects.filter(CreateTime__date=stat_date).delete() except _PLATFORM_LOG_ERRORS as exc: raise RuntimeError(f'platform_income_log 不可用: {exc}') from exc 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, club_id=cid, amount=jine, ) PlatformIncomeLog.objects.filter(pk=log.pk).update(CreateTime=pay_ts) y, m, d = stat_date.year, stat_date.month, stat_date.day by_club = {} for ref, cid, jine, _kind in items: if cid not in by_club: by_club[cid] = {'amt': _ZERO, 'cnt': 0} by_club[cid]['amt'] += jine by_club[cid]['cnt'] += 1 for cid, agg in by_club.items(): DailyIncomeStat.objects.create( date=stat_date, club_id=cid, year=y, month=m, day=d, total_amount=agg['amt'], total_count=agg['cnt'], ) # 只扣掉 repair 多记的部分,不用日志全量重算 szjilu(避免抹掉无日志的历史收入) all_clubs = set(before_daily_by_club.keys()) | set(by_club.keys()) for cid in all_clubs: old_amt = before_daily_by_club.get(cid, {}).get('amt', _ZERO) new_amt = by_club.get(cid, {}).get('amt', _ZERO) excess = old_amt - new_amt if excess <= 0 and stat_date != date.today(): continue szjilu = _get_szjilu_for_update(cid) if excess > 0: szjilu.TotalIncome -= excess szjilu.TotalFlow -= excess if stat_date == date.today(): szjilu.DailyFlow = new_amt szjilu.save(update_fields=['TotalIncome', 'TotalFlow', 'DailyFlow', 'UpdateTime']) return result def apply_szjilu_expense(amount, club_id=None): """【已废弃】szjilu 禁止写入,支出只维护 daily_payout_stat。""" logger.debug('apply_szjilu_expense 已禁用 amount=%s club=%s', amount, club_id) def get_szjilu_snapshot(club_id=None): """读取某俱乐部收支汇总;无记录时返回零值。""" cid = normalize_szjilu_club_id(club_id) row = Szjilu.query.filter(club_id=cid).first() if not row: return { 'zongliushui': 0.0, 'zongshouyi': 0.0, 'zongzhichu': 0.0, 'jrls': 0.0, 'jrzc': 0.0, } return { 'zongliushui': float(row.TotalFlow or _ZERO), 'zongshouyi': float(row.TotalIncome or _ZERO), 'zongzhichu': float(row.TotalExpense or _ZERO), 'jrls': float(row.DailyFlow or _ZERO), 'jrzc': float(row.DailyExpense or _ZERO), } def reset_all_daily_szjilu(): """每日清零:所有俱乐部的今日流水/今日支出。""" updated = 0 with transaction.atomic(): for szjilu in Szjilu.objects.select_for_update().all(): szjilu.DailyExpense = _ZERO szjilu.DailyFlow = _ZERO szjilu.save(update_fields=['DailyExpense', 'DailyFlow', 'UpdateTime']) updated += 1 return updated # 合法微信支付收入(不含退款 Status=5) _LEGIT_ORDER_STATUSES = [1, 2, 3, 4, 6, 7, 8] _LEGIT_CZ_LEIXING = [1, 2, 3, 4, 5] def collect_legitimate_wechat_payments(since_year, until_year): """ 从业务单据重建:每笔 biz_ref 全局只出现一次。 返回 [(biz_ref, club_id, amount, stat_date, pay_datetime, kind), ...] """ from datetime import datetime from orders.models import Order from products.models import Czjilu start = datetime(since_year, 1, 1) end = datetime(until_year + 1, 1, 1) seen = {} items = [] def _add(ref, club_id, amount, pay_dt, kind): ref = (ref or '').strip() if not ref or ref in seen: return jine = Decimal(str(amount or 0)) if jine <= 0: return cid = normalize_szjilu_club_id(club_id) if not pay_dt: pay_dt = datetime.now() stat_d = pay_dt.date() if hasattr(pay_dt, 'date') else pay_dt seen[ref] = True items.append((ref, cid, jine, stat_d, pay_dt, kind)) for od in Order.query.filter( CreateTime__gte=start, CreateTime__lt=end, Status__in=_LEGIT_ORDER_STATUSES, ): _add( f'order:{od.OrderID}', getattr(od, 'ClubID', None), od.Amount, od.UpdateTime or od.CreateTime, 'order', ) for cz in Czjilu.query.filter( CreateTime__gte=start, CreateTime__lt=end, zhuangtai=3, leixing__in=_LEGIT_CZ_LEIXING, ): _add( f'cz:{cz.dingdan_id}', getattr(cz, 'club_id', None), cz.jine, cz.UpdateTime or cz.CreateTime, f'cz_{cz.leixing}', ) return items def rebuild_income_stats_from_scratch(since_year=None, until_year=None, dry_run=True): """ 终极恢复:清空全部收支脏数据 → 按真实微信支付逐笔重建(每笔唯一)。 """ from collections import defaultdict from django.db.models import Sum until_year = until_year or date.today().year since_year = since_year or (until_year - 1) before = { 'daily': float(DailyIncomeStat.objects.aggregate(t=Sum('total_amount'))['t'] or 0), 'logs': 0, 'log_cnt': 0, } try: before['logs'] = float(PlatformIncomeLog.objects.aggregate(t=Sum('amount'))['t'] or 0) before['log_cnt'] = PlatformIncomeLog.objects.count() except _PLATFORM_LOG_ERRORS: pass items = collect_legitimate_wechat_payments(since_year, until_year) total_amt = sum((x[2] for x in items), _ZERO) today = date.today() today_amt = sum((x[2] for x in items if x[3] == today), _ZERO) today_cnt = sum(1 for x in items if x[3] == today) month_amt = sum( (x[2] for x in items if x[3].year == today.year and x[3].month == today.month), _ZERO, ) result = { 'since_year': since_year, 'until_year': until_year, 'before_daily_total': before['daily'], 'before_log_total': before['logs'], 'before_log_count': before['log_cnt'], 'after_total': float(total_amt), 'after_count': len(items), 'today_amount': float(today_amt), 'today_count': today_cnt, 'month_amount': float(month_amt), 'items': items, 'dry_run': dry_run, } if dry_run: return result with transaction.atomic(): try: PlatformIncomeLog.objects.all().delete() except _PLATFORM_LOG_ERRORS as exc: raise RuntimeError(f'platform_income_log 不可用: {exc}') from exc DailyIncomeStat.objects.all().delete() by_day_club = defaultdict(lambda: {'amt': _ZERO, 'cnt': 0}) for ref, cid, jine, stat_d, pay_dt, _kind in items: log = PlatformIncomeLog.objects.create(biz_ref=ref, club_id=cid, amount=jine) if pay_dt: PlatformIncomeLog.objects.filter(pk=log.pk).update(CreateTime=pay_dt) key = (stat_d, cid) by_day_club[key]['amt'] += jine by_day_club[key]['cnt'] += 1 for (stat_d, cid), agg in by_day_club.items(): DailyIncomeStat.objects.create( date=stat_d, club_id=cid, year=stat_d.year, month=stat_d.month, day=stat_d.day, total_amount=agg['amt'], total_count=agg['cnt'], ) result['after_daily_total'] = float( DailyIncomeStat.objects.aggregate(t=Sum('total_amount'))['t'] or 0 ) return result def rebuild_daily_income_stat_table(dry_run=True): """ 恢复 daily_income_stat 全表(财务看板唯一收入数据源)。 每笔 biz_ref 全局只计一次: 1. platform_income_log → 按日志 CreateTime 日期 2. 无日志的已支付订单/充值 → 按 UpdateTime 日期补入 不删日志、不碰 daily_payout_stat、不碰 szjilu。 """ from collections import defaultdict from orders.models import Order from products.models import Czjilu by_ref = {} log_count = 0 try: for log in PlatformIncomeLog.objects.all().iterator(): ref = (log.biz_ref or '').strip() if not ref: continue stat_d = log.CreateTime.date() if log.CreateTime else date.today() by_ref[ref] = { 'club_id': normalize_szjilu_club_id(log.club_id), 'amount': Decimal(str(log.amount or 0)), 'stat_date': stat_d, } log_count += 1 except _PLATFORM_LOG_ERRORS as exc: logger.warning('platform_income_log 不可用,仅从订单补全: %s', exc) gap_orders = 0 for od in Order.query.filter(Status__in=_LEGIT_ORDER_STATUSES): ref = f'order:{od.OrderID}' if ref in by_ref: continue pay_dt = od.UpdateTime or od.CreateTime if not pay_dt: continue amt = Decimal(str(od.Amount or 0)) if amt <= 0: continue by_ref[ref] = { 'club_id': normalize_szjilu_club_id(getattr(od, 'ClubID', None)), 'amount': amt, 'stat_date': pay_dt.date() if hasattr(pay_dt, 'date') else pay_dt, } gap_orders += 1 gap_cz = 0 for cz in Czjilu.query.filter(zhuangtai=3, leixing__in=_LEGIT_CZ_LEIXING): ref = f'cz:{cz.dingdan_id}' if ref in by_ref: continue pay_dt = cz.UpdateTime or cz.CreateTime if not pay_dt: continue amt = Decimal(str(cz.jine or 0)) if amt <= 0: continue by_ref[ref] = { 'club_id': normalize_szjilu_club_id(getattr(cz, 'club_id', None)), 'amount': amt, 'stat_date': pay_dt.date() if hasattr(pay_dt, 'date') else pay_dt, } gap_cz += 1 by_day_club = defaultdict(lambda: {'amt': _ZERO, 'cnt': 0}) today = date.today() today_amt = _ZERO today_cnt = 0 for info in by_ref.values(): key = (info['stat_date'], info['club_id']) by_day_club[key]['amt'] += info['amount'] by_day_club[key]['cnt'] += 1 if info['stat_date'] == today: today_amt += info['amount'] today_cnt += 1 total_amt = sum((v['amt'] for v in by_day_club.values()), _ZERO) before = float(DailyIncomeStat.objects.aggregate(t=Sum('total_amount'))['t'] or 0) result = { 'before_daily_total': before, 'after_daily_total': float(total_amt), 'payment_rows': len(by_ref), 'from_log': log_count, 'gap_orders': gap_orders, 'gap_cz': gap_cz, 'daily_rows': len(by_day_club), 'today_amount': float(today_amt), 'today_count': today_cnt, 'dry_run': dry_run, } if dry_run: return result with transaction.atomic(): DailyIncomeStat.objects.all().delete() for (stat_d, cid), agg in by_day_club.items(): DailyIncomeStat.objects.create( date=stat_d, club_id=cid, year=stat_d.year, month=stat_d.month, day=stat_d.day, total_amount=agg['amt'], total_count=agg['cnt'], ) 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(忽略 since/until,整表重算)。""" return rebuild_daily_income_stat_table(dry_run=dry_run) def collect_today_live_wechat_payments(stat_date=None): """今日实时微信支付(含当日下单或当日完成支付的订单/充值)。""" from datetime import datetime, timedelta from django.db.models import Q from orders.models import Order from products.models import Czjilu stat_date = stat_date or date.today() day_start = datetime.combine(stat_date, datetime.min.time()) day_end = day_start + timedelta(days=1) seen = {} items = [] def _add(ref, club_id, amount, kind): ref = (ref or '').strip() if not ref or ref in seen: return jine = Decimal(str(amount or 0)) if jine <= 0: return cid = normalize_szjilu_club_id(club_id) seen[ref] = True items.append((ref, cid, jine, kind)) for od in Order.query.filter(Status__in=_LEGIT_ORDER_STATUSES).filter( Q(CreateTime__gte=day_start, CreateTime__lt=day_end) | Q(UpdateTime__gte=day_start, UpdateTime__lt=day_end), ): _add(f'order:{od.OrderID}', getattr(od, 'ClubID', None), od.Amount, 'order') for cz in Czjilu.query.filter(zhuangtai=3, leixing__in=_LEGIT_CZ_LEIXING).filter( Q(CreateTime__gte=day_start, CreateTime__lt=day_end) | Q(UpdateTime__gte=day_start, UpdateTime__lt=day_end), ): _add(f'cz:{cz.dingdan_id}', getattr(cz, 'club_id', None), cz.jine, f'cz_{cz.leixing}') return items def restore_daily_income_from_orders(since_year=None, until_year=None, dry_run=True): """ 从订单/充值表重算 daily_income_stat(订单数据不动,只覆盖统计表)。 """ from collections import defaultdict from django.db.models import Sum until_year = until_year or date.today().year since_year = since_year or (until_year - 1) items = collect_legitimate_wechat_payments(since_year, until_year) by_day_club = defaultdict(lambda: {'amt': _ZERO, 'cnt': 0}) for _ref, cid, jine, stat_d, _pay_dt, _kind in items: key = (stat_d, cid) by_day_club[key]['amt'] += jine by_day_club[key]['cnt'] += 1 before = float(DailyIncomeStat.objects.aggregate(t=Sum('total_amount'))['t'] or 0) total_after = sum((v['amt'] for v in by_day_club.values()), _ZERO) result = { 'since_year': since_year, 'until_year': until_year, 'payment_rows': len(items), 'daily_rows': len(by_day_club), 'before_daily_total': before, 'after_daily_total': float(total_after), 'dry_run': dry_run, } if dry_run: return result with transaction.atomic(): DailyIncomeStat.objects.filter( date__gte=date(since_year, 1, 1), date__lte=date(until_year, 12, 31), ).delete() for (stat_d, cid), agg in by_day_club.items(): DailyIncomeStat.objects.create( date=stat_d, club_id=cid, year=stat_d.year, month=stat_d.month, day=stat_d.day, total_amount=agg['amt'], total_count=agg['cnt'], ) result['after_daily_total'] = float( DailyIncomeStat.objects.aggregate(t=Sum('total_amount'))['t'] or 0 ) return result def repopulate_today_income_stats(stat_date=None, dry_run=True): """只重建指定日期的日收入(不删历史其它天),修复今日显示为 0。""" from collections import defaultdict from datetime import datetime, time as dt_time from django.db.models import Sum from django.utils import timezone stat_date = stat_date or date.today() before = DailyIncomeStat.objects.filter(date=stat_date).aggregate( amt=Sum('total_amount'), cnt=Sum('total_count'), ) items = collect_today_live_wechat_payments(stat_date) total = sum((x[2] for x in items), _ZERO) result = { 'date': str(stat_date), 'before_amount': float(before['amt'] or 0), 'before_count': int(before['cnt'] or 0), 'after_amount': float(total), 'after_count': len(items), 'items': items, 'dry_run': dry_run, } if dry_run: return result pay_ts = timezone.make_aware(datetime.combine(stat_date, dt_time(12, 0, 0))) with transaction.atomic(): DailyIncomeStat.objects.filter(date=stat_date).delete() by_club = defaultdict(lambda: {'amt': _ZERO, 'cnt': 0}) for ref, cid, jine, _kind in items: if not income_log_exists(ref): try: log = PlatformIncomeLog.objects.create(biz_ref=ref, club_id=cid, amount=jine) PlatformIncomeLog.objects.filter(pk=log.pk).update(CreateTime=pay_ts) except IntegrityError: pass by_club[cid]['amt'] += jine by_club[cid]['cnt'] += 1 for cid, agg in by_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'], ) return result def fix_income_stat_count_for_date(stat_date=None, dry_run=True): """ 只修正 daily_income_stat.total_count(不动 total_amount、不动其它天)。 repair 会把笔数虚增到 100+;按当日真实支付笔数写回。 """ from collections import defaultdict stat_date = stat_date or date.today() items = collect_today_live_wechat_payments(stat_date) by_club = defaultdict(int) for _ref, cid, _jine, _kind in items: by_club[cid] += 1 before_rows = list(DailyIncomeStat.objects.filter(date=stat_date)) before_count = sum(int(r.total_count or 0) for r in before_rows) after_count = sum(by_club.values()) result = { 'date': str(stat_date), 'before_count': before_count, 'after_count': after_count, 'clubs': dict(by_club), 'dry_run': dry_run, } if dry_run: return result with transaction.atomic(): seen_clubs = set() for cid, cnt in by_club.items(): seen_clubs.add(cid) row = DailyIncomeStat.objects.filter(date=stat_date, club_id=cid).first() if row: row.total_count = cnt row.save(update_fields=['total_count']) for row in before_rows: if row.club_id not in seen_clubs: row.total_count = 0 row.save(update_fields=['total_count']) return result