364 lines
17 KiB
Python
364 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
import csv
|
|
import io
|
|
from dataclasses import dataclass
|
|
from datetime import date, datetime
|
|
from uuid import uuid4
|
|
|
|
from dateutil.relativedelta import relativedelta
|
|
from sqlalchemy import and_, func, or_, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.contact import Contact
|
|
from app.models.customer import Customer
|
|
from app.models.device import Device
|
|
from app.models.equipment import Equipment
|
|
from app.models.location import Location
|
|
from app.models.validation import Validation, ValidationStatus
|
|
|
|
REQUIRED_FIELDS = {
|
|
"report_number": ("Allgemeine Angaben", "Berichtsnummer"),
|
|
"validation_type": ("Allgemeine Angaben", "Validierungsart"),
|
|
"performed_on": ("Allgemeine Angaben", "Pruefdatum"),
|
|
"customer_id": ("Kunde und Standort", "Kunde"),
|
|
"location_id": ("Kunde und Standort", "Standort"),
|
|
"device_id": ("Geraet", "Geraet"),
|
|
"examiner_name": ("Allgemeine Angaben", "Pruefer"),
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ReviewIssue:
|
|
field: str
|
|
message: str
|
|
section: str
|
|
|
|
def as_dict(self) -> dict[str, str]:
|
|
return {"field": self.field, "message": self.message, "section": self.section}
|
|
|
|
|
|
class ValidationWorkflowService:
|
|
def __init__(self, session: Session) -> None:
|
|
self.session = session
|
|
|
|
def review(self, validation: Validation) -> dict:
|
|
errors = self._required_errors(validation) + self._reference_errors(validation)
|
|
warnings = self._warnings(validation)
|
|
complete_sections = self._complete_sections(validation, errors, warnings)
|
|
return {
|
|
"status": validation.status,
|
|
"errors": [issue.as_dict() for issue in errors],
|
|
"warnings": [issue.as_dict() for issue in warnings],
|
|
"complete_sections": complete_sections,
|
|
}
|
|
|
|
def mark_ready_for_review(self, validation: Validation) -> dict:
|
|
review = self.review(validation)
|
|
validation.status = (
|
|
ValidationStatus.ready_for_review.value
|
|
if not review["errors"]
|
|
else ValidationStatus.draft.value
|
|
)
|
|
self.session.flush()
|
|
review["status"] = validation.status
|
|
return review
|
|
|
|
def apply_revalidation_date(self, validation: Validation) -> None:
|
|
if validation.performed_on and not validation.next_validation_manually_overridden:
|
|
validation.next_validation_on = validation.performed_on + relativedelta(
|
|
months=validation.revalidation_interval_months or 24
|
|
)
|
|
|
|
def duplicate(self, validation: Validation) -> Validation:
|
|
clone = Validation(
|
|
report_number=f"{validation.report_number}-KOPIE-{str(uuid4())[:8]}",
|
|
customer_id=validation.customer_id,
|
|
location_id=validation.location_id,
|
|
contact_id=validation.contact_id,
|
|
device_id=validation.device_id,
|
|
validation_type=validation.validation_type,
|
|
project=validation.project,
|
|
test_location=validation.test_location,
|
|
examiner_name=validation.examiner_name,
|
|
participants=validation.participants,
|
|
operator_name=validation.operator_name,
|
|
scheduled_on=validation.scheduled_on,
|
|
performed_on=validation.performed_on,
|
|
next_validation_on=validation.next_validation_on,
|
|
revalidation_interval_months=validation.revalidation_interval_months,
|
|
next_validation_manually_overridden=validation.next_validation_manually_overridden,
|
|
version=validation.version + 1,
|
|
previous_validation_id=validation.id,
|
|
examiner_id=validation.examiner_id,
|
|
status=ValidationStatus.draft.value,
|
|
result=validation.result,
|
|
notes=validation.notes,
|
|
equipment_ids=validation.equipment_ids,
|
|
environment_conditions=validation.environment_conditions,
|
|
documentation_checklist=validation.documentation_checklist,
|
|
performance_checklist=validation.performance_checklist,
|
|
programs=validation.programs,
|
|
loading_patterns=validation.loading_patterns,
|
|
measurement_data=validation.measurement_data,
|
|
drying=validation.drying,
|
|
recommendations=validation.recommendations,
|
|
attachments=validation.attachments,
|
|
)
|
|
self.session.add(clone)
|
|
self.session.flush()
|
|
return clone
|
|
|
|
def create_new_version(self, validation: Validation) -> Validation:
|
|
clone = self.duplicate(validation)
|
|
clone.report_number = f"{validation.report_number}-V{clone.version}"
|
|
return clone
|
|
|
|
def export_json(self, validation: Validation) -> dict:
|
|
return {
|
|
column.name: getattr(validation, column.name)
|
|
for column in Validation.__table__.columns
|
|
if column.name not in {"created_at", "updated_at"}
|
|
}
|
|
|
|
def preview_csv(self, content: bytes) -> dict:
|
|
rows = []
|
|
reader = csv.DictReader(io.StringIO(content.decode("utf-8-sig")))
|
|
for index, row in enumerate(reader, start=2):
|
|
rows.append(self._preview_import_row(index, row))
|
|
return {
|
|
"rows": rows,
|
|
"valid_rows": sum(1 for row in rows if not row["errors"]),
|
|
"invalid_rows": sum(1 for row in rows if row["errors"]),
|
|
"duplicates": sum(1 for row in rows if row["duplicate"]),
|
|
}
|
|
|
|
def import_rows(self, rows: list[dict], duplicate_strategy: str) -> dict:
|
|
summary = {"successful": 0, "skipped": 0, "failed": 0, "errors": []}
|
|
for index, row in enumerate(rows, start=1):
|
|
preview = self._preview_import_row(index, row)
|
|
if preview["errors"]:
|
|
summary["failed"] += 1
|
|
summary["errors"].append(f"Zeile {index}: {', '.join(preview['errors'])}")
|
|
continue
|
|
existing = self.session.scalar(
|
|
select(Validation).where(Validation.report_number == row.get("report_number"))
|
|
)
|
|
if existing and duplicate_strategy == "skip":
|
|
summary["skipped"] += 1
|
|
continue
|
|
target = existing if existing and duplicate_strategy == "update" else Validation()
|
|
target.report_number = (
|
|
f"{row.get('report_number')}-IMPORT-{str(uuid4())[:8]}"
|
|
if existing and duplicate_strategy == "copy"
|
|
else row.get("report_number")
|
|
)
|
|
target.validation_type = row.get("validation_type")
|
|
target.performed_on = self._parse_date(row.get("test_date") or row.get("performed_on"))
|
|
target.customer_id = preview["resolved_customer_id"]
|
|
target.location_id = preview["resolved_location_id"]
|
|
target.contact_id = row.get("contact_id")
|
|
target.device_id = preview["resolved_device_id"]
|
|
target.examiner_name = row.get("examiner") or row.get("examiner_name")
|
|
target.project = row.get("project")
|
|
target.test_location = row.get("test_location")
|
|
target.participants = row.get("participants")
|
|
target.operator_name = row.get("operator_name")
|
|
target.next_validation_on = self._parse_date(row.get("next_validation_on"))
|
|
target.equipment_ids = row.get("equipment_ids") or []
|
|
target.environment_conditions = row.get("environment_conditions") or {}
|
|
target.documentation_checklist = row.get("documentation_checklist") or []
|
|
target.performance_checklist = row.get("performance_checklist") or []
|
|
target.programs = row.get("programs") or []
|
|
target.loading_patterns = row.get("loading_patterns") or []
|
|
target.measurement_data = row.get("measurement_data") or []
|
|
target.drying = row.get("drying") or {}
|
|
target.recommendations = row.get("recommendations") or []
|
|
target.attachments = row.get("attachments") or []
|
|
target.status = row.get("status") or ValidationStatus.draft.value
|
|
target.result = row.get("result")
|
|
target.notes = row.get("notes")
|
|
if target.id is None:
|
|
self.session.add(target)
|
|
summary["successful"] += 1
|
|
self.session.flush()
|
|
return summary
|
|
|
|
def query_validations(
|
|
self,
|
|
search: str | None,
|
|
page: int,
|
|
page_size: int,
|
|
sort_by: str,
|
|
sort_order: str,
|
|
filters: dict,
|
|
) -> dict:
|
|
statement = select(Validation)
|
|
count_statement = select(Validation)
|
|
conditions = []
|
|
if search:
|
|
term = f"%{search}%"
|
|
conditions.append(
|
|
or_(
|
|
Validation.report_number.ilike(term),
|
|
Validation.validation_type.ilike(term),
|
|
Validation.examiner_name.ilike(term),
|
|
Validation.result.ilike(term),
|
|
)
|
|
)
|
|
for key in ["status", "customer_id", "device_id", "validation_type"]:
|
|
if filters.get(key):
|
|
conditions.append(getattr(Validation, key) == filters[key])
|
|
if filters.get("result"):
|
|
result_aliases = {
|
|
"OFFEN": ["OFFEN", "offen", ""],
|
|
"BESTANDEN": ["BESTANDEN", "bestanden"],
|
|
"BESTANDEN_MIT_AUFLAGEN": ["BESTANDEN_MIT_AUFLAGEN", "bestanden_mit_auflagen", "mit_auflagen", "bestanden mit Auflagen"],
|
|
"NICHT_BESTANDEN": ["NICHT_BESTANDEN", "nicht_bestanden", "nicht bestanden"],
|
|
}
|
|
result_values = result_aliases.get(filters["result"], [filters["result"]])
|
|
if filters["result"] == "OFFEN":
|
|
conditions.append(or_(Validation.result.in_(result_values), Validation.result.is_(None)))
|
|
else:
|
|
conditions.append(Validation.result.in_(result_values))
|
|
if filters.get("date_from"):
|
|
conditions.append(Validation.performed_on >= filters["date_from"])
|
|
if filters.get("date_to"):
|
|
conditions.append(Validation.performed_on <= filters["date_to"])
|
|
if filters.get("overdue_only"):
|
|
conditions.append(Validation.next_validation_on < date.today())
|
|
conditions.append(Validation.status != ValidationStatus.cancelled.value)
|
|
if conditions:
|
|
statement = statement.where(and_(*conditions))
|
|
count_statement = count_statement.where(and_(*conditions))
|
|
sort_column = getattr(Validation, sort_by, Validation.updated_at)
|
|
if sort_order == "asc":
|
|
statement = statement.order_by(sort_column.asc())
|
|
else:
|
|
statement = statement.order_by(sort_column.desc())
|
|
total = self.session.scalar(
|
|
select(func.count()).select_from(count_statement.order_by(None).subquery())
|
|
) or 0
|
|
items = list(self.session.scalars(statement.offset((page - 1) * page_size).limit(page_size)))
|
|
return {
|
|
"items": items,
|
|
"total": total,
|
|
"page": page,
|
|
"page_size": page_size,
|
|
"pages": max((total + page_size - 1) // page_size, 1),
|
|
}
|
|
|
|
def revalidation_status(self, validation: Validation) -> str:
|
|
if not validation.next_validation_on:
|
|
return "nicht_erfasst"
|
|
delta = (validation.next_validation_on - date.today()).days
|
|
if delta < 0:
|
|
return "ueberfaellig"
|
|
if delta <= 30:
|
|
return "faellig_30"
|
|
if delta <= 90:
|
|
return "faellig_90"
|
|
return "faellig_spaeter"
|
|
|
|
def _required_errors(self, validation: Validation) -> list[ReviewIssue]:
|
|
issues = []
|
|
for field, (section, label) in REQUIRED_FIELDS.items():
|
|
if not getattr(validation, field):
|
|
issues.append(ReviewIssue(field, f"{label} fehlt.", section))
|
|
return issues
|
|
|
|
def _reference_errors(self, validation: Validation) -> list[ReviewIssue]:
|
|
issues = []
|
|
customer = self.session.get(Customer, validation.customer_id) if validation.customer_id else None
|
|
location = self.session.get(Location, validation.location_id) if validation.location_id else None
|
|
device = self.session.get(Device, validation.device_id) if validation.device_id else None
|
|
if validation.customer_id and customer is None:
|
|
issues.append(ReviewIssue("customer_id", "Kunde existiert nicht.", "Kunde und Standort"))
|
|
if validation.location_id and location is None:
|
|
issues.append(ReviewIssue("location_id", "Standort existiert nicht.", "Kunde und Standort"))
|
|
if validation.device_id and device is None:
|
|
issues.append(ReviewIssue("device_id", "Geraet existiert nicht.", "Geraet"))
|
|
if location and device and device.location_id != location.id:
|
|
issues.append(ReviewIssue("device_id", "Geraet gehoert nicht zum gewaehlten Standort.", "Geraet"))
|
|
return issues
|
|
|
|
def _warnings(self, validation: Validation) -> list[ReviewIssue]:
|
|
warnings = []
|
|
equipment = []
|
|
if validation.equipment_ids:
|
|
equipment = list(self.session.scalars(select(Equipment).where(Equipment.id.in_(validation.equipment_ids))))
|
|
if not validation.equipment_ids:
|
|
warnings.append(ReviewIssue("equipment_ids", "Keine Pruefmittel ausgewaehlt.", "Pruefmittel"))
|
|
for item in equipment:
|
|
if item.calibration_due_on and item.calibration_due_on < date.today():
|
|
warnings.append(ReviewIssue("equipment_ids", f"Pruefmittel {item.serial_number} ist abgelaufen.", "Pruefmittel"))
|
|
if not any(item.get("selected") for item in validation.programs or []):
|
|
warnings.append(ReviewIssue("programs", "Keine Programme ausgewaehlt.", "Programme"))
|
|
if not validation.measurement_data:
|
|
warnings.append(ReviewIssue("measurement_data", "Keine Messdaten vorhanden.", "Messdaten"))
|
|
if not validation.attachments:
|
|
warnings.append(ReviewIssue("attachments", "Keine Bilder oder Anlagen vorhanden.", "Bilder und Anlagen"))
|
|
if not validation.recommendations:
|
|
warnings.append(ReviewIssue("recommendations", "Keine Empfehlungen oder Auflagen erfasst.", "Empfehlungen"))
|
|
winlog_imported = any(item.get("imports") for item in validation.measurement_data or [])
|
|
if not winlog_imported:
|
|
warnings.append(ReviewIssue("measurement_data", "Winlog-Datei noch nicht importiert.", "Messdaten"))
|
|
return warnings
|
|
|
|
def _complete_sections(
|
|
self, validation: Validation, errors: list[ReviewIssue], warnings: list[ReviewIssue]
|
|
) -> list[str]:
|
|
blocked = {issue.section for issue in [*errors, *warnings]}
|
|
sections = [
|
|
"Allgemeine Angaben",
|
|
"Kunde und Standort",
|
|
"Geraet",
|
|
"Pruefmittel",
|
|
"Programme",
|
|
"Messdaten",
|
|
"Bilder und Anlagen",
|
|
"Empfehlungen",
|
|
]
|
|
return [section for section in sections if section not in blocked]
|
|
|
|
def _preview_import_row(self, row_number: int, row: dict) -> dict:
|
|
errors = []
|
|
customer = self.session.get(Customer, row.get("customer_id")) if row.get("customer_id") else self.session.scalar(select(Customer).where(Customer.name == row.get("customer_reference")))
|
|
location = self.session.get(Location, row.get("location_id")) if row.get("location_id") else self.session.scalar(select(Location).where(Location.name == row.get("location_reference")))
|
|
device = self.session.get(Device, row.get("device_id")) if row.get("device_id") else self.session.scalar(select(Device).where(Device.serial_number == row.get("device_serial_number")))
|
|
duplicate = bool(
|
|
row.get("report_number")
|
|
and self.session.scalar(select(Validation).where(Validation.report_number == row.get("report_number")))
|
|
)
|
|
required = [
|
|
("report_number", row.get("report_number")),
|
|
("validation_type", row.get("validation_type")),
|
|
("test_date", row.get("test_date") or row.get("performed_on")),
|
|
("customer_reference", row.get("customer_reference") or row.get("customer_id")),
|
|
("location_reference", row.get("location_reference") or row.get("location_id")),
|
|
("device_serial_number", row.get("device_serial_number") or row.get("device_id")),
|
|
("examiner", row.get("examiner") or row.get("examiner_name")),
|
|
]
|
|
for field, value in required:
|
|
if not value:
|
|
errors.append(f"{field} fehlt")
|
|
if row.get("customer_reference") and not customer:
|
|
errors.append("Kunde nicht gefunden")
|
|
if row.get("location_reference") and not location:
|
|
errors.append("Standort nicht gefunden")
|
|
if row.get("device_serial_number") and not device:
|
|
errors.append("Geraet nicht gefunden")
|
|
return {
|
|
"row_number": row_number,
|
|
"data": row,
|
|
"errors": errors,
|
|
"duplicate": duplicate,
|
|
"resolved_customer_id": customer.id if customer else None,
|
|
"resolved_location_id": location.id if location else None,
|
|
"resolved_device_id": device.id if device else None,
|
|
}
|
|
|
|
def _parse_date(self, value: str | None) -> date | None:
|
|
if not value:
|
|
return None
|
|
return datetime.strptime(value, "%Y-%m-%d").date()
|