from __future__ import annotations import csv import hashlib import re from dataclasses import dataclass from pathlib import Path from sqlalchemy import select from sqlalchemy.orm import Session from app.models.report_template import ( MeasurementImport, MeasurementImportStatus, MeasurementImportType, MeasurementImportValue, ) @dataclass(frozen=True) class MeasurementSeries: headers: list[str] rows: list[dict[str, str]] class HeliosImportService: parser_version = "helios-winlog-pdf-1.0" def __init__(self, session: Session | None = None, upload_root: Path | None = None) -> None: self.session = session self.upload_root = upload_root or Path("/app/uploads") def import_csv(self, path: Path) -> MeasurementSeries: with path.open(newline="", encoding="utf-8-sig") as handle: reader = csv.DictReader(handle) return MeasurementSeries(headers=reader.fieldnames or [], rows=list(reader)) def save_winlog_pdf( self, validation_id: str, filename: str, content: bytes, attachment_only: bool = False ) -> dict: if self.session is None: raise RuntimeError("A database session is required for Winlog imports") safe_name = Path(filename or "winlog.pdf").name target_dir = self.upload_root / "validations" / validation_id / "winlog" target_dir.mkdir(parents=True, exist_ok=True) storage_path = target_dir / safe_name storage_path.write_bytes(content) digest = hashlib.sha256(content).hexdigest() import_row = MeasurementImport( validation_id=validation_id, import_type=( MeasurementImportType.winlog_attachment_only.value if attachment_only else MeasurementImportType.winlog_pdf.value ), original_filename=safe_name, storage_path=str(storage_path), sha256=digest, parser_version=self.parser_version, status=( MeasurementImportStatus.attachment_only.value if attachment_only else MeasurementImportStatus.uploaded.value ), ) self.session.add(import_row) self.session.flush() values: list[MeasurementImportValue] = [] if not attachment_only: values = self._extract_pdf_values(import_row, storage_path) import_row.status = ( MeasurementImportStatus.preview_ready.value if values else MeasurementImportStatus.attachment_only.value ) import_row.import_type = ( MeasurementImportType.winlog_pdf.value if values else MeasurementImportType.winlog_attachment_only.value ) self.session.flush() return self.preview(import_row.id) def preview(self, import_id: str) -> dict: if self.session is None: raise RuntimeError("A database session is required for Winlog imports") import_row = self.session.get(MeasurementImport, import_id) if import_row is None: raise ValueError("Measurement import not found") values = list( self.session.scalars( select(MeasurementImportValue).where(MeasurementImportValue.import_id == import_id) ) ) return { "id": import_row.id, "validation_id": import_row.validation_id, "import_type": import_row.import_type, "original_filename": import_row.original_filename, "sha256": import_row.sha256, "parser_version": import_row.parser_version, "status": import_row.status, "values": [ { "id": value.id, "test_run": value.test_run, "field_name": value.field_name, "raw_value": value.raw_value, "normalized_value": value.normalized_value, "unit": value.unit, "source_page": value.source_page, "source_text": value.source_text, "confidence": value.confidence, "confirmed": value.confirmed, "corrected_value": value.corrected_value, } for value in values ], } def confirm_values(self, import_id: str, values: list[dict]) -> dict: if self.session is None: raise RuntimeError("A database session is required for Winlog imports") import_row = self.session.get(MeasurementImport, import_id) if import_row is None: raise ValueError("Measurement import not found") by_id = { value.id: value for value in self.session.scalars( select(MeasurementImportValue).where(MeasurementImportValue.import_id == import_id) ) } for payload in values: item = by_id.get(payload.get("id")) if item is None: continue item.confirmed = bool(payload.get("confirmed")) item.corrected_value = payload.get("corrected_value") or item.corrected_value import_row.status = MeasurementImportStatus.confirmed.value self.session.flush() return self.preview(import_id) def _extract_pdf_values(self, import_row: MeasurementImport, path: Path) -> list[MeasurementImportValue]: from pypdf import PdfReader values: list[MeasurementImportValue] = [] try: reader = PdfReader(str(path)) pages = [page.extract_text() or "" for page in reader.pages] except Exception: import_row.status = MeasurementImportStatus.error.value return [] for page_index, page_text in enumerate(pages, start=1): if not page_text.strip(): continue test_run = self._detect_test_run(page_text) for field_name, pattern, unit in self._patterns(): match = re.search(pattern, page_text, flags=re.IGNORECASE) if not match: continue raw_value = match.group(1).strip() value = MeasurementImportValue( import_id=import_row.id, test_run=test_run, field_name=field_name, raw_value=raw_value, normalized_value=raw_value, unit=unit, source_page=page_index, source_text=match.group(0)[:500], confidence=80, confirmed=False, corrected_value=None, ) self.session.add(value) values.append(value) return values def _detect_test_run(self, text: str) -> str: lower = text.lower() if "vakuum" in lower: return "Vakuumtest" if "bowie" in lower or "leerkammer" in lower: return "Bowie-Dick / Leerkammerprofil" for index in (1, 2, 3): if f"testlauf {index}" in lower or f"test {index}" in lower: return f"Testlauf {index}" return "nicht zugeordnet" def _patterns(self) -> list[tuple[str, str, str | None]]: return [ ("program_name", r"Programm(?:name)?[:\s]+([^\n]+)", None), ("batch_number", r"Charge(?:nnummer)?[:\s]+([^\n]+)", None), ("start_time", r"Start(?:zeit)?[:\s]+([0-9:.\-\s]+)", None), ("end_time", r"(?:Ende|Endzeit)[:\s]+([0-9:.\-\s]+)", None), ("duration", r"Dauer[:\s]+([0-9:.\-\s]+)", None), ("min_temperature", r"Min(?:dest)?temperatur[:\s]+([0-9,.]+)", "°C"), ("max_temperature", r"Max(?:imal|\.)?temperatur[:\s]+([0-9,.]+)", "°C"), ("temperature_band", r"Temperaturband[:\s]+([0-9,.]+)", "K"), ("holding_time", r"Haltezeit[:\s]+([0-9:.\-\s]+)", None), ("pressure", r"Druck[:\s]+([0-9,.-]+)", "bar"), ("leak_rate", r"Leckrate[:\s]+([0-9,.-]+)", "mbar/min"), ("result", r"Ergebnis[:\s]+([^\n]+)", None), ]