"""„Szabadságon vagyok” (v35) - a felhasználó szabadsága alatt nem kap e-mailt, a kimaradtakról a visszatérésekor EGY összegző levelet kap. MEGRENDELŐI SZABÁLYOK (v35) * A felhasználó a Profilom oldalon (`/profil`) kapcsolja be; opcionálisan megadhatja a VISSZATÉRÉS NAPJÁT - azon a napon (helyi idő szerint 00:00 után, az ütemező 5 percenként nézi) a szabadság magától véget ér. Dátum nélkül addig tart, amíg a felhasználó ki nem kapcsolja. * Amíg be van kapcsolva, a felhasználó NEM kap leveleket (riasztás, napi riport, minden jövőbeli kategória) - KIVÉVE a fiókot / a rendszert védő biztonsági leveleket (jelszó/2FA/mentőkód-értesítők, aktiváló link, bejelentkezési tiltás: `email_sor.SZABADSAG_ALATT_IS_KULDOTT_KATEGORIAK`). * A szabadság VÉGÉN (kézi kikapcsolás vagy a visszatérés napja) egyetlen ÖSSZEGZŐ levél megy: kategóriánként darabszám + a tárgysorok listája időponttal (kategóriánként legfeljebb MAX_SOR_KATEGORIANKENT sor, a többit "és még N db" jelzi). Ha nem maradt ki semmi, levél sem megy. MŰKÖDÉS 1. `email_sor.sorba_allit()` minden címzettnél megnézi, hogy szabadságon van-e (`szabadsagon_levo_cimek`); ha igen (és a kategória nem biztonsági), a sor `visszatartva` állapotba kerül (a kiürítő csak a `varakozik` sorokat küldi), a törzs nem őrződik meg. 2. `kezdes()`: bekapcsolja a jelzőt, és a már sorban álló (még el nem küldött) nem biztonsági leveleket is visszatartja. 3. `befejezes()`: kikapcsolja a jelzőt, a visszatartott sorokból összegzőt készít (a rendszer kategóriában, tehát az sosem tartódik vissza), a sorokat `osszegezve` állapotra állítja. Mindez EGY tranzakció (a hívó commitol): ha bármi elbukik, semmi nem változik, és a szabadság tovább tart (az automatikus vég a következő futáskor újra próbálkozik). 4. `lejarat_ellenorzes()`: az ütemező (main.py, 5 percenként) hívja; a lejárt visszatérési napú szabadságokat lezárja. Minden módosító művelet a Jogosultsági naplóba is ír („Szabadságon vagyok”). Egyik függvény sem commitol, kivéve a `lejarat_ellenorzes()`. """ import html from datetime import date, datetime, timedelta, timezone from typing import Dict, List, Optional, Set, Tuple from sqlalchemy import func, or_ import email_sor import models from database import SessionLocal from templates_env import HELYI_TZ MAX_SOR_KATEGORIANKENT = 30 # az összegzőben kategóriánként ennyi tárgysor jelenik meg MAX_NAP = 366 # a visszatérés napja legfeljebb ennyi nappal lehet a jövőben KATEGORIA_NEVEK = {"riasztas": "Riasztások", "napi_riport": "Napi riportok"} def _utc_most() -> datetime: return datetime.now(timezone.utc).replace(tzinfo=None) # a DB-ben tz nélküli UTC def _helyi_ma() -> date: return datetime.now(HELYI_TZ).date() def _helyi_szoveg(utc_naiv: Optional[datetime]) -> str: if utc_naiv is None: return "?" return utc_naiv.replace(tzinfo=timezone.utc).astimezone(HELYI_TZ).strftime("%Y-%m-%d %H:%M") def szabadsagon_levo_cimek(db, cimek) -> Set[str]: """A megadott címek közül azok (KISBETŰVEL), amelyek tulajdonosa szabadságon van. A sorba_allit hívja minden levélnél.""" kisbetus = sorted({c.lower() for c in cimek if c}) if not kisbetus: return set() sorok = db.query(models.User.email).filter( models.User.szabadsagon.is_(True), func.lower(models.User.email).in_(kisbetus)).all() return {(e or "").lower() for (e,) in sorok} def visszateres_ellenorzese(szoveg: Optional[str]) -> Tuple[Optional[date], Optional[str]]: """A beírt visszatérési nap (ÉÉÉÉ-HH-NN) ellenőrzése. Üres -> (None, None). Hiba esetén (None, hibaüzenet).""" szoveg = (szoveg or "").strip() if not szoveg: return None, None try: nap = date.fromisoformat(szoveg) except ValueError: return None, "Érvénytelen dátum." ma = _helyi_ma() if nap <= ma: return None, "A visszatérés napja a jövőben legyen (legkorábban holnap)." if nap > ma + timedelta(days=MAX_NAP): return None, f"A visszatérés napja legfeljebb {MAX_NAP} nappal lehet a jövőben." return nap, None def _naplo(db, user, regi: str, uj: str) -> None: db.add(models.AuditLog(company_id=user.company_id, user_id=user.id, field_name="Szabadságon vagyok", old_value=regi, new_value=uj)) def _sajat_sorok(db, user, allapot: str): K = models.EmailKimeno return db.query(K).filter(func.lower(K.cimzett) == (user.email or "").lower(), K.allapot == allapot) def visszatartott_darab(db, user) -> int: return _sajat_sorok(db, user, email_sor.ALLAPOT_VISSZATARTVA).count() def allapot(db, user) -> Dict[str, object]: """A Profilom oldal adata: be van-e kapcsolva, mióta, mikor jár le, hány levelet tartottunk vissza eddig.""" aktiv = bool(user.szabadsagon) return { "aktiv": aktiv, "kezdete": _helyi_szoveg(user.szabadsag_kezdete_at) if aktiv else None, "visszateres": user.szabadsag_visszateres if aktiv else None, "visszatartott": visszatartott_darab(db, user) if aktiv else 0, "holnap": (_helyi_ma() + timedelta(days=1)).isoformat(), "max_datum": (_helyi_ma() + timedelta(days=MAX_NAP)).isoformat(), } def kezdes(db, user, visszateres: Optional[date]) -> int: """A szabadság bekapcsolása. Visszatér: a már sorban álló, ezért utólag visszatartott levelek száma. Nem commitol.""" user.szabadsagon = True user.szabadsag_kezdete_at = _utc_most() user.szabadsag_visszateres = visszateres K = models.EmailKimeno # a még el nem küldött (várakozó) nem biztonsági levelek is visszatartódnak, ne menjenek ki a bekapcsolás után darab = _sajat_sorok(db, user, email_sor.ALLAPOT_VARAKOZIK).filter( or_(K.kategoria.is_(None), ~K.kategoria.in_(sorted(email_sor.SZABADSAG_ALATT_IS_KULDOTT_KATEGORIAK))) ).update({"allapot": email_sor.ALLAPOT_VISSZATARTVA, "torzs_html": ""}, synchronize_session=False) _naplo(db, user, "kikapcsolva", "bekapcsolva" + (f" (visszatérés: {visszateres.isoformat()})" if visszateres else " (visszatérési nap nélkül)")) return darab def datum_modositasa(db, user, visszateres: Optional[date]) -> bool: """Bekapcsolt szabadság visszatérési napjának módosítása / törlése. Igaz, ha változott. Nem commitol.""" if user.szabadsag_visszateres == visszateres: return False regi = user.szabadsag_visszateres.isoformat() if user.szabadsag_visszateres else "nincs" user.szabadsag_visszateres = visszateres _naplo(db, user, f"visszatérés: {regi}", f"visszatérés: {visszateres.isoformat() if visszateres else 'nincs'}") return True def osszegzo_epitese(user, sorok, kezdet: Optional[datetime], veg: datetime) -> Tuple[str, str]: """Az összegző levél (tárgy, HTML törzs). Minden változó adat html.escape-pelve kerül a törzsbe.""" e = lambda x: html.escape("" if x is None else str(x), quote=True) # noqa: E731 csoportok: Dict[str, list] = {} for s in sorok: csoportok.setdefault(s.kategoria or "egyeb", []).append(s) nev = lambda k: KATEGORIA_NEVEK.get(k, k) # noqa: E731 osszesen = len(sorok) targy = f"Összegzés: {osszesen} levél maradt ki a szabadságod alatt" reszek = [ "
", "

