"""Per-user FSM state machine for article summary conversations.""" from __future__ import annotations import time from dataclasses import dataclass, field from enum import Enum, auto class ArticleState(Enum): IDLE = auto() URL_DETECTED = auto() AWAITING_LANGUAGE = auto() # Audio flow: waiting for language selection LANGUAGE = auto() DURATION = auto() TOPICS = auto() GENERATING = auto() COMPLETE = auto() @dataclass class ArticleSession: state: ArticleState = ArticleState.IDLE url: str = "" title: str = "" content: str = "" language: str = "" duration_minutes: int = 10 topics: list[str] = field(default_factory=list) detected_topics: list[str] = field(default_factory=list) summary_text: str = "" timestamp: float = field(default_factory=time.time) STATE_TIMEOUT = 300 # 5 minutes class SessionManager: """Manage per-(user, room) article summary sessions.""" def __init__(self) -> None: self._sessions: dict[tuple[str, str], ArticleSession] = {} def get(self, user_id: str, room_id: str) -> ArticleSession: key = (user_id, room_id) session = self._sessions.get(key) if session and time.time() - session.timestamp > STATE_TIMEOUT: session = None self._sessions.pop(key, None) if session is None: session = ArticleSession() self._sessions[key] = session return session def reset(self, user_id: str, room_id: str) -> None: self._sessions.pop((user_id, room_id), None) def touch(self, user_id: str, room_id: str) -> None: key = (user_id, room_id) if key in self._sessions: self._sessions[key].timestamp = time.time()