import os import shutil from fastapi import APIRouter, Request, Depends, Form, File, UploadFile, HTTPException, status from fastapi.responses import HTMLResponse, RedirectResponse, JSONResponse, FileResponse from sqlalchemy import func from sqlalchemy.orm import Session import models import email_feliratkozas from hibakodok import hiba_uzenet, HIBAKODOK from database import get_db from security import get_current_user, is_rendszeradminisztrator, is_safe_extension, sanitize_filename from alap_konfig_feltolto import feltolt_alap_konfigokat from naplo_utils import naploz, be_ki from kozos_dokumentum_utils import kozos_utvonalak, arva_kozos_torlese from ceg_sablon_utils import ceg_sablonok_feltoltese, sajat_sablon_szamok from verzio_utils import jelenlegi_verzio, valtozasnaplo_beolvasasa from templates_env import templates router = APIRouter() def _rendszeradmin_kapu(current_user): return is_rendszeradminisztrator(current_user) @router.get("/rendszeradmin/cegek", response_class=HTMLResponse) def rendszeradmin_cegek( request: Request, sablon: str = "", db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): if not _rendszeradmin_kapu(current_user): return RedirectResponse(url="/valassz-modult", status_code=303) cegek = db.query(models.Company).order_by(models.Company.nev).all() modulok = db.query(models.Modul).filter(models.Modul.aktiv == True).all() ceg_modul_map = {c.id: {m.id for m in c.engedelyezett_modulok} for c in cegek} ai_dict_sorok = db.query(models.Dictionary).filter(models.Dictionary.kategoria == "AiSzkennelesEngedelyezve").all() ai_szkenneles_map = {r.company_id: (r.ertek == "True") for r in ai_dict_sorok} azonositas_dict_sorok = db.query(models.Dictionary).filter(models.Dictionary.kategoria == "AzonositasMod").all() azonositas_mod_map = {r.company_id: r.ertek for r in azonositas_dict_sorok} oe_alapterulet_dict_sorok = db.query(models.Dictionary).filter(models.Dictionary.kategoria == "OEAlapteruletSzamitas").all() oe_alapterulet_map = {r.company_id: (r.ertek == "True") for r in oe_alapterulet_dict_sorok} sugo_dict_sorok = db.query(models.Dictionary).filter(models.Dictionary.kategoria == "SugoEngedelyezve").all() sugo_map = {r.company_id: (r.ertek == "True") for r in sugo_dict_sorok} # Arculat cégenként arculat_map = {} for kategoria, kulcs in [ ("JegyzokonyvLogoUtvonal", "logo_utvonal"), ("JegyzokonyvFejlecSzoveg", "fejlec_szoveg"), ("JegyzokonyvLablecMarka", "lablec_marka"), ]: for r in db.query(models.Dictionary).filter(models.Dictionary.kategoria == kategoria).all(): arculat_map.setdefault(r.company_id, {})[kulcs] = r.ertek # Fő felhasználók: Admin (mindkét modul), Emelőgép szakértő/vizsgáló (Emelőgép), Karbantartó (Tűzvédelem) fo_felhasznalok = db.query(models.User).filter(models.User.role.in_(["Admin", "Emelőgép szakértő", "Emelőgép vizsgáló", "Karbantartó"])).all() fo_felhasznalok_map = {} for u in fo_felhasznalok: fo_felhasznalok_map.setdefault(u.company_id, []).append(u) # Felhasználónkénti modul-hozzárendelés (minden felhasználó, a cég modul-előfizetésén belül) osszes_felhasznalo = db.query(models.User).order_by(models.User.company_id, models.User.username).all() user_modul_map = {u.id: {m.id for m in u.sajat_modulok} for u in osszes_felhasznalo} # Dashboard-statisztika cégenként: hány eszköz van nyilvántartva, és hány felhasználó # van szerepkörönként/összesen - a rendszeradminisztrátor saját fiókja (Admin + # company_id==1) nem számít bele, mert ő nem egy valódi cég munkatársa. gep_szamlalo_map = dict( db.query(models.Machine.company_id, func.count(models.Machine.id)) .group_by(models.Machine.company_id).all() ) felhasznalo_szamlalo_map = {} for u in osszes_felhasznalo: if is_rendszeradminisztrator(u): continue adat = felhasznalo_szamlalo_map.setdefault(u.company_id, {"osszesen": 0, "szerepkorok": {}}) adat["osszesen"] += 1 adat["szerepkorok"][u.role] = adat["szerepkorok"].get(u.role, 0) + 1 # v81: cégenként a SAJÁT ellenőrzőlista-sablonok száma (ha 0, a cég az 1-es cég közös készletét használja) sablon_map = {c.id: sajat_sablon_szamok(db, c.id) for c in cegek} # a sablon-feltöltés eredménye (kizárólag fix kódok jelennek meg, a kérésből szöveg nem kerül az oldalra) sablon_uzenet = { "ok": ("success", "A kijelölt ellenőrzőlista-sablonok feltöltve."), "hiba": ("danger", "A cég létrejött, de a sablonok feltöltése nem sikerült (semmi sem változott a sablonoknál). Próbáld újra a cég kártyáján."), "hiba_feltoltes": ("danger", "A sablonok feltöltése nem sikerült, semmi sem változott."), }.get(sablon) return templates.TemplateResponse( request=request, name="rendszeradmin_cegek.html", context={ "sablon_map": sablon_map, "sablon_uzenet": sablon_uzenet, "user": current_user, "cegek": cegek, "modulok": modulok, "ceg_modul_map": ceg_modul_map, "ai_szkenneles_map": ai_szkenneles_map, "azonositas_mod_map": azonositas_mod_map, "oe_alapterulet_map": oe_alapterulet_map, "sugo_map": sugo_map, "arculat_map": arculat_map, "fo_felhasznalok_map": fo_felhasznalok_map, "osszes_felhasznalo": osszes_felhasznalo, "user_modul_map": user_modul_map, "gep_szamlalo_map": gep_szamlalo_map, "felhasznalo_szamlalo_map": felhasznalo_szamlalo_map, "verzio": jelenlegi_verzio(), } ) @router.get("/rendszeradmin/rendszerinfo", response_class=HTMLResponse) def rendszerinfo( request: Request, current_user: models.User = Depends(get_current_user) ): """Verziótörténet (CHANGELOG.md) + hibakód-magyarázatok egy helyen - annak eldöntésére, hogy a szerveren futó kód valóban a legfrissebb-e (l. 2026-09-29-i user-kérés, ami épp abból a zavarból született, hogy egy már javított hiba még mindig jelentkezett élesben, mert a telepítés/gyorsítótár állapotát nem lehetett egyszerűen ellenőrizni).""" if not _rendszeradmin_kapu(current_user): return RedirectResponse(url="/valassz-modult", status_code=303) return templates.TemplateResponse( request=request, name="rendszeradmin_rendszerinfo.html", context={ "user": current_user, "valtozasnaplo": valtozasnaplo_beolvasasa(), "hibakodok": sorted(HIBAKODOK.items()), } ) @router.post("/rendszeradmin/cegek/uj") def uj_ceg( nev: str = Form(...), adoszam: str = Form(""), sablon_targonca: str = Form(""), sablon_msz9721: str = Form(""), db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): if not _rendszeradmin_kapu(current_user): return JSONResponse(status_code=403, content={"message": "Csak a rendszeradminisztrátor!"}) uj_ceg_obj = models.Company(nev=nev, adoszam=adoszam or None, aktiv=True) db.add(uj_ceg_obj) db.commit() db.refresh(uj_ceg_obj) feltolt_alap_konfigokat(db, uj_ceg_obj.id) szotarak_alap = db.query(models.Dictionary).filter( models.Dictionary.company_id == 1, models.Dictionary.kategoria.notin_(["JegyzokonyvSablon", "NapiEllenorzesSablon"]) ).all() for sz in szotarak_alap: db.add(models.Dictionary(company_id=uj_ceg_obj.id, kategoria=sz.kategoria, ertek=sz.ertek)) konfigok_alap = db.query(models.MachineTypeConfig).filter(models.MachineTypeConfig.company_id == 1).all() for k in konfigok_alap: db.add(models.MachineTypeConfig(company_id=uj_ceg_obj.id, geptipus=k.geptipus, kotelezo_mezok=k.kotelezo_mezok)) # J-20 (v77): a cég létrehozása a Rendszernaplóban és az új cég Jogosultsági naplójában is látszik naploz(db, current_user, uj_ceg_obj.id, "Cég létrehozva", None, f"{nev} (adószám: {adoszam or '-'})") db.commit() # v81: a két jelölőnégyzet - a sablon-feltöltő szkriptek az ÚJ cégnek (saját sablon-példány). A cég ekkor már létezik; a feltöltés külön # tranzakció, hibánál csak a sablonok maradnak ki (a cég és a többi alapadat megmarad), és a hiba naplózódik. kert = _sablon_kerese(sablon_targonca, sablon_msz9721) if not kert: return RedirectResponse(url="/rendszeradmin/cegek", status_code=303) eredmeny = _sablonok_feltoltese_naplozva(db, current_user, uj_ceg_obj.id, kert) return RedirectResponse(url=f"/rendszeradmin/cegek?sablon={'ok' if eredmeny == 'ok' else 'hiba'}", status_code=303) def _sablon_kerese(sablon_targonca: str, sablon_msz9721: str): """A két jelölőnégyzet értéke -> (targonca, msz9721) logikai pár, vagy None, ha egyik sincs bejelölve.""" targonca, msz9721 = (sablon_targonca == "on"), (sablon_msz9721 == "on") return (targonca, msz9721) if (targonca or msz9721) else None def _sablonok_feltoltese_naplozva(db: Session, current_user, ceg_id: int, kert) -> str: """A kijelölt sablon-szkriptek futtatása a cégre + napló, EGY tranzakcióban. Visszatérés: "ok" vagy "hiba_feltoltes" (visszagörgetve, a hiba naplózva).""" targonca, msz9721 = kert elotte = sajat_sablon_szamok(db, ceg_id) try: lefutott = ceg_sablonok_feltoltese(db, ceg_id, targonca=targonca, msz9721=msz9721) utana = sajat_sablon_szamok(db, ceg_id) naploz(db, current_user, ceg_id, "Cég sablonok feltöltve", f"Saját sablonok előtte: {elotte['jegyzokonyv']} jegyzőkönyv, {elotte['napi']} napi", f"Futtatva: {'; '.join(lefutott)}. Saját sablonok utána: {utana['jegyzokonyv']} jegyzőkönyv, {utana['napi']} napi") db.commit() return "ok" except Exception as e: db.rollback() print(f"❌ Hiba a cég sablonjainak feltöltése közben (cég id={ceg_id}): {type(e).__name__}: {e}") naploz(db, current_user, ceg_id, "Cég sablonok feltöltése sikertelen", None, f"Kért: {'targonca ' if targonca else ''}{'MSZ 9721-1/-2' if msz9721 else ''}".strip() + f" - hiba: {type(e).__name__}") db.commit() return "hiba_feltoltes" @router.post("/rendszeradmin/cegek/{ceg_id}/sablonok") def ceg_sablonok_ujrafeltoltese( ceg_id: int, sablon_targonca: str = Form(""), sablon_msz9721: str = Form(""), megertettem: str = Form(""), db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): """MEGLÉVŐ cég ellenőrzőlista-sablonjainak (újra)feltöltése - FELÜLÍRJA a cég azonos nevű saját sablonjait (kézi szerkesztések is elvesznek ezeken a sablonokon). Kizárólag a rendszeradminisztrátor, kötelező tudomásulvételi jelölőnégyzettel.""" if not _rendszeradmin_kapu(current_user): return JSONResponse(status_code=403, content={"message": "Csak a rendszeradminisztrátor!"}) ceg = db.query(models.Company).filter(models.Company.id == ceg_id).first() if not ceg: return JSONResponse(status_code=404, content={"message": "Cég nem található!"}) kert = _sablon_kerese(sablon_targonca, sablon_msz9721) if not kert: return JSONResponse(status_code=400, content={"message": "Jelölj ki legalább egy sablon-készletet!"}) if megertettem != "on": return JSONResponse(status_code=400, content={"message": "A tudomásulvételi jelölőnégyzet bejelölése kötelező (a művelet felülírja a meglévő sablonokat)!"}) eredmeny = _sablonok_feltoltese_naplozva(db, current_user, ceg_id, kert) return RedirectResponse(url=f"/rendszeradmin/cegek?sablon={eredmeny}", status_code=303) @router.post("/rendszeradmin/cegek/{ceg_id}/adatok") def ceg_adatok_mentese( ceg_id: int, nev: str = Form(...), adoszam: str = Form(""), cim: str = Form(""), elerhetoseg: str = Form(""), db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): if not _rendszeradmin_kapu(current_user): return JSONResponse(status_code=403, content={"message": "Csak a rendszeradminisztrátor!"}) ceg = db.query(models.Company).filter(models.Company.id == ceg_id).first() if not ceg: return JSONResponse(status_code=404, content={"message": "Cég nem található!"}) def _adat_szoveg(n, a, c, e): return f"{n}; adószám: {a or '-'}; cím: {c or '-'}; elérhetőség: {e or '-'}" regi_adatok = _adat_szoveg(ceg.nev, ceg.adoszam, ceg.cim, ceg.elerhetoseg) ceg.nev = nev ceg.adoszam = adoszam or None ceg.cim = cim or None ceg.elerhetoseg = elerhetoseg or None naploz(db, current_user, ceg.id, "Cég adatai", regi_adatok, _adat_szoveg(ceg.nev, ceg.adoszam, ceg.cim, ceg.elerhetoseg), csak_ha_valtozott=True) db.commit() return RedirectResponse(url="/rendszeradmin/cegek", status_code=303) @router.post("/rendszeradmin/cegek/{ceg_id}/licenc") def ceg_licenc_mentese( ceg_id: int, licenc_max_gep: str = Form(""), licenc_max_dolgozo: str = Form(""), licenc_max_ugyintezo: str = Form(""), db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): if not _rendszeradmin_kapu(current_user): return JSONResponse(status_code=403, content={"message": "Csak a rendszeradminisztrátor!"}) ceg = db.query(models.Company).filter(models.Company.id == ceg_id).first() if not ceg: return JSONResponse(status_code=404, content={"message": "Cég nem található!"}) def ertelmez(nyers): nyers = (nyers or "").strip() return int(nyers) if nyers else None def _licenc_szoveg(g, d, u): return f"gép: {g if g is not None else 'korlátlan'}; dolgozó: {d if d is not None else 'korlátlan'}; ügyintéző: {u if u is not None else 'korlátlan'}" regi_licenc = _licenc_szoveg(ceg.licenc_max_gep, ceg.licenc_max_dolgozo, ceg.licenc_max_ugyintezo) ceg.licenc_max_gep = ertelmez(licenc_max_gep) ceg.licenc_max_dolgozo = ertelmez(licenc_max_dolgozo) ceg.licenc_max_ugyintezo = ertelmez(licenc_max_ugyintezo) naploz(db, current_user, ceg.id, "Cég licenc", regi_licenc, _licenc_szoveg(ceg.licenc_max_gep, ceg.licenc_max_dolgozo, ceg.licenc_max_ugyintezo), csak_ha_valtozott=True) db.commit() return RedirectResponse(url="/rendszeradmin/cegek", status_code=303) @router.post("/rendszeradmin/cegek/{ceg_id}/aktivalas") def ceg_aktivalasa( ceg_id: int, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): """Cég (ügyfél) inaktiválása/reaktiválása. Törlés helyett - minden történeti adat (gépek, vizsgálatok, naplók) megmarad, csak a szerkesztési jogosultság szűnik meg a cég alá tartozó minden felhasználónál.""" if not _rendszeradmin_kapu(current_user): return JSONResponse(status_code=403, content={"message": "Csak a rendszeradminisztrátor!"}) ceg = db.query(models.Company).filter(models.Company.id == ceg_id).first() if not ceg or ceg_id == 1: return JSONResponse(status_code=403, content={"message": "Ez a cég nem inaktiválható!"}) regi_allapot = "Aktív" if ceg.aktiv else "Inaktív" ceg.aktiv = not ceg.aktiv uj_allapot = "Aktív" if ceg.aktiv else "Inaktív" db.add(models.AuditLog( company_id=ceg.id, user_id=current_user.id, field_name="Cég állapota", old_value=regi_allapot, new_value=uj_allapot )) db.commit() allapot = "aktiválva" if ceg.aktiv else "inaktiválva" return JSONResponse(status_code=200, content={"message": f"Cég sikeresen {allapot}!", "aktiv": ceg.aktiv}) @router.post("/rendszeradmin/cegek/{ceg_id}/berbeadas-kapcsolasa") def ceg_berbeadas_kapcsolasa( ceg_id: int, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): """A bérbeadás/kihelyezés jog cégenkénti be-/kikapcsolása - nem minden cég ad bérbe gépet, ezért ez alapból kikapcsolt, csak a rendszeradmin engedélyezheti cégenként.""" if not _rendszeradmin_kapu(current_user): return JSONResponse(status_code=403, content={"message": "Csak a rendszeradminisztrátor!"}) ceg = db.query(models.Company).filter(models.Company.id == ceg_id).first() if not ceg: return JSONResponse(status_code=404, content={"message": "Cég nem található!"}) regi_allapot = "Engedélyezve" if ceg.berbeadhat else "Letiltva" ceg.berbeadhat = not ceg.berbeadhat uj_allapot = "Engedélyezve" if ceg.berbeadhat else "Letiltva" db.add(models.AuditLog( company_id=ceg.id, user_id=current_user.id, field_name="Bérbeadás/kihelyezés joga", old_value=regi_allapot, new_value=uj_allapot )) db.commit() allapot = "engedélyezve" if ceg.berbeadhat else "letiltva" return JSONResponse(status_code=200, content={"message": f"Bérbeadás/kihelyezés {allapot}!", "berbeadhat": ceg.berbeadhat}) def _ceg_torlesi_szamlalo(ceg_id: int, db: Session) -> dict: """Előzetes összegzés a végleges cégtörlés előtti megerősítéshez: mennyi adat érintett. KIEGÉSZÍTVE (2026-10-03-i külső biztonsági audit, 2. kör): a tervezett vizsgálatok/MSZ 16226 dokumentumok és az EBK/VIVÜ-modul adatai eddig nem jelentek meg itt, pedig a _torli_teljes_ceget() most már ezeket is törli - az admin lássa előre a teljes kört.""" gep_id_lista = [r[0] for r in db.query(models.Machine.id).filter(models.Machine.company_id == ceg_id).all()] naplo_gephez = fajlok = vizsgalatok = hibajegyek = tervezett_vizsgalatok = msz16226_dokumentumok = 0 if gep_id_lista: naplo_gephez = db.query(models.AuditLog).filter(models.AuditLog.machine_id.in_(gep_id_lista)).count() fajlok = db.query(models.File).filter(models.File.machine_id.in_(gep_id_lista)).count() vizsgalatok = db.query(models.Inspection).filter(models.Inspection.machine_id.in_(gep_id_lista)).count() hibajegyek = db.query(models.MaintenanceTicket).filter(models.MaintenanceTicket.machine_id.in_(gep_id_lista)).count() tervezett_vizsgalatok = db.query(models.TervezettVizsgalat).filter(models.TervezettVizsgalat.machine_id.in_(gep_id_lista)).count() msz16226_dokumentumok = db.query(models.Msz16226Dokumentum).filter(models.Msz16226Dokumentum.machine_id.in_(gep_id_lista)).count() naplo_ceghez = db.query(models.AuditLog).filter(models.AuditLog.company_id == ceg_id).count() ebk_projektek = db.query(models.EbkProjekt).filter(models.EbkProjekt.company_id == ceg_id).count() vivu_ugyfelek = db.query(models.VivuUgyfel).filter(models.VivuUgyfel.company_id == ceg_id).count() return { "felhasznalok": db.query(models.User).filter(models.User.company_id == ceg_id).count(), "telephelyek": db.query(models.Site).filter(models.Site.company_id == ceg_id).count(), "gepek": len(gep_id_lista), "fajlok": fajlok, "vizsgalatok": vizsgalatok, "hibajegyek": hibajegyek, "naplo_bejegyzesek": naplo_gephez + naplo_ceghez, "tervezett_vizsgalatok": tervezett_vizsgalatok, "msz16226_dokumentumok": msz16226_dokumentumok, "ebk_projektek": ebk_projektek, "vivu_ugyfelek": vivu_ugyfelek, } def _torli_ebk_adatokat(ceg_id: int, db: Session): """MAGAS (2026-10-03-i külső biztonsági audit, 2. kör): a cégtörlés eddig egyáltalán nem érintette az EBK-modul adatfáját (routers/ebk.py) - egy törölt cég EBK-projektjei, ellenőrzései, katalógusai árván maradtak volna (vagy a DB FK-kényszerei miatt a teljes törlés IntegrityError-ral elakadt volna). A törlési sorrend a modell-fájlban definiált hierarchiát követi, levelektől a gyökér (company_id) felé.""" projekt_id_lista = [r[0] for r in db.query(models.EbkProjekt.id).filter(models.EbkProjekt.company_id == ceg_id).all()] terulet_id_lista = [r[0] for r in db.query(models.EbkTerulet.id).filter(models.EbkTerulet.company_id == ceg_id).all()] epulet_id_lista = [r[0] for r in db.query(models.EbkEpulet.id).filter(models.EbkEpulet.terulet_id.in_(terulet_id_lista)).all()] if terulet_id_lista else [] if projekt_id_lista: ellenorzes_id_lista = [r[0] for r in db.query(models.EbkEllenorzes.id).filter(models.EbkEllenorzes.projekt_id.in_(projekt_id_lista)).all()] projekt_alvallalkozo_id_lista = [r[0] for r in db.query(models.EbkProjektAlvallalkozo.id).filter(models.EbkProjektAlvallalkozo.projekt_id.in_(projekt_id_lista)).all()] if projekt_alvallalkozo_id_lista: db.execute(models.ebk_projekt_alvallalkozo_tevekenysegek.delete().where( models.ebk_projekt_alvallalkozo_tevekenysegek.c.projekt_alvallalkozo_id.in_(projekt_alvallalkozo_id_lista) )) if ellenorzes_id_lista: db.query(models.EbkEllenorzesTetel).filter(models.EbkEllenorzesTetel.ellenorzes_id.in_(ellenorzes_id_lista)).delete(synchronize_session=False) db.query(models.EbkEllenorzes).filter(models.EbkEllenorzes.id.in_(ellenorzes_id_lista)).delete(synchronize_session=False) db.query(models.EbkProjektTetel).filter(models.EbkProjektTetel.projekt_id.in_(projekt_id_lista)).delete(synchronize_session=False) db.query(models.EbkProjektAlvallalkozo).filter(models.EbkProjektAlvallalkozo.projekt_id.in_(projekt_id_lista)).delete(synchronize_session=False) db.execute(models.user_ebk_projektek.delete().where(models.user_ebk_projektek.c.projekt_id.in_(projekt_id_lista))) db.execute(models.ebk_projekt_koordinatorok.delete().where(models.ebk_projekt_koordinatorok.c.projekt_id.in_(projekt_id_lista))) db.execute(models.ebk_projekt_teruletek.delete().where(models.ebk_projekt_teruletek.c.projekt_id.in_(projekt_id_lista))) db.query(models.EbkProjekt).filter(models.EbkProjekt.company_id == ceg_id).delete(synchronize_session=False) if epulet_id_lista: db.query(models.EbkSzint).filter(models.EbkSzint.epulet_id.in_(epulet_id_lista)).delete(synchronize_session=False) if terulet_id_lista: db.query(models.EbkEpulet).filter(models.EbkEpulet.terulet_id.in_(terulet_id_lista)).delete(synchronize_session=False) db.query(models.EbkTerulet).filter(models.EbkTerulet.company_id == ceg_id).delete(synchronize_session=False) db.query(models.EbkUgyfel).filter(models.EbkUgyfel.company_id == ceg_id).delete(synchronize_session=False) db.query(models.EbkEllenorzesiTetel).filter(models.EbkEllenorzesiTetel.company_id == ceg_id).delete(synchronize_session=False) db.query(models.EbkAlvallalkozo).filter(models.EbkAlvallalkozo.company_id == ceg_id).delete(synchronize_session=False) db.query(models.EbkTevekenyseg).filter(models.EbkTevekenyseg.company_id == ceg_id).delete(synchronize_session=False) def _torli_vivu_adatokat(ceg_id: int, db: Session): """MAGAS (2026-10-03-i külső biztonsági audit, 2. kör): ugyanaz a hiányosság, mint az EBK-nál - a VIVÜ-modul (routers/vivu.py) adatfája eddig egyáltalán nem törlődött cégtörlésnél.""" ugyfel_id_lista = [r[0] for r in db.query(models.VivuUgyfel.id).filter(models.VivuUgyfel.company_id == ceg_id).all()] telephely_id_lista = [r[0] for r in db.query(models.VivuTelephely.id).filter(models.VivuTelephely.ugyfel_id.in_(ugyfel_id_lista)).all()] if ugyfel_id_lista else [] epulet_id_lista = [r[0] for r in db.query(models.VivuEpulet.id).filter(models.VivuEpulet.telephely_id.in_(telephely_id_lista)).all()] if telephely_id_lista else [] if epulet_id_lista: db.query(models.VivuSzint).filter(models.VivuSzint.epulet_id.in_(epulet_id_lista)).delete(synchronize_session=False) if telephely_id_lista: ellenorzes_id_lista = [r[0] for r in db.query(models.VivuEllenorzes.id).filter(models.VivuEllenorzes.telephely_id.in_(telephely_id_lista)).all()] if ellenorzes_id_lista: ellenorzes_tetel_id_lista = [r[0] for r in db.query(models.VivuEllenorzesTetel.id).filter(models.VivuEllenorzesTetel.ellenorzes_id.in_(ellenorzes_id_lista)).all()] if ellenorzes_tetel_id_lista: db.query(models.VivuTetelFenykep).filter(models.VivuTetelFenykep.tetel_id.in_(ellenorzes_tetel_id_lista)).delete(synchronize_session=False) db.execute(models.vivu_ellenorzes_epuletek.delete().where(models.vivu_ellenorzes_epuletek.c.ellenorzes_id.in_(ellenorzes_id_lista))) db.query(models.VivuEllenorzesTetel).filter(models.VivuEllenorzesTetel.ellenorzes_id.in_(ellenorzes_id_lista)).delete(synchronize_session=False) db.query(models.VivuEllenorzes).filter(models.VivuEllenorzes.telephely_id.in_(telephely_id_lista)).delete(synchronize_session=False) db.query(models.VivuEpulet).filter(models.VivuEpulet.telephely_id.in_(telephely_id_lista)).delete(synchronize_session=False) if ugyfel_id_lista: db.execute(models.user_vivu_ugyfelek.delete().where(models.user_vivu_ugyfelek.c.ugyfel_id.in_(ugyfel_id_lista))) db.query(models.VivuTelephely).filter(models.VivuTelephely.ugyfel_id.in_(ugyfel_id_lista)).delete(synchronize_session=False) db.query(models.VivuUgyfel).filter(models.VivuUgyfel.company_id == ceg_id).delete(synchronize_session=False) db.query(models.VivuEllenorzesiTetel).filter(models.VivuEllenorzesiTetel.company_id == ceg_id).delete(synchronize_session=False) def _torli_teljes_ceget(ceg_id: int, db: Session): """VÉGLEGES, adatbázis-szintű törlés: egy teljes cég és MINDEN hozzá tartozó adat (felhasználók, telephelyek, gépek, fájlok, vizsgálatok, hibajegyek, naplóbejegyzések, EBK/VIVÜ-modul adatai) visszavonhatatlan törlése. A törlési sorrend a külső kulcs függőségeket követi. MEGJEGYZÉS (2026-10-03-i külső biztonsági audit, 2. kör): a Tűzvédelem-modul (routers/ tuzvedelem.py) adatfája SZÁNDÉKOSAN nincs itt kezelve - a user jelzése szerint ez a modul még nincs kidolgozva és élesben nem használja senki a rendszer-szuperadminon kívül, ezért ez a hiányosság is nyitva marad (ugyanúgy, mint a tuzvedelem.py szerepkör-ellenőrzésének hiánya a 2. köri audit "Közepes" találatai között).""" gep_id_lista = [r[0] for r in db.query(models.Machine.id).filter(models.Machine.company_id == ceg_id).all()] user_id_lista = [r[0] for r in db.query(models.User.id).filter(models.User.company_id == ceg_id).all()] site_id_lista = [r[0] for r in db.query(models.Site.id).filter(models.Site.company_id == ceg_id).all()] if gep_id_lista: db.query(models.MaintenanceTicket).filter(models.MaintenanceTicket.machine_id.in_(gep_id_lista)).delete(synchronize_session=False) db.query(models.File).filter(models.File.machine_id.in_(gep_id_lista)).delete(synchronize_session=False) db.query(models.Inspection).filter(models.Inspection.machine_id.in_(gep_id_lista)).delete(synchronize_session=False) db.query(models.TervezettVizsgalat).filter(models.TervezettVizsgalat.machine_id.in_(gep_id_lista)).delete(synchronize_session=False) db.query(models.Msz16226Dokumentum).filter(models.Msz16226Dokumentum.machine_id.in_(gep_id_lista)).delete(synchronize_session=False) # A MszTelephelyMegjegyzes.audit_log_id ugyanennek a gépnek egy AuditLog sorára hivatkozhat # - ezért ezt MINDIG az AuditLog-törlés ELŐTT kell törölni. db.query(models.MszTelephelyMegjegyzes).filter(models.MszTelephelyMegjegyzes.machine_id.in_(gep_id_lista)).delete(synchronize_session=False) db.query(models.AuditLog).filter(models.AuditLog.machine_id.in_(gep_id_lista)).delete(synchronize_session=False) db.query(models.AuditLog).filter(models.AuditLog.company_id == ceg_id).delete(synchronize_session=False) _torli_ebk_adatokat(ceg_id, db) _torli_vivu_adatokat(ceg_id, db) db.query(models.SzakertoMegbizatas).filter(models.SzakertoMegbizatas.company_id == ceg_id).delete(synchronize_session=False) if user_id_lista: db.query(models.AuditLog).filter(models.AuditLog.user_id.in_(user_id_lista)).update( {models.AuditLog.user_id: None}, synchronize_session=False ) db.query(models.Inspection).filter(models.Inspection.user_id.in_(user_id_lista)).update( {models.Inspection.user_id: None}, synchronize_session=False ) if site_id_lista: db.query(models.User).filter(models.User.telephely_hozzaferes_id.in_(site_id_lista)).update( {models.User.telephely_hozzaferes_id: None}, synchronize_session=False ) if gep_id_lista: db.query(models.Machine).filter(models.Machine.company_id == ceg_id).update( {models.Machine.fo_gep_id: None}, synchronize_session=False ) db.query(models.Machine).filter(models.Machine.company_id == ceg_id).delete(synchronize_session=False) db.query(models.Site).filter(models.Site.company_id == ceg_id).delete(synchronize_session=False) db.query(models.Dictionary).filter(models.Dictionary.company_id == ceg_id).delete(synchronize_session=False) db.query(models.MachineTypeConfig).filter(models.MachineTypeConfig.company_id == ceg_id).delete(synchronize_session=False) db.query(models.DailyCheckQuestion).filter(models.DailyCheckQuestion.company_id == ceg_id).delete(synchronize_session=False) # C szelet: a cég levél-feliratkozás szabályai és a felhasználói választások (az aktiváló tokenekkel együtt) email_feliratkozas.torles_ceghez(db, ceg_id) email_feliratkozas.torles_felhasznalohoz(db, user_id_lista) if user_id_lista: db.query(models.AktivaloToken).filter(models.AktivaloToken.user_id.in_(user_id_lista)).delete(synchronize_session=False) db.query(models.MfaMentokod).filter(models.MfaMentokod.user_id.in_(user_id_lista)).delete(synchronize_session=False) # v34: 2FA mentőkódok db.query(models.JelszoElozmeny).filter(models.JelszoElozmeny.user_id.in_(user_id_lista)).delete(synchronize_session=False) # v36: jelszó-előzmény db.query(models.User).filter(models.User.company_id == ceg_id).delete(synchronize_session=False) db.query(models.Company).filter(models.Company.id == ceg_id).delete(synchronize_session=False) @router.get("/rendszeradmin/cegek/{ceg_id}/torlesi-osszegzes") def ceg_torlesi_osszegzes( ceg_id: int, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): """Előzetes darabszám-összegzés a végleges törlés megerősítő ablakához.""" if not _rendszeradmin_kapu(current_user): return JSONResponse(status_code=403, content={"message": "Csak a rendszeradminisztrátor!"}) ceg = db.query(models.Company).filter(models.Company.id == ceg_id).first() if not ceg or ceg_id == 1: return JSONResponse(status_code=404, content={"message": "Cég nem található!"}) return JSONResponse(status_code=200, content={"nev": ceg.nev, **_ceg_torlesi_szamlalo(ceg_id, db)}) @router.post("/rendszeradmin/cegek/{ceg_id}/torles") def torli_teljes_ceget_vegpont( ceg_id: int, megerositett_nev: str = Form(...), megertettem: str = Form(""), db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): """VÉGLEGES, adatbázis-szintű törlés: kizárólag a rendszeradminisztrátor végezheti, kétszeres megerősítéssel: a cég nevének pontos begépelésével ÉS egy explicit tudomásulvételi jelölőnégyzettel.""" if not _rendszeradmin_kapu(current_user): return JSONResponse(status_code=403, content={"message": "Csak a rendszeradminisztrátor végezheti ezt a műveletet!"}) ceg = db.query(models.Company).filter(models.Company.id == ceg_id).first() if not ceg or ceg_id == 1: return JSONResponse(status_code=403, content={"message": "Ez a cég nem törölhető!"}) if megertettem != "on": return JSONResponse(status_code=400, content={"message": "A tudomásulvételi jelölőnégyzet bejelölése kötelező!"}) if megerositett_nev.strip() != ceg.nev: return JSONResponse(status_code=400, content={"message": "A begépelt cégnév nem egyezik pontosan a cég nevével - a törlés megszakítva, semmi nem változott!"}) ceg_nev = ceg.nev try: _torli_teljes_ceget(ceg_id, db) # J-20 (v77): a törölt cég saját naplósorai is törlődtek, ezért a bejegyzés a rendszeradmin (1-es) céghez kerül; a név és az id a szövegben naploz(db, current_user, 1, "Cég törölve", f"{ceg_nev} (id={ceg_id})", "véglegesen törölve") db.commit() except Exception as e: db.rollback() print(f"❌ Hiba a cég törlése közben: {type(e).__name__}: {e}") return JSONResponse(status_code=500, content={"message": hiba_uzenet("Hiba történt a törlés közben, semmi nem került mentésre!", "E-ADMIN-01")}) print(f"[VÉGLEGES CÉGTÖRLÉS] '{ceg_nev}' (id={ceg_id}) céget és minden adatát véglegesen törölte: {current_user.username} ({current_user.email})") shutil.rmtree(f"uploads/{ceg_id}", ignore_errors=True) return JSONResponse(status_code=200, content={"message": f"'{ceg_nev}' cég és minden hozzá tartozó adat véglegesen törölve!"}) @router.post("/rendszeradmin/cegek/{ceg_id}/ai-szkenneles") def kapcsol_ai_szkenneles( ceg_id: int, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): """AI adattábla-felismerés engedélyezése/tiltása cégenként. Fizetős extra - kizárólag rendszeradmin kapcsolhatja.""" if not _rendszeradmin_kapu(current_user): return JSONResponse(status_code=403, content={"message": "Csak a rendszeradminisztrátor!"}) ceg = db.query(models.Company).filter(models.Company.id == ceg_id).first() if not ceg: return JSONResponse(status_code=404, content={"message": "Cég nem található!"}) beallitas = db.query(models.Dictionary).filter( models.Dictionary.company_id == ceg_id, models.Dictionary.kategoria == "AiSzkennelesEngedelyezve" ).first() regi_allapot = bool(beallitas and beallitas.ertek == "True") if beallitas and beallitas.ertek == "True": beallitas.ertek = "False" uj_allapot = False elif beallitas: beallitas.ertek = "True" uj_allapot = True else: db.add(models.Dictionary(company_id=ceg_id, kategoria="AiSzkennelesEngedelyezve", ertek="True")) uj_allapot = True naploz(db, current_user, ceg_id, "Cég AI-szkennelés (fizetős extra)", be_ki(regi_allapot), be_ki(uj_allapot)) db.commit() allapot_szoveg = "engedélyezve" if uj_allapot else "letiltva" return JSONResponse(status_code=200, content={"message": f"AI szkennelés sikeresen {allapot_szoveg}!", "aktiv": uj_allapot}) @router.post("/rendszeradmin/cegek/{ceg_id}/azonositas-mod") def kapcsol_azonositas_mod( ceg_id: int, mod: str = Form(...), db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): """Cégenkénti géppazonosítási mód (matricák): QR, NFC vagy mindkettő - kizárólag rendszeradmin állíthatja.""" if not _rendszeradmin_kapu(current_user): return JSONResponse(status_code=403, content={"message": "Csak a rendszeradminisztrátor!"}) if mod not in ("QR", "NFC", "Mindketto"): return JSONResponse(status_code=400, content={"message": "Érvénytelen azonosítási mód!"}) ceg = db.query(models.Company).filter(models.Company.id == ceg_id).first() if not ceg: return JSONResponse(status_code=404, content={"message": "Cég nem található!"}) beallitas = db.query(models.Dictionary).filter( models.Dictionary.company_id == ceg_id, models.Dictionary.kategoria == "AzonositasMod" ).first() naploz(db, current_user, ceg_id, "Cég azonosítási mód (QR / NFC)", beallitas.ertek if beallitas else None, mod, csak_ha_valtozott=True) if beallitas: beallitas.ertek = mod else: db.add(models.Dictionary(company_id=ceg_id, kategoria="AzonositasMod", ertek=mod)) db.commit() return JSONResponse(status_code=200, content={"message": f"Azonosítási mód beállítva: {mod}", "mod": mod}) @router.post("/rendszeradmin/cegek/{ceg_id}/oe-alapterulet") def kapcsol_oe_alapterulet( ceg_id: int, ertek: str = Form(...), db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): """Cégenként kapcsolható: a tűzoltó készülékek oltóanyagegység-igényét a rendszer az alapterület alapján (OTSZ 16. melléklet 2. táblázata) is felajánlja - kizárólag rendszeradmin állíthatja.""" if not _rendszeradmin_kapu(current_user): return JSONResponse(status_code=403, content={"message": "Csak a rendszeradminisztrátor!"}) ceg = db.query(models.Company).filter(models.Company.id == ceg_id).first() if not ceg: return JSONResponse(status_code=404, content={"message": "Cég nem található!"}) uj_ertek = "True" if ertek == "on" else "False" beallitas = db.query(models.Dictionary).filter( models.Dictionary.company_id == ceg_id, models.Dictionary.kategoria == "OEAlapteruletSzamitas" ).first() naploz(db, current_user, ceg_id, "Cég OE-alapterület számítás", be_ki(beallitas.ertek == "True") if beallitas else None, be_ki(uj_ertek == "True"), csak_ha_valtozott=True) if beallitas: beallitas.ertek = uj_ertek else: db.add(models.Dictionary(company_id=ceg_id, kategoria="OEAlapteruletSzamitas", ertek=uj_ertek)) db.commit() return JSONResponse(status_code=200, content={"message": "Beállítás mentve.", "ertek": uj_ertek}) @router.post("/rendszeradmin/cegek/{ceg_id}/sugo") def kapcsol_sugo( ceg_id: int, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): """Használati útmutató (súgó buborékok + útmutató oldal) engedélyezése/tiltása cégenként - kizárólag rendszeradmin kapcsolhatja.""" if not _rendszeradmin_kapu(current_user): return JSONResponse(status_code=403, content={"message": "Csak a rendszeradminisztrátor!"}) ceg = db.query(models.Company).filter(models.Company.id == ceg_id).first() if not ceg: return JSONResponse(status_code=404, content={"message": "Cég nem található!"}) beallitas = db.query(models.Dictionary).filter( models.Dictionary.company_id == ceg_id, models.Dictionary.kategoria == "SugoEngedelyezve" ).first() regi_allapot = bool(beallitas and beallitas.ertek == "True") if beallitas and beallitas.ertek == "True": beallitas.ertek = "False" uj_allapot = False elif beallitas: beallitas.ertek = "True" uj_allapot = True else: db.add(models.Dictionary(company_id=ceg_id, kategoria="SugoEngedelyezve", ertek="True")) uj_allapot = True naploz(db, current_user, ceg_id, "Cég használati útmutató (súgó)", be_ki(regi_allapot), be_ki(uj_allapot)) db.commit() allapot_szoveg = "engedélyezve" if uj_allapot else "letiltva" return JSONResponse(status_code=200, content={"message": f"Használati útmutató sikeresen {allapot_szoveg}!", "aktiv": uj_allapot}) @router.post("/rendszeradmin/cegek/{ceg_id}/logo") async def feltolt_ceg_logo( ceg_id: int, logo_file: UploadFile = File(...), db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): if not _rendszeradmin_kapu(current_user): return JSONResponse(status_code=403, content={"message": "Csak a rendszeradminisztrátor!"}) ceg = db.query(models.Company).filter(models.Company.id == ceg_id).first() if not ceg: return JSONResponse(status_code=404, content={"message": "Cég nem található!"}) import io from PIL import Image from image_utils import kep_e if not is_safe_extension(logo_file.filename) or not kep_e(logo_file.filename): return JSONResponse(status_code=400, content={"message": "Nem engedélyezett fájlformátum! Csak kép tölthető fel."}) # Nem a tomorit_kepet()-en megy át (az mindig JPEG-re konvertálna, ami a PNG logók # átlátszóságát elveszítené) - itt csak azt ellenőrizzük, hogy a tartalom valóban # megnyitható-e képként, az eredeti fájlformátumot megőrizve mentjük. logo_bytes = await logo_file.read() try: Image.open(io.BytesIO(logo_bytes)).verify() except Exception: return JSONResponse(status_code=400, content={"message": "A feltöltött fájl nem érvényes kép!"}) mappa = f"uploads/{ceg_id}/arculat" os.makedirs(mappa, exist_ok=True) safe_filename = sanitize_filename(logo_file.filename) file_loc = f"{mappa}/logo_{safe_filename}" with open(file_loc, "wb+") as f: f.write(logo_bytes) db_utvonal = "/" + file_loc beallitas = db.query(models.Dictionary).filter( models.Dictionary.company_id == ceg_id, models.Dictionary.kategoria == "JegyzokonyvLogoUtvonal" ).first() naploz(db, current_user, ceg_id, "Cég jegyzőkönyv-logó", beallitas.ertek if beallitas else None, db_utvonal) if beallitas: beallitas.ertek = db_utvonal else: db.add(models.Dictionary(company_id=ceg_id, kategoria="JegyzokonyvLogoUtvonal", ertek=db_utvonal)) db.commit() return JSONResponse(status_code=200, content={"message": "Logó sikeresen feltöltve!", "utvonal": db_utvonal}) @router.get("/uploads/{ceg_id}/arculat/{filename}") def arculati_logo_kiszolgalas( ceg_id: int, filename: str, current_user: models.User = Depends(get_current_user) ): """A feltöltés maga eddig is működött (fájl a lemezre íródott, DB-be az útvonal bekerült), de erre az útvonal-mintára nem volt kiszolgáló route - a rendszeradmin felület logó-előnézete (és bármi más, ami közvetlenül ezt az URL-t hívná) mindig 404-et kapott, ami épp úgy nézett ki, mintha a mentés sose történt volna meg (2026-09-30-i user-jelzés).""" if not _rendszeradmin_kapu(current_user): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Nincs jogosultságod a kért fájl eléréséhez!") safe_filename = sanitize_filename(filename) file_path = f"uploads/{ceg_id}/arculat/{safe_filename}" if not os.path.exists(file_path): raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="A keresett fájl nem található a szerveren.") return FileResponse(file_path) @router.post("/rendszeradmin/cegek/{ceg_id}/arculat") def ment_jegyzokonyv_arculat( ceg_id: int, fejlec_szoveg: str = Form(""), lablec_marka: str = Form("Defensys"), db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): if not _rendszeradmin_kapu(current_user): return JSONResponse(status_code=403, content={"message": "Csak a rendszeradminisztrátor!"}) ceg = db.query(models.Company).filter(models.Company.id == ceg_id).first() if not ceg: return JSONResponse(status_code=404, content={"message": "Cég nem található!"}) if lablec_marka not in ["Defensys", "EmeloReg"]: lablec_marka = "Defensys" regi_arculat = [] uj_arculat = [] for kategoria, ertek in [("JegyzokonyvFejlecSzoveg", fejlec_szoveg.strip()), ("JegyzokonyvLablecMarka", lablec_marka)]: beallitas = db.query(models.Dictionary).filter( models.Dictionary.company_id == ceg_id, models.Dictionary.kategoria == kategoria ).first() regi_arculat.append(f"{kategoria.replace('Jegyzokonyv', '')}: {beallitas.ertek if beallitas else '-'}") uj_arculat.append(f"{kategoria.replace('Jegyzokonyv', '')}: {ertek or '-'}") if beallitas: beallitas.ertek = ertek else: db.add(models.Dictionary(company_id=ceg_id, kategoria=kategoria, ertek=ertek)) naploz(db, current_user, ceg_id, "Cég jegyzőkönyv-arculat", "; ".join(regi_arculat), "; ".join(uj_arculat), csak_ha_valtozott=True) db.commit() return RedirectResponse(url="/rendszeradmin/cegek", status_code=303) # --- RENDSZER-TISZTÍTÁS: csak-olvasó audit a fejlesztés/tesztelés során felgyülemlett, zavaró # adatokról (modul nélküli felhasználók, üres cégek, stb.), a törléshez a cégtörléssel azonos, # kétszeres megerősítésű mintát követve. --- _GYANUS_CEGNEV_KULCSSZAVAK = ["teszt", "test", "demo", "minta", "sample", "próba", "proba"] # Minden tábla, amiben van company_id (vagy azzal egyenértékű) mező - az árva-adat kereső # ezeket nézi végig, és minden olyan sort kilistáz, aminek a company_id-ja NEM egyezik # egyetlen jelenleg létező cég azonosítójával sem (a NULL company_id jogos, "globális/közös" # adatot jelölhet egyes táblákban, ezért azt a kereső nem tekinti árvának). _ARVA_ADAT_TABLAK = [ (models.Site, "Telephely (Emelőgép)", "nev", "company_id"), (models.User, "Felhasználó", "username", "company_id"), (models.Machine, "Gép (Emelőgép)", "megnevezes", "company_id"), (models.Dictionary, "Beállítás (szótár)", "kategoria", "company_id"), (models.MachineTypeConfig, "Géptípus-konfiguráció", "geptipus", "company_id"), (models.DailyCheckQuestion, "Napi ellenőrző kérdés", "szoveg", "company_id"), (models.KarbantartoCeg, "Karbantartó cég (üzemeltető)", "nev", "company_id"), (models.TuzvedelmiTelephely, "Telephely (Tűzvédelem)", "nev", "company_id"), (models.KeszenletiHely, "Készenléti hely", "hely_leiras", "company_id"), (models.TuzoltoKeszulek, "Tűzoltó készülék", "gyari_szam", "company_id"), (models.EbkTerulet, "EBK terület", "nev", "company_id"), (models.EbkAlvallalkozo, "EBK alvállalkozó", "nev", "company_id"), (models.EbkTevekenyseg, "EBK tevékenység", "nev", "company_id"), (models.EbkEllenorzesiTetel, "EBK ellenőrzési tétel", "szoveg", "company_id"), (models.EbkUgyfel, "EBK ügyfél", "nev", "company_id"), (models.EbkProjekt, "EBK projekt", "megrendelo_nev", "company_id"), ] def _arva_company_id_rekordok(db: Session) -> list: """Végignézi az _ARVA_ADAT_TABLAK összes tábláját, és minden olyan sort visszaad, aminek a company_id mezője nem NULL, de nem is egyezik egyetlen jelenleg létező cég azonosítójával - ezek a fejlesztés közbeni átszervezések/cégtörlések után "lebegve" maradt, egy admin/ szuperfelhasználó számára soha meg nem jeleníthető rekordok (a hozzájuk rendelt felhasználó saját, cégtől független kapcsolatain - pl. EBK koordinátor-projekt - keresztül még elérhetők).""" ervenyes_ceg_idk = {row[0] for row in db.query(models.Company.id).all()} talalatok = [] for model, cim, nev_mezo, company_mezo in _ARVA_ADAT_TABLAK: company_oszlop = getattr(model, company_mezo) nev_oszlop = getattr(model, nev_mezo) sorok = db.query(model.id, company_oszlop, nev_oszlop).filter( company_oszlop.isnot(None) ).all() for sor_id, cid, nev in sorok: if cid not in ervenyes_ceg_idk: talalatok.append({ "tabla": cim, "id": sor_id, "nev": nev or "-", "arva_company_id": cid, }) return talalatok def _felhasznalo_torlesi_szamlalo(user_id: int, db: Session) -> dict: """Előzetes darabszám-összegzés a felhasználó végleges törlése előtti megerősítéshez: mennyi (más rekordokon lévő) hivatkozás érintett - ezek a törléskor NEM törlődnek, csak a rájuk mutató felhasználó-hivatkozás NULL-ra áll.""" return { "vizsgalatok": db.query(models.Inspection).filter(models.Inspection.user_id == user_id).count(), "naplo_bejegyzesek": db.query(models.AuditLog).filter(models.AuditLog.user_id == user_id).count(), "szakvizsgak": db.query(models.Szakvizsga).filter(models.Szakvizsga.user_id == user_id).count(), "ebk_projektek": db.query(models.EbkProjekt).filter(models.EbkProjekt.letrehozo_user_id == user_id).count(), "ebk_ellenorzesek": db.query(models.EbkEllenorzes).filter(models.EbkEllenorzes.ellenorzo_user_id == user_id).count(), } def _torli_felhasznalot(user_id: int, db: Session): """VÉGLEGES felhasználó-törlés: a rá hivatkozó rekordokat (vizsgálatok, naplóbejegyzések, szakvizsgák, EBK-projektek/ellenőrzések stb.) megőrzi, csak a felhasználóra mutató hivatkozást állítja NULL-ra rajtuk, majd magát a felhasználót (és a many-to-many kapcsolótábla-sorait - modulok, engedélyezett cégek/telephelyek/gépek) törli.""" hivatkozo_oszlopok = [ models.BerletiIdoszak.letrehozo_user_id, models.Inspection.user_id, models.AuditLog.user_id, models.KarbantartoSzerzodes.letrehozo_user_id, models.Szakvizsga.user_id, models.TuzoltoKeszulek.letrehozo_user_id, models.KeszulekKolcsonzes.letrehozo_user_id, models.TuzvedelmiEllenorzes.rogzito_user_id, models.TuzvedelmiNaplo.lezaro_user_id, models.EbkProjekt.letrehozo_user_id, models.EbkEllenorzes.ellenorzo_user_id, ] for oszlop in hivatkozo_oszlopok: db.query(oszlop.class_).filter(oszlop == user_id).update({oszlop: None}, synchronize_session=False) # C szelet: a felhasználó levél-feliratkozásai és aktiváló tokenjei is törlődnek email_feliratkozas.torles_felhasznalohoz(db, [user_id]) db.query(models.AktivaloToken).filter(models.AktivaloToken.user_id == user_id).delete(synchronize_session=False) db.query(models.MfaMentokod).filter(models.MfaMentokod.user_id == user_id).delete(synchronize_session=False) # v34: 2FA mentőkódok user_obj = db.query(models.User).filter(models.User.id == user_id).first() if user_obj: db.delete(user_obj) @router.get("/rendszeradmin/tisztitas", response_class=HTMLResponse) def rendszeradmin_tisztitas( request: Request, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): if not _rendszeradmin_kapu(current_user): return RedirectResponse(url="/valassz-modult", status_code=303) osszes_felhasznalo = db.query(models.User).order_by(models.User.company_id, models.User.username).all() modul_nelkuli_felhasznalok = [ u for u in osszes_felhasznalo if u.aktiv and not is_rendszeradminisztrator(u) and len(u.sajat_modulok) == 0 ] inaktiv_modullal_felhasznalok = [ u for u in osszes_felhasznalo if not u.aktiv and len(u.sajat_modulok) > 0 ] cegek = db.query(models.Company).filter(models.Company.id != 1).order_by(models.Company.nev).all() aktiv_felhasznalo_szam = {} for u in osszes_felhasznalo: if u.aktiv: aktiv_felhasznalo_szam[u.company_id] = aktiv_felhasznalo_szam.get(u.company_id, 0) + 1 ures_cegek = [c for c in cegek if aktiv_felhasznalo_szam.get(c.id, 0) == 0] gyanus_nevu_cegek = [ c for c in cegek if any(kulcsszo in c.nev.lower() for kulcsszo in _GYANUS_CEGNEV_KULCSSZAVAK) ] arva_rekordok = _arva_company_id_rekordok(db) return templates.TemplateResponse( request=request, name="rendszeradmin_tisztitas.html", context={ "user": current_user, "modul_nelkuli_felhasznalok": modul_nelkuli_felhasznalok, "inaktiv_modullal_felhasznalok": inaktiv_modullal_felhasznalok, "ures_cegek": ures_cegek, "gyanus_nevu_cegek": gyanus_nevu_cegek, "arva_rekordok": arva_rekordok, } ) @router.get("/rendszeradmin/felhasznalo/{user_id}/torlesi-osszegzes") def felhasznalo_torlesi_osszegzes( user_id: int, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): """Előzetes darabszám-összegzés a felhasználó végleges törlésének megerősítő ablakához.""" if not _rendszeradmin_kapu(current_user): return JSONResponse(status_code=403, content={"message": "Csak a rendszeradminisztrátor!"}) cel_user = db.query(models.User).filter(models.User.id == user_id).first() if not cel_user or is_rendszeradminisztrator(cel_user): return JSONResponse(status_code=404, content={"message": "Felhasználó nem található!"}) return JSONResponse(status_code=200, content={"email": cel_user.email, **_felhasznalo_torlesi_szamlalo(user_id, db)}) @router.post("/rendszeradmin/felhasznalo/{user_id}/torles") def torli_felhasznalot_vegpont( user_id: int, megerositett_email: str = Form(...), megertettem: str = Form(""), db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): """VÉGLEGES, adatbázis-szintű felhasználó-törlés: kizárólag a rendszeradminisztrátor végezheti, kétszeres megerősítéssel: az e-mail cím pontos begépelésével ÉS egy explicit tudomásulvételi jelölőnégyzettel - a cégtörlés mintáját követve.""" if not _rendszeradmin_kapu(current_user): return JSONResponse(status_code=403, content={"message": "Csak a rendszeradminisztrátor végezheti ezt a műveletet!"}) cel_user = db.query(models.User).filter(models.User.id == user_id).first() if not cel_user or is_rendszeradminisztrator(cel_user): return JSONResponse(status_code=403, content={"message": "Ez a felhasználó nem törölhető!"}) if megertettem != "on": return JSONResponse(status_code=400, content={"message": "A tudomásulvételi jelölőnégyzet bejelölése kötelező!"}) if megerositett_email.strip() != cel_user.email: return JSONResponse(status_code=400, content={"message": "A begépelt e-mail cím nem egyezik pontosan - a törlés megszakítva, semmi nem változott!"}) user_email = cel_user.email user_ceg_id = cel_user.company_id user_leiras = f"{cel_user.username} ({user_email}, {cel_user.role})" try: _torli_felhasznalot(user_id, db) # J-20 (v77): a felhasználó cége megmarad, ezért a bejegyzés a cég Jogosultsági naplójában is látszik naploz(db, current_user, user_ceg_id, "Felhasználó törölve (rendszeradmin)", user_leiras, "véglegesen törölve") db.commit() except Exception as e: db.rollback() print(f"❌ Hiba a felhasználó törlése közben: {type(e).__name__}: {e}") return JSONResponse(status_code=500, content={"message": hiba_uzenet("Hiba történt a törlés közben, semmi nem került mentésre!", "E-ADMIN-02")}) print(f"[VÉGLEGES FELHASZNÁLÓ-TÖRLÉS] '{user_email}' fiókot véglegesen törölte: {current_user.username} ({current_user.email})") return JSONResponse(status_code=200, content={"message": f"'{user_email}' felhasználó véglegesen törölve!"}) @router.post("/rendszeradmin/felhasznalo/{user_id}/modulok-torlese") def torli_felhasznalo_modul_jogosultsagait( user_id: int, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): """Könnyű, nem-végleges takarítás: egy inaktív felhasználó minden modul-jogosultságát törli (a fiókot magát nem), hogy ne maradjon feleslegesen bekapcsolt hozzáférés.""" if not _rendszeradmin_kapu(current_user): return JSONResponse(status_code=403, content={"message": "Csak a rendszeradminisztrátor!"}) cel_user = db.query(models.User).filter(models.User.id == user_id).first() if not cel_user: return JSONResponse(status_code=404, content={"message": "Felhasználó nem található!"}) regi_modulok = ", ".join(sorted(m.kod for m in cel_user.sajat_modulok)) or "-" cel_user.sajat_modulok = [] naploz(db, current_user, cel_user.company_id, "Modulok igazítása (rendszeradmin, modul-jogok törlése)", f"{cel_user.username}: {regi_modulok}", f"{cel_user.username}: -", csak_ha_valtozott=True) db.commit() return JSONResponse(status_code=200, content={"message": f"'{cel_user.email}' modul-jogosultságai törölve!"}) # --- EGYEDI GÉPTÖRLÉS: kizárólag rendszeradminisztrátori jogosultsággal, ha egyébként nincs más # módja egy hibásan/tévesen felvitt gép és minden hozzá tartozó adat végleges eltávolításának # (a "Végleg kivonva/Selejtezve" státusz csak megjelöli, nem törli a gépet). A cégtörléssel # azonos, kétszeres megerősítésű mintát követi. --- def _gep_torlesi_azonosito(gep) -> str: """A megerősítéshez begépelendő szöveg: elsőként a gyári szám (ha van), különben egy determinisztikus GÉP-{id} azonosító, hogy mindig legyen egyértelmű, begépelhető érték.""" return (gep.gyari_szam or "").strip() or f"GÉP-{gep.id}" def _gep_torlesi_szamlalo(gep_id: int, db: Session) -> dict: """Előzetes összegzés a gép végleges törlése előtti megerősítéshez.""" return { "fajlok": db.query(models.File).filter(models.File.machine_id == gep_id).count(), "vizsgalatok": db.query(models.Inspection).filter(models.Inspection.machine_id == gep_id).count(), "hibajegyek": db.query(models.MaintenanceTicket).filter(models.MaintenanceTicket.machine_id == gep_id).count(), "berleti_idoszakok": db.query(models.BerletiIdoszak).filter(models.BerletiIdoszak.machine_id == gep_id).count(), "tartozekok": db.query(models.Machine).filter(models.Machine.fo_gep_id == gep_id).count(), "naplo_bejegyzesek": db.query(models.AuditLog).filter(models.AuditLog.machine_id == gep_id).count(), "tervezett_vizsgalatok": db.query(models.TervezettVizsgalat).filter(models.TervezettVizsgalat.machine_id == gep_id).count(), "msz16226_dokumentumok": db.query(models.Msz16226Dokumentum).filter(models.Msz16226Dokumentum.machine_id == gep_id).count(), } def _torli_egy_gepet(gep_id: int, db: Session): """VÉGLEGES, adatbázis-szintű törlés: egy gép és MINDEN hozzá tartozó adat (fájlok, vizsgálatok, hibajegyek, bérleti időszakok, naplóbejegyzések, tervezett vizsgálatok, MSZ 16226 üzemviteli dokumentum) visszavonhatatlan törlése. A hozzá tartozékként kapcsolt gépeket nem törli, csak leválasztja róla (fo_gep_id = NULL). JAVÍTVA (2026-10-03-i külső biztonsági audit, 2. kör): a TervezettVizsgalat/ Msz16226Dokumentum/MszTelephelyMegjegyzes táblák eddig hiányoztak innen - egy gép törlésekor árván maradtak volna (vagy a DB FK-kényszere miatt a törlés elakadt volna).""" db.query(models.Machine).filter(models.Machine.fo_gep_id == gep_id).update( {models.Machine.fo_gep_id: None}, synchronize_session=False ) db.query(models.MaintenanceTicket).filter(models.MaintenanceTicket.machine_id == gep_id).delete(synchronize_session=False) db.query(models.File).filter(models.File.machine_id == gep_id).delete(synchronize_session=False) db.query(models.Inspection).filter(models.Inspection.machine_id == gep_id).delete(synchronize_session=False) db.query(models.BerletiIdoszak).filter(models.BerletiIdoszak.machine_id == gep_id).delete(synchronize_session=False) db.query(models.TervezettVizsgalat).filter(models.TervezettVizsgalat.machine_id == gep_id).delete(synchronize_session=False) db.query(models.Msz16226Dokumentum).filter(models.Msz16226Dokumentum.machine_id == gep_id).delete(synchronize_session=False) # A MszTelephelyMegjegyzes.audit_log_id ugyanennek a gépnek egy AuditLog sorára hivatkozhat # - ezért ezt MINDIG az AuditLog-törlés ELŐTT kell törölni. db.query(models.MszTelephelyMegjegyzes).filter(models.MszTelephelyMegjegyzes.machine_id == gep_id).delete(synchronize_session=False) db.query(models.AuditLog).filter(models.AuditLog.machine_id == gep_id).delete(synchronize_session=False) db.execute(models.user_machines.delete().where(models.user_machines.c.machine_id == gep_id)) db.query(models.Machine).filter(models.Machine.id == gep_id).delete(synchronize_session=False) @router.get("/rendszeradmin/gep/{gep_id}/torlesi-osszegzes") def gep_torlesi_osszegzes( gep_id: int, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): if not _rendszeradmin_kapu(current_user): return JSONResponse(status_code=403, content={"message": "Csak a rendszeradminisztrátor!"}) gep = db.query(models.Machine).filter(models.Machine.id == gep_id).first() if not gep: return JSONResponse(status_code=404, content={"message": "Gép nem található!"}) return JSONResponse(status_code=200, content={ "azonosito": _gep_torlesi_azonosito(gep), "nev": gep.megnevezes or "Ismeretlen gép", **_gep_torlesi_szamlalo(gep_id, db) }) @router.post("/rendszeradmin/gep/{gep_id}/torles") def torli_gepet_vegpont( gep_id: int, megerositett_azonosito: str = Form(...), megertettem: str = Form(""), db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): """VÉGLEGES, adatbázis-szintű gép-törlés: kizárólag a rendszeradminisztrátor végezheti, kétszeres megerősítéssel: a gyári szám (vagy GÉP-{id}) pontos begépelésével ÉS egy explicit tudomásulvételi jelölőnégyzettel.""" if not _rendszeradmin_kapu(current_user): return JSONResponse(status_code=403, content={"message": "Csak a rendszeradminisztrátor végezheti ezt a műveletet!"}) gep = db.query(models.Machine).filter(models.Machine.id == gep_id).first() if not gep: return JSONResponse(status_code=404, content={"message": "Gép nem található!"}) if megertettem != "on": return JSONResponse(status_code=400, content={"message": "A tudomásulvételi jelölőnégyzet bejelölése kötelező!"}) if megerositett_azonosito.strip() != _gep_torlesi_azonosito(gep): return JSONResponse(status_code=400, content={"message": "A begépelt azonosító nem egyezik pontosan - a törlés megszakítva, semmi nem változott!"}) gep_nev = gep.megnevezes or "Ismeretlen gép" company_id = gep.company_id gep_azonosito = _gep_torlesi_azonosito(gep) kozos_fajlok = kozos_utvonalak(db, [gep_id]) # v80: a gép megosztott dokumentumai - a törlés után az árvák fizikai fájlja is törlődik try: _torli_egy_gepet(gep_id, db) # J-20 (v77): a gép saját naplósorai törlődtek, ezért a bejegyzés a gép CÉGÉHEZ kerül (machine_id nélkül) naploz(db, current_user, company_id, "Gép törölve (rendszeradmin)", f"{gep_nev} ({gep_azonosito}, id={gep_id})", "véglegesen törölve") db.commit() except Exception as e: db.rollback() print(f"❌ Hiba a gép törlése közben: {type(e).__name__}: {e}") return JSONResponse(status_code=500, content={"message": hiba_uzenet("Hiba történt a törlés közben, semmi nem került mentésre!", "E-ADMIN-03")}) print(f"[VÉGLEGES GÉPTÖRLÉS] '{gep_nev}' (id={gep_id}, company_id={company_id}) gépet és minden adatát véglegesen törölte: {current_user.username} ({current_user.email})") shutil.rmtree(f"uploads/{company_id}/{gep_id}", ignore_errors=True) arva_kozos_torlese(db, kozos_fajlok) # v80: csak azok a közös fájlok törlődnek, amelyekre nem mutat más gép hivatkozása return JSONResponse(status_code=200, content={"message": f"'{gep_nev}' gép és minden hozzá tartozó adat véglegesen törölve!"}) # --- TELJES GÉPADAT-EXPORT (rendszeradminisztrátori szintű, cégek közötti) --- @router.get("/rendszeradmin/gepek-export", response_class=HTMLResponse) def gepek_export_oldal( request: Request, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): if not _rendszeradmin_kapu(current_user): return RedirectResponse(url="/valassz-modult", status_code=303) cegek = db.query(models.Company).filter(models.Company.id != 1).order_by(models.Company.nev).all() return templates.TemplateResponse( request=request, name="rendszeradmin_gepek_export.html", context={"user": current_user, "cegek": cegek} ) @router.get("/rendszeradmin/gepek-export/futtatas") def gepek_export_futtatas( company_id: int = 0, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): if not _rendszeradmin_kapu(current_user): return JSONResponse(status_code=403, content={"message": "Csak a rendszeradminisztrátor!"}) import io import urllib.parse from datetime import date import openpyxl from openpyxl.styles import Font from sqlalchemy.orm import selectinload from fastapi.responses import StreamingResponse from msz_logic import compute_vizsgalati_esedekesseg query = db.query(models.Machine).options( selectinload(models.Machine.ceg), selectinload(models.Machine.telephely), selectinload(models.Machine.fo_gep), selectinload(models.Machine.tartozekok), ) if company_id: query = query.filter(models.Machine.company_id == company_id) gepek = query.order_by(models.Machine.company_id, models.Machine.id).all() gep_idk = [g.id for g in gepek] fajlok_map = {} if gep_idk: for machine_id, file_type, file_path in db.query( models.File.machine_id, models.File.file_type, models.File.file_path ).filter(models.File.machine_id.in_(gep_idk)).all(): fajlnev = (file_path or "").rsplit("/", 1)[-1] fajlok_map.setdefault(machine_id, []).append((fajlnev, file_type)) berlet_map = {} if gep_idk: for berlet in db.query(models.BerletiIdoszak).filter( models.BerletiIdoszak.machine_id.in_(gep_idk), models.BerletiIdoszak.vege.is_(None) ).all(): berlet_map[berlet.machine_id] = berlet mai_nap = date.today() def format_esedekesseg(bejegyzes): if bejegyzes["allapot"] == "hianyzik": return "Nincs rögzített vizsgálat" if bejegyzes["allapot"] == "ismeretlen": return f"Utolsó: {bejegyzes['utolso']} (ciklus ismeretlen)" allapot_cimke = {"lejart": "LEJÁRT", "hamarosan": "hamarosan esedékes", "rendben": "rendben"} return f"Utolsó: {bejegyzes['utolso']} → Esedékes: {bejegyzes['esedekes']} ({allapot_cimke.get(bejegyzes['allapot'], bejegyzes['allapot'])})" def format_csatolt_fajlok(gep): fajlok = fajlok_map.get(gep.id, []) if not fajlok: return "Nincs csatolt fájl" felsorolas = "; ".join(f"{fajlnev} ({tipus})" for fajlnev, tipus in fajlok) return f"{len(fajlok)} db: {felsorolas}" def format_kapcsolodo_gepek(gep): reszek = [] if gep.fo_gep: reszek.append(f"Tartozéka ennek: {gep.fo_gep.megnevezes} (ID {gep.fo_gep.id})") if gep.tartozekok: nevek = ", ".join(f"{t.megnevezes} (ID {t.id})" for t in gep.tartozekok) reszek.append(f"Tartozékai: {nevek}") return " | ".join(reszek) if reszek else "-" def format_berlet(gep): reszek = [] if gep.berelt_eszkoz: resz = f"Bérelt eszköz ({gep.berbeado_neve or 'ismeretlen bérbeadó'})" if gep.berlet_lejarat_datum: resz += f", lejárat: {gep.berlet_lejarat_datum.strftime('%Y-%m-%d')}" reszek.append(resz) aktiv_berlet = berlet_map.get(gep.id) if aktiv_berlet: reszek.append(f"Jelenleg kihelyezve ide: {aktiv_berlet.berlo_ceg.nev if aktiv_berlet.berlo_ceg else (aktiv_berlet.berlo_nev_kulso or '-')} ({aktiv_berlet.kezdete})") return " | ".join(reszek) if reszek else "-" def format_datum(d): return d.strftime("%Y-%m-%d") if d else "-" col_map = [ ("Cég", lambda g: g.ceg.nev if g.ceg else "-"), ("ID", lambda g: g.id), ("Telephely", lambda g: g.telephely.nev if g.telephely else "-"), ("Készenléti hely", lambda g: g.megnevezes or "-"), ("Gyártó", lambda g: g.gyarto or "-"), ("Típus", lambda g: g.tipus or "-"), ("Gyári szám", lambda g: g.gyari_szam or "-"), ("Géptípus", lambda g: g.geptipus or "-"), ("Üzemóra", lambda g: g.uzemora), ("Leltári szám", lambda g: g.leltari_szam or "-"), ("Teherbírás", lambda g: f"{g.teherbiras} kg" if g.teherbiras else "-"), ("Gyártás dátuma", lambda g: format_datum(g.gyartas_datum)), ("Üzembe helyezés", lambda g: format_datum(g.uzembe_helyezes_datum)), ("Státusz", lambda g: g.statusz), ("Szerkezeti vizsgálat", lambda g: format_esedekesseg(compute_vizsgalati_esedekesseg(g, mai_nap)[0])), ("Fővizsgálat", lambda g: format_esedekesseg(compute_vizsgalati_esedekesseg(g, mai_nap)[1])), ("Időszakos biztonsági felülvizsgálat (EBSZ)", lambda g: format_esedekesseg(compute_vizsgalati_esedekesseg(g, mai_nap)[2])), ("Csatolt fájlok", format_csatolt_fajlok), ("Kapcsolódó gépek", format_kapcsolodo_gepek), ("Bérlet / kölcsönzés", format_berlet), ] wb = openpyxl.Workbook() ws = wb.active ws.title = "Gépadat export" ws.append([cim for cim, _ in col_map]) header_font = Font(bold=True) for cell in ws[1]: cell.font = header_font for gep in gepek: ws.append([fuggveny(gep) for _, fuggveny in col_map]) for col in ws.columns: max_length = max((len(str(cell.value)) for cell in col if cell.value is not None), default=0) ws.column_dimensions[col[0].column_letter].width = min(max_length + 2, 60) output = io.BytesIO() wb.save(output) output.seek(0) filename = urllib.parse.quote(f"EmeLOG_Teljes_Gepadat_Export_{mai_nap.strftime('%Y%m%d')}.xlsx") return StreamingResponse( output, media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={'Content-Disposition': f"attachment; filename*=UTF-8''{filename}"} ) # --- VIVÜ: REJTETT KARBANTARTÓ FELÜLET (kizárólag rendszeradminisztrátor) --- # Egy már mentett VIVÜ ellenőrzés (jegyzőkönyv) utólagos, teljes körű javítására - nincs link rá # a normál navigációban, kizárólag közvetlen URL-ről érhető el, és kizárólag a valódi # rendszeradminisztrátornak (a cégek saját Admin-jai nem férnek hozzá). A user kifejezett # kérésére NINCS külön módosítási napló hozzá (2026-09-13). @router.get("/rendszeradmin/vivu/ellenorzesek", response_class=HTMLResponse) def rendszeradmin_vivu_ellenorzesek( request: Request, ceg_id: int = 0, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): if not _rendszeradmin_kapu(current_user): return RedirectResponse(url="/valassz-modult", status_code=303) vivu_modul = db.query(models.Modul).filter(models.Modul.kod == "VIVU").first() cegek = [] if vivu_modul: cegek = [ c for c in db.query(models.Company).order_by(models.Company.nev).all() if vivu_modul in c.engedelyezett_modulok ] ellenorzesek = [] if ceg_id: ellenorzesek = db.query(models.VivuEllenorzes).join(models.VivuTelephely).join(models.VivuUgyfel).filter( models.VivuUgyfel.company_id == ceg_id ).order_by(models.VivuEllenorzes.datum.desc(), models.VivuEllenorzes.id.desc()).all() return templates.TemplateResponse( request=request, name="rendszeradmin_vivu_ellenorzesek.html", context={ "user": current_user, "cegek": cegek, "ceg_id": ceg_id, "ellenorzesek": ellenorzesek, } ) @router.get("/rendszeradmin/vivu/ellenorzes/{ellenorzes_id}/karbantartas", response_class=HTMLResponse) def rendszeradmin_vivu_karbantartas_oldal( ellenorzes_id: int, request: Request, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): if not _rendszeradmin_kapu(current_user): return RedirectResponse(url="/valassz-modult", status_code=303) from routers.vivu import SULYOSSAGOK, SZAKTERULET_MEZOK, _epuletek_json_ugyfelhez ellenorzes = db.query(models.VivuEllenorzes).join(models.VivuTelephely).join(models.VivuUgyfel).filter( models.VivuEllenorzes.id == ellenorzes_id ).first() if not ellenorzes: return RedirectResponse(url="/rendszeradmin/vivu/ellenorzesek", status_code=303) ugyfel = ellenorzes.telephely.ugyfel telephelyek = [t for t in ugyfel.telephelyek if t.aktiv or t.id == ellenorzes.telephely_id] tetel_eredmenyek = sorted(ellenorzes.tetel_eredmenyek, key=lambda t: t.id) return templates.TemplateResponse( request=request, name="rendszeradmin_vivu_karbantartas.html", context={ "user": current_user, "ellenorzes": ellenorzes, "ugyfel": ugyfel, "telephelyek": telephelyek, "epuletek_json": _epuletek_json_ugyfelhez(ugyfel, db), "tetel_eredmenyek": tetel_eredmenyek, "sulyossagok": SULYOSSAGOK, "szakterulet_mezok": SZAKTERULET_MEZOK, "erintett_epulet_idk": [e.id for e in ellenorzes.epuletek], } ) @router.post("/rendszeradmin/vivu/ellenorzes/{ellenorzes_id}/karbantartas") async def rendszeradmin_vivu_karbantartas_mentese( ellenorzes_id: int, request: Request, telephely_id: int = Form(...), erintett_epuletek_json: str = Form("[]"), celellenorzes: str = Form(""), datum: str = Form(""), mindenrendben: str = Form(""), tetel_eredmenyek_json: str = Form(...), torolt_fenykepek_json: str = Form("[]"), db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): if not _rendszeradmin_kapu(current_user): return JSONResponse(status_code=403, content={"message": "Csak a rendszeradminisztrátor jogosult erre!"}) import json from datetime import datetime from routers.vivu import SULYOSSAGOK, SZAKTERULET_MEZOK, _parse_date, vivu_karbantartas_hivatkozasok_ellenorzese ellenorzes = db.query(models.VivuEllenorzes).join(models.VivuTelephely).join(models.VivuUgyfel).filter( models.VivuEllenorzes.id == ellenorzes_id ).first() if not ellenorzes: return JSONResponse(status_code=404, content={"message": "Az ellenőrzés nem található!"}) ugyfel = ellenorzes.telephely.ugyfel if telephely_id not in [t.id for t in ugyfel.telephelyek]: return JSONResponse(status_code=400, content={"message": "A telephely nem ehhez a Megrendelőhöz tartozik!"}) try: tetel_sorok = json.loads(tetel_eredmenyek_json) erintett_epulet_idk = json.loads(erintett_epuletek_json) torolt_fenykep_idk = json.loads(torolt_fenykepek_json) if not isinstance(erintett_epulet_idk, list) or not isinstance(torolt_fenykep_idk, list): raise ValueError() except Exception: return JSONResponse(status_code=400, content={"message": "Hibás adatformátum!"}) company_id = ugyfel.company_id # J-13 (v71): a sorok / érintett épületek / törölt fényképek azonosítóinak ellenőrzése MENTÉS ELŐTT (hibánál 400, semmi sem változik) hivatkozas_hiba = vivu_karbantartas_hivatkozasok_ellenorzese( db, company_id, telephely_id, erintett_epulet_idk, torolt_fenykep_idk, tetel_sorok) if hivatkozas_hiba: return JSONResponse(status_code=400, content={"message": hivatkozas_hiba}) form_data = await request.form() try: ellenorzes.telephely_id = telephely_id ellenorzes.celellenorzes = celellenorzes or None ellenorzes.datum = _parse_date(datum) or ellenorzes.datum ellenorzes.mindenrendben = (mindenrendben == "true") ellenorzes.epuletek = db.query(models.VivuEpulet).join(models.VivuTelephely).filter( models.VivuEpulet.id.in_(erintett_epulet_idk), models.VivuTelephely.id == telephely_id, ).all() if erintett_epulet_idk else [] if torolt_fenykep_idk: torlendo_kepek = db.query(models.VivuTetelFenykep).join(models.VivuEllenorzesTetel).filter( models.VivuTetelFenykep.id.in_(torolt_fenykep_idk), models.VivuEllenorzesTetel.ellenorzes_id == ellenorzes.id, ).all() for kep in torlendo_kepek: fajl = kep.file_path.lstrip("/") if kep.file_path else None if fajl and os.path.exists(fajl): try: os.remove(fajl) except OSError: pass db.delete(kep) db.flush() meglevo_eredmenyek = {t.id: t for t in ellenorzes.tetel_eredmenyek} mappa = f"uploads/{company_id}/vivu/{ellenorzes.id}" for sor in tetel_sorok: eredmeny_id = sor.get("id") uj_tetel = eredmeny_id is None if uj_tetel: # "+ Újabb szabálytalanság rögzítése erre a tételre" a karbantartó oldalon - # a tetel_id-t a kliens a meglévő panel data-tetel-id attribútumából klónozza, # tehát mindig egy már a jegyzőkönyvben szereplő katalógus-tételre mutat. tetel_id = sor.get("tetel_id") if not tetel_id: continue eredmeny = models.VivuEllenorzesTetel( ellenorzes_id=ellenorzes.id, tetel_id=tetel_id, statusz="SZABALYTALANSAG", ) db.add(eredmeny) db.flush() photo_key = sor.get("temp_id") else: eredmeny = meglevo_eredmenyek.get(eredmeny_id) if not eredmeny: continue # ismeretlen/idegen tétel-id - kihagyjuk, nem hozunk létre újat if sor.get("torles"): db.delete(eredmeny) continue eredmeny.statusz = sor.get("statusz") photo_key = eredmeny_id statusz = eredmeny.statusz if statusz == "SZABALYTALANSAG": eredmeny.epulet_id = sor.get("epulet_id") or None eredmeny.szint_id = sor.get("szint_id") or None eredmeny.helyiseg = (sor.get("helyiseg") or "").strip() or None if not eredmeny.epulet_id: raise ValueError("Minden szabálytalanságnál meg kell adni az épületet!") eredmeny.sulyossag = sor.get("sulyossag") if sor.get("sulyossag") in SULYOSSAGOK else None eredmeny.leiras = sor.get("leiras") or None for mezo, _cimke in SZAKTERULET_MEZOK: setattr(eredmeny, mezo, bool(sor.get(mezo))) else: eredmeny.epulet_id = None eredmeny.szint_id = None eredmeny.helyiseg = None eredmeny.sulyossag = None eredmeny.leiras = None for mezo, _cimke in SZAKTERULET_MEZOK: setattr(eredmeny, mezo, False) fajl_index = 0 while True: fenykep = form_data.get(f"uj_fenykep_{photo_key}_{fajl_index}") if fenykep is None: break if getattr(fenykep, "filename", "") != "": if not is_safe_extension(fenykep.filename): raise ValueError("Nem engedélyezett fájlformátum a fényképnél!") from image_utils import tomorit_kepet safe_filename = sanitize_filename(fenykep.filename).rsplit(".", 1)[0] + ".jpg" kep_bytes = tomorit_kepet(await fenykep.read()) if kep_bytes is None: raise ValueError("A feltöltött fájl nem érvényes kép!") os.makedirs(mappa, exist_ok=True) file_loc = f"{mappa}/{datetime.now().strftime('%Y%m%d%H%M%S')}_karb_{photo_key}_{fajl_index}_{safe_filename}" with open(file_loc, "wb") as f: f.write(kep_bytes) db.add(models.VivuTetelFenykep( tetel_id=eredmeny.id, file_path="/" + file_loc, feltolto_user_id=current_user.id, )) fajl_index += 1 db.commit() except Exception as e: db.rollback() print(f"❌ Hiba a mentés közben (rendszeradmin karbantartó felület): {type(e).__name__}: {e}") return JSONResponse(status_code=400, content={"message": hiba_uzenet("Hiba történt a mentés közben!", "E-ADMIN-04")}) return JSONResponse(status_code=200, content={"message": "Jegyzőkönyv frissítve.", "ellenorzes_id": ellenorzes.id})