Üdv újra!

", f"

Kedves {e(user.username)}! A szabadságod alatt ({e(_helyi_szoveg(kezdet))} - {e(_helyi_szoveg(veg))}) {osszesen} levelet nem küldtünk ki neked. " "Ahelyett, hogy egyenként megkapnád őket, itt az összegzésük:

", "", ] for kat, lista in csoportok.items(): reszek.append(f"") reszek.append("
{e(nev(kat))}{len(lista)} db
") for kat, lista in csoportok.items(): reszek.append(f"

{e(nev(kat))} ({len(lista)} db)

") reszek.append("

A részleteket az EmeLOG-ban tekintheted meg. A szabadság alatt a fiókodat védő biztonsági értesítők továbbra is megérkeztek.

") return targy, "".join(reszek) def befejezes(db, user, ok: str = "kézi kikapcsolás") -> int: """A szabadság lezárása + az összegző levél sorba állítása. Visszatér: az összegzett levelek száma. Nem commitol (egy tranzakció).""" veg = _utc_most() kezdet = user.szabadsag_kezdete_at sorok = _sajat_sorok(db, user, email_sor.ALLAPOT_VISSZATARTVA).order_by(models.EmailKimeno.id).all() user.szabadsagon = False user.szabadsag_kezdete_at = None user.szabadsag_visszateres = None db.flush() if sorok: targy, torzs = osszegzo_epitese(user, sorok, kezdet, veg) # kategória: "rendszer" - ez sosem tartódik vissza; ha a sorba állítás nem sikerül (0), kivételt dobunk, hogy az egész tranzakció visszagördüljön if email_sor.sorba_allit([user.email], targy, torzs, kategoria="rendszer", prioritas=email_sor.PRIORITAS_NORMAL, company_id=user.company_id, db=db) != 1: raise RuntimeError("az összegző levelet nem sikerült sorba állítani") for s in sorok: s.allapot = email_sor.ALLAPOT_OSSZEGEZVE _naplo(db, user, "bekapcsolva", f"kikapcsolva ({ok}); összegző levélben: {len(sorok)} db") return len(sorok) def lejarat_ellenorzes() -> int: """Az ütemező hívja (5 percenként): a lejárt visszatérési napú szabadságokat lezárja, és elküldi az összegzőt. Visszatér: a lezárt szabadságok száma. SOSEM dob kivételt; egy felhasználó hibája nem akadályozza a többit (a hibás a következő futáskor újra próbálkozik).""" db = SessionLocal() lezart = 0 try: ma = _helyi_ma() azonositok = [i for (i,) in db.query(models.User.id).filter( models.User.szabadsagon.is_(True), models.User.szabadsag_visszateres.isnot(None), models.User.szabadsag_visszateres <= ma).all()] for uid in azonositok: try: befejezes(db, db.get(models.User, uid), "a visszatérés napja elérkezett (automatikus)") db.commit() lezart += 1 except Exception as ex: db.rollback() print(f"❌ A szabadság automatikus lezárása sikertelen (felhasználó #{uid}): {type(ex).__name__}: {ex}") except Exception as ex: db.rollback() print(f"❌ Hiba a szabadságok ellenőrzésekor: {type(ex).__name__}: {ex}") finally: db.close() return lezart