rpa_vision_v3/visual_workflow_builder/backend/models.py

"""backend/models.py

Modèles "visuels" minimalistes pour le Visual Workflow Builder.

Auteur : Dom, Alice, Kiro - 08 janvier 2026

Objectif du Patch #1:
- Fournir des structures de données stables (to_dict/from_dict)
- Permettre la persistance disque (JSON/YAML) via services.serialization
- Permettre au backend de démarrer même si le reste du core RPA n'est pas branché

NB: Ces modèles sont volontairement permissifs (ils conservent les champs inconnus).
"""

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Dict, List, Optional
from uuid import uuid4


def generate_id(prefix: str = "wf") -> str:
    """Génère un identifiant court et lisible."""
    return f"{prefix}_{uuid4().hex[:12]}"


@dataclass
class WorkflowSettings:
    """Sac de paramètres pour un workflow."""
    data: Dict[str, Any] = field(default_factory=dict)

    @classmethod
    def from_dict(cls, d: Any) -> "WorkflowSettings":
        if isinstance(d, WorkflowSettings):
            return d
        if not isinstance(d, dict):
            return cls({})
        return cls(dict(d))

    def to_dict(self) -> Dict[str, Any]:
        return dict(self.data)


@dataclass
class VisualNode:
    """Représentation d'un nœud dans le canvas visuel."""
    id: str
    type: str = "unknown"
    position: Dict[str, Any] = field(default_factory=dict)
    data: Dict[str, Any] = field(default_factory=dict)
    style: Dict[str, Any] = field(default_factory=dict)
    extra: Dict[str, Any] = field(default_factory=dict)

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "VisualNode":
        if not isinstance(d, dict):
            raise ValueError("Le nœud doit être un dictionnaire")
        node_id = d.get("id") or generate_id("node")
        node_type = d.get("type") or d.get("node_type") or "unknown"
        known = {"id", "type", "node_type", "position", "data", "style"}
        extra = {k: v for k, v in d.items() if k not in known}
        return cls(
            id=str(node_id),
            type=str(node_type),
            position=dict(d.get("position") or {}),
            data=dict(d.get("data") or {}),
            style=dict(d.get("style") or {}),
            extra=extra,
        )

    def to_dict(self) -> Dict[str, Any]:
        out = {
            "id": self.id,
            "type": self.type,
            "position": self.position,
            "data": self.data,
            "style": self.style,
        }
        out.update(self.extra)
        return out


@dataclass
class VisualEdge:
    """Représentation d'une connexion entre nœuds."""
    id: str
    source: str
    target: str
    type: str = "default"
    label: Optional[str] = None
    data: Dict[str, Any] = field(default_factory=dict)
    extra: Dict[str, Any] = field(default_factory=dict)

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "VisualEdge":
        if not isinstance(d, dict):
            raise ValueError("La connexion doit être un dictionnaire")
        edge_id = d.get("id") or generate_id("edge")
        source = d.get("source")
        target = d.get("target")
        if not source or not target:
            raise ValueError("La connexion nécessite 'source' et 'target'")
        known = {"id", "source", "target", "type", "label", "data"}
        extra = {k: v for k, v in d.items() if k not in known}
        return cls(
            id=str(edge_id),
            source=str(source),
            target=str(target),
            type=str(d.get("type") or "default"),
            label=d.get("label"),
            data=dict(d.get("data") or {}),
            extra=extra,
        )

    def to_dict(self) -> Dict[str, Any]:
        out = {
            "id": self.id,
            "source": self.source,
            "target": self.target,
            "type": self.type,
            "label": self.label,
            "data": self.data,
        }
        out.update(self.extra)
        return out


