from argparse import ArgumentParser from pathlib import Path import sys from sqlalchemy import select from sqlalchemy.orm import selectinload ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) from app.database import SessionLocal # noqa: E402 from app.models import ProductionReport, WorkSession # noqa: E402 from app.services.metrics import calculate_report_metrics # noqa: E402 from app.services.work_schedule import get_work_schedule_config # noqa: E402 def apply_metrics(report: ProductionReport, schedule) -> None: metrics = calculate_report_metrics( report.start_at, report.end_at, list(report.items), devices=report.session.devices if report.session else None, schedule=schedule, ) report.duration_minutes = metrics["duration_minutes"] report.effective_minutes = metrics["effective_minutes"] report.total_good_qty = metrics["total_good_qty"] report.total_output_qty = metrics["total_output_qty"] report.actual_beat = metrics["actual_beat"] report.standard_beat = metrics["standard_beat"] report.expected_workload = metrics["expected_workload"] report.pace_rate = metrics["pace_rate"] report.workload_rate = metrics["workload_rate"] def main() -> None: parser = ArgumentParser(description="Recalculate production report metrics.") parser.add_argument("--report-id", type=int, help="只重算指定报工 ID") args = parser.parse_args() query = select(ProductionReport).options( selectinload(ProductionReport.items), selectinload(ProductionReport.session).selectinload(WorkSession.devices), ) if args.report_id: query = query.where(ProductionReport.id == args.report_id) with SessionLocal() as db: reports = db.scalars(query).all() for report in reports: schedule = get_work_schedule_config(db, report.attendance_point_name) apply_metrics(report, schedule) db.commit() print(f"recalculated reports: {len(reports)}") if __name__ == "__main__": main()