Initial Validation Suite V2 report engine
This commit is contained in:
commit
2b5c765e41
79 changed files with 2072 additions and 0 deletions
272
app/app.py
Normal file
272
app/app.py
Normal file
|
|
@ -0,0 +1,272 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from datetime import date, datetime
|
||||
from pathlib import Path
|
||||
|
||||
from flask import Flask, flash, redirect, render_template, request, send_from_directory, url_for
|
||||
from flask_login import LoginManager, UserMixin, current_user, login_required, login_user, logout_user
|
||||
from flask_sqlalchemy import SQLAlchemy
|
||||
from sqlalchemy import func, select
|
||||
from werkzeug.security import check_password_hash, generate_password_hash
|
||||
|
||||
from reporting.report_builder import build_validation_report
|
||||
|
||||
db = SQLAlchemy()
|
||||
login_manager = LoginManager()
|
||||
BASE_DIR = Path(__file__).resolve().parent
|
||||
GENERATED_DIR = BASE_DIR / "generated"
|
||||
GENERATED_DIR.mkdir(exist_ok=True)
|
||||
|
||||
|
||||
class User(UserMixin, db.Model):
|
||||
__tablename__ = "users"
|
||||
id = db.Column(db.Integer, primary_key=True)
|
||||
username = db.Column(db.String(100), unique=True, nullable=False)
|
||||
password_hash = db.Column(db.String(255), nullable=False)
|
||||
|
||||
def set_password(self, password):
|
||||
self.password_hash = generate_password_hash(password)
|
||||
|
||||
def check_password(self, password):
|
||||
return check_password_hash(self.password_hash, password)
|
||||
|
||||
|
||||
class Customer(db.Model):
|
||||
__tablename__ = "customers"
|
||||
id = db.Column(db.Integer, primary_key=True)
|
||||
practice_name = db.Column(db.String(180), nullable=False)
|
||||
operator_name = db.Column(db.String(180), nullable=False)
|
||||
contact_person = db.Column(db.String(180))
|
||||
hygiene_contact = db.Column(db.String(180))
|
||||
street = db.Column(db.String(180))
|
||||
postal_code = db.Column(db.String(20))
|
||||
city = db.Column(db.String(120))
|
||||
email = db.Column(db.String(180))
|
||||
phone = db.Column(db.String(80))
|
||||
|
||||
@property
|
||||
def address(self):
|
||||
return ", ".join(x for x in [
|
||||
self.street or "",
|
||||
f"{self.postal_code or ''} {self.city or ''}".strip()
|
||||
] if x)
|
||||
|
||||
|
||||
class Device(db.Model):
|
||||
__tablename__ = "devices"
|
||||
id = db.Column(db.Integer, primary_key=True)
|
||||
customer_id = db.Column(db.Integer, db.ForeignKey("customers.id"), nullable=False)
|
||||
manufacturer = db.Column(db.String(120), nullable=False)
|
||||
model = db.Column(db.String(120), nullable=False)
|
||||
serial_number = db.Column(db.String(120), unique=True, nullable=False)
|
||||
year_built = db.Column(db.String(20))
|
||||
commissioning_date = db.Column(db.Date)
|
||||
chamber_volume = db.Column(db.String(50))
|
||||
steam_generation = db.Column(db.String(180))
|
||||
water_supply = db.Column(db.String(180))
|
||||
documentation = db.Column(db.String(180))
|
||||
supplier = db.Column(db.String(180))
|
||||
customer = db.relationship("Customer", backref="devices")
|
||||
|
||||
|
||||
class Equipment(db.Model):
|
||||
__tablename__ = "equipment"
|
||||
id = db.Column(db.Integer, primary_key=True)
|
||||
designation = db.Column(db.String(150), nullable=False)
|
||||
manufacturer = db.Column(db.String(120))
|
||||
model = db.Column(db.String(120))
|
||||
serial_number = db.Column(db.String(120), unique=True, nullable=False)
|
||||
measurement_range = db.Column(db.String(150))
|
||||
calibration_date = db.Column(db.Date)
|
||||
calibration_due = db.Column(db.Date)
|
||||
equipment_type = db.Column(db.String(80))
|
||||
|
||||
|
||||
class Validation(db.Model):
|
||||
__tablename__ = "validations"
|
||||
id = db.Column(db.Integer, primary_key=True)
|
||||
report_number = db.Column(db.String(60), unique=True, nullable=False)
|
||||
customer_id = db.Column(db.Integer, db.ForeignKey("customers.id"), nullable=False)
|
||||
device_id = db.Column(db.Integer, db.ForeignKey("devices.id"), nullable=False)
|
||||
validation_type = db.Column(db.String(60), nullable=False)
|
||||
project = db.Column(db.String(120), nullable=False)
|
||||
test_date = db.Column(db.Date, nullable=False)
|
||||
test_location = db.Column(db.String(120), nullable=False)
|
||||
examiner = db.Column(db.String(150), nullable=False)
|
||||
participants = db.Column(db.String(255))
|
||||
status = db.Column(db.String(30), default="Entwurf", nullable=False)
|
||||
result = db.Column(db.String(30), default="Bestanden", nullable=False)
|
||||
room_temperature = db.Column(db.String(20))
|
||||
room_humidity = db.Column(db.String(20))
|
||||
vacuum_leak_rate = db.Column(db.String(30))
|
||||
next_validation = db.Column(db.Date)
|
||||
recommendations = db.Column(db.Text)
|
||||
notes = db.Column(db.Text)
|
||||
customer = db.relationship("Customer")
|
||||
device = db.relationship("Device")
|
||||
|
||||
|
||||
@login_manager.user_loader
|
||||
def load_user(user_id):
|
||||
try:
|
||||
return db.session.get(User, int(user_id))
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def parse_date(value):
|
||||
if not value:
|
||||
return None
|
||||
try:
|
||||
return datetime.strptime(value, "%Y-%m-%d").date()
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def next_report_number():
|
||||
count = db.session.scalar(select(func.count()).select_from(Validation)) or 0
|
||||
return f"VAL-{datetime.now().year}-{count + 1:04d}"
|
||||
|
||||
|
||||
def seed_demo_data():
|
||||
if db.session.scalar(select(func.count()).select_from(Customer)):
|
||||
return
|
||||
|
||||
customer = Customer(
|
||||
practice_name="Urologische Praxis Muster",
|
||||
operator_name="Dr. Max Mustermann",
|
||||
contact_person="Dr. Max Mustermann",
|
||||
hygiene_contact="Frau Beispiel",
|
||||
street="Musterstraße 5",
|
||||
postal_code="90459",
|
||||
city="Nürnberg",
|
||||
email="praxis@example.de",
|
||||
phone="0911 123456",
|
||||
)
|
||||
db.session.add(customer)
|
||||
db.session.flush()
|
||||
|
||||
device = Device(
|
||||
customer_id=customer.id,
|
||||
manufacturer="Euronda",
|
||||
model="E10.7",
|
||||
serial_number="DEMO-EXN250688",
|
||||
year_built="2025",
|
||||
commissioning_date=date(2025, 12, 5),
|
||||
chamber_volume="23 Liter",
|
||||
steam_generation="Eigendampferzeugung",
|
||||
water_supply="Automatische Versorgung über Aquafilter",
|
||||
documentation="Interne CF-Card, Ausgabe am PC möglich",
|
||||
supplier="Medizintechnik Muster",
|
||||
)
|
||||
db.session.add(device)
|
||||
|
||||
equipment = [
|
||||
Equipment(designation="Temperatur Datenlogger 1", manufacturer="Ebro", model="EBI 11 T235", serial_number="15102807", measurement_range="0 °C bis +150 °C", calibration_date=date(2025,1,17), calibration_due=date(2027,1,17), equipment_type="Temperatur"),
|
||||
Equipment(designation="Temperatur Datenlogger 2", manufacturer="Ebro", model="EBI 11 T240", serial_number="15211066", measurement_range="0 °C bis +150 °C", calibration_date=date(2025,1,17), calibration_due=date(2027,1,17), equipment_type="Temperatur"),
|
||||
Equipment(designation="Temperatur Datenlogger 3", manufacturer="Ebro", model="EBI 11 T240", serial_number="15102538", measurement_range="0 °C bis +150 °C", calibration_date=date(2025,1,17), calibration_due=date(2027,1,17), equipment_type="Temperatur"),
|
||||
Equipment(designation="Temperatur Datenlogger 4", manufacturer="Ebro", model="EBI 11 T240", serial_number="15211067", measurement_range="0 °C bis +150 °C", calibration_date=date(2025,1,17), calibration_due=date(2027,1,17), equipment_type="Temperatur"),
|
||||
Equipment(designation="Temperatur Datenlogger 5", manufacturer="Ebro", model="EBI 11 T240", serial_number="15125738", measurement_range="0 °C bis +150 °C", calibration_date=date(2025,1,17), calibration_due=date(2027,1,17), equipment_type="Temperatur"),
|
||||
Equipment(designation="Drucklogger", manufacturer="Ebro", model="EBI 11 P111", serial_number="15092368", measurement_range="-100 mbar bis 4 bar abs.", calibration_date=date(2025,1,17), calibration_due=date(2027,1,17), equipment_type="Druck"),
|
||||
]
|
||||
db.session.add_all(equipment)
|
||||
db.session.flush()
|
||||
|
||||
validation = Validation(
|
||||
report_number="VAL-DEMO-0001",
|
||||
customer_id=customer.id,
|
||||
device_id=device.id,
|
||||
validation_type="Erstvalidierung",
|
||||
project="Erst-Validierung",
|
||||
test_date=date(2025, 12, 5),
|
||||
test_location="Nürnberg",
|
||||
examiner="Andreas Schubert",
|
||||
participants="Frau Beispiel, Dr. Max Mustermann",
|
||||
status="Entwurf",
|
||||
result="Bestanden",
|
||||
room_temperature="21 °C",
|
||||
room_humidity="41 %",
|
||||
vacuum_leak_rate="0,2 mbar/min",
|
||||
next_validation=date(2027, 11, 1),
|
||||
recommendations="Auflagen und Empfehlungen werden im Kapitel 4.2 dokumentiert.",
|
||||
notes="V2-Vorschau: Winlog-Messwerte, Originalbilder, Zertifikate und Logo werden später dynamisch ersetzt.",
|
||||
)
|
||||
db.session.add(validation)
|
||||
db.session.commit()
|
||||
|
||||
|
||||
def create_app():
|
||||
app = Flask(__name__)
|
||||
app.config.update(
|
||||
SECRET_KEY=os.getenv("SECRET_KEY", "validation-v2-local-secret"),
|
||||
SQLALCHEMY_DATABASE_URI=os.getenv("DATABASE_URL", "sqlite:///validation-v2.db"),
|
||||
SQLALCHEMY_TRACK_MODIFICATIONS=False,
|
||||
)
|
||||
db.init_app(app)
|
||||
login_manager.init_app(app)
|
||||
login_manager.login_view = "login"
|
||||
|
||||
@app.route("/register", methods=["GET", "POST"])
|
||||
def register():
|
||||
if current_user.is_authenticated:
|
||||
return redirect(url_for("dashboard"))
|
||||
if request.method == "POST":
|
||||
username = (request.form.get("username") or "").strip()
|
||||
password = request.form.get("password") or ""
|
||||
repeat = request.form.get("password_repeat") or ""
|
||||
if len(username) < 3 or len(password) < 6 or password != repeat:
|
||||
flash("Bitte gültige Zugangsdaten eingeben.", "error")
|
||||
elif db.session.scalar(select(User).where(func.lower(User.username) == username.lower())):
|
||||
flash("Benutzername bereits vergeben.", "error")
|
||||
else:
|
||||
user = User(username=username)
|
||||
user.set_password(password)
|
||||
db.session.add(user)
|
||||
db.session.commit()
|
||||
return redirect(url_for("login"))
|
||||
return render_template("register.html")
|
||||
|
||||
@app.route("/login", methods=["GET", "POST"])
|
||||
def login():
|
||||
if current_user.is_authenticated:
|
||||
return redirect(url_for("dashboard"))
|
||||
if request.method == "POST":
|
||||
username = (request.form.get("username") or "").strip()
|
||||
user = db.session.scalar(select(User).where(func.lower(User.username) == username.lower()))
|
||||
if user and user.check_password(request.form.get("password") or ""):
|
||||
login_user(user)
|
||||
return redirect(url_for("dashboard"))
|
||||
flash("Anmeldung fehlgeschlagen.", "error")
|
||||
return render_template("login.html")
|
||||
|
||||
@app.route("/logout", methods=["POST"])
|
||||
@login_required
|
||||
def logout():
|
||||
logout_user()
|
||||
return redirect(url_for("login"))
|
||||
|
||||
@app.route("/")
|
||||
@login_required
|
||||
def dashboard():
|
||||
validations = list(db.session.scalars(select(Validation).order_by(Validation.id.desc())))
|
||||
return render_template("dashboard.html", validations=validations)
|
||||
|
||||
@app.route("/validations/<int:item_id>/report")
|
||||
@login_required
|
||||
def report(item_id):
|
||||
validation = db.session.get(Validation, item_id)
|
||||
if not validation:
|
||||
return "Nicht gefunden", 404
|
||||
equipment = list(db.session.scalars(select(Equipment).order_by(Equipment.id)))
|
||||
filename = build_validation_report(validation, equipment, GENERATED_DIR)
|
||||
return send_from_directory(GENERATED_DIR, filename, as_attachment=True)
|
||||
|
||||
with app.app_context():
|
||||
db.create_all()
|
||||
seed_demo_data()
|
||||
|
||||
return app
|
||||
|
||||
|
||||
app = create_app()
|
||||
Loading…
Add table
Add a link
Reference in a new issue