admin.schubert/Validation_Suite
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(navigation): add prominent validation quickstart action

Browse source
This commit is contained in:
Schubert Ferenc 2026-07-11 18:02:53 +02:00
parent 0bbcaba211
commit 9c6b184e39
28614 changed files with 4356173 additions and 23 deletions
Download patch file Download diff file Expand all files Collapse all files

BIN
validation-suite/backend/mercury/app/services/__pycache__/demo_data.cpython-313.pyc Normal file
View file

Binary file not shown.

BIN
validation-suite/backend/mercury/app/services/__pycache__/domain_service.cpython-313.pyc
View file

Binary file not shown.

BIN
validation-suite/backend/mercury/app/services/__pycache__/quickstart_service.cpython-313.pyc Normal file
View file

Binary file not shown.

BIN
validation-suite/backend/mercury/app/services/__pycache__/validation_workflow.cpython-313.pyc
View file

Binary file not shown.

757
validation-suite/backend/mercury/app/services/demo_data.py Normal file
View file

@ -0,0 +1,757 @@
from __future__ import annotations
from copy import deepcopy
from dataclasses import dataclass
from datetime import date
from pathlib import Path
import shutil
from typing import Any
from dateutil.relativedelta import relativedelta
from sqlalchemy import delete, select
from sqlalchemy.orm import Session
from app.models.contact import Contact
from app.models.customer import Customer, CustomerType
from app.models.device import Device
from app.models.equipment import Equipment, EquipmentKind, EquipmentStatus
from app.models.location import Location
from app.models.validation import Validation, ValidationStatus
from app.modules.orion.service import OrionReportService
from app.modules.orion.template_service import ReportTemplateService
from app.services.validation_workflow import ValidationWorkflowService
DEMO_TAG = "DEMO_DATA"
DEMO_REPORT_PREFIX = "DEMO-VAL"
DEMO_DEVICE_PREFIX = "DEMO-DEV"
DEMO_EQUIPMENT_PREFIX = "DEMO-EQ"
DEMO_UPLOAD_ROOT = Path("/app/uploads/demo-data")
DEMO_REPORT_ROOT = Path("/app/reports")
@dataclass(frozen=True)
class DemoCustomerTemplate:
name: str
specialty: str
street: str
postal_code: str
city: str
phone: str
email: str
hygiene_officer: str
quality_manager: str
operator: str
locations: list[dict[str, Any]]
contacts: list[dict[str, Any]]
device: dict[str, Any]
CUSTOMER_TEMPLATES: list[DemoCustomerTemplate] = [
DemoCustomerTemplate(
name="Zahnzentrum Alster",
specialty="Zahnarzt",
street="Alsterufer 18",
postal_code="20354",
city="Hamburg",
phone="040 5550010",
email="kontakt@zahnzentrum-alster.schubamed.de",
hygiene_officer="Dr. Jana Peters",
quality_manager="M. Kruse",
operator="Dr. Jana Peters",
locations=[
{"name": "Hauptpraxis", "street": "Alsterufer 18", "postal_code": "20354", "city": "Hamburg", "room": "Steri 1"},
{"name": "OP-Bereich", "street": "Alsterufer 18", "postal_code": "20354", "city": "Hamburg", "room": "OP 2"},
],
contacts=[
{"full_name": "Dr. Jana Peters", "function": "Betreiberin", "email": "jana.peters@schubamed.de", "phone": "040 5550011"},
{"full_name": "M. Kruse", "function": "QM-Beauftragte", "email": "qm.zahnzentrum.alster@schubamed.de", "phone": "040 5550012"},
],
device={
"manufacturer": "Euronda SpA",
"model": "E10.7",
"serial_number": "SN-DEMO-001",
"year_built": 2023,
"commissioned_on": date(2023, 6, 14),
"chamber_volume_liters": 24,
"steam_generation": "Eigendampferzeugung",
"water_treatment": "Vollentsalzung",
"documentation": "CF-Karte und PDF-Protokoll",
"supplier": "schubamed Medizintechnik",
},
),
DemoCustomerTemplate(
name="Urologie Zentrum Elbe",
specialty="Urologie",
street="Mönckebergstr. 11",
postal_code="20095",
city="Hamburg",
phone="040 5550020",
email="kontakt@urologie-elbe.schubamed.de",
hygiene_officer="Dr. Felix Brandt",
quality_manager="S. Lange",
operator="Dr. Felix Brandt",
locations=[
{"name": "Praxis Mitte", "street": "Mönckebergstr. 11", "postal_code": "20095", "city": "Hamburg", "room": "Steri A"},
],
contacts=[
{"full_name": "Dr. Felix Brandt", "function": "Betreiber", "email": "felix.brandt@schubamed.de", "phone": "040 5550021"},
],
device={
"manufacturer": "Euronda SpA",
"model": "E9",
"serial_number": "SN-DEMO-002",
"year_built": 2022,
"commissioned_on": date(2022, 11, 2),
"chamber_volume_liters": 18,
"steam_generation": "Generator integriert",
"water_treatment": "Wasserkonditionierung",
"documentation": "Digitale Dokumentation",
"supplier": "schubamed Medizintechnik",
},
),
DemoCustomerTemplate(
name="Orthopädie am Park",
specialty="Orthopädie",
street="Königsallee 44",
postal_code="40212",
city="Düsseldorf",
phone="0211 5550030",
email="kontakt@orthopaedie-park.schubamed.de",
hygiene_officer="Dr. Lea Sommer",
quality_manager="A. Hoffmann",
operator="Dr. Lea Sommer",
locations=[
{"name": "Ambulanz", "street": "Königsallee 44", "postal_code": "40212", "city": "Düsseldorf", "room": "Steri 1"},
{"name": "Behandlungszentrum", "street": "Kaiserswerther Str. 99", "postal_code": "40474", "city": "Düsseldorf", "room": "Steri 2"},
],
contacts=[
{"full_name": "Dr. Lea Sommer", "function": "Betreiberin", "email": "lea.sommer@schubamed.de", "phone": "0211 5550031"},
{"full_name": "A. Hoffmann", "function": "Hygienebeauftragte", "email": "hygiene.orthopaedie@schubamed.de", "phone": "0211 5550032"},
],
device={
"manufacturer": "MELAG",
"model": "Vacuklav 41 B+",
"serial_number": "SN-DEMO-003",
"year_built": 2021,
"commissioned_on": date(2021, 9, 8),
"chamber_volume_liters": 22,
"steam_generation": "Dampferzeuger intern",
"water_treatment": "VE-Wasser",
"documentation": "CF-Karte, USB-Export",
"supplier": "schubamed Medizintechnik",
},
),
DemoCustomerTemplate(
name="Dermatologie Rhein",
specialty="Dermatologie",
street="Breite Str. 8",
postal_code="50667",
city="Köln",
phone="0221 5550040",
email="kontakt@dermatologie-rhein.schubamed.de",
hygiene_officer="Dr. Sophie Keller",
quality_manager="D. Neumann",
operator="Dr. Sophie Keller",
locations=[
{"name": "Hautpraxis", "street": "Breite Str. 8", "postal_code": "50667", "city": "Köln", "room": "Steri"},
],
contacts=[
{"full_name": "Dr. Sophie Keller", "function": "Betreiberin", "email": "sophie.keller@schubamed.de", "phone": "0221 5550041"},
{"full_name": "D. Neumann", "function": "QM", "email": "qm.dermatologie.rhein@schubamed.de", "phone": "0221 5550042"},
],
device={
"manufacturer": "MELAG",
"model": "Vacuklav 44 B+",
"serial_number": "SN-DEMO-004",
"year_built": 2020,
"commissioned_on": date(2020, 4, 16),
"chamber_volume_liters": 24,
"steam_generation": "Eigendampferzeugung",
"water_treatment": "Osmose",
"documentation": "Digitale Dokumentation und Ausdruck",
"supplier": "schubamed Medizintechnik",
},
),
DemoCustomerTemplate(
name="HNO Zentrum Forum",
specialty="HNO",
street="Theatinerstr. 19",
postal_code="80333",
city="München",
phone="089 5550050",
email="kontakt@hno-forum.schubamed.de",
hygiene_officer="Dr. Tom Berger",
quality_manager="R. Wagner",
operator="Dr. Tom Berger",
locations=[
{"name": "Forum Praxis", "street": "Theatinerstr. 19", "postal_code": "80333", "city": "München", "room": "Steri 1"},
],
contacts=[
{"full_name": "Dr. Tom Berger", "function": "Betreiber", "email": "tom.berger@schubamed.de", "phone": "089 5550051"},
],
device={
"manufacturer": "MELAG",
"model": "PrimeLine",
"serial_number": "SN-DEMO-005",
"year_built": 2024,
"commissioned_on": date(2024, 3, 1),
"chamber_volume_liters": 29,
"steam_generation": "Generator integriert",
"water_treatment": "VE-Wasser",
"documentation": "USB-Export",
"supplier": "schubamed Medizintechnik",
},
),
DemoCustomerTemplate(
name="Gynäkologie Marienhof",
specialty="Gynäkologie",
street="Marienplatz 3",
postal_code="86150",
city="Augsburg",
phone="0821 5550060",
email="kontakt@gyn-marienhof.schubamed.de",
hygiene_officer="Dr. Miriam Wolf",
quality_manager="K. Braun",
operator="Dr. Miriam Wolf",
locations=[
{"name": "Frauenpraxis", "street": "Marienplatz 3", "postal_code": "86150", "city": "Augsburg", "room": "Steri"},
{"name": "Ambulanz Süd", "street": "Bürgermeister-Fischer-Str. 12", "postal_code": "86150", "city": "Augsburg", "room": "Steri 2"},
],
contacts=[
{"full_name": "Dr. Miriam Wolf", "function": "Betreiberin", "email": "miriam.wolf@schubamed.de", "phone": "0821 5550061"},
],
device={
"manufacturer": "MELAG",
"model": "ProLine",
"serial_number": "SN-DEMO-006",
"year_built": 2019,
"commissioned_on": date(2019, 8, 23),
"chamber_volume_liters": 18,
"steam_generation": "Kompaktgenerator",
"water_treatment": "VE-Wasser",
"documentation": "CF-Card",
"supplier": "schubamed Medizintechnik",
},
),
DemoCustomerTemplate(
name="Chirurgie Ost",
specialty="Chirurgie",
street="Lindenstr. 7",
postal_code="04109",
city="Leipzig",
phone="0341 5550070",
email="kontakt@chirurgie-ost.schubamed.de",
hygiene_officer="Dr. Paul Richter",
quality_manager="F. Scholz",
operator="Dr. Paul Richter",
locations=[
{"name": "OP-Zentrum", "street": "Lindenstr. 7", "postal_code": "04109", "city": "Leipzig", "room": "Steri OP"},
],
contacts=[
{"full_name": "Dr. Paul Richter", "function": "Betreiber", "email": "paul.richter@schubamed.de", "phone": "0341 5550071"},
{"full_name": "F. Scholz", "function": "Hygiene", "email": "hygiene.chirurgie.ost@schubamed.de", "phone": "0341 5550072"},
],
device={
"manufacturer": "W&H",
"model": "Lara XL",
"serial_number": "SN-DEMO-007",
"year_built": 2023,
"commissioned_on": date(2023, 2, 15),
"chamber_volume_liters": 17,
"steam_generation": "Dampferzeuger intern",
"water_treatment": "Mikrofiltration",
"documentation": "USB und Ausdruck",
"supplier": "schubamed Medizintechnik",
},
),
DemoCustomerTemplate(
name="Allgemeinmedizin West",
specialty="Allgemeinmedizin",
street="Bergstr. 25",
postal_code="70173",
city="Stuttgart",
phone="0711 5550080",
email="kontakt@allgemeinmedizin-west.schubamed.de",
hygiene_officer="Dr. Nina Beck",
quality_manager="J. Hartmann",
operator="Dr. Nina Beck",
locations=[
{"name": "Hausarztpraxis", "street": "Bergstr. 25", "postal_code": "70173", "city": "Stuttgart", "room": "Steri"},
],
contacts=[
{"full_name": "Dr. Nina Beck", "function": "Betreiberin", "email": "nina.beck@schubamed.de", "phone": "0711 5550081"},
],
device={
"manufacturer": "W&H",
"model": "Lisa",
"serial_number": "SN-DEMO-008",
"year_built": 2022,
"commissioned_on": date(2022, 7, 12),
"chamber_volume_liters": 22,
"steam_generation": "Eigendampferzeugung",
"water_treatment": "VE-Wasser",
"documentation": "Digital und Papier",
"supplier": "schubamed Medizintechnik",
},
),
DemoCustomerTemplate(
name="Augenärzte am Ring",
specialty="Augenarzt",
street="Ringstr. 52",
postal_code="50672",
city="Köln",
phone="0221 5550090",
email="kontakt@augenaerzte-ring.schubamed.de",
hygiene_officer="Dr. Karin Otto",
quality_manager="M. Franke",
operator="Dr. Karin Otto",
locations=[
{"name": "Augenzentrum", "street": "Ringstr. 52", "postal_code": "50672", "city": "Köln", "room": "Steri 1"},
{"name": "Laserzentrum", "street": "Ringstr. 54", "postal_code": "50672", "city": "Köln", "room": "Steri 2"},
],
contacts=[
{"full_name": "Dr. Karin Otto", "function": "Betreiberin", "email": "karin.otto@schubamed.de", "phone": "0221 5550091"},
{"full_name": "M. Franke", "function": "QM", "email": "qm.augen@schubamed.de", "phone": "0221 5550092"},
],
device={
"manufacturer": "Mocom",
"model": "B Classic",
"serial_number": "SN-DEMO-009",
"year_built": 2021,
"commissioned_on": date(2021, 5, 20),
"chamber_volume_liters": 12,
"steam_generation": "Kompaktgenerator",
"water_treatment": "Vollentsalzung",
"documentation": "USB-Export",
"supplier": "schubamed Medizintechnik",
},
),
DemoCustomerTemplate(
name="MVZ Medikon",
specialty="MVZ",
street="Forum 9",
postal_code="90402",
city="Nürnberg",
phone="0911 5550100",
email="kontakt@mvz-medikon.schubamed.de",
hygiene_officer="Dr. Lara Stein",
quality_manager="C. Meier",
operator="Dr. Lara Stein",
locations=[
{"name": "Hauptstandort", "street": "Forum 9", "postal_code": "90402", "city": "Nürnberg", "room": "Steri A"},
{"name": "Nebenstandort", "street": "Forum 11", "postal_code": "90402", "city": "Nürnberg", "room": "Steri B"},
],
contacts=[
{"full_name": "Dr. Lara Stein", "function": "Betreiberin", "email": "lara.stein@schubamed.de", "phone": "0911 5550101"},
{"full_name": "C. Meier", "function": "Hygiene", "email": "hygiene.mvz@schubamed.de", "phone": "0911 5550102"},
],
device={
"manufacturer": "Kronos",
"model": "B23",
"serial_number": "SN-DEMO-010",
"year_built": 2024,
"commissioned_on": date(2024, 1, 11),
"chamber_volume_liters": 23,
"steam_generation": "Eigendampferzeugung",
"water_treatment": "VE-Wasser",
"documentation": "Hybrid",
"supplier": "schubamed Medizintechnik",
},
),
]
EQUIPMENT_TEMPLATES: list[dict[str, Any]] = [
{
"kind": EquipmentKind.temperature_logger,
"manufacturer": "EBRO",
"model": "EBI 11",
"serial_number": "DEMO-EBI11-001",
"calibrated_on": date(2025, 1, 17),
"calibration_due_on": date(2026, 1, 17),
"status": EquipmentStatus.green,
"notes": "Demo-Temperaturlogger",
},
{
"kind": EquipmentKind.temperature_logger,
"manufacturer": "EBRO",
"model": "EBI 11",
"serial_number": "DEMO-EBI11-002",
"calibrated_on": date(2025, 2, 2),
"calibration_due_on": date(2026, 2, 2),
"status": EquipmentStatus.green,
"notes": "Demo-Temperaturlogger",
},
{
"kind": EquipmentKind.pressure_logger,
"manufacturer": "EBRO",
"model": "Drucklogger",
"serial_number": "DEMO-PR-001",
"calibrated_on": date(2025, 1, 21),
"calibration_due_on": date(2026, 1, 21),
"status": EquipmentStatus.green,
"notes": "Demo-Drucklogger",
},
{
"kind": EquipmentKind.sensor,
"manufacturer": "Mettler Toledo",
"model": "Waage",
"serial_number": "DEMO-SCALE-001",
"calibrated_on": date(2024, 12, 12),
"calibration_due_on": date(2025, 12, 12),
"status": EquipmentStatus.yellow,
"notes": "Demo-Waage",
},
{
"kind": EquipmentKind.sensor,
"manufacturer": "Mettler Toledo",
"model": "Leitwertmessgerät",
"serial_number": "DEMO-COND-001",
"calibrated_on": date(2025, 3, 4),
"calibration_due_on": date(2026, 3, 4),
"status": EquipmentStatus.green,
"notes": "Demo-Leitwertmessgerät",
},
{
"kind": EquipmentKind.sensor,
"manufacturer": "Trotec",
"model": "Raumklimamessgerät",
"serial_number": "DEMO-CLIMATE-001",
"calibrated_on": date(2024, 11, 1),
"calibration_due_on": date(2025, 11, 1),
"status": EquipmentStatus.red,
"notes": "Demo-Raumklimamessgerät",
},
]
class DemoDataService:
def __init__(self, session: Session, upload_root: Path | None = None, report_root: Path | None = None) -> None:
self.session = session
self.upload_root = upload_root or DEMO_UPLOAD_ROOT.parent
self.report_root = report_root or DEMO_REPORT_ROOT
self.demo_upload_root = self.upload_root / "demo-data"
def run(
self,
*,
customers: int = 10,
devices: int = 10,
validations: int = 20,
reports: bool = False,
images: bool = False,
reset: bool = False,
) -> dict[str, int]:
if reset:
self.reset()
created_customers = self.ensure_customers(max(customers, 10))
created_equipment = self.ensure_equipment()
created_devices = self.ensure_devices(max(devices, 10), created_customers)
created_validations = self.ensure_validations(max(validations, 20), created_customers, created_devices, images=images)
self.session.commit()
if reports:
self._generate_reports(created_validations)
return {
"customers": len(created_customers),
"devices": len(created_devices),
"validations": len(created_validations),
"equipment": len(created_equipment),
}
def reset(self) -> None:
demo_validation_ids = self.session.scalars(
select(Validation.id).where(
Validation.report_number.like(f"{DEMO_REPORT_PREFIX}%")
| Validation.notes.like(f"%{DEMO_TAG}%")
)
).all()
if demo_validation_ids:
self.session.execute(delete(Validation).where(Validation.id.in_(demo_validation_ids)))
self.session.execute(
delete(Equipment).where(Equipment.serial_number.like("DEMO-%") | Equipment.notes.like(f"%{DEMO_TAG}%"))
)
self.session.execute(
delete(Device).where(Device.serial_number.like("SN-DEMO-%") | Device.notes.like(f"%{DEMO_TAG}%"))
)
self.session.execute(
delete(Contact).where(
Contact.email.like("demo-%@schubamed.de") | Contact.notes.like(f"%{DEMO_TAG}%")
)
)
self.session.execute(
delete(Customer).where(
Customer.email.like("demo-%@schubamed.de") | Customer.notes.like(f"%{DEMO_TAG}%")
)
)
self.session.commit()
if self.demo_upload_root.exists():
shutil.rmtree(self.demo_upload_root)
self.report_root.mkdir(parents=True, exist_ok=True)
for pdf in self.report_root.glob(f"{DEMO_REPORT_PREFIX}*.pdf"):
pdf.unlink()
def ensure_customers(self, count: int) -> list[Customer]:
ensured: list[Customer] = []
for index in range(count):
template = CUSTOMER_TEMPLATES[index % len(CUSTOMER_TEMPLATES)]
occurrence = index // len(CUSTOMER_TEMPLATES)
customer = self._ensure_customer(template, occurrence, index)
ensured.append(customer)
return ensured
def ensure_equipment(self) -> list[Equipment]:
items: list[Equipment] = []
for template in EQUIPMENT_TEMPLATES:
equipment = self.session.scalar(
select(Equipment).where(Equipment.serial_number == template["serial_number"])
)
if equipment is None:
equipment = Equipment(**template)
equipment.notes = f"{template['notes']} {DEMO_TAG}"
self.session.add(equipment)
self.session.flush()
items.append(equipment)
return items
def ensure_devices(self, count: int, customers: list[Customer]) -> list[Device]:
devices: list[Device] = []
all_locations = {customer.id: list(self.session.scalars(select(Location).where(Location.customer_id == customer.id))) for customer in customers}
for index in range(count):
customer = customers[index % len(customers)]
template = CUSTOMER_TEMPLATES[index % len(CUSTOMER_TEMPLATES)].device
occurrence = index // len(CUSTOMER_TEMPLATES)
serial_number = template["serial_number"] if occurrence == 0 else f"{template['serial_number']}-{occurrence + 1}"
device = self.session.scalar(select(Device).where(Device.serial_number == serial_number))
if device is None:
location_candidates = all_locations.get(customer.id, [])
location_id = location_candidates[0].id if location_candidates else None
device_payload = deepcopy(template)
device_payload["serial_number"] = serial_number
device_payload["customer_id"] = customer.id
device_payload["location_id"] = location_id
device = Device(**device_payload)
device.notes = f"{DEMO_TAG} {customer.name}"
self.session.add(device)
self.session.flush()
devices.append(device)
return devices
def ensure_validations(
self,
count: int,
customers: list[Customer],
devices: list[Device],
*,
images: bool = False,
) -> list[Validation]:
validations: list[Validation] = []
report_number_index = 1
for index in range(count):
template = CUSTOMER_TEMPLATES[index % len(CUSTOMER_TEMPLATES)]
customer = customers[index % len(customers)]
device = devices[index % len(devices)]
locations = list(self.session.scalars(select(Location).where(Location.customer_id == customer.id)))
contacts = list(self.session.scalars(select(Contact).where(Contact.customer_id == customer.id)))
location = locations[index % len(locations)] if locations else None
contact = contacts[index % len(contacts)] if contacts else None
report_number = f"{DEMO_REPORT_PREFIX}-{report_number_index:03d}"
report_number_index += 1
validation = self.session.scalar(select(Validation).where(Validation.report_number == report_number))
if validation is None:
validation = Validation(
report_number=report_number,
customer_id=customer.id,
location_id=location.id if location else None,
contact_id=contact.id if contact else None,
device_id=device.id,
validation_type="Erstvalidierung" if index % 2 == 0 else "Revalidierung",
performed_on=date.today() - relativedelta(months=6 + index),
examiner_name="Dr. Lara Stein",
operator_name=template.operator,
participants=f"{customer.name} - Team",
status=[
ValidationStatus.draft.value,
ValidationStatus.approved.value,
ValidationStatus.completed.value,
][index % 3],
result="bestanden" if index % 3 else "bestanden mit Auflagen",
revalidation_interval_months=24,
equipment_ids=[equipment.id for equipment in self.session.scalars(select(Equipment).limit(3))],
environment_conditions={
"room_temperature": 22 + (index % 3),
"humidity": 48 + (index % 5),
"test_time": f"{8 + (index % 3)}:30",
"checks": [
{"text": "Raumbedingungen stabil", "value": "yes", "comment": ""},
{"text": "Aufstellort frei zugänglich", "value": "yes", "comment": ""},
],
},
documentation_checklist=self._checklist("documentation_control", index),
performance_checklist=self._checklist("sterilizer_description", index),
programs=[
{"name": "Vakuumtest", "selected": True, "custom": False},
{"name": "Bowie-Dick / Leerkammerprofil", "selected": index % 2 == 0, "custom": False},
{"name": "134 C hohl verpackt", "selected": True, "custom": False},
],
loading_patterns=self._loading_patterns(index),
measurement_data=self._measurement_data(index),
drying={"start_weight": 12.5, "end_weight": 12.1, "difference": 0.4, "assessment": "in Ordnung", "comment": ""},
recommendations=[
{"number": 1, "text": "Routinekontrolle dokumentieren", "deadline": "6 Monate", "status": "offen"}
] if index % 4 == 0 else [],
attachments=self._attachments_for_validation(index, images=images),
notes=DEMO_TAG,
)
self.session.add(validation)
self.session.flush()
ValidationWorkflowService(self.session).apply_revalidation_date(validation)
validations.append(validation)
return validations
def _ensure_customer(self, template: DemoCustomerTemplate, occurrence: int, index: int) -> Customer:
email = template.email if occurrence == 0 else f"demo-{index + 1:02d}@schubamed.de"
customer = self.session.scalar(select(Customer).where(Customer.email == email))
if customer is None:
customer = Customer(
customer_type=CustomerType.practice,
name=template.name if occurrence == 0 else f"{template.name} {occurrence + 1}",
street=template.street,
postal_code=template.postal_code,
city=template.city,
phone=template.phone,
email=email,
hygiene_officer=template.hygiene_officer,
quality_manager=template.quality_manager,
notes=f"{DEMO_TAG} {template.specialty}",
)
self.session.add(customer)
self.session.flush()
for loc_index, location_payload in enumerate(template.locations, start=1):
location = self.session.scalar(
select(Location).where(
Location.customer_id == customer.id,
Location.name == location_payload["name"],
)
)
if location is None:
self.session.add(Location(customer_id=customer.id, **location_payload))
for contact_payload in template.contacts:
contact = self.session.scalar(
select(Contact).where(
Contact.customer_id == customer.id,
Contact.full_name == contact_payload["full_name"],
)
)
if contact is None:
self.session.add(
Contact(
customer_id=customer.id,
full_name=contact_payload["full_name"],
function=contact_payload["function"],
email=contact_payload["email"],
phone=contact_payload["phone"],
notes=DEMO_TAG,
)
)
self.session.flush()
return customer
def _checklist(self, key: str, index: int) -> list[dict[str, Any]]:
bundle = ReportTemplateService(self.session).ensure_default_template()
template = next((item for item in bundle.checklists if item.checklist_key == key), None)
if template is None:
return []
values = ["yes", "yes", "na"] if index % 2 == 0 else ["yes", "na", "yes"]
return [
{
"number": item["number"],
"text": item["text"],
"value": values[(item["number"] - 1) % len(values)],
"comment": "",
}
for item in template.items
]
def _loading_patterns(self, index: int) -> list[dict[str, Any]]:
return [
{
"run": 1,
"pattern": "Standard",
"description": f"Demo-Beladung {index + 1}",
"images": [],
},
{
"run": 2,
"pattern": "Standard",
"description": "",
"images": [],
},
{
"run": 3,
"pattern": "Standard",
"description": "",
"images": [],
},
]
def _measurement_data(self, index: int) -> list[dict[str, Any]]:
base_temperature = 132 + (index % 3)
return [
{
"test_run": "Vakuumtest",
"start_time": "08:00",
"end_time": "08:08",
"duration": "00:08",
"leak_rate": "0.4",
"result": "bestanden",
},
{
"test_run": "Testlauf 1",
"start_time": "09:00",
"end_time": "09:31",
"duration": "00:31",
"minimum_temperature": base_temperature,
"maximum_temperature": base_temperature + 2,
"temperature_band": "2.0",
"holding_time": "3:00",
"pressure": "2.1",
"result": "bestanden",
},
]
def _attachments_for_validation(self, index: int, *, images: bool) -> list[dict[str, Any]]:
if not images or index >= 5:
return []
image_dir = self.demo_upload_root / "images"
image_dir.mkdir(parents=True, exist_ok=True)
image_path = image_dir / f"demo-figure-{index + 1}.svg"
if not image_path.exists():
image_path.write_text(
"\n".join(
[
'<svg xmlns="http://www.w3.org/2000/svg" width="1200" height="800" viewBox="0 0 1200 800">',
'<rect width="1200" height="800" fill="#f6f7f2"/>',
'<rect x="60" y="60" width="1080" height="680" rx="18" fill="#ffffff" stroke="#bfc5bd" stroke-width="3"/>',
'<text x="100" y="150" font-family="Arial, sans-serif" font-size="44" fill="#2f3b35">SCHUBAMED Demo-Bild</text>',
f'<text x="100" y="230" font-family="Arial, sans-serif" font-size="32" fill="#4d5b54">Validierung {index + 1}</text>',
"</svg>",
]
),
encoding="utf-8",
)
return [
{
"category": "Beladung",
"filename": image_path.name,
"content_type": "image/svg+xml",
"description": f"Demo-Abbildung {index + 1}",
"order": 1,
"storage_path": str(image_path),
}
]
def _generate_reports(self, validations: list[Validation]) -> None:
self.report_root.mkdir(parents=True, exist_ok=True)
ReportTemplateService(self.session).ensure_default_template()
for validation in validations:
OrionReportService(self.session, self.report_root).render_pdf(validation.id)

