95 lines
3.1 KiB
Python
95 lines
3.1 KiB
Python
from datetime import timedelta
|
||
|
||
from fastapi import HTTPException, status
|
||
from sqlalchemy import delete, select
|
||
from sqlalchemy.orm import Session, selectinload
|
||
|
||
from app.models import (
|
||
MoldLockFeedback,
|
||
Personnel,
|
||
ProductionReport,
|
||
ProductionReportItem,
|
||
ReportAuditLog,
|
||
WorkSession,
|
||
)
|
||
from app.timezone import now
|
||
|
||
|
||
VOID_RETENTION_DAYS = 30
|
||
APPROVED_EDIT_DAYS = 30
|
||
|
||
|
||
def void_deadline(voided_at=None):
|
||
return (voided_at or now()) + timedelta(days=VOID_RETENTION_DAYS)
|
||
|
||
|
||
def can_edit_report(report: ProductionReport) -> bool:
|
||
if bool(report.is_voided):
|
||
return False
|
||
if str(report.status) == "pending":
|
||
return True
|
||
if str(report.status) != "approved" or report.reviewed_at is None:
|
||
return False
|
||
return report.reviewed_at >= now() - timedelta(days=APPROVED_EDIT_DAYS)
|
||
|
||
|
||
def can_unvoid_report(report: ProductionReport) -> bool:
|
||
if not bool(report.is_voided) or report.voided_at is None:
|
||
return False
|
||
return report.voided_at > now() - timedelta(days=VOID_RETENTION_DAYS)
|
||
|
||
|
||
def mark_report_voided(report: ProductionReport, user: Personnel) -> None:
|
||
if bool(report.is_voided):
|
||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="该报工记录已作废")
|
||
timestamp = now()
|
||
report.is_voided = True
|
||
report.voided_at = timestamp
|
||
report.voided_by = user.phone
|
||
report.unvoid_deadline_at = void_deadline(timestamp)
|
||
|
||
|
||
def restore_voided_report(report: ProductionReport) -> None:
|
||
if not bool(report.is_voided):
|
||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="该报工记录未作废")
|
||
if not can_unvoid_report(report):
|
||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="作废已超过30天，不能撤销作废")
|
||
report.is_voided = False
|
||
report.voided_at = None
|
||
report.voided_by = None
|
||
report.unvoid_deadline_at = None
|
||
|
||
|
||
def purge_expired_voided_reports(db: Session) -> int:
|
||
cutoff = now() - timedelta(days=VOID_RETENTION_DAYS)
|
||
reports = db.scalars(
|
||
select(ProductionReport)
|
||
.options(selectinload(ProductionReport.session).selectinload(WorkSession.devices))
|
||
.where(
|
||
ProductionReport.is_voided.is_(True),
|
||
ProductionReport.voided_at.is_not(None),
|
||
ProductionReport.voided_at <= cutoff,
|
||
)
|
||
.limit(100)
|
||
).all()
|
||
if not reports:
|
||
return 0
|
||
|
||
for report in reports:
|
||
session = report.session
|
||
if session is not None:
|
||
session_device_ids = [device.id for device in session.devices if device.id is not None]
|
||
if session_device_ids:
|
||
db.execute(
|
||
delete(MoldLockFeedback).where(
|
||
MoldLockFeedback.session_device_id.in_(session_device_ids)
|
||
)
|
||
)
|
||
db.execute(delete(ReportAuditLog).where(ReportAuditLog.report_id == report.id))
|
||
db.execute(delete(ProductionReportItem).where(ProductionReportItem.report_id == report.id))
|
||
db.delete(report)
|
||
if session is not None:
|
||
db.delete(session)
|
||
db.commit()
|
||
return len(reports)
|