204 lines
8.1 KiB
Python
204 lines
8.1 KiB
Python
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),
|
|
]
|