4
validation-suite/backend/mercury/app/services/domain_service.py
View file

@ -36,11 +36,13 @@ class CrudService:
safe_page = max(page, 1)
safe_page_size = min(max(page_size, 1), 100)
offset = (safe_page - 1) * safe_page_size
total = self.repository.count(search)
return {
"items": self.repository.list(safe_page_size, offset, search),
"total": self.repository.count(search),
"total": total,
"page": safe_page,
"page_size": safe_page_size,
"pages": max((total + safe_page_size - 1) // safe_page_size, 1),
}
def create(self, data: dict) -> ModelT:

256
validation-suite/backend/mercury/app/services/quickstart_service.py Normal file
View file

@ -0,0 +1,256 @@
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.")
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)
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 if location else (locations[0].id if locations else None),
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": []},
]

14
validation-suite/backend/mercury/app/services/validation_workflow.py
View file

@ -7,7 +7,7 @@ from datetime import date, datetime
from uuid import uuid4
from dateutil.relativedelta import relativedelta
from sqlalchemy import and_, or_, select
from sqlalchemy import and_, func, or_, select
from sqlalchemy.orm import Session
from app.models.contact import Contact
@ -224,9 +224,17 @@ class ValidationWorkflowService:
statement = statement.order_by(sort_column.asc())
else:
statement = statement.order_by(sort_column.desc())
total = len(list(self.session.scalars(count_statement)))
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}
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: