From 600188a99a8dae7c0ad5875e7bcd5d5930d5908b Mon Sep 17 00:00:00 2001 From: XingQue Date: Wed, 24 Jun 2026 21:18:57 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=81=A2=E5=A4=8D=20repair=20=E9=87=8D?= =?UTF-8?q?=E5=A4=8D=E8=AE=B0=E8=B4=A6=20+=20=E7=A6=81=E7=94=A8=E9=94=99?= =?UTF-8?q?=E8=AF=AF=E8=A1=A5=E8=B4=A6=E5=91=BD=E4=BB=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../commands/recover_wechat_income_stats.py | 85 ++++++ .../commands/repair_wechat_income.py | 103 +------ .../commands/sync_daily_income_from_log.py | 68 +++++ jituan/services/szjilu_accounting.py | 289 ++++++++++++++++-- 4 files changed, 433 insertions(+), 112 deletions(-) create mode 100644 config/management/commands/recover_wechat_income_stats.py create mode 100644 config/management/commands/sync_daily_income_from_log.py diff --git a/config/management/commands/recover_wechat_income_stats.py b/config/management/commands/recover_wechat_income_stats.py new file mode 100644 index 0000000..4532579 --- /dev/null +++ b/config/management/commands/recover_wechat_income_stats.py @@ -0,0 +1,85 @@ +""" +恢复被 repair_wechat_income 搞乱的收入统计。 + +repair 会对已入账订单重复加钱（200 变 900 就是这个）。 +本命令：删掉当日错误日志 → 按真实微信支付重建 → 重算日统计和 szjilu。 + +先不加 --yes 可预览重建后金额。 +""" +from datetime import date, timedelta + +from django.core.management.base import BaseCommand + +from jituan.services.szjilu_accounting import recover_income_stats_for_date + + +class Command(BaseCommand): + 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='从指定日期起向前恢复几天，默认 1', + ) + parser.add_argument( + '--yes', + action='store_true', + help='确认执行（不加则仅预览，不改数据库）', + ) + + def handle(self, *args, **options): + if options['date']: + end_day = date.fromisoformat(options['date']) + else: + end_day = date.today() + days = max(1, int(options['days'] or 1)) + start_day = end_day - timedelta(days=days - 1) + dry_run = not options['yes'] + + if dry_run: + self.stdout.write(self.style.WARNING( + '【预览模式】未改数据库。确认金额无误后加 --yes 执行。\n' + )) + else: + 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['before_daily_amount']:.2f} 元 / {r['before_daily_count']} 笔" + ) + self.stdout.write( + f" 当前幂等日志: {r['before_log_amount']:.2f} 元 / {r['before_log_count']} 笔" + ) + self.stdout.write(self.style.SUCCESS( + f" 重建后（真实支付）: {r['after_amount']:.2f} 元 / {r['after_count']} 笔" + )) + if r['items'] and dry_run: + self.stdout.write(' 明细（前20笔）:') + for ref, cid, amt, kind in r['items'][:20]: + self.stdout.write(f' {ref} {amt}元 club={cid} ({kind})') + if len(r['items']) > 20: + self.stdout.write(f' ... 共 {len(r["items"])} 笔') + + if dry_run: + 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 --date 2026-06-24 --yes' + )) + else: + self.stdout.write(self.style.SUCCESS( + '\n恢复完成。请刷新后台财务页核对今日收入。' + '\n勿再执行 repair_wechat_income（已禁用）。' + )) diff --git a/config/management/commands/repair_wechat_income.py b/config/management/commands/repair_wechat_income.py index 6daf177..df23ac6 100644 --- a/config/management/commands/repair_wechat_income.py +++ b/config/management/commands/repair_wechat_income.py @@ -1,102 +1,15 @@ -"""补记漏掉的微信收入统计（今日或指定日期已支付但未入账的单据）。""" -from datetime import date, datetime, timedelta - +"""【已废弃】请用 sync_daily_income_from_log，本命令曾导致重复记账。""" from django.core.management.base import BaseCommand -from django.db.models import Q - -from config.models import PlatformIncomeLog -from jituan.constants import CLUB_ID_DEFAULT -from jituan.services.szjilu_accounting import repair_wechat_income_if_missing -from orders.models import Order -from products.models import Czjilu class Command(BaseCommand): - help = '扫描已支付订单/充值单，补记 daily_income_stat（幂等，可重复执行）' - - 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='从指定日期起向前补几天，默认 1（仅当天）', - ) + help = '已废弃：勿使用。请改用 python manage.py sync_daily_income_from_log --reset-day' def handle(self, *args, **options): - if options['date']: - end_day = date.fromisoformat(options['date']) - else: - end_day = date.today() - days = max(1, int(options['days'] or 1)) - start_day = end_day - timedelta(days=days - 1) - - repaired = 0 - skipped = 0 - errors = 0 - - for day in range(days): - d = start_day + timedelta(days=day) - day_start = datetime.combine(d, datetime.min.time()) - day_end = day_start + timedelta(days=1) - self.stdout.write(f'扫描 {d} ...') - - orders = Order.query.filter( - CreateTime__gte=day_start, - CreateTime__lt=day_end, - Status__in=[1, 2, 3, 4, 5, 6, 7, 8], - ) - for od in orders: - club_id = getattr(od, 'ClubID', None) or CLUB_ID_DEFAULT - ref = f'order:{od.OrderID}' - try: - ok = self._repair_one(od.Amount, club_id, biz_ref=ref) - if ok: - repaired += 1 - self.stdout.write(f' + 订单 {od.OrderID} {od.Amount}元 club={club_id}') - else: - skipped += 1 - except Exception as exc: - errors += 1 - self.stderr.write(f' ! 订单 {od.OrderID} 失败: {exc}') - - 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: - club_id = getattr(cz, 'club_id', None) or CLUB_ID_DEFAULT - ref = f'cz:{cz.dingdan_id}' - try: - ok = self._repair_one(cz.jine, club_id, biz_ref=ref) - if ok: - repaired += 1 - self.stdout.write(f' + 充值 {cz.dingdan_id} {cz.jine}元 leixing={cz.leixing} club={club_id}') - else: - skipped += 1 - except Exception as exc: - errors += 1 - self.stderr.write(f' ! 充值 {cz.dingdan_id} 失败: {exc}') - - self.stdout.write(self.style.SUCCESS( - f'完成：新补记 {repaired} 笔，已存在跳过 {skipped} 笔，失败 {errors} 笔', + self.stderr.write(self.style.ERROR( + 'repair_wechat_income 已禁用（会对已入账订单重复加钱）。\n' + '纠正今日虚高请执行：\n' + ' python manage.py sync_daily_income_from_log --reset-day\n' + '可选同时重算累计：\n' + ' python manage.py sync_daily_income_from_log --reset-day --sync-szjilu' )) - - def _repair_one(self, amount, club_id, biz_ref): - """幂等补记；若仅有孤儿幂等日志无日统计，删日志后重试一次。""" - ref = (biz_ref or '').strip() - ok = repair_wechat_income_if_missing(amount, club_id, biz_ref=ref) - if ok: - return True - if PlatformIncomeLog.objects.filter(biz_ref=ref).exists(): - PlatformIncomeLog.objects.filter(biz_ref=ref).delete() - return repair_wechat_income_if_missing(amount, club_id, biz_ref=ref) - return False diff --git a/config/management/commands/sync_daily_income_from_log.py b/config/management/commands/sync_daily_income_from_log.py new file mode 100644 index 0000000..8b3b5ba --- /dev/null +++ b/config/management/commands/sync_daily_income_from_log.py @@ -0,0 +1,68 @@ +""" +用 platform_income_log 重写 daily_income_stat，纠正重复记账后的虚高数字。 +权威数据源是 platform_income_log（每笔支付一条，biz_ref 唯一），不是扫订单重算。 +""" +from datetime import date, timedelta + +from django.core.management.base import BaseCommand + +from jituan.services.szjilu_accounting import ( + sync_daily_income_stat_for_date, + sync_szjilu_income_from_logs, +) + + +class Command(BaseCommand): + help = ( + '按 platform_income_log 同步 daily_income_stat（纠正今日收入虚高/笔数不准）。' + '不要用 repair_wechat_income，那个会重复加钱。' + ) + + 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='从指定日期起向前同步几天，默认 1', + ) + parser.add_argument( + '--reset-day', + action='store_true', + help='同步前先删除该日全部日统计行，再按日志重建（纠正重复记账必选）', + ) + parser.add_argument( + '--sync-szjilu', + action='store_true', + help='同时按日志全量重算 szjilu 累计收入/今日流水', + ) + + def handle(self, *args, **options): + if options['date']: + end_day = date.fromisoformat(options['date']) + else: + end_day = date.today() + days = max(1, int(options['days'] or 1)) + start_day = end_day - timedelta(days=days - 1) + reset_day = bool(options['reset_day']) + + total = 0 + for i in range(days): + d = start_day + timedelta(days=i) + n = sync_daily_income_stat_for_date(d, reset_day=reset_day) + total += n + self.stdout.write(f'{d}: 已同步 {n} 个俱乐部日统计行') + + if options['sync_szjilu']: + sz = sync_szjilu_income_from_logs() + self.stdout.write(f'szjilu: 已重算 {sz} 个俱乐部累计收入') + + self.stdout.write(self.style.SUCCESS( + f'完成。今日收入/笔数现与 platform_income_log 一致。' + f'若仍有漏记支付，请确认 migrate config 后等新支付自动入账。' + )) diff --git a/jituan/services/szjilu_accounting.py b/jituan/services/szjilu_accounting.py index 2b0367e..de6b210 100644 --- a/jituan/services/szjilu_accounting.py +++ b/jituan/services/szjilu_accounting.py @@ -1,11 +1,13 @@ """各俱乐部全局收支流水（szjilu）统一记账。""" 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 PlatformIncomeLog, Szjilu +from config.models import DailyIncomeStat, PlatformIncomeLog, Szjilu from jituan.constants import CLUB_ID_DEFAULT logger = logging.getLogger(__name__) @@ -26,6 +28,16 @@ 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( @@ -47,7 +59,7 @@ def apply_szjilu_income(amount, club_id=None): def _apply_income_stats(amount, club_id): - """写入 daily_income_stat + szjilu（核心统计，必须成功）。""" + """写入 daily_income_stat + szjilu（仅在新 platform_income_log 落库后调用）。""" from orders.utils import update_daily_income cid = normalize_szjilu_club_id(club_id) @@ -58,13 +70,15 @@ def _apply_income_stats(amount, club_id): def record_wechat_income_once(amount, club_id=None, biz_ref=None): """ - 微信入账：同一 biz_ref 只记一次 daily_income_stat + szjilu。 - 返回 True=本次新入账；False=已记过（重复回调）。 - 幂等日志与日统计在同一事务内提交，避免「已支付但未入账」。 + 微信入账：同一 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)) @@ -77,34 +91,275 @@ def record_wechat_income_once(amount, club_id=None, biz_ref=None): ) _apply_income_stats(jine, cid) except IntegrityError: - logger.info('平台入账已存在，跳过重复记账 biz_ref=%s', ref) return False except _PLATFORM_LOG_ERRORS as exc: logger.warning( - 'platform_income_log 不可用(%s)，跳过幂等直接记入收入 biz_ref=%s', + 'platform_income_log 不可用(%s)，无法幂等入账 biz_ref=%s，请 migrate config 后执行 sync_daily_income_from_log', exc, ref, ) - _apply_income_stats(jine, cid) - return True + return False except Exception as exc: logger.error( - 'platform_income_log 入账异常，仍尝试记入收入 biz_ref=%s err=%s', + 'platform_income_log 入账失败 biz_ref=%s err=%s', ref, exc, exc_info=True, ) - _apply_income_stats(jine, cid) - return 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: - return record_wechat_income_once(amount, club_id, biz_ref) - except Exception as exc: - logger.error('补记收支失败 biz_ref=%s err=%s', biz_ref, exc, exc_info=True) - return False + 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(): + """按 platform_income_log 全量重算各俱乐部 szjilu 累计收入/流水（纠正重复累加）。""" + try: + rows = PlatformIncomeLog.objects.values('club_id').annotate( + total=Sum('amount'), + cnt=Count('id'), + ) + except _PLATFORM_LOG_ERRORS as exc: + logger.error('platform_income_log 不可用: %s', exc) + return 0 + + today = date.today() + updated = 0 + with transaction.atomic(): + for row in rows: + cid = normalize_szjilu_club_id(row['club_id']) + total = row['total'] or _ZERO + today_agg = PlatformIncomeLog.objects.filter( + CreateTime__date=today, club_id=cid, + ).aggregate(t=Sum('amount')) + today_flow = today_agg['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']) + updated += 1 + return updated + + +# 订单已付款且非退款（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): + """ + 按真实支付时间（UpdateTime）收集当日微信收入，每笔 biz_ref 唯一。 + 返回 [(biz_ref, club_id, amount, kind), ...] + """ + 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) + 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( + UpdateTime__gte=day_start, + UpdateTime__lt=day_end, + Status__in=_ORDER_INCOME_STATUSES, + ): + _add(f'order:{od.OrderID}', getattr(od, 'ClubID', None), od.Amount, 'order') + + for cz in Czjilu.query.filter( + UpdateTime__gte=day_start, + UpdateTime__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 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 + + items = collect_canonical_wechat_income_for_date(stat_date) + 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() + + 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):