@dataclass
class Variable:
    """Variable de workflow."""
    name: str
    value: Any = None
    var_type: str = "any"
    scope: str = "workflow"
    description: str = ""
    required: bool = False
    extra: Dict[str, Any] = field(default_factory=dict)

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "Variable":
        if not isinstance(d, dict):
            raise ValueError("La variable doit être un dictionnaire")
        name = d.get("name") or d.get("key")
        if not name:
            raise ValueError("La variable nécessite un 'name'")
        known = {"name", "key", "value", "type", "var_type", "scope", "description", "required"}
        extra = {k: v for k, v in d.items() if k not in known}
        return cls(
            name=str(name),
            value=d.get("value"),
            var_type=str(d.get("type") or d.get("var_type") or "any"),
            scope=str(d.get("scope") or "workflow"),
            description=str(d.get("description") or ""),
            required=bool(d.get("required") or False),
            extra=extra,
        )

    def to_dict(self) -> Dict[str, Any]:
        out = {
            "name": self.name,
            "value": self.value,
            "type": self.var_type,
            "scope": self.scope,
            "description": self.description,
            "required": self.required,
        }
        out.update(self.extra)
        return out


@dataclass
class VisualWorkflow:
    """Workflow visuel complet."""
    id: str
    name: str
    description: str = ""
    created_by: str = "unknown"
    created_at: str = field(default_factory=lambda: datetime.now().isoformat())
    updated_at: str = field(default_factory=lambda: datetime.now().isoformat())
    nodes: List[VisualNode] = field(default_factory=list)
    edges: List[VisualEdge] = field(default_factory=list)
    variables: List[Variable] = field(default_factory=list)
    settings: WorkflowSettings = field(default_factory=WorkflowSettings)
    tags: List[str] = field(default_factory=list)
    category: str = "default"
    is_template: bool = False
    extra: Dict[str, Any] = field(default_factory=dict)

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "VisualWorkflow":
        if not isinstance(d, dict):
            raise ValueError("Le workflow doit être un dictionnaire")

        wf_id = d.get("id") or generate_id("wf")
        name = d.get("name") or "Sans titre"

        known = {
            "id", "name", "description", "created_by", "created_at", "updated_at",
            "nodes", "edges", "variables", "settings", "tags", "category", "is_template"
        }
        extra = {k: v for k, v in d.items() if k not in known}

        nodes = [VisualNode.from_dict(n) for n in (d.get("nodes") or [])]
        edges = [VisualEdge.from_dict(e) for e in (d.get("edges") or [])]
        variables = [Variable.from_dict(v) for v in (d.get("variables") or [])]
        settings = WorkflowSettings.from_dict(d.get("settings") or {})

        return cls(
            id=str(wf_id),
            name=str(name),
            description=str(d.get("description") or ""),
            created_by=str(d.get("created_by") or "unknown"),
            created_at=str(d.get("created_at") or datetime.now().isoformat()),
            updated_at=str(d.get("updated_at") or datetime.now().isoformat()),
            nodes=nodes,
            edges=edges,
            variables=variables,
            settings=settings,
            tags=list(d.get("tags") or []),
            category=str(d.get("category") or "default"),
            is_template=bool(d.get("is_template") or False),
            extra=extra,
        )

    def to_dict(self) -> Dict[str, Any]:
        out = {
            "id": self.id,
            "name": self.name,
            "description": self.description,
            "created_by": self.created_by,
            "created_at": self.created_at,
            "updated_at": self.updated_at,
            "nodes": [n.to_dict() for n in self.nodes],
            "edges": [e.to_dict() for e in self.edges],
            "variables": [v.to_dict() for v in self.variables],
            "settings": self.settings.to_dict(),
            "tags": self.tags,
            "category": self.category,
            "is_template": self.is_template,
        }
        out.update(self.extra)
        return out

    def validate(self) -> List[str]:
        """Valide le workflow et retourne la liste des erreurs."""
        errors: List[str] = []

        if not self.name or not str(self.name).strip():
            errors.append("le nom est requis")

        node_ids = [n.id for n in self.nodes]
        if len(node_ids) != len(set(node_ids)):
            errors.append("identifiants de nœuds dupliqués")

        # Les connexions doivent référencer des nœuds existants
        nodes_set = set(node_ids)
        for e in self.edges:
            if e.source not in nodes_set:
                errors.append(f"connexion {e.id} source '{e.source}' n'existe pas")
            if e.target not in nodes_set:
                errors.append(f"connexion {e.id} target '{e.target}' n'existe pas")

        # Les variables doivent être uniques par nom
        var_names = [v.name for v in self.variables]
        if len(var_names) != len(set(var_names)):
            errors.append("noms de variables dupliqués")

        return errors