dibang-xiaoshou/backend/scripts/init_vehicles_from_history.py

#!/usr/bin/env python3
"""
从历史出入库表(inbound_outbound)、地磅单表(weighing_slips)中提取不重复的车牌号，
初始化写入车牌号基础资料表(vehicles)。

使用方式（在 backend 目录下执行）：
  python -m scripts.init_vehicles_from_history
或：
  uv run python -m scripts.init_vehicles_from_history
"""
from __future__ import annotations

import asyncio
import sys
from pathlib import Path

# 将 backend 根目录加入 path，以便 import app
backend_root = Path(__file__).resolve().parent.parent
if str(backend_root) not in sys.path:
    sys.path.insert(0, str(backend_root))


async def main() -> None:
    from sqlmodel import select
    from sqlmodel.ext.asyncio.session import AsyncSession

    from app.db.session import init_models, AsyncSessionLocal
    from app.models.inventory import InboundOutbound
    from app.models.master_data import Vehicle
    from app.models.weighing import WeighingSlip

    # 1. 确保所有表已创建（含 vehicles）
    await init_models()

    # 2. 从出入库表、地磅单表收集不重复车牌号（去空、去重，按原始大小写保留一个）
    seen_normalized: set[str] = set()
    vehicle_nos: list[str] = []

    async with AsyncSessionLocal() as session:
        # 出入库表：未删除记录的车牌号
        q_inv = select(InboundOutbound.vehicle_no).where(InboundOutbound.is_deleted == 0)
        result_inv = await session.exec(q_inv)
        for (v,) in result_inv.all():
            v = (v or "").strip()
            if not v:
                continue
            key = v.lower()
            if key not in seen_normalized:
                seen_normalized.add(key)
                vehicle_nos.append(v)

        # 地磅单表：未作废记录的车牌号
        q_ws = select(WeighingSlip.vehicle_no).where(WeighingSlip.is_voided == 0)
        result_ws = await session.exec(q_ws)
        for (v,) in result_ws.all():
            v = (v or "").strip()
            if not v:
                continue
            key = v.lower()
            if key not in seen_normalized:
                seen_normalized.add(key)
                vehicle_nos.append(v)

    if not vehicle_nos:
        print("未从出入库/地磅单表中发现任何车牌号，无需导入。")
        return

    # 3. 写入 vehicles 表（仅插入尚未存在的车牌号）
    from datetime import datetime

    added = 0
    skipped = 0
    async with AsyncSessionLocal() as session:
        existing = await session.exec(
            select(Vehicle.vehicle_no).where(Vehicle.is_deleted == 0)
        )
        existing_set = {row.lower() for row in existing.all()}

        for v in vehicle_nos:
            if v.lower() in existing_set:
                skipped += 1
                continue
            vehicle = Vehicle(
                vehicle_no=v,
                remarks=None,
                is_deleted=0,
                created_at=datetime.now().isoformat(),
            )
            session.add(vehicle)
            existing_set.add(v.lower())
            added += 1

        await session.commit()

    print(f"历史车牌号统计：共发现 {len(vehicle_nos)} 个不重复车牌号。")
    print(f"本次新增写入 vehicles 表：{added} 条；已存在跳过：{skipped} 条。")


if __name__ == "__main__":
    asyncio.run(main())