feat(validation): add editing versioning revalidation and aligned reports
This commit is contained in:
Schubert Ferenc
parent
f73a24df13
commit
302e542fda
28 changed files with 2691 additions and 406 deletions
|
|
@ -0,0 +1,42 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from alembic import op
|
||||
|
||||
revision = "202607110001"
|
||||
down_revision = "202607100002"
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.execute("ALTER TABLE validations ALTER COLUMN status TYPE varchar(40) USING status::text")
|
||||
op.execute(
|
||||
"""
|
||||
UPDATE validations
|
||||
SET status = CASE status
|
||||
WHEN 'draft' THEN 'ENTWURF'
|
||||
WHEN 'ready_for_report' THEN 'BEREIT_ZUR_PRUEFUNG'
|
||||
WHEN 'in_progress' THEN 'IN_PRUEFUNG'
|
||||
WHEN 'completed' THEN 'ABGESCHLOSSEN'
|
||||
ELSE status
|
||||
END
|
||||
"""
|
||||
)
|
||||
op.execute("ALTER TABLE validations ALTER COLUMN status SET DEFAULT 'ENTWURF'")
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.execute(
|
||||
"""
|
||||
UPDATE validations
|
||||
SET status = CASE status
|
||||
WHEN 'ENTWURF' THEN 'draft'
|
||||
WHEN 'BEREIT_ZUR_PRUEFUNG' THEN 'ready_for_report'
|
||||
WHEN 'IN_PRUEFUNG' THEN 'in_progress'
|
||||
WHEN 'FREIGEGEBEN' THEN 'ready_for_report'
|
||||
WHEN 'ABGESCHLOSSEN' THEN 'completed'
|
||||
WHEN 'STORNIERT' THEN 'completed'
|
||||
ELSE status
|
||||
END
|
||||
"""
|
||||
)
|
||||
|
|
@ -0,0 +1,25 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
|
||||
revision = "202607110002"
|
||||
down_revision = "202607110001"
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.add_column("validations", sa.Column("revalidation_interval_months", sa.Integer(), nullable=False, server_default="24"))
|
||||
op.add_column("validations", sa.Column("next_validation_manually_overridden", sa.Boolean(), nullable=False, server_default=sa.false()))
|
||||
op.add_column("validations", sa.Column("version", sa.Integer(), nullable=False, server_default="1"))
|
||||
op.add_column("validations", sa.Column("previous_validation_id", sa.String(), nullable=True))
|
||||
op.create_foreign_key("fk_validations_previous_validation_id_validations", "validations", "validations", ["previous_validation_id"], ["id"])
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.drop_constraint("fk_validations_previous_validation_id_validations", "validations", type_="foreignkey")
|
||||
op.drop_column("validations", "previous_validation_id")
|
||||
op.drop_column("validations", "version")
|
||||
op.drop_column("validations", "next_validation_manually_overridden")
|
||||
op.drop_column("validations", "revalidation_interval_months")
|
||||
|
|
@ -2,8 +2,13 @@ from __future__ import annotations
|
|||
|
||||
from typing import Any
|
||||
|
||||
from fastapi import APIRouter, Depends, Query, Response
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
from fastapi import APIRouter, Depends, Query, Response, UploadFile
|
||||
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.api.dependencies import current_user
|
||||
|
|
@ -14,6 +19,7 @@ from app.models.device import Device
|
|||
from app.models.equipment import Equipment
|
||||
from app.models.location import Location
|
||||
from app.models.validation import Validation
|
||||
from app.modules.orion.service import OrionReportService
|
||||
from app.schemas.common import PaginatedResponse
|
||||
from app.schemas.domain import (
|
||||
ContactCreate,
|
||||
|
|
@ -32,16 +38,22 @@ from app.schemas.domain import (
|
|||
LocationRead,
|
||||
LocationUpdate,
|
||||
ValidationCreate,
|
||||
ValidationImportPreview,
|
||||
ValidationImportRequest,
|
||||
ValidationImportSummary,
|
||||
ValidationRead,
|
||||
ValidationReview,
|
||||
ValidationUpdate,
|
||||
)
|
||||
from app.services.domain_service import CrudService, DomainServices
|
||||
from app.services.validation_workflow import ValidationWorkflowService
|
||||
|
||||
router = APIRouter(tags=["domain"], dependencies=[Depends(current_user)])
|
||||
|
||||
|
||||
@router.get("/dashboard")
|
||||
def dashboard(session: Session = Depends(get_session)) -> dict[str, int]:
|
||||
today = func.current_date()
|
||||
return {
|
||||
"customers": session.scalar(select(func.count()).select_from(Customer)) or 0,
|
||||
"locations": session.scalar(select(func.count()).select_from(Location)) or 0,
|
||||
|
|
@ -49,6 +61,11 @@ def dashboard(session: Session = Depends(get_session)) -> dict[str, int]:
|
|||
"devices": session.scalar(select(func.count()).select_from(Device)) or 0,
|
||||
"equipment": session.scalar(select(func.count()).select_from(Equipment)) or 0,
|
||||
"validations": session.scalar(select(func.count()).select_from(Validation)) or 0,
|
||||
"validation_drafts": session.scalar(select(func.count()).select_from(Validation).where(Validation.status == "ENTWURF")) or 0,
|
||||
"validation_ready": session.scalar(select(func.count()).select_from(Validation).where(Validation.status == "BEREIT_ZUR_PRUEFUNG")) or 0,
|
||||
"validation_in_review": session.scalar(select(func.count()).select_from(Validation).where(Validation.status == "IN_PRUEFUNG")) or 0,
|
||||
"validation_approved": session.scalar(select(func.count()).select_from(Validation).where(Validation.status == "FREIGEGEBEN")) or 0,
|
||||
"validation_overdue": session.scalar(select(func.count()).select_from(Validation).where(Validation.next_validation_on < today)) or 0,
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -61,17 +78,33 @@ def paging(
|
|||
|
||||
|
||||
def commit_create(session: Session, service: CrudService, payload):
|
||||
item = service.create(payload.model_dump())
|
||||
session.commit()
|
||||
session.refresh(item)
|
||||
return item
|
||||
try:
|
||||
item = service.create(payload.model_dump())
|
||||
if isinstance(item, Validation):
|
||||
ValidationWorkflowService(session).apply_revalidation_date(item)
|
||||
session.commit()
|
||||
session.refresh(item)
|
||||
return item
|
||||
except IntegrityError as exc:
|
||||
session.rollback()
|
||||
from fastapi import HTTPException
|
||||
|
||||
raise HTTPException(status_code=409, detail="Datensatz verletzt Datenbankbeziehungen.") from exc
|
||||
|
||||
|
||||
def commit_update(session: Session, service: CrudService, item_id: str, payload):
|
||||
item = service.update(item_id, payload.model_dump())
|
||||
session.commit()
|
||||
session.refresh(item)
|
||||
return item
|
||||
try:
|
||||
item = service.update(item_id, payload.model_dump())
|
||||
if isinstance(item, Validation):
|
||||
ValidationWorkflowService(session).apply_revalidation_date(item)
|
||||
session.commit()
|
||||
session.refresh(item)
|
||||
return item
|
||||
except IntegrityError as exc:
|
||||
session.rollback()
|
||||
from fastapi import HTTPException
|
||||
|
||||
raise HTTPException(status_code=409, detail="Datensatz verletzt Datenbankbeziehungen.") from exc
|
||||
|
||||
|
||||
def commit_delete(session: Session, service: CrudService, item_id: str) -> Response:
|
||||
|
|
@ -181,8 +214,39 @@ def delete_equipment(item_id: str, session: Session = Depends(get_session)):
|
|||
|
||||
|
||||
@router.get("/validations", response_model=PaginatedResponse[ValidationRead])
|
||||
def list_validations(params: dict = Depends(paging), session: Session = Depends(get_session)):
|
||||
return DomainServices(session).validations.list(**params)
|
||||
def list_validations(
|
||||
search: str | None = Query(default=None, max_length=120),
|
||||
page: int = Query(default=1, ge=1),
|
||||
page_size: int = Query(default=20, ge=1, le=100),
|
||||
sort_by: str = Query(default="updated_at"),
|
||||
sort_order: str = Query(default="desc", pattern="^(asc|desc)$"),
|
||||
status: str | None = None,
|
||||
customer_id: str | None = None,
|
||||
device_id: str | None = None,
|
||||
validation_type: str | None = None,
|
||||
result: str | None = None,
|
||||
date_from: str | None = None,
|
||||
date_to: str | None = None,
|
||||
overdue_only: bool = False,
|
||||
session: Session = Depends(get_session),
|
||||
):
|
||||
return ValidationWorkflowService(session).query_validations(
|
||||
search=search,
|
||||
page=page,
|
||||
page_size=page_size,
|
||||
sort_by=sort_by,
|
||||
sort_order=sort_order,
|
||||
filters={
|
||||
"status": status,
|
||||
"customer_id": customer_id,
|
||||
"device_id": device_id,
|
||||
"validation_type": validation_type,
|
||||
"result": result,
|
||||
"date_from": date_from,
|
||||
"date_to": date_to,
|
||||
"overdue_only": overdue_only,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@router.get("/validations/next-report-number")
|
||||
|
|
@ -213,4 +277,101 @@ def update_validation(item_id: str, payload: ValidationUpdate, session: Session
|
|||
|
||||
@router.delete("/validations/{item_id}", status_code=204)
|
||||
def delete_validation(item_id: str, session: Session = Depends(get_session)):
|
||||
item = DomainServices(session).validations.repository.get(item_id)
|
||||
if item and item.status != "ENTWURF":
|
||||
from fastapi import HTTPException
|
||||
|
||||
raise HTTPException(status_code=409, detail="Only draft validations can be deleted")
|
||||
return commit_delete(session, DomainServices(session).validations, item_id)
|
||||
|
||||
|
||||
@router.post("/validations/{item_id}/review", response_model=ValidationReview)
|
||||
def review_validation(item_id: str, session: Session = Depends(get_session)):
|
||||
item = DomainServices(session).validations.repository.get(item_id)
|
||||
if item is None:
|
||||
from fastapi import HTTPException
|
||||
|
||||
raise HTTPException(status_code=404, detail="Resource not found")
|
||||
review = ValidationWorkflowService(session).mark_ready_for_review(item)
|
||||
session.commit()
|
||||
return review
|
||||
|
||||
|
||||
@router.post("/validations/{item_id}/duplicate", response_model=ValidationRead, status_code=201)
|
||||
def duplicate_validation(item_id: str, session: Session = Depends(get_session)):
|
||||
item = DomainServices(session).validations.repository.get(item_id)
|
||||
if item is None:
|
||||
from fastapi import HTTPException
|
||||
|
||||
raise HTTPException(status_code=404, detail="Resource not found")
|
||||
clone = ValidationWorkflowService(session).duplicate(item)
|
||||
session.commit()
|
||||
session.refresh(clone)
|
||||
return clone
|
||||
|
||||
|
||||
@router.post("/validations/{item_id}/new-version", response_model=ValidationRead, status_code=201)
|
||||
def new_validation_version(item_id: str, session: Session = Depends(get_session)):
|
||||
item = DomainServices(session).validations.repository.get(item_id)
|
||||
if item is None:
|
||||
from fastapi import HTTPException
|
||||
|
||||
raise HTTPException(status_code=404, detail="Resource not found")
|
||||
clone = ValidationWorkflowService(session).create_new_version(item)
|
||||
session.commit()
|
||||
session.refresh(clone)
|
||||
return clone
|
||||
|
||||
|
||||
@router.post("/validations/{item_id}/cancel", response_model=ValidationRead)
|
||||
def cancel_validation(item_id: str, session: Session = Depends(get_session)):
|
||||
item = DomainServices(session).validations.repository.get(item_id)
|
||||
if item is None:
|
||||
from fastapi import HTTPException
|
||||
|
||||
raise HTTPException(status_code=404, detail="Resource not found")
|
||||
item.status = "STORNIERT"
|
||||
session.commit()
|
||||
session.refresh(item)
|
||||
return item
|
||||
|
||||
|
||||
@router.get("/validations/{item_id}/export.json")
|
||||
def export_validation_json(item_id: str, session: Session = Depends(get_session)):
|
||||
item = DomainServices(session).validations.repository.get(item_id)
|
||||
if item is None:
|
||||
from fastapi import HTTPException
|
||||
|
||||
raise HTTPException(status_code=404, detail="Resource not found")
|
||||
data = ValidationWorkflowService(session).export_json(item)
|
||||
return JSONResponse(
|
||||
content=json.loads(json.dumps(data, default=str)),
|
||||
headers={"Content-Disposition": f'attachment; filename="{item.report_number}.json"'},
|
||||
)
|
||||
|
||||
|
||||
@router.post("/validations/import/csv-preview", response_model=ValidationImportPreview)
|
||||
async def preview_validation_csv(file: UploadFile, session: Session = Depends(get_session)):
|
||||
return ValidationWorkflowService(session).preview_csv(await file.read())
|
||||
|
||||
|
||||
@router.post("/validations/import/json", response_model=ValidationImportSummary)
|
||||
def import_validation_json(payload: ValidationImportRequest, session: Session = Depends(get_session)):
|
||||
summary = ValidationWorkflowService(session).import_rows(payload.rows, payload.duplicate_strategy)
|
||||
session.commit()
|
||||
return summary
|
||||
|
||||
|
||||
@router.get("/validations/{item_id}/report.html", response_class=HTMLResponse)
|
||||
def validation_report_preview(item_id: str, session: Session = Depends(get_session)):
|
||||
return OrionReportService(session, Path("/app/reports")).render_html(item_id)
|
||||
|
||||
|
||||
@router.get("/validations/{item_id}/report.pdf")
|
||||
def validation_report_pdf(item_id: str, session: Session = Depends(get_session)):
|
||||
report_path = OrionReportService(session, Path("/app/reports")).render_pdf(item_id)
|
||||
return FileResponse(
|
||||
report_path,
|
||||
media_type="application/pdf",
|
||||
filename=report_path.name,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -3,17 +3,19 @@ from __future__ import annotations
|
|||
import enum
|
||||
from datetime import date
|
||||
|
||||
from sqlalchemy import Date, Enum, ForeignKey, JSON, String, Text
|
||||
from sqlalchemy import Boolean, Date, ForeignKey, Integer, JSON, String, Text
|
||||
from sqlalchemy.orm import Mapped, mapped_column
|
||||
|
||||
from app.db.base import Base, TimestampMixin, UUIDMixin
|
||||
|
||||
|
||||
class ValidationStatus(str, enum.Enum):
|
||||
draft = "draft"
|
||||
in_progress = "in_progress"
|
||||
ready_for_report = "ready_for_report"
|
||||
completed = "completed"
|
||||
draft = "ENTWURF"
|
||||
ready_for_review = "BEREIT_ZUR_PRUEFUNG"
|
||||
in_review = "IN_PRUEFUNG"
|
||||
approved = "FREIGEGEBEN"
|
||||
completed = "ABGESCHLOSSEN"
|
||||
cancelled = "STORNIERT"
|
||||
|
||||
|
||||
class Validation(Base, UUIDMixin, TimestampMixin):
|
||||
|
|
@ -22,7 +24,7 @@ class Validation(Base, UUIDMixin, TimestampMixin):
|
|||
report_number: Mapped[str] = mapped_column(String(80), unique=True, index=True)
|
||||
customer_id: Mapped[str] = mapped_column(ForeignKey("customers.id"), index=True)
|
||||
location_id: Mapped[str | None] = mapped_column(ForeignKey("locations.id"), index=True)
|
||||
contact_id: Mapped[str | None] = mapped_column(ForeignKey("contacts.id"), index=True)
|
||||
contact_id: Mapped[str | None] = mapped_column(ForeignKey("contacts.id"), index=True, nullable=True)
|
||||
device_id: Mapped[str | None] = mapped_column(ForeignKey("devices.id"), index=True)
|
||||
validation_type: Mapped[str] = mapped_column(String(120))
|
||||
project: Mapped[str | None] = mapped_column(String(180))
|
||||
|
|
@ -33,8 +35,12 @@ class Validation(Base, UUIDMixin, TimestampMixin):
|
|||
scheduled_on: Mapped[date | None] = mapped_column(Date)
|
||||
performed_on: Mapped[date | None] = mapped_column(Date)
|
||||
next_validation_on: Mapped[date | None] = mapped_column(Date)
|
||||
examiner_id: Mapped[str | None] = mapped_column(ForeignKey("users.id"))
|
||||
status: Mapped[ValidationStatus] = mapped_column(Enum(ValidationStatus), default=ValidationStatus.draft)
|
||||
revalidation_interval_months: Mapped[int] = mapped_column(Integer, default=24)
|
||||
next_validation_manually_overridden: Mapped[bool] = mapped_column(Boolean, default=False)
|
||||
version: Mapped[int] = mapped_column(Integer, default=1)
|
||||
previous_validation_id: Mapped[str | None] = mapped_column(ForeignKey("validations.id"), nullable=True)
|
||||
examiner_id: Mapped[str | None] = mapped_column(ForeignKey("users.id"), nullable=True)
|
||||
status: Mapped[str] = mapped_column(String(40), default=ValidationStatus.draft.value)
|
||||
result: Mapped[str | None] = mapped_column(String(120))
|
||||
notes: Mapped[str | None] = mapped_column(Text)
|
||||
equipment_ids: Mapped[list[str]] = mapped_column(JSON, default=list)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,37 @@
|
|||
from app.modules.orion.components.base import ReportComponent
|
||||
from app.modules.orion.components.chapters import (
|
||||
AttachmentComponent,
|
||||
ChecklistComponent,
|
||||
CoverComponent,
|
||||
CustomerComponent,
|
||||
DeviceComponent,
|
||||
DryingComponent,
|
||||
EnvironmentComponent,
|
||||
EquipmentComponent,
|
||||
LoadingComponent,
|
||||
MeasurementComponent,
|
||||
ProgramComponent,
|
||||
RecommendationComponent,
|
||||
StaticTextComponent,
|
||||
SummaryComponent,
|
||||
TocComponent,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"AttachmentComponent",
|
||||
"ChecklistComponent",
|
||||
"CoverComponent",
|
||||
"CustomerComponent",
|
||||
"DeviceComponent",
|
||||
"DryingComponent",
|
||||
"EnvironmentComponent",
|
||||
"EquipmentComponent",
|
||||
"LoadingComponent",
|
||||
"MeasurementComponent",
|
||||
"ProgramComponent",
|
||||
"RecommendationComponent",
|
||||
"StaticTextComponent",
|
||||
"ReportComponent",
|
||||
"SummaryComponent",
|
||||
"TocComponent",
|
||||
]
|
||||
|
|
@ -0,0 +1,15 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
from app.modules.orion.context import ReportContext
|
||||
|
||||
|
||||
class ReportComponent(ABC):
|
||||
anchor: str
|
||||
title: str
|
||||
|
||||
@abstractmethod
|
||||
def render(self, context: ReportContext) -> str:
|
||||
raise NotImplementedError
|
||||
|
||||
|
|
@ -0,0 +1,325 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from app.modules.orion.components.base import ReportComponent
|
||||
from app.modules.orion.context import ReportContext
|
||||
from app.modules.orion.html import definition_list, paragraph, section, table, text, yes_no
|
||||
|
||||
|
||||
class CoverComponent(ReportComponent):
|
||||
anchor = "cover"
|
||||
title = "Deckblatt"
|
||||
|
||||
def render(self, context: ReportContext) -> str:
|
||||
validation = context.validation
|
||||
rows = definition_list(
|
||||
[
|
||||
("Berichtsnummer", validation.report_number),
|
||||
("Validierungsart", validation.validation_type),
|
||||
("Projekt", validation.project),
|
||||
("Pruefdatum", validation.performed_on),
|
||||
("Pruefungsort", validation.test_location),
|
||||
("Pruefer", validation.examiner_name),
|
||||
("Gesamtergebnis", validation.result),
|
||||
("Status", validation.status),
|
||||
("Ansprechpartner", context.contact.full_name if context.contact else "nicht erfasst"),
|
||||
("Mitwirkende Personen", validation.participants or "nicht erfasst"),
|
||||
]
|
||||
)
|
||||
return (
|
||||
"<section class=\"cover-page\" id=\"cover\">"
|
||||
"<div class=\"cover-kicker\">Neutraler Logo-Platzhalter · Validation Suite</div>"
|
||||
"<h1>Pruefbericht zur Validierung</h1>"
|
||||
"<p class=\"cover-subtitle\">Funktions- und Leistungsqualifikation Klein-Sterilisator</p>"
|
||||
f"<p class=\"cover-subtitle\">{text(context.customer.name)}</p>"
|
||||
f"{rows}"
|
||||
"<div class=\"signature-grid\"><div>Unterschrift technische Validierung</div><div>Unterschrift Auftraggeber</div></div>"
|
||||
"</section>"
|
||||
)
|
||||
|
||||
|
||||
class TocComponent(ReportComponent):
|
||||
anchor = "toc"
|
||||
title = "Inhaltsverzeichnis"
|
||||
|
||||
def __init__(self, components: list[ReportComponent]) -> None:
|
||||
self.components = components
|
||||
|
||||
def render(self, context: ReportContext) -> str:
|
||||
links = "".join(
|
||||
f"<li><a href=\"#{component.anchor}\">{text(component.title)}</a></li>"
|
||||
for component in self.components
|
||||
if component.anchor not in {"cover", "toc"}
|
||||
)
|
||||
return section(self.anchor, self.title, f"<ol class=\"toc-list\">{links}</ol>")
|
||||
|
||||
|
||||
class SummaryComponent(ReportComponent):
|
||||
anchor = "summary"
|
||||
title = "Zusammenfassendes Ergebnis der Validierung"
|
||||
|
||||
def render(self, context: ReportContext) -> str:
|
||||
validation = context.validation
|
||||
body = definition_list(
|
||||
[
|
||||
("Kunde", context.customer.name),
|
||||
("Standort", context.location.name if context.location else None),
|
||||
("Geraet", f"{context.device.manufacturer} {context.device.model}" if context.device else None),
|
||||
("Pruefmittel", len(context.equipment)),
|
||||
("Ergebnis", validation.result),
|
||||
("Mitwirkende Personen", validation.participants),
|
||||
("Hinweis auf naechste Leistungsbeurteilung", validation.next_validation_on or "nicht erfasst"),
|
||||
]
|
||||
)
|
||||
return section(self.anchor, self.title, body)
|
||||
|
||||
|
||||
class StaticTextComponent(ReportComponent):
|
||||
def __init__(self, anchor: str, title: str, body: str = "nicht erfasst") -> None:
|
||||
self.anchor = anchor
|
||||
self.title = title
|
||||
self.body = body
|
||||
|
||||
def render(self, context: ReportContext) -> str:
|
||||
return section(self.anchor, self.title, f"<p>{text(self.body)}</p>")
|
||||
|
||||
|
||||
class CustomerComponent(ReportComponent):
|
||||
anchor = "customer"
|
||||
title = "Kunde, Standort und Ansprechpartner"
|
||||
|
||||
def render(self, context: ReportContext) -> str:
|
||||
customer = context.customer
|
||||
location = context.location
|
||||
contact = context.contact
|
||||
body = "<h3>Kunde</h3>" + definition_list(
|
||||
[
|
||||
("Name", customer.name),
|
||||
("Typ", customer.customer_type.value),
|
||||
("Adresse", " ".join(filter(None, [customer.street, customer.postal_code, customer.city]))),
|
||||
("Telefon", customer.phone),
|
||||
("Mail", customer.email),
|
||||
("Betreiber", context.validation.operator_name),
|
||||
("QM", customer.quality_manager),
|
||||
("Hygienebeauftragter", customer.hygiene_officer),
|
||||
]
|
||||
)
|
||||
if location:
|
||||
body += "<h3>Standort</h3>" + definition_list(
|
||||
[
|
||||
("Name", location.name),
|
||||
("Adresse", " ".join(filter(None, [location.street, location.postal_code, location.city]))),
|
||||
("Raum", location.room),
|
||||
]
|
||||
)
|
||||
if contact:
|
||||
body += "<h3>Ansprechpartner</h3>" + definition_list(
|
||||
[
|
||||
("Name", contact.full_name),
|
||||
("Funktion", contact.function),
|
||||
("Mail", contact.email),
|
||||
("Telefon", contact.phone),
|
||||
]
|
||||
)
|
||||
return section(self.anchor, self.title, body)
|
||||
|
||||
|
||||
class DeviceComponent(ReportComponent):
|
||||
anchor = "device"
|
||||
title = "Geraet"
|
||||
|
||||
def render(self, context: ReportContext) -> str:
|
||||
device = context.device
|
||||
if device is None:
|
||||
return section(self.anchor, self.title, "")
|
||||
body = definition_list(
|
||||
[
|
||||
("Hersteller", device.manufacturer),
|
||||
("Modell", device.model),
|
||||
("Typ", device.device_type),
|
||||
("Seriennummer", device.serial_number),
|
||||
("Baujahr", device.year_built),
|
||||
("Inbetriebnahme", device.commissioned_on),
|
||||
("Kammervolumen", f"{device.chamber_volume_liters} Liter" if device.chamber_volume_liters else None),
|
||||
("Dampferzeugung", device.steam_generation),
|
||||
("Wasseraufbereitung", device.water_treatment),
|
||||
("Dokumentation", device.documentation),
|
||||
("Lieferant", device.supplier),
|
||||
]
|
||||
)
|
||||
return section(self.anchor, self.title, body)
|
||||
|
||||
|
||||
class EquipmentComponent(ReportComponent):
|
||||
anchor = "equipment"
|
||||
title = "Pruefmittel"
|
||||
|
||||
def render(self, context: ReportContext) -> str:
|
||||
rows = [
|
||||
[
|
||||
item.kind.value,
|
||||
item.manufacturer,
|
||||
item.model,
|
||||
item.serial_number,
|
||||
item.calibrated_on,
|
||||
item.calibration_due_on,
|
||||
item.status.value,
|
||||
]
|
||||
for item in context.equipment
|
||||
]
|
||||
return section(
|
||||
self.anchor,
|
||||
self.title,
|
||||
table(
|
||||
["Art", "Hersteller", "Modell", "Seriennummer", "Kalibriert", "Gueltig bis", "Status"],
|
||||
rows,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
class EnvironmentComponent(ReportComponent):
|
||||
anchor = "environment"
|
||||
title = "Umgebungsbedingungen"
|
||||
|
||||
def render(self, context: ReportContext) -> str:
|
||||
data = context.validation.environment_conditions or {}
|
||||
body = definition_list(
|
||||
[
|
||||
("Raumtemperatur", data.get("room_temperature")),
|
||||
("relative Luftfeuchtigkeit", data.get("humidity")),
|
||||
("Pruefzeit", data.get("test_time")),
|
||||
]
|
||||
)
|
||||
rows = [[item.get("text"), yes_no(item.get("value")), item.get("comment")] for item in data.get("checks", [])]
|
||||
if rows:
|
||||
body += table(["Pruefpunkt", "Bewertung", "Kommentar"], rows)
|
||||
return section(self.anchor, self.title, body)
|
||||
|
||||
|
||||
class ChecklistComponent(ReportComponent):
|
||||
anchor = "checklists"
|
||||
title = "Dokumentations- und Leistungschecklisten"
|
||||
|
||||
def render(self, context: ReportContext) -> str:
|
||||
documentation = context.validation.documentation_checklist or []
|
||||
performance = context.validation.performance_checklist or []
|
||||
body = "<h3>Dokumentation</h3>" + self._render_items(documentation)
|
||||
body += "<h3>Leistung</h3>" + self._render_items(performance)
|
||||
return section(self.anchor, self.title, body)
|
||||
|
||||
def _render_items(self, items: list[dict]) -> str:
|
||||
rows = [
|
||||
[item.get("number"), item.get("text"), yes_no(item.get("value")), item.get("comment")]
|
||||
for item in items
|
||||
]
|
||||
return table(["Nr.", "Pruefpunkt", "Bewertung", "Kommentar"], rows)
|
||||
|
||||
|
||||
class ProgramComponent(ReportComponent):
|
||||
anchor = "programs"
|
||||
title = "Programme"
|
||||
|
||||
def render(self, context: ReportContext) -> str:
|
||||
programs = [item for item in (context.validation.programs or []) if item.get("selected")]
|
||||
rows = [[index + 1, item.get("name"), "Eigenes Programm" if item.get("custom") else "Standard"] for index, item in enumerate(programs)]
|
||||
return section(self.anchor, self.title, table(["Nr.", "Programm", "Typ"], rows))
|
||||
|
||||
|
||||
class LoadingComponent(ReportComponent):
|
||||
anchor = "loading"
|
||||
title = "Beladungsmuster"
|
||||
|
||||
def render(self, context: ReportContext) -> str:
|
||||
rows = [
|
||||
[item.get("run"), item.get("pattern"), item.get("description"), len(item.get("images", []))]
|
||||
for item in context.validation.loading_patterns or []
|
||||
]
|
||||
return section(self.anchor, self.title, table(["Testlauf", "Beladungsmuster", "Beschreibung", "Bilder"], rows))
|
||||
|
||||
|
||||
class MeasurementComponent(ReportComponent):
|
||||
anchor = "measurements"
|
||||
title = "Messdaten"
|
||||
|
||||
def render(self, context: ReportContext) -> str:
|
||||
rows = [
|
||||
[
|
||||
item.get("name"),
|
||||
item.get("start_time"),
|
||||
item.get("end_time"),
|
||||
item.get("duration"),
|
||||
item.get("leak_rate"),
|
||||
item.get("min_temperature"),
|
||||
item.get("max_temperature"),
|
||||
item.get("temperature_band"),
|
||||
item.get("holding_time"),
|
||||
item.get("pressure"),
|
||||
item.get("result"),
|
||||
]
|
||||
for item in context.validation.measurement_data or []
|
||||
]
|
||||
body = table(
|
||||
[
|
||||
"Bereich",
|
||||
"Start",
|
||||
"Ende",
|
||||
"Dauer",
|
||||
"Leckrate",
|
||||
"Min. Temp.",
|
||||
"Max. Temp.",
|
||||
"Band",
|
||||
"Haltezeit",
|
||||
"Druck",
|
||||
"Ergebnis",
|
||||
],
|
||||
rows,
|
||||
"compact",
|
||||
)
|
||||
winlog_rows = []
|
||||
for item in context.validation.measurement_data or []:
|
||||
for imported in item.get("imports", []):
|
||||
winlog_rows.append([item.get("name"), imported.get("filename"), imported.get("content_type")])
|
||||
if winlog_rows:
|
||||
body += "<h3>Winlog-Dateien</h3>" + table(["Bereich", "Datei", "Typ"], winlog_rows)
|
||||
return section(self.anchor, self.title, body)
|
||||
|
||||
|
||||
class DryingComponent(ReportComponent):
|
||||
anchor = "drying"
|
||||
title = "Trocknung"
|
||||
|
||||
def render(self, context: ReportContext) -> str:
|
||||
data = context.validation.drying or {}
|
||||
body = definition_list(
|
||||
[
|
||||
("Startgewicht", data.get("start_weight")),
|
||||
("Endgewicht", data.get("end_weight")),
|
||||
("Differenz", data.get("difference")),
|
||||
("Bewertung", data.get("rating")),
|
||||
("Bemerkung", data.get("comment")),
|
||||
]
|
||||
)
|
||||
return section(self.anchor, self.title, body)
|
||||
|
||||
|
||||
class RecommendationComponent(ReportComponent):
|
||||
anchor = "recommendations"
|
||||
title = "Empfehlungen und Auflagen"
|
||||
|
||||
def render(self, context: ReportContext) -> str:
|
||||
rows = [
|
||||
[item.get("number"), item.get("text"), item.get("deadline"), item.get("status")]
|
||||
for item in context.validation.recommendations or []
|
||||
]
|
||||
return section(self.anchor, self.title, table(["Nr.", "Text", "Frist", "Status"], rows))
|
||||
|
||||
|
||||
class AttachmentComponent(ReportComponent):
|
||||
anchor = "attachments"
|
||||
title = "Bilder und Anlagen"
|
||||
|
||||
def render(self, context: ReportContext) -> str:
|
||||
rows = [
|
||||
[item.get("order"), item.get("category"), item.get("filename"), item.get("description")]
|
||||
for item in sorted(context.validation.attachments or [], key=lambda row: row.get("order") or 0)
|
||||
]
|
||||
return section(self.anchor, self.title, table(["Reihenfolge", "Kategorie", "Datei", "Beschreibung"], rows))
|
||||
|
|
@ -0,0 +1,64 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
from fastapi import HTTPException, status
|
||||
from sqlalchemy import 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
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ReportContext:
|
||||
validation: Validation
|
||||
customer: Customer
|
||||
location: Location | None
|
||||
contact: Contact | None
|
||||
device: Device | None
|
||||
equipment: list[Equipment]
|
||||
generated_dir: Path
|
||||
|
||||
|
||||
class OrionContextBuilder:
|
||||
def __init__(self, session: Session, generated_dir: Path) -> None:
|
||||
self.session = session
|
||||
self.generated_dir = generated_dir
|
||||
|
||||
def build(self, validation_id: str) -> ReportContext:
|
||||
validation = self.session.get(Validation, validation_id)
|
||||
if validation is None:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Validation not found")
|
||||
|
||||
customer = self.session.get(Customer, validation.customer_id)
|
||||
if customer is None:
|
||||
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Validation has no customer")
|
||||
|
||||
location = self.session.get(Location, validation.location_id) if validation.location_id else None
|
||||
contact = self.session.get(Contact, validation.contact_id) if validation.contact_id else None
|
||||
device = self.session.get(Device, validation.device_id) if validation.device_id else None
|
||||
equipment = []
|
||||
if validation.equipment_ids:
|
||||
equipment = list(
|
||||
self.session.scalars(
|
||||
select(Equipment).where(Equipment.id.in_(validation.equipment_ids))
|
||||
)
|
||||
)
|
||||
|
||||
self.generated_dir.mkdir(parents=True, exist_ok=True)
|
||||
return ReportContext(
|
||||
validation=validation,
|
||||
customer=customer,
|
||||
location=location,
|
||||
contact=contact,
|
||||
device=device,
|
||||
equipment=equipment,
|
||||
generated_dir=self.generated_dir,
|
||||
)
|
||||
|
||||
45
validation-suite/backend/mercury/app/modules/orion/html.py
Normal file
45
validation-suite/backend/mercury/app/modules/orion/html.py
Normal file
|
|
@ -0,0 +1,45 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from datetime import date, datetime
|
||||
from html import escape
|
||||
from typing import Any
|
||||
|
||||
|
||||
def text(value: Any) -> str:
|
||||
if value is None or value == "":
|
||||
return "nicht erfasst"
|
||||
if isinstance(value, (date, datetime)):
|
||||
return value.strftime("%d.%m.%Y")
|
||||
return escape(str(value))
|
||||
|
||||
|
||||
def paragraph(value: Any) -> str:
|
||||
content = text(value)
|
||||
return content.replace("\n", "<br>")
|
||||
|
||||
|
||||
def yes_no(value: Any) -> str:
|
||||
labels = {"yes": "Ja", "no": "Nein", "na": "Nicht zutreffend", True: "Ja", False: "Nein"}
|
||||
return text(labels.get(value, value))
|
||||
|
||||
|
||||
def definition_list(rows: list[tuple[str, Any]]) -> str:
|
||||
items = "".join(
|
||||
f"<div class=\"definition-row\"><dt>{text(label)}</dt><dd>{paragraph(value)}</dd></div>"
|
||||
for label, value in rows
|
||||
if value not in (None, "", [])
|
||||
)
|
||||
return f"<dl class=\"definition-list\">{items}</dl>"
|
||||
|
||||
|
||||
def table(headers: list[str], rows: list[list[Any]], css_class: str = "") -> str:
|
||||
head = "".join(f"<th>{text(header)}</th>" for header in headers)
|
||||
body = "".join(
|
||||
"<tr>" + "".join(f"<td>{paragraph(cell)}</td>" for cell in row) + "</tr>"
|
||||
for row in rows
|
||||
)
|
||||
return f"<table class=\"data-table {css_class}\"><thead><tr>{head}</tr></thead><tbody>{body}</tbody></table>"
|
||||
|
||||
|
||||
def section(chapter_id: str, title: str, body: str) -> str:
|
||||
return f"<section class=\"chapter\" id=\"{text(chapter_id)}\"><h2>{text(title)}</h2>{body}</section>"
|
||||
|
|
@ -2,29 +2,75 @@ from __future__ import annotations
|
|||
|
||||
from pathlib import Path
|
||||
|
||||
from weasyprint import HTML
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.modules.orion.components import (
|
||||
AttachmentComponent,
|
||||
ChecklistComponent,
|
||||
CoverComponent,
|
||||
CustomerComponent,
|
||||
DeviceComponent,
|
||||
DryingComponent,
|
||||
EnvironmentComponent,
|
||||
EquipmentComponent,
|
||||
LoadingComponent,
|
||||
MeasurementComponent,
|
||||
ProgramComponent,
|
||||
RecommendationComponent,
|
||||
ReportComponent,
|
||||
StaticTextComponent,
|
||||
SummaryComponent,
|
||||
TocComponent,
|
||||
)
|
||||
from app.modules.orion.context import OrionContextBuilder, ReportContext
|
||||
from app.modules.orion.templates.report import render_document
|
||||
|
||||
|
||||
class OrionReportService:
|
||||
chapters = [
|
||||
"Deckblatt",
|
||||
"Inhaltsverzeichnis",
|
||||
"Zusammenfassung",
|
||||
"Gerät",
|
||||
"Kunde",
|
||||
"Normen",
|
||||
"Prüfmittel",
|
||||
"Programme",
|
||||
"Beladung",
|
||||
"Messungen",
|
||||
"Diagramme",
|
||||
"Empfehlungen",
|
||||
"Anlagen",
|
||||
]
|
||||
def __init__(self, session: Session, generated_dir: Path | None = None) -> None:
|
||||
self.session = session
|
||||
self.generated_dir = generated_dir or Path("/app/reports")
|
||||
|
||||
def render_pdf(self, title: str, output_path: Path) -> Path:
|
||||
chapter_markup = "".join(f"<section><h2>{chapter}</h2></section>" for chapter in self.chapters)
|
||||
html = f"<html><body><h1>{title}</h1>{chapter_markup}</body></html>"
|
||||
HTML(string=html).write_pdf(output_path)
|
||||
def render_html(self, validation_id: str) -> str:
|
||||
context = OrionContextBuilder(self.session, self.generated_dir).build(validation_id)
|
||||
components = self._components()
|
||||
chapters = [component.render(context) for component in components]
|
||||
return render_document(context, chapters)
|
||||
|
||||
def render_pdf(self, validation_id: str) -> Path:
|
||||
from weasyprint import HTML
|
||||
|
||||
context = OrionContextBuilder(self.session, self.generated_dir).build(validation_id)
|
||||
output_path = context.generated_dir / f"{context.validation.report_number}.pdf"
|
||||
html = self.render_html(validation_id)
|
||||
HTML(string=html, base_url=str(context.generated_dir)).write_pdf(output_path)
|
||||
return output_path
|
||||
|
||||
def _components(self) -> list[ReportComponent]:
|
||||
chapters: list[ReportComponent] = [
|
||||
StaticTextComponent("bq", "1. Funktionsqualifikation (BQ)", "Anlass, Ziel und gesetzliche Grundlagen werden anhand der erfassten Validierungsdaten bewertet."),
|
||||
StaticTextComponent("bq-goal", "1.1 Anlass und Ziel der Pruefung"),
|
||||
StaticTextComponent("legal", "1.2 Gesetzliche Grundlagen"),
|
||||
DeviceComponent(),
|
||||
StaticTextComponent("performance", "1.4 Leistungsueberpruefung"),
|
||||
ChecklistComponent(),
|
||||
StaticTextComponent("work-instructions", "Arbeitsanweisungen"),
|
||||
EnvironmentComponent(),
|
||||
StaticTextComponent("batch-control", "1.9 Chargenkontrolle"),
|
||||
ProgramComponent(),
|
||||
LoadingComponent(),
|
||||
StaticTextComponent("reference-load", "1.12 Referenzbeladung"),
|
||||
EquipmentComponent(),
|
||||
StaticTextComponent("equipment-thermo", "2.2 Pruefmittel zur thermoelektrischen Untersuchung"),
|
||||
StaticTextComponent("configuration", "3. Pruefkonfiguration"),
|
||||
MeasurementComponent(),
|
||||
StaticTextComponent("results", "4. Ergebnisse der Validierung"),
|
||||
DryingComponent(),
|
||||
RecommendationComponent(),
|
||||
AttachmentComponent(),
|
||||
StaticTextComponent("cycles", "6. Programmablaeufe / Zyklen"),
|
||||
StaticTextComponent("risk", "7. Risikoeinstufung und Abschlussgespraech"),
|
||||
StaticTextComponent("certificates", "8. Zertifikate"),
|
||||
StaticTextComponent("calibration-certificates", "9. Kalibrierzertifikate"),
|
||||
]
|
||||
return [CoverComponent(), SummaryComponent(), TocComponent(chapters), CustomerComponent(), *chapters]
|
||||
|
|
|
|||
|
|
@ -0,0 +1,201 @@
|
|||
@page {
|
||||
size: A4;
|
||||
margin: 24mm 16mm 22mm 16mm;
|
||||
@top-left {
|
||||
content: "Validation Suite";
|
||||
color: #4F6A74;
|
||||
font-size: 9pt;
|
||||
font-weight: 700;
|
||||
}
|
||||
@top-right {
|
||||
content: string(report-number);
|
||||
color: #6B7C85;
|
||||
font-size: 9pt;
|
||||
}
|
||||
@bottom-left {
|
||||
content: "Schubamed";
|
||||
color: #6B7C85;
|
||||
font-size: 8pt;
|
||||
}
|
||||
@bottom-right {
|
||||
content: "Seite " counter(page) " von " counter(pages);
|
||||
color: #6B7C85;
|
||||
font-size: 8pt;
|
||||
}
|
||||
}
|
||||
|
||||
@page:first {
|
||||
margin: 20mm 16mm 18mm 16mm;
|
||||
@top-left { content: ""; }
|
||||
@top-right { content: ""; }
|
||||
}
|
||||
|
||||
* {
|
||||
box-sizing: border-box;
|
||||
}
|
||||
|
||||
html {
|
||||
color: #2E3B40;
|
||||
font-family: Inter, Arial, sans-serif;
|
||||
font-size: 10.5pt;
|
||||
line-height: 1.45;
|
||||
}
|
||||
|
||||
body {
|
||||
margin: 0;
|
||||
}
|
||||
|
||||
.report-meta {
|
||||
string-set: report-number attr(data-report-number);
|
||||
}
|
||||
|
||||
.cover-page {
|
||||
min-height: 245mm;
|
||||
display: flex;
|
||||
flex-direction: column;
|
||||
justify-content: center;
|
||||
page-break-after: always;
|
||||
}
|
||||
|
||||
.cover-kicker {
|
||||
color: #6C8A96;
|
||||
font-size: 11pt;
|
||||
font-weight: 700;
|
||||
letter-spacing: .08em;
|
||||
margin-bottom: 14mm;
|
||||
text-transform: uppercase;
|
||||
}
|
||||
|
||||
h1 {
|
||||
color: #2E3B40;
|
||||
font-size: 34pt;
|
||||
line-height: 1.05;
|
||||
margin: 0 0 8mm 0;
|
||||
}
|
||||
|
||||
.cover-subtitle {
|
||||
color: #4F6A74;
|
||||
font-size: 16pt;
|
||||
margin: 0 0 18mm 0;
|
||||
}
|
||||
|
||||
.signature-grid {
|
||||
display: grid;
|
||||
grid-template-columns: 1fr 1fr;
|
||||
gap: 16mm;
|
||||
margin-top: 18mm;
|
||||
}
|
||||
|
||||
.signature-grid div {
|
||||
border-top: 1px solid #6B7C85;
|
||||
color: #6B7C85;
|
||||
padding-top: 3mm;
|
||||
}
|
||||
|
||||
.chapter {
|
||||
break-before: page;
|
||||
}
|
||||
|
||||
h2 {
|
||||
border-bottom: 1px solid #E6EAEA;
|
||||
color: #2E3B40;
|
||||
font-size: 18pt;
|
||||
margin: 0 0 8mm 0;
|
||||
padding-bottom: 4mm;
|
||||
}
|
||||
|
||||
h3 {
|
||||
color: #4F6A74;
|
||||
font-size: 12pt;
|
||||
margin: 8mm 0 3mm 0;
|
||||
}
|
||||
|
||||
.definition-list {
|
||||
display: block;
|
||||
margin: 0;
|
||||
}
|
||||
|
||||
.definition-row {
|
||||
border-bottom: 1px solid #E6EAEA;
|
||||
display: grid;
|
||||
grid-template-columns: 42mm 1fr;
|
||||
gap: 6mm;
|
||||
padding: 2.5mm 0;
|
||||
}
|
||||
|
||||
dt {
|
||||
color: #6B7C85;
|
||||
font-weight: 700;
|
||||
}
|
||||
|
||||
dd {
|
||||
margin: 0;
|
||||
}
|
||||
|
||||
.data-table {
|
||||
border-collapse: collapse;
|
||||
margin-top: 4mm;
|
||||
table-layout: fixed;
|
||||
width: 100%;
|
||||
}
|
||||
|
||||
.data-table th {
|
||||
background: #F7F8F8;
|
||||
color: #4F6A74;
|
||||
font-size: 8.5pt;
|
||||
font-weight: 700;
|
||||
text-align: left;
|
||||
}
|
||||
|
||||
.data-table th,
|
||||
.data-table td {
|
||||
border: 1px solid #E6EAEA;
|
||||
padding: 2.4mm;
|
||||
vertical-align: top;
|
||||
word-wrap: break-word;
|
||||
}
|
||||
|
||||
.data-table.compact {
|
||||
font-size: 8.5pt;
|
||||
}
|
||||
|
||||
.toc-list {
|
||||
counter-reset: toc;
|
||||
list-style: none;
|
||||
margin: 0;
|
||||
padding: 0;
|
||||
}
|
||||
|
||||
.toc-list li {
|
||||
border-bottom: 1px solid #E6EAEA;
|
||||
padding: 3mm 0;
|
||||
}
|
||||
|
||||
.toc-list a {
|
||||
color: #2E3B40;
|
||||
text-decoration: none;
|
||||
}
|
||||
|
||||
@media screen {
|
||||
body {
|
||||
background: #F7F8F8;
|
||||
padding: 24px;
|
||||
}
|
||||
|
||||
.report-document {
|
||||
background: #FFFFFF;
|
||||
box-shadow: 0 14px 40px rgba(46, 59, 64, 0.08);
|
||||
margin: 0 auto;
|
||||
max-width: 960px;
|
||||
padding: 48px;
|
||||
}
|
||||
|
||||
.cover-page {
|
||||
min-height: auto;
|
||||
}
|
||||
|
||||
.chapter {
|
||||
break-before: auto;
|
||||
margin-top: 48px;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,28 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from app.modules.orion.context import ReportContext
|
||||
from app.modules.orion.html import text
|
||||
|
||||
|
||||
def render_document(context: ReportContext, chapters: list[str]) -> str:
|
||||
css = (Path(__file__).resolve().parent / "report.css").read_text(encoding="utf-8")
|
||||
title = f"Validierungsbericht {context.validation.report_number}"
|
||||
chapter_markup = "\n".join(chapters)
|
||||
return f"""<!doctype html>
|
||||
<html lang="de">
|
||||
<head>
|
||||
<meta charset="utf-8">
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1">
|
||||
<title>{text(title)}</title>
|
||||
<style>{css}</style>
|
||||
</head>
|
||||
<body>
|
||||
<article class="report-document">
|
||||
<div class="report-meta" data-report-number="{text(context.validation.report_number)}"></div>
|
||||
{chapter_markup}
|
||||
</article>
|
||||
</body>
|
||||
</html>"""
|
||||
|
||||
|
|
@ -1,8 +1,9 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from datetime import date
|
||||
from uuid import UUID
|
||||
|
||||
from pydantic import EmailStr, Field
|
||||
from pydantic import EmailStr, Field, field_validator
|
||||
|
||||
from app.models.customer import CustomerType
|
||||
from app.models.equipment import EquipmentKind, EquipmentStatus
|
||||
|
|
@ -108,12 +109,12 @@ class EquipmentUpdate(EquipmentCreate):
|
|||
|
||||
|
||||
class ValidationCreate(ORMModel):
|
||||
report_number: str
|
||||
customer_id: str
|
||||
location_id: str | None = None
|
||||
contact_id: str | None = None
|
||||
device_id: str | None = None
|
||||
validation_type: str
|
||||
report_number: str | None = None
|
||||
customer_id: UUID | None = None
|
||||
location_id: UUID | None = None
|
||||
contact_id: UUID | None = None
|
||||
device_id: UUID | None = None
|
||||
validation_type: str | None = None
|
||||
project: str | None = None
|
||||
test_location: str | None = None
|
||||
examiner_name: str | None = None
|
||||
|
|
@ -122,8 +123,12 @@ class ValidationCreate(ORMModel):
|
|||
scheduled_on: date | None = None
|
||||
performed_on: date | None = None
|
||||
next_validation_on: date | None = None
|
||||
examiner_id: str | None = None
|
||||
status: ValidationStatus = ValidationStatus.draft
|
||||
revalidation_interval_months: int = 24
|
||||
next_validation_manually_overridden: bool = False
|
||||
version: int = 1
|
||||
previous_validation_id: UUID | None = None
|
||||
examiner_id: UUID | None = None
|
||||
status: ValidationStatus | str = ValidationStatus.draft
|
||||
result: str | None = None
|
||||
notes: str | None = None
|
||||
equipment_ids: list[str] = Field(default_factory=list)
|
||||
|
|
@ -137,6 +142,21 @@ class ValidationCreate(ORMModel):
|
|||
recommendations: list[dict] = Field(default_factory=list)
|
||||
attachments: list[dict] = Field(default_factory=list)
|
||||
|
||||
@field_validator(
|
||||
"customer_id",
|
||||
"location_id",
|
||||
"contact_id",
|
||||
"device_id",
|
||||
"examiner_id",
|
||||
"previous_validation_id",
|
||||
mode="before",
|
||||
)
|
||||
@classmethod
|
||||
def empty_string_to_none(cls, value):
|
||||
if value == "":
|
||||
return None
|
||||
return value
|
||||
|
||||
|
||||
class ValidationRead(ValidationCreate, EntityRead):
|
||||
pass
|
||||
|
|
@ -144,3 +164,45 @@ class ValidationRead(ValidationCreate, EntityRead):
|
|||
|
||||
class ValidationUpdate(ValidationCreate):
|
||||
pass
|
||||
|
||||
|
||||
class ValidationIssue(ORMModel):
|
||||
field: str
|
||||
message: str
|
||||
section: str
|
||||
|
||||
|
||||
class ValidationReview(ORMModel):
|
||||
status: str
|
||||
errors: list[ValidationIssue]
|
||||
warnings: list[ValidationIssue]
|
||||
complete_sections: list[str]
|
||||
|
||||
|
||||
class ValidationImportPreviewRow(ORMModel):
|
||||
row_number: int
|
||||
data: dict
|
||||
errors: list[str]
|
||||
duplicate: bool
|
||||
resolved_customer_id: str | None = None
|
||||
resolved_location_id: str | None = None
|
||||
resolved_device_id: str | None = None
|
||||
|
||||
|
||||
class ValidationImportPreview(ORMModel):
|
||||
rows: list[ValidationImportPreviewRow]
|
||||
valid_rows: int
|
||||
invalid_rows: int
|
||||
duplicates: int
|
||||
|
||||
|
||||
class ValidationImportRequest(ORMModel):
|
||||
rows: list[dict]
|
||||
duplicate_strategy: str = "skip"
|
||||
|
||||
|
||||
class ValidationImportSummary(ORMModel):
|
||||
successful: int
|
||||
skipped: int
|
||||
failed: int
|
||||
errors: list[str] = Field(default_factory=list)
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from typing import TypeVar
|
||||
from uuid import UUID
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
|
|
@ -41,16 +42,33 @@ class CrudService:
|
|||
}
|
||||
|
||||
def create(self, data: dict) -> ModelT:
|
||||
data = self._normalize(data)
|
||||
return self.repository.add(self.repository.model(**data))
|
||||
|
||||
def update(self, item_id: str, data: dict) -> ModelT:
|
||||
data = self._normalize(data)
|
||||
item = self.repository.get(item_id)
|
||||
if item is None:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Resource not found")
|
||||
if isinstance(item, Validation) and item.status in {"FREIGEGEBEN", "ABGESCHLOSSEN"}:
|
||||
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Freigegebene oder abgeschlossene Validierungen sind schreibgeschuetzt")
|
||||
for key, value in data.items():
|
||||
setattr(item, key, value)
|
||||
return item
|
||||
|
||||
def _normalize(self, data: dict) -> dict:
|
||||
normalized = {}
|
||||
for key, value in data.items():
|
||||
if key.endswith("_id") and value == "":
|
||||
normalized[key] = None
|
||||
elif isinstance(value, UUID):
|
||||
normalized[key] = str(value)
|
||||
elif isinstance(value, list):
|
||||
normalized[key] = [str(item) if isinstance(item, UUID) else item for item in value]
|
||||
else:
|
||||
normalized[key] = value
|
||||
return normalized
|
||||
|
||||
def delete(self, item_id: str) -> None:
|
||||
item = self.repository.get(item_id)
|
||||
if item is None:
|
||||
|
|
|
|||
|
|
@ -0,0 +1,344 @@
|
|||
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_, 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", "result"]:
|
||||
if filters.get(key):
|
||||
conditions.append(getattr(Validation, key) == filters[key])
|
||||
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 = len(list(self.session.scalars(count_statement)))
|
||||
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}
|
||||
|
||||
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()
|
||||
|
|
@ -13,13 +13,14 @@ dependencies = [
|
|||
"pydantic[email]==2.11.7",
|
||||
"python-jose[cryptography]==3.5.0",
|
||||
"python-multipart==0.0.20",
|
||||
"python-dateutil==2.9.0.post0",
|
||||
"sqlalchemy==2.0.41",
|
||||
"uvicorn[standard]==0.35.0",
|
||||
"weasyprint==62.3"
|
||||
]
|
||||
|
||||
[project.optional-dependencies]
|
||||
dev = ["black==25.1.0", "isort==6.0.1", "ruff==0.12.4"]
|
||||
dev = ["black==25.1.0", "isort==6.0.1", "pytest==8.3.4", "ruff==0.12.4"]
|
||||
|
||||
[tool.black]
|
||||
line-length = 100
|
||||
|
|
@ -31,3 +32,5 @@ profile = "black"
|
|||
line-length = 100
|
||||
target-version = "py312"
|
||||
|
||||
[tool.setuptools.packages.find]
|
||||
include = ["app*"]
|
||||
|
|
|
|||
|
|
@ -0,0 +1,225 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from datetime import date
|
||||
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.db.base import Base
|
||||
from app.models.customer import Customer, CustomerType
|
||||
from app.models.contact import Contact
|
||||
from app.models.device import Device
|
||||
from app.models.location import Location
|
||||
from app.models.validation import Validation, ValidationStatus
|
||||
from app.services.validation_workflow import ValidationWorkflowService
|
||||
from app.schemas.domain import ValidationCreate
|
||||
from app.modules.orion.service import OrionReportService
|
||||
|
||||
|
||||
def session() -> Session:
|
||||
engine = create_engine("sqlite+pysqlite:///:memory:")
|
||||
Base.metadata.create_all(engine)
|
||||
return Session(engine)
|
||||
|
||||
|
||||
def seed(session: Session):
|
||||
customer = Customer(customer_type=CustomerType.practice, name="Praxis Test")
|
||||
session.add(customer)
|
||||
session.flush()
|
||||
location = Location(customer_id=customer.id, name="OP")
|
||||
session.add(location)
|
||||
session.flush()
|
||||
device = Device(
|
||||
customer_id=customer.id,
|
||||
location_id=location.id,
|
||||
manufacturer="MELAG",
|
||||
model="Vacuklav",
|
||||
serial_number="SN-1",
|
||||
)
|
||||
session.add(device)
|
||||
session.flush()
|
||||
return customer, location, device
|
||||
|
||||
|
||||
def seed_contact(session: Session, customer: Customer) -> Contact:
|
||||
contact = Contact(customer_id=customer.id, full_name="Kontakt")
|
||||
session.add(contact)
|
||||
session.flush()
|
||||
return contact
|
||||
|
||||
|
||||
def valid_validation(customer: Customer, location: Location, device: Device) -> Validation:
|
||||
return Validation(
|
||||
report_number="VAL-TEST-1",
|
||||
validation_type="Erstvalidierung",
|
||||
performed_on=date.today(),
|
||||
customer_id=customer.id,
|
||||
location_id=location.id,
|
||||
device_id=device.id,
|
||||
examiner_name="Pruefer",
|
||||
status=ValidationStatus.draft.value,
|
||||
)
|
||||
|
||||
|
||||
def test_required_fields_are_reported():
|
||||
db = session()
|
||||
validation = Validation(status=ValidationStatus.draft.value)
|
||||
|
||||
review = ValidationWorkflowService(db).review(validation)
|
||||
|
||||
assert {item["field"] for item in review["errors"]} >= {
|
||||
"report_number",
|
||||
"validation_type",
|
||||
"performed_on",
|
||||
"customer_id",
|
||||
"location_id",
|
||||
"device_id",
|
||||
"examiner_name",
|
||||
}
|
||||
|
||||
|
||||
def test_status_changes_to_ready_when_no_errors():
|
||||
db = session()
|
||||
customer, location, device = seed(db)
|
||||
validation = valid_validation(customer, location, device)
|
||||
db.add(validation)
|
||||
db.flush()
|
||||
|
||||
review = ValidationWorkflowService(db).mark_ready_for_review(validation)
|
||||
|
||||
assert review["errors"] == []
|
||||
assert validation.status == ValidationStatus.ready_for_review.value
|
||||
|
||||
|
||||
def test_search_finds_report_number():
|
||||
db = session()
|
||||
customer, location, device = seed(db)
|
||||
db.add(valid_validation(customer, location, device))
|
||||
db.flush()
|
||||
|
||||
result = ValidationWorkflowService(db).query_validations(
|
||||
search="VAL-TEST",
|
||||
page=1,
|
||||
page_size=10,
|
||||
sort_by="updated_at",
|
||||
sort_order="desc",
|
||||
filters={},
|
||||
)
|
||||
|
||||
assert result["total"] == 1
|
||||
|
||||
|
||||
def test_csv_import_preview_detects_duplicate_and_missing_fields():
|
||||
db = session()
|
||||
customer, location, device = seed(db)
|
||||
db.add(valid_validation(customer, location, device))
|
||||
db.flush()
|
||||
csv_content = (
|
||||
"report_number,validation_type,test_date,customer_reference,location_reference,"
|
||||
"device_serial_number,examiner,status,result,notes\n"
|
||||
"VAL-TEST-1,Erstvalidierung,2026-07-11,Praxis Test,OP,SN-1,Pruefer,ENTWURF,offen,\n"
|
||||
"VAL-NEW,,,,,,,\n"
|
||||
).encode()
|
||||
|
||||
preview = ValidationWorkflowService(db).preview_csv(csv_content)
|
||||
|
||||
assert preview["duplicates"] == 1
|
||||
assert preview["invalid_rows"] == 1
|
||||
|
||||
|
||||
def test_validation_without_contact_id_can_be_saved():
|
||||
db = session()
|
||||
customer, location, device = seed(db)
|
||||
validation = valid_validation(customer, location, device)
|
||||
validation.contact_id = None
|
||||
db.add(validation)
|
||||
db.flush()
|
||||
|
||||
assert validation.contact_id is None
|
||||
|
||||
|
||||
def test_validation_with_valid_contact_id_can_be_saved():
|
||||
db = session()
|
||||
customer, location, device = seed(db)
|
||||
contact = seed_contact(db, customer)
|
||||
validation = valid_validation(customer, location, device)
|
||||
validation.contact_id = contact.id
|
||||
db.add(validation)
|
||||
db.flush()
|
||||
|
||||
assert validation.contact_id == contact.id
|
||||
|
||||
|
||||
def test_empty_contact_id_string_is_normalized_to_none():
|
||||
payload = ValidationCreate(
|
||||
report_number="VAL-EMPTY",
|
||||
validation_type="Erstvalidierung",
|
||||
performed_on=date.today(),
|
||||
customer_id=None,
|
||||
location_id="",
|
||||
contact_id="",
|
||||
device_id=None,
|
||||
examiner_name="Pruefer",
|
||||
)
|
||||
|
||||
assert payload.contact_id is None
|
||||
assert payload.location_id is None
|
||||
|
||||
|
||||
def test_invalid_uuid_returns_validation_error():
|
||||
try:
|
||||
ValidationCreate(
|
||||
report_number="VAL-BAD",
|
||||
validation_type="Erstvalidierung",
|
||||
performed_on=date.today(),
|
||||
customer_id="not-a-uuid",
|
||||
device_id=None,
|
||||
examiner_name="Pruefer",
|
||||
)
|
||||
except Exception as exc:
|
||||
assert "uuid" in str(exc).lower()
|
||||
else:
|
||||
raise AssertionError("Invalid UUID was accepted")
|
||||
|
||||
|
||||
def test_revalidation_date_uses_calendar_months():
|
||||
db = session()
|
||||
customer, location, device = seed(db)
|
||||
validation = valid_validation(customer, location, device)
|
||||
validation.performed_on = date(2026, 1, 31)
|
||||
validation.revalidation_interval_months = 1
|
||||
|
||||
ValidationWorkflowService(db).apply_revalidation_date(validation)
|
||||
|
||||
assert validation.next_validation_on == date(2026, 2, 28)
|
||||
|
||||
|
||||
def test_new_version_keeps_structured_json_and_previous_reference():
|
||||
db = session()
|
||||
customer, location, device = seed(db)
|
||||
validation = valid_validation(customer, location, device)
|
||||
validation.status = ValidationStatus.approved.value
|
||||
validation.documentation_checklist = [{"text": "A", "value": "yes"}]
|
||||
db.add(validation)
|
||||
db.flush()
|
||||
|
||||
clone = ValidationWorkflowService(db).create_new_version(validation)
|
||||
|
||||
assert clone.previous_validation_id == validation.id
|
||||
assert clone.version == 2
|
||||
assert clone.documentation_checklist == validation.documentation_checklist
|
||||
|
||||
|
||||
def test_orion_renders_reference_main_chapters(tmp_path):
|
||||
db = session()
|
||||
customer, location, device = seed(db)
|
||||
validation = valid_validation(customer, location, device)
|
||||
db.add(validation)
|
||||
db.flush()
|
||||
|
||||
html = OrionReportService(db, tmp_path).render_html(validation.id)
|
||||
|
||||
assert "1. Funktionsqualifikation" in html
|
||||
assert "2.2 Pruefmittel" in html
|
||||
assert "4. Ergebnisse der Validierung" in html
|
||||
assert "9. Kalibrierzertifikate" in html
|
||||
Loading…
Add table
Add a link
Reference in a new issue