agent_v0/raw_session.py

# raw_session.py
"""
Modèle RawSession v1 pour l'agent_v0.

Ce module définit la structure des données RawSession et
la sérialisation en JSON au format rawsession_v1.
"""

from __future__ import annotations

import json
import os
import time
import uuid
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional

from config import AGENT_VERSION, SESSIONS_ROOT


@dataclass
class ScreenshotMeta:
    screenshot_id: str
    relative_path: str
    captured_at: str


@dataclass
class Event:
    t: float
    type: str
    window: Dict[str, Any]
    screenshot_id: Optional[str] = None
    # champs optionnels selon type
    button: Optional[str] = None
    pos: Optional[List[int]] = None
    keys: Optional[List[str]] = None
    idle_ms: Optional[int] = None
    scroll_delta: Optional[List[int]] = None  # [dx, dy] pour la molette


@dataclass
class RawSession:
    schema_version: str
    session_id: str
    agent_version: str
    environment: Dict[str, Any]
    user: Dict[str, Any]
    context: Dict[str, Any]
    started_at: str
    ended_at: Optional[str] = None

    events: List[Event] = field(default_factory=list)
    screenshots: List[ScreenshotMeta] = field(default_factory=list)

    _start_time_monotonic: float = field(default_factory=time.monotonic, repr=False)

    @classmethod
    def create(
        cls,
        user_id: str,
        user_label: str = "",
        customer: str = "",
        training_label: str = "",
        notes: str = "",
        platform: str = "unknown",
        hostname: str = "unknown",
        screen_resolution: Optional[List[int]] = None,
        display_scale: float = 1.0,
    ) -> "RawSession":
        """Fabrique une nouvelle session avec les champs de base remplis."""
        session_id = f"sess_{time.strftime('%Y%m%dT%H%M%S')}_{uuid.uuid4().hex[:8]}"
        started_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())

        environment = {
            "platform": platform,
            "hostname": hostname,
            "screen": {
                "primary_resolution": screen_resolution or [0, 0],
                "display_scale": display_scale,
            },
        }
        user = {
            "id": user_id,
            "label": user_label or user_id,
        }
        context = {
            "customer": customer,
            "training_label": training_label,
            "notes": notes,
        }

        return cls(
            schema_version="rawsession_v1",
            session_id=session_id,
            agent_version=AGENT_VERSION,
            environment=environment,
            user=user,
            context=context,
            started_at=started_at,
        )

    # --- API temps relatif ---

    def now_relative(self) -> float:
        """Temps relatif (en secondes) depuis le début de la session."""
        return time.monotonic() - self._start_time_monotonic

    # --- API screenshots ---

    def add_screenshot(
        self,
        screenshot_id: str,
        relative_path: str,
        captured_at: Optional[str] = None,
    ) -> None:
        """Enregistre les métadonnées d'un screenshot."""
        captured_at = captured_at or time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
        self.screenshots.append(
            ScreenshotMeta(
                screenshot_id=screenshot_id,
                relative_path=relative_path,
                captured_at=captured_at,
            )
        )

    # --- API événements ---

    def add_mouse_click_event(
        self,
        button: str,
        pos: List[int],
        window_title: str,
        app_name: str,
        screenshot_id: Optional[str],
    ) -> None:
        self.events.append(
            Event(
                t=self.now_relative(),
                type="mouse_click",
                button=button,
                pos=pos,
                window={"title": window_title, "app_name": app_name},
                screenshot_id=screenshot_id,
            )
        )

    def add_key_combo_event(
        self,
        keys: List[str],
        window_title: str,
        app_name: str,
        screenshot_id: Optional[str],
    ) -> None:
        self.events.append(
            Event(
                t=self.now_relative(),
                type="key_combo",
                keys=keys,
                window={"title": window_title, "app_name": app_name},
                screenshot_id=screenshot_id,
            )
        )

    def add_hover_idle_event(
        self,
        pos: List[int],
        idle_ms: int,
        window_title: str,
        app_name: str,
        screenshot_id: Optional[str],
    ) -> None:
        self.events.append(
            Event(
                t=self.now_relative(),
                type="hover_idle",
                pos=pos,
                idle_ms=idle_ms,
                window={"title": window_title, "app_name": app_name},
                screenshot_id=screenshot_id,
            )
        )

    def add_scroll_event(
        self,
        pos: List[int],
        delta: List[int],
        window_title: str,
        app_name: str,
        screenshot_id: Optional[str],
    ) -> None:
        """delta = [dx, dy]"""
        self.events.append(
            Event(
                t=self.now_relative(),
                type="scroll",
                pos=pos,
                scroll_delta=delta,
                window={"title": window_title, "app_name": app_name},
                screenshot_id=screenshot_id,
            )
        )

    def close(self) -> None:
        """Marque la fin de la session."""
        self.ended_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())

    # --- Sérialisation ---

    def to_dict(self) -> Dict[str, Any]:
        return {
            "schema_version": self.schema_version,
            "session_id": self.session_id,
            "agent_version": self.agent_version,
            "environment": self.environment,
            "user": self.user,
            "context": self.context,
            "started_at": self.started_at,
            "ended_at": self.ended_at,
            "events": [e.__dict__ for e in self.events],
            "screenshots": [s.__dict__ for s in self.screenshots],
        }

    def save_json(self, base_dir: Optional[str] = None) -> str:
        """
        Sauvegarde le JSON rawsession_v1 dans un sous-dossier de SESSIONS_ROOT.

        Retourne le chemin du fichier JSON.
        """
        base_dir = base_dir or SESSIONS_ROOT
        session_dir = os.path.join(base_dir, self.session_id)
        os.makedirs(session_dir, exist_ok=True)
        json_path = os.path.join(session_dir, f"{self.session_id}.json")

        with open(json_path, "w", encoding="utf-8") as f:
            json.dump(self.to_dict(), f, ensure_ascii=False, indent=2)

        return json_path