"""Levél-feliratkozások hierarchiával (2026-10-04, e-mail rendszer C szelete). MIÉRT KELL? Korábban a riasztásokat és a riportot MINDIG az adott cég összes Admin és Emelőgép ügyintéző (EÜI) szerepkörű felhasználója megkapta - a kódban kódolva, nem lehetett kikapcsolni, és más szerepkör nem is kérhette. Mostantól: 1. A CÉGES ADMIN szerepkörönként és kategóriánként megadja a szabályt (EmailFeliratkozasSzabaly): kotelezo - mindig megkapja, a felhasználó nem kapcsolhatja ki alapbol_be - alapból megkapja, a felhasználó kikapcsolhatja alapbol_ki - alapból nem kapja, a felhasználó bekapcsolhatja tiltott - nem kaphatja (és nem is kérheti) 2. A FELHASZNÁLÓ a saját "Értesítéseim" oldalán azt állítja át, amit a szabály megenged (EmailFeliratkozas) - kötelező vagy tiltott kategóriát nem módosíthat. 3. A levél küldésekor a címzettek listáját EZ A MODUL állítja elő (cimzettek()). KATEGÓRIÁK (a rendszer-/biztonsági és az aktiváló levelek NEM ide tartoznak: azok mindig mennek): gep_riasztas - hiba- és kritikus gép-értesítések (napi ellenőrzés hibája, automatikus besorolás-ellenőrzés) lejarati_riport - a napi/heti riport a lejárt és közelgő vizsgálatokról sikertelen_vizsgalat - "Korlátozással alkalmas" / "Nem alkalmas" minősítés utáni értesítés berlet - a kihelyezett (bérelt) gépek hibája / sikertelen vizsgálata miatt a BÉRLŐ cég felhasználóinak küldött értesítések (a bérbeadó cégnek a saját kategóriája szerint megy; a bérelt gépek lejáratai a bérlő lejarati_riport-jában vannak) HATÓKÖR: az Admin és az EÜI változatlanul (régi viselkedés); minden más szerepkör csak akkor kaphat levelet, ha CÉG szintű hozzáférése van és hozzáfér az Emelőgép modulhoz (hatokor_megfelelo_e). A napi riport dolgozói (HR) szakasza - orvosi alkalmasság, jogosítvány - kizárólag az Admin/EÜI címzetteknek megy (l. email_utils._ceg_riport). ALAPÉRTELMEZÉS (nincs sor a szabály-táblában): Admin és Emelőgép ügyintéző = alapbol_be (pontosan a korábbi működés, migráció nélkül), minden más szerepkör = tiltott (a cégadmin engedélyezheti). A CÉGES FŐKAPCSOLÓK (Dictionary: EmailRiportGyakorisag, VizsgalatSikertelenEmailErtesites) VÁLTOZATLANOK és ÁLTALUK FELÜLBÍRÁLTAK: ha a cég kikapcsolta a riportot / a sikertelen-vizsgálat levelet, senki nem kapja, akármit állít a hierarchia. Ezt a hívó (email_utils / msz_logic) ellenőrzi, nem ez a modul. Nem commitol: a hívó dönt a tranzakcióról (kivéve ahol a docstring mást mond). """ from datetime import datetime, timezone from typing import Dict, Iterable, List, Optional, Tuple import models # --- kategóriák (kód, név, rövid leírás) - a sorrend a felületen is ez --- KATEGORIAK: List[Tuple[str, str, str]] = [ ("gep_riasztas", "Gép-riasztások", "Hiba- és kritikus gép-értesítések: a napi ellenőrzés során talált hiba, az automatikus besorolás-ellenőrzés."), ("lejarati_riport", "Lejárati riport", "A napi (vagy heti) összefoglaló a lejárt és hamarosan lejáró vizsgálatokról. " "A gyakoriságot a cég állítja a Beállításokban; ha ott ki van kapcsolva, senki nem kapja."), ("sikertelen_vizsgalat", "Sikertelen vizsgálat", "Azonnali értesítés, ha egy vizsgálat vagy dokumentum \"Korlátozással alkalmas\" vagy \"Nem alkalmas\" " "minősítést kap. A Beállításokban a cég ki is kapcsolhatja - akkor senki nem kapja."), ("berlet", "Bérelt gépek", "Hiba- és sikertelen-vizsgálat értesítések azokról a gépekről, amelyeket más cégtől bérelsz (hozzád kihelyezett gépek). " "A bérelt gépek lejáratai a Lejárati riportban érkeznek."), ] KATEGORIA_KODOK = [k[0] for k in KATEGORIAK] KATEGORIA_NEVEK = {k[0]: k[1] for k in KATEGORIAK} # --- módok --- KOTELEZO = "kotelezo" ALAPBOL_BE = "alapbol_be" ALAPBOL_KI = "alapbol_ki" TILTOTT = "tiltott" MODOK = [KOTELEZO, ALAPBOL_BE, ALAPBOL_KI, TILTOTT] MOD_CIMKEK = { KOTELEZO: "Kötelező (nem kapcsolható ki)", ALAPBOL_BE: "Alapból be (kikapcsolható)", ALAPBOL_KI: "Alapból ki (bekapcsolható)", TILTOTT: "Tiltott", } # --- szerepkörök, amikre szabály adható (a rendszerben előforduló összes szerepkör) --- SZEREPKOROK = [ "Admin", "Emelőgép ügyintéző", "Emelőgép szakértő", "Emelőgép vizsgáló", "Felhasználó", "Karbantartó", "VIVU", "EBK", "Megtekintő", ] # Alapból ezek kapnak levelet (a korábbi, kódba égetett viselkedés) ALAP_CIMZETT_SZEREPKOROK = ("Admin", "Emelőgép ügyintéző") def alap_mod(szerepkor: str) -> str: """Ha nincs tárolt szabály: Admin/EÜI alapból kapja (kikapcsolható), a többi szerepkör tiltott.""" return ALAPBOL_BE if szerepkor in ALAP_CIMZETT_SZEREPKOROK else TILTOTT def _utc_most() -> datetime: return datetime.now(timezone.utc).replace(tzinfo=None) # ============================================================================================= # 1. A SZABÁLYOK ÉS A VÁLASZTÁSOK BETÖLTÉSE # ============================================================================================= def szabalyok_betoltese(db, company_id: int) -> Dict[Tuple[str, str], str]: """{(szerepkör, kategória): mód} - a cég TÁROLT szabályai, kiegészítve az alapértelmezéssel, így MINDEN szerepkör x kategória pár szerepel benne.""" szabalyok = {(sz, kat): alap_mod(sz) for sz in SZEREPKOROK for kat in KATEGORIA_KODOK} sorok = db.query(models.EmailFeliratkozasSzabaly).filter( models.EmailFeliratkozasSzabaly.company_id == company_id).all() for s in sorok: if s.mod in MODOK and s.kategoria in KATEGORIA_KODOK: szabalyok[(s.szerepkor, s.kategoria)] = s.mod return szabalyok def valasztasok_betoltese(db, user_idk: Iterable[int]) -> Dict[Tuple[int, str], bool]: """{(user_id, kategória): bekapcsolva} - a felhasználók TÁROLT saját választásai.""" user_idk = list(user_idk) if not user_idk: return {} sorok = db.query(models.EmailFeliratkozas).filter(models.EmailFeliratkozas.user_id.in_(user_idk)).all() return {(s.user_id, s.kategoria): bool(s.bekapcsolva) for s in sorok} def hatarozat(mod: str, valasztas: Optional[bool]) -> bool: """A VÉGLEGES döntés: kap-e levelet. kotelezo -> igen; tiltott -> nem; alapbol_be/ki -> a felhasználó tárolt választása, ennek hiányában az alapérték.""" if mod == KOTELEZO: return True if mod == TILTOTT: return False if valasztas is None: return mod == ALAPBOL_BE return bool(valasztas) # ============================================================================================= # 2. CÍMZETTEK (ezt hívják a levélküldő helyek) # ============================================================================================= def hatokor_megfelelo_e(user, company_modul_kodok) -> bool: """Az Admin és az EÜI (a korábbi, kódba égetett címzettek) változatlanul megfelelnek. Minden MÁS szerepkörnél (amit a cégadmin kapcsolt be) csak akkor kaphat levelet, ha CÉG szintű hozzáférése van (nem telephelyre/gépre korlátozott) ÉS hozzáfér az Emelőgép modulhoz - különben a saját hatóköre feletti gépekről és a teljes riportról kapna információt (az alkalmazás egyébként elrejtené előle).""" if user.role in ALAP_CIMZETT_SZEREPKOROK: return True if (user.hozzaferes_szint or "CEG") != "CEG": return False sajat = {m.kod for m in user.sajat_modulok if m.aktiv} return "EMELOGEP" in (company_modul_kodok & sajat) def _ceg_modul_kodok(db, company_id: int) -> set: ceg = db.get(models.Company, company_id) return {m.kod for m in ceg.engedelyezett_modulok if m.aktiv} if ceg is not None else set() def _kuldheto_felhasznalok(db, company_id: int): """A cég olyan felhasználói, akiknek egyáltalán küldhető levél: aktív, aktivált (a még meg nem erősített cím nem kap semmit - E szelet) és van e-mail címük.""" return db.query(models.User).filter( models.User.company_id == company_id, models.User.aktiv.isnot(False), models.User.aktivalva.isnot(False), models.User.email.isnot(None), ).all() def cimzett_felhasznalok(db, company_id: int, kategoria: str) -> list: """A cég azon felhasználói (User objektumok, e-mail cím szerint egyedi), akik a hierarchia szerint kapják a megadott kategóriát. A cég főkapcsolóit a HÍVÓ ellenőrzi.""" if kategoria not in KATEGORIA_KODOK: raise ValueError(f"ismeretlen levélkategória: {kategoria}") felhasznalok = _kuldheto_felhasznalok(db, company_id) if not felhasznalok: return [] szabalyok = szabalyok_betoltese(db, company_id) valasztasok = valasztasok_betoltese(db, [u.id for u in felhasznalok]) modulok = _ceg_modul_kodok(db, company_id) eredmeny, cimek = [], set() for u in felhasznalok: mod = szabalyok.get((u.role, kategoria), TILTOTT) if (hatarozat(mod, valasztasok.get((u.id, kategoria))) and hatokor_megfelelo_e(u, modulok) and u.email and u.email not in cimek): eredmeny.append(u) cimek.add(u.email) return eredmeny def cimzettek(db, company_id: int, kategoria: str) -> List[str]: """A cimzett_felhasznalok e-mail címei (a felhasználók sorrendjében).""" return [u.email for u in cimzett_felhasznalok(db, company_id, kategoria)] def feliratkozva_e(db, user, kategoria: str) -> bool: """Egy felhasználóra: kap-e levelet az adott kategóriából (a cég szabálya + a saját választása).""" mod = szabalyok_betoltese(db, user.company_id).get((user.role, kategoria), TILTOTT) v = valasztasok_betoltese(db, [user.id]).get((user.id, kategoria)) return hatarozat(mod, v) # ============================================================================================= # 3. A FELHASZNÁLÓ SAJÁT OLDALA # ============================================================================================= def felhasznaloi_allapot(db, user) -> List[dict]: """A felhasználó "Értesítéseim" oldalához: kategóriánként a szabály, a tényleges állapot és hogy módosíthatja-e. Csak a tiltott kategóriák 'lathato' = False (nem érdekesek a felhasználónak).""" szabalyok = szabalyok_betoltese(db, user.company_id) valasztasok = valasztasok_betoltese(db, [user.id]) hatokor_ok = hatokor_megfelelo_e(user, _ceg_modul_kodok(db, user.company_id)) eredmeny = [] for kod, nev, leiras in KATEGORIAK: mod = szabalyok.get((user.role, kod), TILTOTT) v = valasztasok.get((user.id, kod)) lathato = mod != TILTOTT and hatokor_ok # a hatókör (telephely-/gép-korlát, modul) miatt nem kapható: nem mutatjuk eredmeny.append({ "kod": kod, "nev": nev, "leiras": leiras, "mod": mod, "mod_cimke": MOD_CIMKEK[mod], "kap": lathato and hatarozat(mod, v), "modosithato": lathato and mod in (ALAPBOL_BE, ALAPBOL_KI), "lathato": lathato, }) return eredmeny def valasztas_beallitasa(db, user, kategoria: str, bekapcsolva: bool) -> bool: """A felhasználó saját választásának mentése. Igaz, ha mentettük; hamis, ha a cég szabálya nem engedi (kötelező / tiltott kategória, ismeretlen kategória). NEM commitol.""" if kategoria not in KATEGORIA_KODOK: return False mod = szabalyok_betoltese(db, user.company_id).get((user.role, kategoria), TILTOTT) if mod not in (ALAPBOL_BE, ALAPBOL_KI): return False sor = db.query(models.EmailFeliratkozas).filter_by(user_id=user.id, kategoria=kategoria).first() ma = _utc_most() if sor is None: db.add(models.EmailFeliratkozas(user_id=user.id, kategoria=kategoria, bekapcsolva=bool(bekapcsolva), modositva_at=ma)) elif bool(sor.bekapcsolva) != bool(bekapcsolva): sor.bekapcsolva = bool(bekapcsolva) sor.modositva_at = ma return True # ============================================================================================= # 4. A CÉGES ADMIN OLDALA # ============================================================================================= def szabaly_ervenyes_e(szerepkor: str, kategoria: str, mod: str) -> bool: """Elfogadható-e a (szerepkör, kategória, mód) hármas. Az Admin szerepkör nem tiltható le (különben egy elgépelt beállítással a cégben senki nem látná a riasztást).""" if szerepkor not in SZEREPKOROK or kategoria not in KATEGORIA_KODOK or mod not in MODOK: return False if szerepkor == "Admin" and mod == TILTOTT: return False return True def szabaly_beallitasa(db, company_id: int, szerepkor: str, kategoria: str, mod: str, kuldo) -> bool: """Egy szabály mentése (a cég Admin-jának kérésére) + napló. Igaz, ha ÉRDEMBEN változott (a naplóba csak a valódi változás kerül). Érvénytelen hármasra ValueError. NEM commitol.""" if not szabaly_ervenyes_e(szerepkor, kategoria, mod): raise ValueError("érvénytelen feliratkozás-szabály") regi = szabalyok_betoltese(db, company_id)[(szerepkor, kategoria)] sor = db.query(models.EmailFeliratkozasSzabaly).filter_by( company_id=company_id, szerepkor=szerepkor, kategoria=kategoria).first() if regi == mod and sor is None: return False # megegyezik az alapértelmezéssel és nincs tárolt sor: nincs mit menteni if sor is not None and sor.mod == mod: return False ma = _utc_most() if sor is None: db.add(models.EmailFeliratkozasSzabaly(company_id=company_id, szerepkor=szerepkor, kategoria=kategoria, mod=mod, modositva_at=ma, modosito_user_id=getattr(kuldo, "id", None))) else: sor.mod = mod sor.modositva_at = ma sor.modosito_user_id = getattr(kuldo, "id", None) db.add(models.AuditLog( company_id=company_id, user_id=getattr(kuldo, "id", None), field_name="Levél-feliratkozás szabály", old_value=f"{szerepkor} / {KATEGORIA_NEVEK[kategoria]}: {MOD_CIMKEK[regi]}", new_value=f"{szerepkor} / {KATEGORIA_NEVEK[kategoria]}: {MOD_CIMKEK[mod]}")) return True def ceges_attekintes(db, company_id: int) -> List[dict]: """A cég felhasználónkénti állapota (csak olvasható táblázathoz): ki melyik kategóriát kapja most.""" felhasznalok = db.query(models.User).filter( models.User.company_id == company_id, models.User.aktiv.isnot(False)).order_by(models.User.username).all() szabalyok = szabalyok_betoltese(db, company_id) valasztasok = valasztasok_betoltese(db, [u.id for u in felhasznalok]) modulok = _ceg_modul_kodok(db, company_id) sorok = [] for u in felhasznalok: allapot = {} kuldheto = bool(u.email) and u.aktivalva is not False and hatokor_megfelelo_e(u, modulok) for kod in KATEGORIA_KODOK: mod = szabalyok.get((u.role, kod), TILTOTT) allapot[kod] = kuldheto and mod != TILTOTT and hatarozat(mod, valasztasok.get((u.id, kod))) sorok.append({ "id": u.id, "username": u.username, "email": u.email, "role": u.role, "varo": u.aktivalva is False, "allapot": allapot, "hatokor_korlat": not hatokor_megfelelo_e(u, modulok), }) return sorok def cimzett_darabszamok(attekintes: List[dict]) -> Dict[str, int]: """Kategóriánként hány felhasználó kapja most (a ceges_attekintes() eredményéből) - a céges oldalon figyelmeztetés, ha egy kategóriának nincs címzettje (pl. mindenki kikapcsolta).""" return {kod: sum(1 for s in attekintes if s["allapot"].get(kod)) for kod in KATEGORIA_KODOK} def torles_felhasznalohoz(db, user_idk: Iterable[int]) -> None: """Felhasználó(k) törlésekor a saját választásaik törlése (a DB-ben ON DELETE CASCADE is van; ez a SQLite/tesztek és a védelmi mélység miatt). NEM commitol.""" user_idk = list(user_idk) if user_idk: db.query(models.EmailFeliratkozas).filter(models.EmailFeliratkozas.user_id.in_(user_idk)).delete(synchronize_session=False) def torles_ceghez(db, company_id: int) -> None: """Cég törlésekor a szabályok törlése. NEM commitol.""" db.query(models.EmailFeliratkozasSzabaly).filter( models.EmailFeliratkozasSzabaly.company_id == company_id).delete(synchronize_session=False)