"""Постоянно открытые Outlook и Teams в Chromium (Playwright, persistent profile).""" from __future__ import annotations import asyncio import json import logging import time from pathlib import Path from typing import Awaitable, Callable from urllib.parse import urlparse from playwright.async_api import BrowserContext, Page, async_playwright from ..config import Settings from .login import is_login_url, login_flow log = logging.getLogger(__name__) WATCHER_JS = (Path(__file__).parent / "watcher.js").read_text(encoding="utf-8") OnEvent = Callable[[dict], Awaitable[None]] Say = Callable[..., Awaitable[None]] class BrowserManager: def __init__(self, s: Settings, on_event: OnEvent, say: Say): self.s = s self.on_event = on_event self.say = say self.cfg = json.loads(s.selectors_path.read_text(encoding="utf-8")) self.hosts = {src["name"]: src["host"] for src in self.cfg["sources"]} self.urls = {"mail": s.mail_url, "teams": s.teams_url} self.pw = None self.ctx: BrowserContext | None = None self.pages: dict[str, Page] = {} self.last_event: dict[str, float] = {} self.login_lock = asyncio.Lock() self._state: dict[str, str] = {} self._last_reload = time.time() self.password_cache: str | None = s.uni_password # ---------- запуск ---------- async def start(self) -> None: self.pw = await async_playwright().start() self.ctx = await self.pw.chromium.launch_persistent_context( user_data_dir=str(self.s.profile_dir), headless=self.s.headless, executable_path=self.s.chromium_path, viewport={"width": 1400, "height": 900}, locale="en-US", timezone_id=self.s.tz, permissions=["notifications"], args=["--disable-dev-shm-usage", "--no-first-run", "--no-default-browser-check", "--disable-blink-features=AutomationControlled"], ) await self.ctx.expose_binding("__uniEmit", self._binding) init = f"window.__UNI_CFG = {json.dumps(self.cfg)};\n{WATCHER_JS}" await self.ctx.add_init_script(script=init) existing = list(self.ctx.pages) for name in ("mail", "teams"): page = existing.pop(0) if existing else await self.ctx.new_page() await self._attach(name, page) for extra in existing: await extra.close() for name in self.pages: await self._goto(name) async def _attach(self, name: str, page: Page) -> None: self.pages[name] = page page.on("crash", lambda _p, n=name: asyncio.create_task(self._recreate(n, "crash"))) async def _recreate(self, name: str, why: str) -> None: log.warning("Пересоздаю вкладку %s (%s)", name, why) try: await self.pages[name].close() except Exception: pass await self._attach(name, await self.ctx.new_page()) await self._goto(name) async def _goto(self, name: str) -> None: try: await self.pages[name].goto(self.urls[name], wait_until="domcontentloaded", timeout=60000) except Exception as e: log.warning("Не открылась %s: %s", name, e) async def _binding(self, source, payload: str) -> None: try: ev = json.loads(payload) except Exception: return self.last_event[ev.get("source", "?")] = time.time() await self.on_event(ev) # ---------- состояние ---------- def state(self, name: str) -> str: page = self.pages.get(name) if not page or page.is_closed(): return "closed" url = page.url if is_login_url(url): return "login" if self.hosts[name] in (urlparse(url).hostname or ""): return "ok" return "other" def states(self) -> dict[str, str]: return {n: self.state(n) for n in self.pages} async def screenshot(self, name: str) -> bytes: return await self.pages[name].screenshot() async def debug(self, name: str) -> dict: try: return await self.pages[name].evaluate("window.__uniDebug ? window.__uniDebug() : null") or {} except Exception as e: return {"error": str(e)} async def reload(self, name: str | None = None) -> None: for n in [name] if name else list(self.pages): try: await self.pages[n].reload(wait_until="domcontentloaded", timeout=60000) except Exception as e: log.warning("reload %s: %s", n, e) await self._recreate(n, "reload failed") # ---------- ручное управление (для нестандартного SSO) ---------- async def click(self, name: str, x: float, y: float) -> None: await self.pages[name].mouse.click(x, y) async def type(self, name: str, text: str) -> None: await self.pages[name].keyboard.type(text, delay=30) async def key(self, name: str, key: str) -> None: await self.pages[name].keyboard.press(key) async def goto(self, name: str, url: str | None = None) -> None: await self.pages[name].goto(url or self.urls[name], wait_until="domcontentloaded", timeout=60000) # ---------- вход ---------- async def login(self, ask) -> bool: if self.login_lock.locked(): await self.say("Вход уже идёт.") return False async def ask_cached(prompt: str, secret: bool = False, timeout: float = 300): v = await ask(prompt, secret=secret, timeout=timeout) if secret and v: self.password_cache = v # только в памяти, на диск не пишется return v async with self.login_lock: ask = ask_cached ok = await login_flow( self.pages["mail"], target_url=self.urls["mail"], target_host=self.hosts["mail"], email=self.s.uni_email, password=self.password_cache, say=self.say, ask=ask, ) if not ok: return False await self.say("✅ Почта открыта. Захожу в Teams…") # Teams обычно подхватывает SSO-сессию сам, максимум просит выбрать аккаунт ok_t = await login_flow( self.pages["teams"], target_url=self.urls["teams"], target_host=self.hosts["teams"], email=self.s.uni_email, password=self.password_cache, say=self.say, ask=ask, timeout=180, ) await self.say("✅ Teams открыт. Слежу за событиями." if ok_t else "⚠️ Почта работает, а Teams — нет. /screen teams") return ok_t # ---------- фоновые циклы ---------- async def health_loop(self) -> None: await asyncio.sleep(45) while True: try: for name in list(self.pages): st = self.state(name) prev = self._state.get(name) self._state[name] = st if st == "closed": await self._recreate(name, "closed") elif st == "login" and prev != "login": await self.say( f"⚠️ {'Почта' if name == 'mail' else 'Teams'}: сессия слетела, нужен вход. Жми /login" ) elif st == "ok" and prev == "login": await self.say(f"✅ {'Почта' if name == 'mail' else 'Teams'} снова на связи.") if self.s.reload_hours > 0 and time.time() - self._last_reload > self.s.reload_hours * 3600: if not self.login_lock.locked(): log.info("Плановая перезагрузка вкладок") self._last_reload = time.time() await self.reload() except Exception: log.exception("health_loop") await asyncio.sleep(60) async def scan_loop(self) -> None: """Страховка: раз в N секунд просим страницу пересмотреть списки (на случай пропущенных мутаций).""" while True: await asyncio.sleep(self.s.scan_interval) for name, page in list(self.pages.items()): if self.state(name) != "ok": continue try: await page.evaluate("window.__uniScan && window.__uniScan(false)") except Exception: pass async def close(self) -> None: try: if self.ctx: await self.ctx.close() finally: if self.pw: await self.pw.stop()