dy-dashboard/repair_arrival_totals.py

#!/usr/bin/env python3
"""到货批次汇总修复：列出/备份/修复 实际到货数量、未到货数、到货日期

根因：2026-07-27 前的旧版 backfill_arrival_sync.py 按 order_id 汇总到货批次时未按
order_type 过滤，且只更新第一个命中的订单表（domestic → overseas → packaging 后 return），
导致 domestic/overseas 表中与其它类型订单 id 相同的行被写入别类型批次的合计。

修复口径（与详情弹窗一致）：实际到货数量 = 本订单本类型批次合计，未到货数 = 数量 - 该合计。
"""
import json
import os
import sys
from datetime import datetime

sys.path.insert(0, '/home/gxd/.openclaw/workspace/dy-dashboard')
os.chdir('/home/gxd/.openclaw/workspace/dy-dashboard')

from services.db import query, get_db

TABLE_MAP = {
    'domestic': ('domestic_orders', '实际到货日期', '货品编号'),
    'packaging': ('packaging_orders', '到货日期', '编码'),
    'overseas': ('overseas_orders', '实际到货日期', '产品编码'),
}


def batch_truth(order_type):
    rows = query("""
        SELECT order_id, SUM(arrival_qty) AS total, MAX(arrival_date) AS latest_date, COUNT(*) AS cnt
        FROM arrival_batches WHERE order_type=%s GROUP BY order_id
    """, (order_type,))
    return {r['order_id']: r for r in rows}


def all_type_totals():
    rows = query("""
        SELECT order_id, SUM(arrival_qty) AS total FROM arrival_batches GROUP BY order_id
    """)
    return {r['order_id']: int(r['total'] or 0) for r in rows}


def single_batch_qtys():
    """order_id -> 该 id 下各类型批次单条数量集合（用于识别旧脚本写入的单批次值）"""
    rows = query("SELECT order_id, arrival_qty FROM arrival_batches")
    out = {}
    for r in rows:
        out.setdefault(r['order_id'], set()).add(int(r['arrival_qty'] or 0))
    return out


def batch_dates():
    """order_id -> 该 id 下全部批次的到货日期集合（用于确认日期也来自别类型批次）"""
    rows = query("SELECT order_id, arrival_date FROM arrival_batches WHERE arrival_date IS NOT NULL")
    out = {}
    for r in rows:
        d = r['arrival_date']
        d = d.strftime('%Y-%m-%d') if hasattr(d, 'strftime') else str(d)[:10]
        out.setdefault(r['order_id'], set()).add(d)
    return out


def build_plan():
    plan = []
    all_totals = all_type_totals()
    single_qtys = single_batch_qtys()
    other_dates = batch_dates()
    for order_type, (table, date_col, code_col) in TABLE_MAP.items():
        truth = batch_truth(order_type)
        orders = query("SELECT id, `数量`, `实际到货数量`, `未到货数`, `" + date_col + "` AS arr_date FROM " + table)
        for o in orders:
            oid = o['id']
            qty = int(o['数量'] or 0)
            stored = int(o['实际到货数量'] or 0)
            stored_unf = o['未到货数']
            stored_unf = None if stored_unf is None else int(stored_unf)
            cur_date = o['arr_date']
            if cur_date and hasattr(cur_date, 'strftime'):
                cur_date = cur_date.strftime('%Y-%m-%d')
            elif cur_date:
                cur_date = str(cur_date)[:10]
            t = truth.get(oid)
            if t:
                new_total = int(t['total'] or 0)
                nd = t['latest_date']
                new_date = nd.strftime('%Y-%m-%d') if hasattr(nd, 'strftime') else (str(nd)[:10] if nd else None)
                new_unf = qty - new_total
                if stored != new_total or stored_unf != new_unf or cur_date != new_date:
                    plan.append({'type': order_type, 'reason': 'batch-sync', 'id': oid,
                                 'before': {'实际到货数量': stored, '未到货数': stored_unf, '到货日期': cur_date},
                                 'after': {'实际到货数量': new_total, '未到货数': new_unf, '到货日期': new_date}})
            else:
                # 无本类型批次：仅修复可证明来自其它类型批次的污染值
                # 签名：存储值 = 该 order_id 全部批次合计，或 = 其中某一条批次的数量
                if stored > 0 and (all_totals.get(oid) == stored or stored in single_qtys.get(oid, set())):
                    # 到货日期只在能证明来自同 id 其它类型批次时才清空
                    new_date = None if (cur_date and cur_date in other_dates.get(oid, set())) or not cur_date else cur_date
                    plan.append({'type': order_type, 'reason': 'cross-type-contamination', 'id': oid,
                                 'before': {'实际到货数量': stored, '未到货数': stored_unf, '到货日期': cur_date},
                                 'after': {'实际到货数量': 0, '未到货数': qty, '到货日期': new_date}})
    return plan


def main():
    apply = '--apply' in sys.argv
    plan = build_plan()
    print('待修复行数：%d' % len(plan))
    for p in plan:
        print('  [%s] %s id=%s %s  %s -> %s' % (p['reason'], p['type'], p['id'],
              '', json.dumps(p['before'], ensure_ascii=False), json.dumps(p['after'], ensure_ascii=False)))
    if not plan:
        return
    ts = datetime.now().strftime('%Y%m%d_%H%M%S')
    outdir = 'backups/arrival_totals_repair_%s' % ts
    if apply:
        os.makedirs(outdir, exist_ok=True)
        with open(os.path.join(outdir, 'before.json'), 'w', encoding='utf-8') as f:
            json.dump(plan, f, ensure_ascii=False, indent=2, default=str)
        conn = get_db()
        try:
            with conn.cursor() as cur:
                for p in plan:
                    table, date_col, _ = TABLE_MAP[p['type']]
                    cur.execute(
                        "UPDATE " + table + " SET `实际到货数量`=%s, `未到货数`=%s, `" + date_col + "`=%s WHERE id=%s",
                        (p['after']['实际到货数量'], p['after']['未到货数'], p['after']['到货日期'], p['id'])
                    )
            conn.commit()
        finally:
            conn.close()
        with open(os.path.join(outdir, 'applied.json'), 'w', encoding='utf-8') as f:
            json.dump(plan, f, ensure_ascii=False, indent=2, default=str)
        print('已应用修复，备份目录：%s' % outdir)
        try:
            from services.cache import refresh_all_caches
            refresh_all_caches()
            print('缓存已刷新')
        except Exception as e:
            print('缓存刷新失败：%s' % e)


if __name__ == '__main__':
    main()