268 lines
11 KiB
Python
268 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
from copy import deepcopy
|
|
from datetime import date
|
|
|
|
from sqlalchemy import and_, func, 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, EquipmentKind
|
|
from app.models.location import Location
|
|
from app.models.user import User
|
|
from app.models.validation import Validation, ValidationStatus
|
|
from app.modules.orion.template_service import ReportTemplateService
|
|
from app.schemas.domain import QuickStartCreateRequest, QuickStartCustomerData, QuickStartValidationSummary
|
|
from app.services.validation_workflow import ValidationWorkflowService
|
|
|
|
|
|
class QuickStartService:
|
|
def __init__(self, session: Session) -> None:
|
|
self.session = session
|
|
|
|
def customer_data(self, customer_id: str) -> QuickStartCustomerData:
|
|
customer = self._customer(customer_id)
|
|
locations = list(
|
|
self.session.scalars(
|
|
select(Location).where(Location.customer_id == customer.id).order_by(Location.name.asc())
|
|
)
|
|
)
|
|
contacts = list(
|
|
self.session.scalars(
|
|
select(Contact).where(Contact.customer_id == customer.id).order_by(Contact.full_name.asc())
|
|
)
|
|
)
|
|
devices = list(
|
|
self.session.scalars(
|
|
select(Device).where(Device.customer_id == customer.id).order_by(Device.serial_number.asc())
|
|
)
|
|
)
|
|
validations = self._validation_summaries([device.id for device in devices])
|
|
return QuickStartCustomerData(
|
|
customer=customer,
|
|
locations=locations,
|
|
contacts=contacts,
|
|
devices=devices,
|
|
validations=validations,
|
|
)
|
|
|
|
def device_last_validation(self, device_id: str) -> QuickStartValidationSummary | None:
|
|
return self._last_validation(device_id)
|
|
|
|
def create_validation(self, payload: QuickStartCreateRequest, examiner: User) -> Validation:
|
|
customer = self._customer(str(payload.customer_id))
|
|
devices = list(
|
|
self.session.scalars(
|
|
select(Device).where(Device.customer_id == customer.id).order_by(Device.serial_number.asc())
|
|
)
|
|
)
|
|
device = self._device(str(payload.device_id)) if payload.device_id else self._single_or_default(devices)
|
|
if device is None:
|
|
raise ValueError("Kein Gerät verfügbar.")
|
|
if device.customer_id != customer.id:
|
|
raise ValueError("Gerät gehört nicht zum ausgewählten Kunden.")
|
|
locations = list(
|
|
self.session.scalars(
|
|
select(Location).where(Location.customer_id == customer.id).order_by(Location.name.asc())
|
|
)
|
|
)
|
|
contacts = list(
|
|
self.session.scalars(
|
|
select(Contact).where(Contact.customer_id == customer.id).order_by(Contact.full_name.asc())
|
|
)
|
|
)
|
|
location = self._location(str(payload.location_id)) if payload.location_id else self._single_or_default(locations)
|
|
contact = self._contact(str(payload.contact_id)) if payload.contact_id else self._single_or_default(contacts)
|
|
if location is None:
|
|
if locations:
|
|
raise ValueError("Bitte Standort auswählen.")
|
|
raise ValueError("Kein Standort verfügbar.")
|
|
if location.customer_id != customer.id:
|
|
raise ValueError("Standort gehört nicht zum ausgewählten Kunden.")
|
|
if device.location_id and device.location_id != location.id:
|
|
raise ValueError("Gerät gehört nicht zum gewählten Standort.")
|
|
if contact and contact.customer_id != customer.id:
|
|
raise ValueError("Ansprechpartner gehört nicht zum ausgewählten Kunden.")
|
|
|
|
base = self._base_validation(device.id, payload.validation_type)
|
|
bundle = ReportTemplateService(self.session).ensure_default_template()
|
|
validation = Validation(
|
|
report_number=self._next_report_number(),
|
|
customer_id=customer.id,
|
|
location_id=location.id,
|
|
contact_id=contact.id if contact else None,
|
|
device_id=device.id,
|
|
validation_type=payload.validation_type,
|
|
performed_on=date.today(),
|
|
examiner_name=examiner.full_name,
|
|
examiner_id=examiner.id,
|
|
operator_name=customer.quality_manager or customer.hygiene_officer or examiner.full_name,
|
|
status=ValidationStatus.draft.value,
|
|
result="offen",
|
|
revalidation_interval_months=base.revalidation_interval_months if base else 24,
|
|
next_validation_manually_overridden=False,
|
|
version=1 if base is None else base.version + 1,
|
|
previous_validation_id=base.id if base else None,
|
|
equipment_ids=list(base.equipment_ids) if base and base.equipment_ids else self._default_equipment_ids(),
|
|
environment_conditions=deepcopy(base.environment_conditions) if base else {},
|
|
documentation_checklist=deepcopy(base.documentation_checklist) if base else self._template_checklist(bundle, "documentation_checklist"),
|
|
performance_checklist=deepcopy(base.performance_checklist) if base else self._template_checklist(bundle, "performance_checklist"),
|
|
programs=deepcopy(base.programs) if base else self._default_programs(),
|
|
loading_patterns=deepcopy(base.loading_patterns) if base else self._default_loading_patterns(),
|
|
measurement_data=[],
|
|
drying={},
|
|
recommendations=deepcopy(base.recommendations) if base else [],
|
|
attachments=[],
|
|
)
|
|
self.session.add(validation)
|
|
self.session.flush()
|
|
ValidationWorkflowService(self.session).apply_revalidation_date(validation)
|
|
return validation
|
|
|
|
def _customer(self, customer_id: str) -> Customer:
|
|
customer = self.session.get(Customer, customer_id)
|
|
if customer is None:
|
|
raise ValueError("Kunde nicht gefunden.")
|
|
return customer
|
|
|
|
def _device(self, device_id: str) -> Device:
|
|
device = self.session.get(Device, device_id)
|
|
if device is None:
|
|
raise ValueError("Gerät nicht gefunden.")
|
|
return device
|
|
|
|
def _location(self, location_id: str) -> Location:
|
|
location = self.session.get(Location, location_id)
|
|
if location is None:
|
|
raise ValueError("Standort nicht gefunden.")
|
|
return location
|
|
|
|
def _contact(self, contact_id: str) -> Contact:
|
|
contact = self.session.get(Contact, contact_id)
|
|
if contact is None:
|
|
raise ValueError("Ansprechpartner nicht gefunden.")
|
|
return contact
|
|
|
|
def _single_or_default(self, items):
|
|
if len(items) == 1:
|
|
return items[0]
|
|
return None
|
|
|
|
def _last_validation(self, device_id: str) -> QuickStartValidationSummary | None:
|
|
validation = self.session.scalar(
|
|
select(Validation)
|
|
.where(
|
|
and_(
|
|
Validation.device_id == device_id,
|
|
Validation.status.in_([ValidationStatus.approved.value, ValidationStatus.completed.value]),
|
|
)
|
|
)
|
|
.order_by(Validation.performed_on.desc().nulls_last(), Validation.updated_at.desc())
|
|
)
|
|
if validation is None:
|
|
return None
|
|
return self._summary(validation)
|
|
|
|
def _validation_summaries(self, device_ids: list[str]) -> list[QuickStartValidationSummary]:
|
|
if not device_ids:
|
|
return []
|
|
validations = list(
|
|
self.session.scalars(
|
|
select(Validation)
|
|
.where(Validation.device_id.in_(device_ids))
|
|
.order_by(Validation.performed_on.desc().nulls_last(), Validation.updated_at.desc())
|
|
.limit(10)
|
|
)
|
|
)
|
|
return [self._summary(item) for item in validations]
|
|
|
|
def _summary(self, validation: Validation) -> QuickStartValidationSummary:
|
|
return QuickStartValidationSummary(
|
|
id=validation.id,
|
|
device_id=validation.device_id,
|
|
report_number=validation.report_number,
|
|
performed_on=validation.performed_on,
|
|
result=validation.result,
|
|
next_validation_on=validation.next_validation_on,
|
|
status=validation.status,
|
|
validation_type=validation.validation_type,
|
|
equipment_ids=list(validation.equipment_ids or []),
|
|
)
|
|
|
|
def _base_validation(self, device_id: str, validation_type: str) -> Validation | None:
|
|
if validation_type not in {"Revalidierung", "Leistungsbeurteilung"}:
|
|
return None
|
|
return self.session.scalar(
|
|
select(Validation)
|
|
.where(
|
|
and_(
|
|
Validation.device_id == device_id,
|
|
Validation.status.in_([ValidationStatus.approved.value, ValidationStatus.completed.value]),
|
|
)
|
|
)
|
|
.order_by(Validation.performed_on.desc().nulls_last(), Validation.updated_at.desc())
|
|
)
|
|
|
|
def _next_report_number(self) -> str:
|
|
total = self.session.scalar(select(func.count()).select_from(Validation)) or 0
|
|
return f"VAL-{total + 1:05d}"
|
|
|
|
def _default_equipment_ids(self) -> list[str]:
|
|
equipment = list(
|
|
self.session.scalars(
|
|
select(Equipment).where(Equipment.kind == EquipmentKind.temperature_logger).order_by(Equipment.serial_number.asc())
|
|
)
|
|
)
|
|
equipment.extend(
|
|
list(
|
|
self.session.scalars(
|
|
select(Equipment).where(Equipment.kind == EquipmentKind.pressure_logger).order_by(Equipment.serial_number.asc())
|
|
)
|
|
)
|
|
)
|
|
selected: list[str] = []
|
|
temp_count = 0
|
|
pressure_count = 0
|
|
for item in equipment:
|
|
if item.kind == EquipmentKind.temperature_logger and temp_count < 5:
|
|
selected.append(item.id)
|
|
temp_count += 1
|
|
elif item.kind == EquipmentKind.pressure_logger and pressure_count < 1:
|
|
selected.append(item.id)
|
|
pressure_count += 1
|
|
return selected
|
|
|
|
def _template_checklist(self, bundle, key: str) -> list[dict]:
|
|
checklist_key = {
|
|
"documentation_checklist": "documentation_control",
|
|
"performance_checklist": "sterilizer_description",
|
|
}.get(key, key)
|
|
template = next((item for item in bundle.checklists if item.checklist_key == checklist_key), None)
|
|
if template is None:
|
|
return []
|
|
return [
|
|
{
|
|
"template_key": template.checklist_key,
|
|
"template_title": template.title,
|
|
"text": item,
|
|
"value": "na",
|
|
"comment": "",
|
|
}
|
|
for item in template.items
|
|
]
|
|
|
|
def _default_programs(self) -> list[dict]:
|
|
return [
|
|
{"name": "Vakuumtest", "selected": True, "custom": False},
|
|
{"name": "Bowie-Dick / Leerkammerprofil", "selected": True, "custom": False},
|
|
{"name": "134 C hohl verpackt", "selected": True, "custom": False},
|
|
]
|
|
|
|
def _default_loading_patterns(self) -> list[dict]:
|
|
return [
|
|
{"run": 1, "pattern": "Standardbeladung", "description": "", "images": []},
|
|
{"run": 2, "pattern": "Standardbeladung", "description": "", "images": []},
|
|
{"run": 3, "pattern": "Standardbeladung", "description": "", "images": []},
|
|
]
|