60 lines
2.0 KiB
Python
60 lines
2.0 KiB
Python
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()
|