1021 lines
34 KiB
Python
1021 lines
34 KiB
Python
"""各俱乐部全局收支流水（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 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):
|
||
"""平台收入：总收益、总流水、今日流水增加。"""
|
||
jine = Decimal(str(amount))
|
||
with transaction.atomic():
|
||
szjilu = _get_szjilu_for_update(club_id)
|
||
szjilu.TotalIncome += jine
|
||
szjilu.TotalFlow += jine
|
||
szjilu.DailyFlow += jine
|
||
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
|
||
|
||
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):
|
||
"""
|
||
微信入账：同一 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():
|
||
"""按 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, 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):
|
||
"""平台出款/支出：总收益减少，总支出、今日支出增加。"""
|
||
jine = Decimal(str(amount))
|
||
with transaction.atomic():
|
||
szjilu = _get_szjilu_for_update(club_id)
|
||
szjilu.TotalIncome -= jine
|
||
szjilu.TotalExpense += jine
|
||
szjilu.DailyExpense += jine
|
||
szjilu.save()
|
||
|
||
|
||
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.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.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'],
|
||
)
|
||
|
||
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 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 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'],
|
||
)
|
||
|
||
today_d = date.today()
|
||
for cid, agg in by_club.items():
|
||
szjilu, _ = Szjilu.objects.select_for_update().get_or_create(
|
||
club_id=cid, defaults=dict(_SZJILU_DEFAULTS),
|
||
)
|
||
if stat_date == today_d:
|
||
szjilu.DailyFlow = agg['amt']
|
||
total_all = PlatformIncomeLog.objects.filter(club_id=cid).aggregate(
|
||
t=Sum('amount'),
|
||
)['t'] or _ZERO
|
||
szjilu.TotalIncome = total_all
|
||
szjilu.TotalFlow = total_all
|
||
szjilu.save(update_fields=['TotalIncome', 'TotalFlow', 'DailyFlow', 'UpdateTime'])
|
||
|
||
return result
|