From bb616d3ca84e8b8cf56911a051e4eeffc4e4b0bd Mon Sep 17 00:00:00 2001
From: Ewald <129623145+MajoorWaldi@users.noreply.github.com>
Date: Tue, 9 Dec 2025 17:53:10 +0100
Subject: [PATCH] Add files via upload

---
 server/__init__.py            |   5 +
 server/collections.py         |  38 ++
 server/config.py              |  23 +
 server/generation_metadata.py | 783 ++++++++++++++++++++++++++++++++++
 server/metadata.py            |  40 ++
 server/routes.py              | 549 ++++++++++++++++++++++++
 server/utils.py               | 446 +++++++++++++++++++
 server/version.py             |   1 +
 8 files changed, 1885 insertions(+)
 create mode 100644 server/__init__.py
 create mode 100644 server/collections.py
 create mode 100644 server/config.py
 create mode 100644 server/generation_metadata.py
 create mode 100644 server/metadata.py
 create mode 100644 server/routes.py
 create mode 100644 server/utils.py
 create mode 100644 server/version.py

diff --git a/server/__init__.py b/server/__init__.py
new file mode 100644
index 00000000..81337889
--- /dev/null
+++ b/server/__init__.py
@@ -0,0 +1,5 @@
+"""
+Backend entry for Majoor File Manager.
+Just importing this module will register our HTTP routes on PromptServer.
+"""
+from . import routes  # noqa: F401
diff --git a/server/collections.py b/server/collections.py
new file mode 100644
index 00000000..86c4d52f
--- /dev/null
+++ b/server/collections.py
@@ -0,0 +1,38 @@
+import os, json
+from .config import COLLECTIONS_DIR
+from .utils import ensure_dir
+
+def _path(name):
+    return os.path.join(COLLECTIONS_DIR, name + ".json")
+
+def get_collections():
+    ensure_dir(COLLECTIONS_DIR)
+    out = []
+    for f in os.listdir(COLLECTIONS_DIR):
+        if f.endswith(".json"):
+            out.append(f[:-5])
+    return out
+
+def load_collection(name):
+    fp = _path(name)
+    if not os.path.exists(fp):
+        return []
+    with open(fp, "r", encoding="utf-8") as f:
+        return json.load(f)
+
+def save_collection(name, files):
+    ensure_dir(COLLECTIONS_DIR)
+    with open(_path(name), "w", encoding="utf-8") as f:
+        json.dump(files, f, indent=2)
+
+def add_to_collection(name, path):
+    data = load_collection(name)
+    if path not in data:
+        data.append(path)
+    save_collection(name, data)
+
+def remove_from_collection(name, path):
+    data = load_collection(name)
+    if path in data:
+        data.remove(path)
+    save_collection(name, data)
diff --git a/server/config.py b/server/config.py
new file mode 100644
index 00000000..a5b1e693
--- /dev/null
+++ b/server/config.py
@@ -0,0 +1,23 @@
+import os
+import platform
+from folder_paths import get_output_directory
+
+OUTPUT_ROOT = get_output_directory()
+THUMB_SIZE = (320, 320)
+
+# Sidecar JSON files are optional; default to Windows metadata on Windows
+_DEFAULT_SIDECAR = platform.system().lower() != "windows"
+_SIDECAR_ENV = os.environ.get("MJR_ENABLE_SIDECAR")
+if _SIDECAR_ENV is None:
+    ENABLE_JSON_SIDECAR = _DEFAULT_SIDECAR
+else:
+    ENABLE_JSON_SIDECAR = str(_SIDECAR_ENV).strip().lower() not in (
+        "0",
+        "false",
+        "no",
+        "off",
+    )
+
+METADATA_EXT = ".mjr.json"
+COLLECTIONS_DIR = os.path.join(OUTPUT_ROOT, "_mjr_collections")
+os.makedirs(COLLECTIONS_DIR, exist_ok=True)
diff --git a/server/generation_metadata.py b/server/generation_metadata.py
new file mode 100644
index 00000000..2387c211
--- /dev/null
+++ b/server/generation_metadata.py
@@ -0,0 +1,783 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Set, Tuple
+
+try:
+    from PIL import Image  # type: ignore
+except Exception:  # Pillow non dispo -> pas de parsing PNG
+    Image = None
+try:
+    from .utils import load_metadata  # type: ignore
+except Exception:
+    load_metadata = None  # sidecar fallback option
+
+
+# Samplers que l'on considère comme "source de vérité" (lowercase)
+SAMPLER_CLASSES: Set[str] = {
+    "ksampler",
+    "samplercustom",
+    "ksampleradvanced",
+    # WAN / Kijai wrappers
+    "wanvideosampler",
+    "wanvideoksampler",
+    "wanvideoksampler",
+    "wanmoeksampler",
+}
+
+# Loaders de modèle (diffusion / checkpoint)
+CHECKPOINT_LOADER_CLASSES: Set[str] = {
+    "CheckpointLoaderSimple",
+    "CheckpointLoader",
+}
+UNET_LOADER_CLASSES: Set[str] = {
+    "UNETLoader",
+    "LoadDiffusionModel",  # certains wrappers utilisent ce label
+}
+DIFFUSERS_LOADER_CLASSES: Set[str] = {
+    "DiffusersLoader",
+}
+
+# LoRA
+LORA_CLASSES: Set[str] = {
+    "LoraLoader",
+    "LoraLoaderModelOnly",
+}
+
+
+# --------- Helpers généraux ---------
+
+
+def _ensure_dict_from_json(value: Any) -> Optional[Dict[str, Any]]:
+    """
+    Accepte soit un dict, soit une string JSON.
+    Retourne un dict ou None.
+    """
+    if isinstance(value, dict):
+        return value
+    if isinstance(value, str):
+        try:
+            data = json.loads(value)
+            if isinstance(data, dict):
+                return data
+        except Exception:
+            return None
+    return None
+
+
+def _extract_link_node_id(val: Any) -> Optional[str]:
+    """
+    Récupère l'id de node à partir d'un input de type:
+    - ["node_id", output_idx]
+    - [[ "node_id", output_idx ], ...]
+    - str/int direct (cas où le graph stocke juste l'id)
+    """
+    if isinstance(val, (str, int)):
+        return str(val)
+    if isinstance(val, tuple) and val and isinstance(val[0], (str, int)):
+        return str(val[0])
+    # cas ["5", 0]
+    if isinstance(val, list) and len(val) >= 2 and isinstance(val[0], (str, int)):
+        return str(val[0])
+
+    # cas [[ "5", 0 ], ...]
+    if (
+        isinstance(val, list)
+        and val
+        and isinstance(val[0], list)
+        and len(val[0]) >= 2
+        and isinstance(val[0][0], (str, int))
+    ):
+        return str(val[0][0])
+
+    return None
+
+
+# --------- Lecture du graph PROMPT depuis le PNG ---------
+
+
+def _normalize_workflow_to_prompt_graph(workflow: Any) -> Optional[Dict[str, Any]]:
+    """
+    Convertit un workflow (dict avec 'nodes': [...]) en map id->node pour parsing.
+    """
+    if isinstance(workflow, dict) and isinstance(workflow.get("nodes"), list):
+        nodes_map: Dict[str, Any] = {}
+        for n in workflow.get("nodes", []):
+            if not isinstance(n, dict):
+                continue
+            nid = n.get("id")
+            if nid is None:
+                continue
+            ctype = n.get("class_type") or n.get("type") or n.get("title")
+            nodes_map[str(nid)] = {
+                "class_type": ctype,
+                "inputs": n.get("inputs", {}) or {},
+                "title": n.get("title"),
+                "widgets_values": n.get("widgets_values"),
+            }
+        return nodes_map if nodes_map else None
+    if isinstance(workflow, dict):
+        # Peut déjà être map id -> node
+        return {str(k): v for k, v in workflow.items() if isinstance(v, dict)}
+    return None
+
+
+def _prompt_graph_to_workflow(prompt_graph: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    Reconstruit un workflow (nodes: [...]) minimal à partir du prompt_graph map id->node.
+    - Ajoute un champ version numérique attendu par ComfyUI.
+    - Normalise les ids en int quand possible pour éviter les erreurs "expected number".
+    - Fournit toujours les clés last_node_id / last_link_id / links pour le loader.
+    """
+    nodes: List[Dict[str, Any]] = []
+    links: List[List[Any]] = []
+    max_id: int = -1
+    last_link_id: int = 0
+
+    for nid, node in prompt_graph.items():
+        if not isinstance(node, dict):
+            continue
+
+        # Normalise l'id
+        try:
+            nid_int: Any = int(nid)
+        except Exception:
+            nid_int = nid
+        if isinstance(nid_int, int):
+            max_id = max(max_id, nid_int)
+
+        inputs_dict = node.get("inputs", {}) if isinstance(node.get("inputs"), dict) else {}
+        inputs_arr: List[Dict[str, Any]] = []
+
+        for idx, (iname, ival) in enumerate(inputs_dict.items()):
+            inp_entry: Dict[str, Any] = {
+                "name": iname,
+                "type": "ANY",
+            }
+            # Si l'entrée ressemble à un lien ["node_id", slot]
+            if isinstance(ival, (list, tuple)) and len(ival) >= 2 and isinstance(ival[0], (str, int)):
+                try:
+                    src_id = int(ival[0])
+                except Exception:
+                    src_id = ival[0] if ival[0] is not None else 0
+                try:
+                    src_slot = int(ival[1])
+                except Exception:
+                    src_slot = 0
+                last_link_id += 1
+                link_id = last_link_id
+                # lien: [id, fromNode, fromSlot, toNode, toSlot, type]
+                dest_id = nid_int if isinstance(nid_int, int) else (nid if nid is not None else 0)
+                links.append([link_id, src_id, src_slot, dest_id, idx, "ANY"])
+                inp_entry["link"] = link_id
+            else:
+                inp_entry["value"] = ival
+            inputs_arr.append(inp_entry)
+
+        entry: Dict[str, Any] = {
+            "id": nid_int,
+            "class_type": node.get("class_type"),
+            "type": node.get("class_type") or node.get("title") or "Unknown",
+            "inputs": inputs_arr,
+            "outputs": [],
+        }
+        if node.get("title"):
+            entry["title"] = node.get("title")
+        if node.get("widgets_values") is not None:
+            entry["widgets_values"] = node.get("widgets_values")
+
+        nodes.append(entry)
+
+    workflow = {
+        "last_node_id": max_id if max_id >= 0 else 0,
+        "last_link_id": last_link_id,
+        "links": links,
+        "nodes": nodes,
+        "version": 1,
+    }
+    return workflow
+
+
+def _normalize_workflow_dict(wf: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    Normalise un workflow brut pour éviter les erreurs de loader ComfyUI.
+    - force version numérique
+    - ajoute les champs last_node_id / last_link_id / links si absents
+    - convertit les ids de node en int quand possible
+    """
+    out = dict(wf) if isinstance(wf, dict) else {}
+
+    # version
+    try:
+        out["version"] = int(out.get("version", 1))
+    except Exception:
+        out["version"] = 1
+
+    # last_node_id / last_link_id
+    try:
+        out["last_node_id"] = int(out.get("last_node_id", 0))
+    except Exception:
+        out["last_node_id"] = 0
+    try:
+        out["last_link_id"] = int(out.get("last_link_id", 0))
+    except Exception:
+        out["last_link_id"] = 0
+
+    # links: filtre et assainit
+    clean_links: List[List[Any]] = []
+    links_val = out.get("links")
+    if isinstance(links_val, list):
+        for link in links_val:
+            if not isinstance(link, (list, tuple)):
+                continue
+            # pad to length 6
+            link_list = list(link) + [None] * (6 - len(link))
+            idv, from_node, from_slot, to_node, to_slot, ltype = link_list[:6]
+            try:
+                idv = int(idv)
+            except Exception:
+                continue
+            try:
+                from_node = int(from_node)
+            except Exception:
+                if from_node is None:
+                    from_node = 0
+            try:
+                from_slot = int(from_slot)
+            except Exception:
+                from_slot = 0
+            try:
+                to_node = int(to_node)
+            except Exception:
+                if to_node is None:
+                    to_node = 0
+            try:
+                to_slot = int(to_slot)
+            except Exception:
+                to_slot = 0
+            ltype = ltype or "ANY"
+            clean_links.append([idv, from_node, from_slot, to_node, to_slot, ltype])
+    out["links"] = clean_links
+    if clean_links:
+        try:
+            max_link = max(int(l[0]) for l in clean_links)
+            out["last_link_id"] = max(out["last_link_id"], max_link)
+        except Exception:
+            pass
+
+    # nodes
+    nodes = out.get("nodes")
+    if not isinstance(nodes, list):
+        out["nodes"] = []
+        return out
+
+    norm_nodes: List[Dict[str, Any]] = []
+    max_id = out["last_node_id"]
+    for node in nodes:
+        if not isinstance(node, dict):
+            continue
+        n = dict(node)
+        nid = n.get("id")
+        try:
+            nid_int = int(nid)
+            n["id"] = nid_int
+            max_id = max(max_id, nid_int)
+        except Exception:
+            pass  # keep as-is
+        # inputs: normalise en tableau attendu par le front
+        inputs = n.get("inputs")
+        if isinstance(inputs, dict):
+            inputs_arr = []
+            for iname, ival in inputs.items():
+                inputs_arr.append({"name": iname, "type": "ANY", "value": ival})
+            n["inputs"] = inputs_arr
+        elif isinstance(inputs, list):
+            n["inputs"] = inputs
+        else:
+            n["inputs"] = []
+        # outputs: assure une liste
+        if not isinstance(n.get("outputs"), list):
+            n["outputs"] = [] if n.get("outputs") is None else n.get("outputs")
+        norm_nodes.append(n)
+    out["nodes"] = norm_nodes
+    out["last_node_id"] = max_id
+    return out
+
+
+def load_prompt_graph_from_png(path: str | Path) -> Optional[Dict[str, Any]]:
+    """
+    Charge le graph 'prompt' stocké dans les metadata ComfyUI du PNG.
+
+    - On suppose que ComfyUI a écrit un champ 'prompt' (format graph backend).
+    - Pas de fallback exotique: si 'prompt' n'est pas là, on renvoie None.
+    """
+    if Image is None:
+        return None
+
+    p = Path(path)
+    if not p.exists():
+        return None
+
+    img = Image.open(p)
+    info = getattr(img, "info", {}) or {}
+
+    # 1) prompt au top-level
+    prompt = _ensure_dict_from_json(info.get("prompt"))
+    if prompt is not None:
+        # certains dumps contiennent une liste de nodes plutôt qu'un map
+        if isinstance(prompt, list):
+            prompt = _normalize_workflow_to_prompt_graph({"nodes": prompt})
+        return prompt
+
+    # 1bis) workflow au top-level
+    wf_raw = _ensure_dict_from_json(info.get("workflow"))
+    if wf_raw is not None:
+        pg = _normalize_workflow_to_prompt_graph(wf_raw)
+        if pg:
+            return pg
+
+    # 2) prompt dans extra_pnginfo (cas courant)
+    extra = info.get("extra_pnginfo")
+    if isinstance(extra, str):
+        try:
+            extra = json.loads(extra)
+        except Exception:
+            extra = None
+
+    if isinstance(extra, dict):
+        prompt = _ensure_dict_from_json(extra.get("prompt"))
+        if prompt is not None:
+            if isinstance(prompt, list):
+                prompt = _normalize_workflow_to_prompt_graph({"nodes": prompt})
+            return prompt
+        wf_raw = _ensure_dict_from_json(extra.get("workflow"))
+        if wf_raw is not None:
+            pg = _normalize_workflow_to_prompt_graph(wf_raw)
+            if pg:
+                return pg
+    if isinstance(extra, list):
+        for item in extra:
+            if not isinstance(item, dict):
+                continue
+            prompt = _ensure_dict_from_json(item.get("prompt"))
+            if prompt is not None:
+                return prompt
+            wf_raw = _ensure_dict_from_json(item.get("workflow"))
+            if wf_raw is not None:
+                pg = _normalize_workflow_to_prompt_graph(wf_raw)
+                if pg:
+                    return pg
+
+    # Pas de prompt exploitable → on arrête là
+    return None
+
+
+def load_raw_workflow_from_png(path: str | Path) -> Optional[Dict[str, Any]]:
+    """
+    Récupère le workflow brut (tel qu'exporté par ComfyUI) s'il est présent dans les metadata PNG.
+    Ne tente pas de normalisation: on renvoie tel quel pour que le front puisse le charger.
+    """
+    if Image is None:
+        return None
+
+    p = Path(path)
+    if not p.exists():
+        return None
+
+    img = Image.open(p)
+    info = getattr(img, "info", {}) or {}
+
+    def _maybe_workflow(val: Any) -> Optional[Dict[str, Any]]:
+        wf = _ensure_dict_from_json(val)
+        if isinstance(wf, dict) and wf.get("nodes"):
+            return wf
+        return None
+
+    # Top-level
+    wf = _maybe_workflow(info.get("workflow"))
+    if wf:
+        return wf
+
+    extra = info.get("extra_pnginfo")
+    if isinstance(extra, str):
+        try:
+            extra = json.loads(extra)
+        except Exception:
+            extra = None
+
+    if isinstance(extra, dict):
+        wf = _maybe_workflow(extra.get("workflow"))
+        if wf:
+            return wf
+    if isinstance(extra, list):
+        for item in extra:
+            if not isinstance(item, dict):
+                continue
+            wf = _maybe_workflow(item.get("workflow"))
+            if wf:
+                return wf
+    return None
+
+
+# --------- Extraction sampler-centric ---------
+
+
+def _pick_sampler_node(prompt_graph: Dict[str, Any]) -> Optional[str]:
+    """
+    Sélectionne UN sampler dans le graph.
+    Stratégie simple et déterministe:
+    - on prend le sampler avec l'id numérique le plus grand.
+    """
+    sampler_ids: List[int] = []
+
+    for node_id, node in prompt_graph.items():
+        node_type = (node.get("class_type") or "").lower()
+        if node_type in SAMPLER_CLASSES or "sampler" in node_type:
+            try:
+                sampler_ids.append(int(node_id))
+            except Exception:
+                # si l'id n'est pas un int, on ignore
+                continue
+
+    if not sampler_ids:
+        return None
+
+    best_id = max(sampler_ids)
+    return str(best_id)
+
+
+def _extract_clip_text(prompt_graph: Dict[str, Any], node_id: Optional[str]) -> Optional[str]:
+    """
+    Récupère le texte d'un node de type encodeur de texte (CLIP / SDXL / Flux, etc.).
+    On reste générique: on regarde les champs 'text', 'text_g', 'text_l'.
+    """
+    if node_id is None:
+        return None
+
+    node = prompt_graph.get(str(node_id))
+    if not node:
+        return None
+
+    inputs: Dict[str, Any] = node.get("inputs", {}) or {}
+
+    texts: List[str] = []
+    for key in ("text", "text_g", "text_l"):
+        v = inputs.get(key)
+        if isinstance(v, str) and v.strip():
+            texts.append(v.strip())
+
+    if not texts:
+        return None
+
+    # jointure simple; tu peux raffiner (multi-lignes, etc.)
+    return " | ".join(texts)
+
+
+def _collect_all_texts(prompt_graph: Dict[str, Any]) -> Dict[str, List[str]]:
+    """
+    Collecte tous les textes présents dans les nodes potentiels (encodeurs, prompts divers).
+    Essaie de séparer positif/négatif selon class_type/title.
+    """
+    pos_list: List[str] = []
+    neg_list: List[str] = []
+
+    for node_id, node in prompt_graph.items():
+        if not isinstance(node, dict):
+            continue
+        ctype = str(node.get("class_type") or "").lower()
+        title = str(node.get("title") or "").lower()
+        inputs = node.get("inputs") or {}
+        if not isinstance(inputs, dict):
+            continue
+
+        text_keywords = [
+            "cliptextencode",
+            "clip_text_encode",
+            "textencode",
+            "prompt",
+            "t5textencode",
+            "llama",
+            "qwen",
+            "gpt",
+            "textencoder",
+            "customtext",
+            "text_node",
+            "text",  # generic text node name
+        ]
+        is_text_node = any(kw in ctype for kw in text_keywords)
+        if not is_text_node:
+            continue
+
+        texts = []
+        for key in ("text", "text_g", "text_l", "prompt", "positive", "negative"):
+            v = inputs.get(key)
+            if isinstance(v, str) and v.strip():
+                texts.append(v.strip())
+
+        if not texts:
+            continue
+
+        is_neg = "negative" in title or "neg" in title or "negative" in ctype
+        if is_neg:
+            neg_list.extend(texts)
+        else:
+            pos_list.extend(texts)
+
+    def _dedup(seq: List[str]) -> List[str]:
+        out: List[str] = []
+        seen = set()
+        for x in seq:
+            if x in seen:
+                continue
+            seen.add(x)
+            out.append(x)
+        return out
+
+    return {"positive": _dedup(pos_list), "negative": _dedup(neg_list)}
+
+
+def _walk_model_chain(
+    prompt_graph: Dict[str, Any],
+    start_id: Optional[str],
+) -> Tuple[Optional[str], Optional[str], List[Dict[str, Any]]]:
+    """
+    Remonte la chaîne 'model' à partir du sampler pour récupérer:
+    - model_name  : ckpt / diffusion model / unet
+    - vae_name    : si dispo (CheckpointLoaderSimple)
+    - loras       : liste [{name, strength_model, strength_clip}, ...]
+
+    On suit uniquement la branche 'model' (ce qui est raccord avec ta logique).
+    """
+    model_name: Optional[str] = None
+    vae_name: Optional[str] = None
+    loras: List[Dict[str, Any]] = []
+    visited: Set[str] = set()
+
+    current_id = str(start_id) if start_id is not None else None
+
+    while current_id and current_id not in visited:
+        visited.add(current_id)
+        node = prompt_graph.get(current_id)
+        if not node:
+            break
+
+        node_type = node.get("class_type")
+        inputs: Dict[str, Any] = node.get("inputs", {}) or {}
+
+        # --- Checkpoints ---
+        if node_type in CHECKPOINT_LOADER_CLASSES:
+            ckpt = inputs.get("ckpt_name") or inputs.get("ckpt_file")
+            if isinstance(ckpt, str):
+                model_name = ckpt
+            vae = inputs.get("vae_name")
+            if isinstance(vae, str):
+                vae_name = vae
+            break
+
+        # --- UNET / Diffusion model loaders ---
+        if node_type in UNET_LOADER_CLASSES:
+            unet = inputs.get("unet_name") or inputs.get("model_name")
+            if isinstance(unet, str):
+                model_name = unet
+            break
+
+        if node_type in DIFFUSERS_LOADER_CLASSES:
+            m = inputs.get("model_path") or inputs.get("model_name")
+            if isinstance(m, str):
+                model_name = m
+            break
+
+        # --- LoRA chain ---
+        if node_type in LORA_CLASSES:
+            lora_name = inputs.get("lora_name")
+            if isinstance(lora_name, str):
+                l_entry = {
+                    "name": lora_name,
+                    "strength_model": inputs.get("strength_model"),
+                    "strength_clip": inputs.get("strength_clip"),
+                }
+                loras.append(l_entry)
+
+            next_id = _extract_link_node_id(inputs.get("model"))
+            if next_id is None:
+                break
+            current_id = next_id
+            continue
+
+        # --- Autres maillons de la chaîne 'model' ---
+        next_id = _extract_link_node_id(inputs.get("model"))
+        if next_id is None:
+            break
+        current_id = next_id
+
+    return model_name, vae_name, loras
+
+
+def extract_generation_params_from_prompt_graph(
+    prompt_graph: Dict[str, Any],
+    raw_workflow: Optional[Dict[str, Any]] = None,
+    *,
+    reconstruct_allowed: bool = False,
+) -> Dict[str, Any]:
+    """
+    Extraction centrale des paramÇùtres de gǸnǸration depuis un graph de prompt (map id -> node).
+
+    - `raw_workflow` peut Ǧtre fourni pour le renvoyer tel quel.
+    - si `reconstruct_allowed` est True et qu'aucun workflow n'est fourni, on reconstruit un workflow minimal.
+    """
+    if not prompt_graph:
+        return {}
+
+    sampler_id = _pick_sampler_node(prompt_graph)
+
+    if sampler_id is None:
+        collected = _collect_all_texts(prompt_graph)
+        wf_fallback = None
+        if isinstance(raw_workflow, dict):
+            wf_fallback = raw_workflow
+        elif reconstruct_allowed:
+            wf_fallback = _prompt_graph_to_workflow(prompt_graph)
+        return {
+            "positive_prompt": " | ".join(collected.get("positive") or []),
+            "negative_prompt": " | ".join(collected.get("negative") or []),
+            "seed": None,
+            "steps": None,
+            "cfg": None,
+            "sampler_name": None,
+            "scheduler": None,
+            "model": None,
+            "vae": None,
+            "loras": [],
+            "workflow": wf_fallback,
+        }
+
+    sampler = prompt_graph.get(sampler_id, {})
+    inputs: Dict[str, Any] = sampler.get("inputs", {}) or {}
+
+    # seed / cfg / steps / sampler / scheduler
+    seed = inputs.get("seed") or inputs.get("noise_seed")
+    cfg = (
+        inputs.get("cfg")
+        or inputs.get("cfg_scale")
+        or inputs.get("guidance")
+        or inputs.get("guidance_scale")
+    )
+    steps = inputs.get("steps")
+    sampler_name = inputs.get("sampler_name") or inputs.get("sampler")
+    scheduler = inputs.get("scheduler")
+
+    # Prompts
+    pos_input = inputs.get("positive")
+    neg_input = inputs.get("negative")
+    pos_id = _extract_link_node_id(pos_input)
+    neg_id = _extract_link_node_id(neg_input)
+    if pos_id is None and isinstance(pos_input, str):
+        positive_prompt = pos_input
+    else:
+        positive_prompt = _extract_clip_text(prompt_graph, pos_id)
+    if neg_id is None and isinstance(neg_input, str):
+        negative_prompt = neg_input
+    else:
+        negative_prompt = _extract_clip_text(prompt_graph, neg_id)
+
+    # Fallback: collect any text nodes if prompts are missing
+    if not positive_prompt or not negative_prompt:
+        collected = _collect_all_texts(prompt_graph)
+        if not positive_prompt and collected.get("positive"):
+            positive_prompt = " | ".join(collected["positive"])
+        if not negative_prompt and collected.get("negative"):
+            negative_prompt = " | ".join(collected["negative"])
+
+    # ChaÇ©ne modÇùle
+    model_link_id = (
+        _extract_link_node_id(inputs.get("model"))
+        or _extract_link_node_id(inputs.get("guider"))
+    )
+    model_name, vae_name, loras = _walk_model_chain(prompt_graph, model_link_id)
+
+    if isinstance(raw_workflow, dict):
+        workflow = raw_workflow
+    elif reconstruct_allowed:
+        workflow = _prompt_graph_to_workflow(prompt_graph)
+    else:
+        workflow = None
+
+    result = {
+        "positive_prompt": positive_prompt,
+        "negative_prompt": negative_prompt,
+        "seed": seed,
+        "steps": steps,
+        "cfg": cfg,
+        "sampler_name": sampler_name,
+        "scheduler": scheduler,
+        "model": model_name,
+        "vae": vae_name,
+        "loras": loras,
+        "workflow": workflow,
+    }
+    return result
+
+
+def extract_generation_params_from_workflow(workflow: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    Variante qui prend un workflow ComfyUI brut (dict avec 'nodes': [...]).
+    """
+    prompt_graph = _normalize_workflow_to_prompt_graph(workflow)
+    if prompt_graph is None:
+        return {}
+    return extract_generation_params_from_prompt_graph(
+        prompt_graph, raw_workflow=workflow, reconstruct_allowed=False
+    )
+
+
+def extract_generation_params_from_png(path: str | Path) -> Dict[str, Any]:
+    """
+    Lecture de toutes les infos de génération à partir d'un output ComfyUI.
+
+    - Si path est un MP4, on cherche le PNG sibling (même nom, extension .png).
+    - On lit le graph 'prompt' depuis les metadata du PNG.
+    - On choisit un sampler (KSampler / SamplerCustom / WAN...).
+    - On remonte la chaîne 'model' et les branches positive/negative.
+    """
+    p = Path(path)
+
+    original_ext = p.suffix.lower()
+
+    # MP4 -> PNG sibling
+    if original_ext == ".mp4":
+        png_sibling = p.with_suffix(".png")
+        target = png_sibling
+    else:
+        target = p
+
+    if not target.exists():
+        return {}
+
+    raw_workflow = load_raw_workflow_from_png(target)
+    prompt_graph = load_prompt_graph_from_png(target)
+    if prompt_graph is None and load_metadata is not None:
+        try:
+            side_meta = load_metadata(str(target))
+            if isinstance(side_meta, dict):
+                # accepte 'prompt' ou 'workflow' dans le sidecar
+                prompt = _ensure_dict_from_json(side_meta.get("prompt"))
+                if prompt is not None:
+                    prompt_graph = prompt
+                else:
+                    wf_raw = _ensure_dict_from_json(side_meta.get("workflow"))
+                    if wf_raw is not None:
+                        prompt_graph = _normalize_workflow_to_prompt_graph(wf_raw)
+                        if raw_workflow is None and isinstance(wf_raw, dict):
+                            raw_workflow = wf_raw
+        except Exception:
+            prompt_graph = None
+    if not prompt_graph:
+        return {}
+
+    reconstruct_allowed = original_ext != ".png"
+    return extract_generation_params_from_prompt_graph(
+        prompt_graph,
+        raw_workflow=raw_workflow,
+        reconstruct_allowed=reconstruct_allowed,
+    )
diff --git a/server/metadata.py b/server/metadata.py
new file mode 100644
index 00000000..014759b9
--- /dev/null
+++ b/server/metadata.py
@@ -0,0 +1,40 @@
+from __future__ import annotations
+import logging
+from typing import Dict
+
+from .utils import load_metadata, save_metadata, apply_system_metadata
+
+logger = logging.getLogger(__name__)
+
+
+def get_metadata(path: str) -> Dict:
+    """Lecture simple du sidecar JSON ou fallback."""
+    return load_metadata(path)
+
+
+def update_metadata(path: str, updates: dict) -> Dict:
+    """Mise à jour du sidecar JSON."""
+    meta = load_metadata(path)
+    meta.update(updates or {})
+    save_metadata(path, meta)
+    return meta
+
+
+def update_metadata_with_windows(path: str, updates: dict) -> Dict:
+    """
+    Met à jour les métadonnées locales + Windows (rating/tags).
+    Fonction principale utilisée par routes.py.
+    """
+    meta = load_metadata(path)
+    meta.update(updates or {})
+
+    if isinstance(updates, dict) and ("rating" in updates or "tags" in updates):
+        applied = apply_system_metadata(path, updates.get("rating"), updates.get("tags") or [])
+        meta["rating"] = applied.get("rating", meta.get("rating", 0))
+        meta["tags"] = applied.get("tags", meta.get("tags", []))
+
+    save_metadata(path, meta)
+    return meta
+
+
+__all__ = ["get_metadata", "update_metadata", "update_metadata_with_windows"]
diff --git a/server/routes.py b/server/routes.py
new file mode 100644
index 00000000..7e23f35b
--- /dev/null
+++ b/server/routes.py
@@ -0,0 +1,549 @@
+import os
+import subprocess
+import asyncio
+import time  # ← AJOUT
+import threading  # ← AJOUT
+from pathlib import Path
+from typing import Any, Dict, List
+
+try:
+    from send2trash import send2trash  # type: ignore
+except Exception:
+    send2trash = None
+
+from aiohttp import web
+
+import folder_paths
+from server import PromptServer
+from .utils import (
+    IMAGE_EXTS,
+    VIDEO_EXTS,
+    AUDIO_EXTS,
+    MODEL3D_EXTS,
+    classify_ext,
+    get_system_metadata,
+    load_metadata,
+)
+from .metadata import update_metadata_with_windows
+from .generation_metadata import extract_generation_params_from_png
+from .config import OUTPUT_ROOT
+
+# ---------------------------------------------------------------------------
+# In-memory cache pour le listing des outputs
+# Évite de refaire un os.walk complet à chaque auto-refresh.
+# ---------------------------------------------------------------------------
+
+_FILE_CACHE: List[Dict[str, Any]] = []
+_LAST_SCAN_TS: float = 0.0
+_LAST_FOLDER_MTIME: float = 0.0
+
+try:
+    _SCAN_MIN_INTERVAL = float(os.environ.get("MJR_SCAN_MIN_INTERVAL", "5.0"))
+except Exception:
+    _SCAN_MIN_INTERVAL = 5.0  # fallback 5s si variable mal formée / absente
+
+_CACHE_LOCK = threading.Lock()
+
+
+def _folder_changed(root_path: Path) -> bool:
+    """Vérifie rapidement si le dossier a changé (modification de fichier ou ajout)."""
+    try:
+        # Sur Linux/Mac, mtime du dossier change quand on ajoute/supprime un fichier.
+        # Sur Windows, c'est moins fiable pour les sous-dossiers, mais ça aide.
+        current_mtime = root_path.stat().st_mtime
+        global _LAST_FOLDER_MTIME
+        if current_mtime != _LAST_FOLDER_MTIME:
+          _LAST_FOLDER_MTIME = current_mtime
+          return True
+        return False
+    except:
+        return True
+
+
+def _get_output_root() -> Path:
+    """Utilise le dossier d'output ComfyUI (respecte --output-directory)."""
+    return Path(folder_paths.get_output_directory()).resolve()
+
+
+def _metadata_target(path: Path, kind: str) -> Path:
+    """
+    Pour les vidéos/audio/3D, on tente un PNG jumeau pour récupérer les métadonnées.
+    """
+    if kind != "image":
+        sibling = path.with_suffix(".png")
+        if sibling.exists():
+            try:
+                sibling.relative_to(_get_output_root())
+                return sibling
+            except ValueError:
+                pass
+    return path
+
+
+def _open_in_explorer(path: Path) -> bool:
+    """Ouvre l'explorateur Windows en sélectionnant le fichier."""
+    try:
+        # Tentative 1 : forcer une nouvelle fenêtre (et focus) via "start".
+        start_cmd = f'start "" explorer.exe /select,"{str(path)}"'
+        res_start = subprocess.run(
+            start_cmd,
+            shell=True,
+            check=False,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        )
+        if res_start.returncode in (0, 1):
+            return True
+
+        # Tentative 2 : forme la plus compatible : chaîne avec /select,"<path>"
+        cmd = f'explorer.exe /select,"{str(path)}"'
+        res = subprocess.run(
+            cmd,
+            shell=True,  # explorer.exe requiert souvent le parsing shell pour /select,
+            check=False,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        )
+        if res.returncode in (0, 1):
+            return True
+
+        # Tentative 3 : mode argument list (certains environnements)
+        res2 = subprocess.run(
+            ["explorer.exe", "/select,", str(path)],
+            shell=False,
+            check=False,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        )
+        return res2.returncode in (0, 1)
+    except Exception:
+        return False
+
+
+def _format_size(num_bytes: int) -> str:
+    if not isinstance(num_bytes, (int, float)) or num_bytes < 0:
+        return ""
+    units = ["B", "KB", "MB", "GB", "TB"]
+    value = float(num_bytes)
+    i = 0
+    while value >= 1024.0 and i < len(units) - 1:
+        value /= 1024.0
+        i += 1
+    if value >= 10:
+        return f"{value:.1f} {units[i]}"
+    return f"{value:.2f} {units[i]}"
+
+
+def _format_date(ts: float) -> str:
+    from datetime import datetime
+
+    try:
+        return datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M")
+    except Exception:
+        return ""
+
+
+def _build_view_url(filename: str, subfolder: str, kind: str) -> str:
+    """
+    Construit l’URL /view standard de ComfyUI.
+    type=output fonctionne pour image/vidéo/audio/models.
+    """
+    from urllib.parse import urlencode
+
+    params = {"filename": filename, "type": "output"}
+    if subfolder:
+        params["subfolder"] = subfolder
+    return f"/view?{urlencode(params)}"
+
+
+def _scan_outputs() -> List[Dict[str, Any]]:
+    """
+    Parcourt le dossier d'outputs et renvoie une liste de fichiers
+    (listing rapide, sans lecture des metadonnees Exif/Windows).
+    """
+    root = _get_output_root()
+    files: List[Dict[str, Any]] = []
+
+    if not root.exists():
+        return files
+
+    for dirpath, dirnames, filenames in os.walk(root):
+        dirnames[:] = [d for d in dirnames if not d.startswith(".")]
+
+        for name in filenames:
+            lower = name.lower()
+            kind = classify_ext(lower)
+            if kind == "other":
+                continue
+
+            full = Path(dirpath) / name
+            try:
+                stat = full.stat()
+            except OSError:
+                continue
+
+            rel = full.relative_to(root)
+            subfolder = str(rel.parent) if rel.parent != Path(".") else ""
+
+            mtime = stat.st_mtime
+            ext = (Path(name).suffix[1:] or "").upper()
+
+            item: Dict[str, Any] = {
+                "filename": name,
+                "name": name,
+                "subfolder": subfolder,
+                "relpath": str(rel).replace("\\", "/"),
+                "mtime": mtime,
+                "date": _format_date(mtime),
+                "size": stat.st_size,
+                "size_readable": _format_size(stat.st_size),
+                "kind": kind,
+                "ext": ext,
+                "url": _build_view_url(name, subfolder, kind),
+            }
+
+            files.append(item)
+
+    files.sort(key=lambda f: f["mtime"], reverse=True)
+    return files
+
+
+def _scan_outputs_cached(force: bool = False) -> List[Dict[str, Any]]:
+    """
+    Wrapper autour de _scan_outputs avec cache mémoire.
+    Évite de refaire un scan disque complet à chaque auto-refresh.
+    """
+    global _FILE_CACHE, _LAST_SCAN_TS
+
+    now = time.time()
+    # Fast path : cache encore "frais" et pas de force → on renvoie directement
+    if not force and _FILE_CACHE and (now - _LAST_SCAN_TS) < _SCAN_MIN_INTERVAL:
+        # On renvoie une copie superficielle pour éviter que l'appelant ne modifie
+        # la liste interne par accident.
+        return list(_FILE_CACHE)
+
+    # On prend un verrou pour éviter les scans concurrents
+    with _CACHE_LOCK:
+        now = time.time()
+        if not force and _FILE_CACHE and (now - _LAST_SCAN_TS) < _SCAN_MIN_INTERVAL:
+            return list(_FILE_CACHE)
+
+        # Fast check dossier : si rien n'a changé et on a déjà un cache, on renvoie direct
+        root = _get_output_root()
+        if not force and _FILE_CACHE and not _folder_changed(root):
+            return list(_FILE_CACHE)
+
+        files = _scan_outputs()
+        _FILE_CACHE = files
+        _LAST_SCAN_TS = now
+        return list(_FILE_CACHE)
+
+
+async def _scan_outputs_async(force: bool = False) -> List[Dict[str, Any]]:
+    """Exécute le scan (éventuellement caché) dans un thread pour ne pas bloquer l'event loop."""
+    loop = asyncio.get_running_loop()
+    return await loop.run_in_executor(None, _scan_outputs_cached, force)
+
+
+@PromptServer.instance.routes.get("/mjr/filemanager/files")
+async def list_files(request: web.Request) -> web.Response:
+    # force=1 / true / yes / force → on ignore le cache et on rescane vraiment
+    force_param = (request.query.get("force") or "").lower()
+    force = force_param in ("1", "true", "yes", "force")
+
+    files = await _scan_outputs_async(force=force)
+    return web.json_response({"files": files})
+
+
+@PromptServer.instance.routes.post("/mjr/filemanager/metadata/batch")
+async def batch_metadata(request: web.Request) -> web.Response:
+    """
+    Charge rating/tags de faÇßon paresseuse pour une liste de fichiers.
+    """
+    root = _get_output_root()
+
+    try:
+        payload = await request.json()
+    except Exception:
+        return web.json_response({"ok": False, "error": "Invalid JSON"}, status=400)
+
+    items = payload.get("items") or []
+
+    loop = asyncio.get_running_loop()
+
+    def _fetch_batch():
+        results: List[Dict[str, Any]] = []
+        errors: List[Dict[str, Any]] = []
+        for item in items:
+            filename = (item or {}).get("filename")
+            subfolder = (item or {}).get("subfolder", "")
+
+            if not filename:
+                errors.append({"filename": filename, "error": "Missing filename"})
+                continue
+
+            target = (root / subfolder / filename).resolve()
+            try:
+                target.relative_to(root)
+            except ValueError:
+                continue
+
+            if not target.exists():
+                continue
+
+            kind = classify_ext(filename.lower())
+            meta_target = _metadata_target(target, kind)
+            sys_meta = get_system_metadata(str(meta_target))
+            json_meta = load_metadata(str(meta_target))
+
+            results.append(
+                {
+                    "filename": filename,
+                    "subfolder": subfolder,
+                    "rating": sys_meta.get("rating") or json_meta.get("rating", 0),
+                    "tags": sys_meta.get("tags") or json_meta.get("tags", []),
+                }
+            )
+        return results, errors
+
+    results, errors = await loop.run_in_executor(None, _fetch_batch)
+
+    return web.json_response({"ok": True, "metadatas": results, "errors": errors})
+
+
+@PromptServer.instance.routes.post("/mjr/filemanager/delete")
+async def delete_files(request: web.Request) -> web.Response:
+    """
+    Supprime une liste de fichiers dans le dossier output.
+
+    Body JSON:
+    {
+      "items": [
+        {"subfolder": "2025-12-03/flux", "filename": "ComfyUI_00001_.png"},
+        ...
+      ]
+    }
+    """
+    root = _get_output_root()
+
+    try:
+        payload = await request.json()
+    except Exception:
+        return web.json_response(
+            {"ok": False, "error": "Invalid JSON body"}, status=400
+        )
+
+    items = payload.get("items") or []
+    deleted = []
+    errors = []
+
+    for item in items:
+        filename = (item or {}).get("filename")
+        subfolder = (item or {}).get("subfolder", "")
+
+        if not filename:
+            continue
+
+        candidate = (root / subfolder / filename).resolve()
+
+        # sécurité: pas de delete hors du dossier d'output
+        try:
+            candidate.relative_to(root)
+        except ValueError:
+            errors.append(
+                {
+                    "filename": filename,
+                    "subfolder": subfolder,
+                    "error": "Outside output directory",
+                }
+            )
+            continue
+
+        if not candidate.exists():
+            errors.append(
+                {"filename": filename, "subfolder": subfolder, "error": "Not found"}
+            )
+            continue
+
+        try:
+            if send2trash:
+                send2trash(str(candidate))
+            else:
+                candidate.unlink()
+            deleted.append({"filename": filename, "subfolder": subfolder})
+        except Exception as exc:
+            errors.append(
+                {"filename": filename, "subfolder": subfolder, "error": str(exc)}
+            )
+
+    return web.json_response({"ok": True, "deleted": deleted, "errors": errors})
+
+
+@PromptServer.instance.routes.post("/mjr/filemanager/open_explorer")
+async def open_explorer(request: web.Request) -> web.Response:
+    """
+    Ouvre l'explorateur Windows et sélectionne le fichier demandé.
+    """
+    root = _get_output_root()
+    try:
+        payload = await request.json()
+    except Exception:
+        return web.json_response({"ok": False, "error": "Invalid JSON body"}, status=400)
+
+    filename = payload.get("filename")
+    subfolder = payload.get("subfolder", "")
+    if not filename:
+        return web.json_response({"ok": False, "error": "Missing filename"}, status=400)
+
+    target = (root / subfolder / filename).resolve()
+    try:
+        target.relative_to(root)
+    except ValueError:
+        return web.json_response({"ok": False, "error": "File is outside output directory"}, status=400)
+
+    if not target.exists():
+        return web.json_response({"ok": False, "error": "File not found"}, status=404)
+
+    if _open_in_explorer(target):
+        return web.json_response({"ok": True})
+
+    # Fallback: open folder if select fails (some Windows setups)
+    try:
+        subprocess.run(
+            ["explorer.exe", str(target.parent)],
+            shell=False,
+            check=False,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        )
+        return web.json_response(
+            {"ok": True, "warning": "Opened folder (selection may not be highlighted)"},
+            status=200,
+        )
+    except Exception as exc:
+        return web.json_response(
+            {"ok": False, "error": f"Failed to open explorer: {exc}"},
+            status=500,
+        )
+
+
+# ---------------------------------------------------------------------------
+# METADATA / WORKFLOW EXTRACTION
+# ---------------------------------------------------------------------------
+
+@PromptServer.instance.routes.get("/mjr/filemanager/metadata")
+async def get_metadata(request: web.Request) -> web.Response:
+    """
+    Endpoint minimal : paramètres de génération extraits du PNG (ou sibling PNG).
+    """
+    root = _get_output_root()
+
+    filename = request.query.get("filename")
+    subfolder = request.query.get("subfolder", "")
+
+    if not filename:
+        return web.json_response(
+            {"ok": False, "error": "missing filename"}, status=400
+        )
+
+    target = (root / subfolder / filename).resolve()
+
+    try:
+        target.relative_to(root)
+    except ValueError:
+        return web.json_response(
+            {"ok": False, "error": "File is outside output directory"}, status=400
+        )
+
+    if not target.exists():
+        return web.json_response(
+            {"ok": False, "error": "File not found"}, status=404
+        )
+
+    # Si vidéo, forcer la présence du PNG sibling
+    if target.suffix.lower() == ".mp4":
+        png_sibling = target.with_suffix(".png")
+        if not png_sibling.exists():
+            return web.json_response(
+                {"ok": False, "error": "PNG sibling not found for video"}, status=404
+            )
+
+    # Si mp4, lecture sur le PNG sibling pour refléter les infos de génération
+    if target.suffix.lower() == ".mp4":
+        png_sibling = target.with_suffix(".png")
+        meta_target = png_sibling if png_sibling.exists() else target
+    else:
+        meta_target = target
+
+    try:
+        params = extract_generation_params_from_png(meta_target)
+    except Exception as e:
+        # Empêche un 500 et remonte l'erreur côté UI pour debug
+        return web.json_response(
+            {"ok": False, "error": f"metadata parsing failed: {e}"}, status=500
+        )
+
+    return web.json_response({"ok": True, "generation": params})
+
+
+@PromptServer.instance.routes.post("/mjr/filemanager/metadata/update")
+async def update_metadata(request: web.Request) -> web.Response:
+    """
+    Met à jour les métadonnées Windows (rating + tags) pour un fichier d'output.
+    """
+    root = _get_output_root()
+
+    try:
+        payload = await request.json()
+    except Exception:
+        return web.json_response(
+            {"ok": False, "error": "Invalid JSON body"}, status=400
+        )
+
+    filename = payload.get("filename")
+    subfolder = payload.get("subfolder", "")
+    rating = payload.get("rating")
+    tags = payload.get("tags") or []
+
+    if not filename:
+        return web.json_response(
+            {"ok": False, "error": "Missing filename"}, status=400
+        )
+
+    target = (root / subfolder / filename).resolve()
+    try:
+        target.relative_to(root)
+    except ValueError:
+        return web.json_response(
+            {"ok": False, "error": "File is outside output directory"}, status=400
+        )
+
+    if not target.exists():
+        return web.json_response(
+            {"ok": False, "error": "File not found"}, status=404
+        )
+
+    try:
+        meta = update_metadata_with_windows(str(target), {"rating": rating, "tags": tags})
+
+        # Si on a un sibling PNG, on propage la note/tag pour rester aligné
+        sibling = (target.parent / (target.stem + ".png")).resolve()
+        if sibling.exists():
+            try:
+                sibling.relative_to(root)
+                update_metadata_with_windows(str(sibling), {"rating": rating, "tags": tags})
+            except Exception:
+                pass
+    except Exception as exc:
+        return web.json_response(
+            {
+                "ok": False,
+                "file": {"filename": filename, "subfolder": subfolder},
+                "error": str(exc),
+            },
+            status=500,
+        )
+
+    return web.json_response({"ok": True, "rating": meta.get("rating", 0), "tags": meta.get("tags", [])})
+
+
diff --git a/server/utils.py b/server/utils.py
new file mode 100644
index 00000000..cecc9661
--- /dev/null
+++ b/server/utils.py
@@ -0,0 +1,446 @@
+import json
+import os
+import platform
+import shutil
+import subprocess
+import unicodedata
+from typing import Dict, List, Optional
+from .config import ENABLE_JSON_SIDECAR, METADATA_EXT
+
+# Extensions gérées (partagées)
+IMAGE_EXTS = (".png", ".jpg", ".jpeg", ".webp", ".gif")
+VIDEO_EXTS = (".mp4", ".mov", ".webm", ".mkv")
+AUDIO_EXTS = (".wav", ".mp3", ".flac", ".ogg", ".m4a", ".aac")
+MODEL3D_EXTS = (".obj", ".fbx", ".glb", ".gltf", ".stl")
+WINDOWS_RATING_INDEX = 14  # fallback
+WINDOWS_TAGS_INDEX = 21    # fallback
+STAR_CHARS = ("\u2605", "\u2606", "\u22c6", "\u272d", "\u272e", "\u2730")
+RATING_KEYS = (
+    "rating",
+    "note",
+    "notation",
+    "evaluation",
+    "evaluations",
+    "etoile",
+    "etoiles",
+    "star",
+    "stars",
+)
+TAG_KEYS = ("tag", "mot-cle", "mot cle", "mots cle", "keywords", "keyword")
+
+_WIN32COM: Optional[object] = None
+_WIN32COM_CHECKED = False
+_CACHED_RATING_INDEX: Optional[int] = None
+_CACHED_TAGS_INDEX: Optional[int] = None
+_EXIFTOOL_PATH: Optional[str] = None
+_EXIFTOOL_CHECKED = False
+_META_CACHE: Dict[str, tuple[Optional[float], int, dict]] = {}
+_CACHE_EPOCH = 0
+
+
+def _normalize_label(text) -> str:
+    try:
+        normalized = unicodedata.normalize("NFKD", str(text or ""))
+        normalized = "".join(ch for ch in normalized if not unicodedata.combining(ch))
+        return normalized.lower().strip()
+    except Exception:
+        return str(text or "").lower().strip()
+
+
+def _get_mtime_safe(path: str) -> Optional[float]:
+    try:
+        return os.path.getmtime(path)
+    except Exception:
+        return None
+
+
+def safe_import_win32com():
+    global _WIN32COM, _WIN32COM_CHECKED
+    if _WIN32COM_CHECKED:
+        return _WIN32COM
+    _WIN32COM_CHECKED = True
+    if platform.system().lower() == "windows":
+        try:
+            import win32com.client  # type: ignore
+
+            _WIN32COM = win32com.client
+        except ImportError:
+            print("pywin32 is not installed. Windows-specific metadata will not be available.")
+            _WIN32COM = None
+    return _WIN32COM
+
+
+def _get_exiftool_path() -> Optional[str]:
+    global _EXIFTOOL_PATH, _EXIFTOOL_CHECKED
+    if _EXIFTOOL_CHECKED:
+        return _EXIFTOOL_PATH
+    _EXIFTOOL_CHECKED = True
+    _EXIFTOOL_PATH = shutil.which("exiftool") or shutil.which("exiftool.exe")
+    return _EXIFTOOL_PATH
+
+
+def _parse_rating_value(raw) -> int:
+    if raw in (None, ""):
+        return 0
+    s_raw = str(raw).strip()
+    if not s_raw:
+        return 0
+
+    star_count = sum(s_raw.count(ch) for ch in STAR_CHARS)
+    if star_count:
+        return max(0, min(star_count, 5))
+
+    try:
+        numeric = float(s_raw.replace(",", ".").split()[0])
+    except Exception:
+        return 0
+
+    if numeric <= 0:
+        return 0
+    if numeric > 5:
+        if numeric <= 25:
+            return 2
+        if numeric <= 50:
+            return 3
+        if numeric <= 75:
+            return 4
+        return 5
+    return int(round(min(numeric, 5)))
+
+
+def get_windows_metadata(file_path: str) -> dict:
+    """Récupère les métadonnées de fichier via l'API Windows Shell."""
+    try:
+        win32com = safe_import_win32com()
+        if not win32com:
+            return {"rating": 0, "tags": []}
+
+        shell = win32com.Dispatch("Shell.Application")
+        folder = shell.Namespace(os.path.dirname(file_path))
+        file = folder.ParseName(os.path.basename(file_path))
+
+        rating_idx, tags_idx = _resolve_shell_indices(folder)
+
+        rating_raw = folder.GetDetailsOf(file, rating_idx)
+        tags_raw = folder.GetDetailsOf(file, tags_idx)
+
+        rating = _parse_rating_value(rating_raw)
+
+        tags = []
+        if tags_raw:
+            tag_text = str(tags_raw)
+            for sep in [",", ";"]:
+                if sep in tag_text:
+                    tags = [t.strip() for t in tag_text.split(sep) if t.strip()]
+                    break
+            else:
+                t = tag_text.strip()
+                if t:
+                    tags = [t]
+
+        return {
+            "rating": rating,
+            "tags": tags,
+        }
+    except Exception as e:
+        print(f"Error reading metadata for {file_path}: {e}")
+        return {"rating": 0, "tags": []}
+
+
+def get_exif_metadata(file_path: str) -> dict:
+    """
+    Lecture rapide via exiftool si disponible.
+    Retourne {"rating": int, "tags": List[str]} ou {} en fallback.
+    """
+    exe = _get_exiftool_path()
+    if not exe:
+        return {}
+    try:
+        cmd = [
+            exe,
+            "-n",
+            "-json",
+            "-Rating",
+            "-RatingPercent",
+            "-Subject",
+            "-Keywords",
+            "-XPKeywords",
+            file_path,
+        ]
+        out = subprocess.check_output(cmd, timeout=2)
+        data = json.loads(out)[0]
+        rating_raw = data.get("Rating") or data.get("RatingPercent")
+        rating = _parse_rating_value(rating_raw)
+
+        tags: List[str] = []
+        for key in ("Subject", "Keywords", "XPKeywords"):
+            val = data.get(key)
+            if isinstance(val, list):
+                tags.extend([str(v).strip() for v in val if str(v).strip()])
+            elif isinstance(val, str):
+                tags.extend([t.strip() for t in val.split(",") if t.strip()])
+
+        if tags:
+            tags = sorted(set(tags))
+
+        return {"rating": rating, "tags": tags}
+    except Exception:
+        return {}
+
+
+def set_exif_metadata(file_path: str, rating: int, tags: list) -> bool:
+    """
+    Écrit Rating/Keywords via exiftool si dispo. Retourne True si OK.
+    """
+    exe = _get_exiftool_path()
+    if not exe:
+        return False
+
+    global _CACHE_EPOCH
+    _CACHE_EPOCH += 1
+    epoch = _CACHE_EPOCH
+    original_mtime = _get_mtime_safe(file_path)
+
+    try:
+        r = max(0, min(int(rating), 5))
+        win_r = {1: 1, 2: 25, 3: 50, 4: 75, 5: 99}.get(r, r if r <= 5 else 99)
+
+        cmd = [
+            exe,
+            "-overwrite_original",
+            f"-xmp:rating={r}",
+            f"-rating={r}",
+            f"-ratingpercent={win_r}",
+            "-Subject=",
+            "-Keywords=",
+            "-XPKeywords=",
+        ]
+
+        clean_tags = [str(t).strip() for t in tags if str(t).strip()]
+        if clean_tags:
+            for t in clean_tags:
+                cmd.append(f"-xmp:subject+={t}")
+                cmd.append(f"-iptc:keywords+={t}")
+            # XPKeywords attend une chaîne séparée par ; côté Windows
+            cmd.append(f"-xpkeywords={'; '.join(clean_tags)}")
+
+        cmd.append(file_path)
+        subprocess.run(cmd, check=True, capture_output=True, timeout=3)
+
+        # Restaure l'mtime pour éviter de réordonner le grid après une note/tag
+        if original_mtime is not None:
+            try:
+                os.utime(file_path, (original_mtime, original_mtime))
+            except Exception:
+                pass
+
+        # Met à jour le cache avec l'epoch courant
+        _META_CACHE[file_path] = (
+            original_mtime if original_mtime is not None else _get_mtime_safe(file_path),
+            epoch,
+            {"rating": rating, "tags": tags},
+        )
+        return True
+    except Exception:
+        return False
+
+
+def set_windows_metadata(file_path: str, rating: int, tags: list) -> bool:
+    """Écrit les métadonnées via l'API Windows Shell."""
+    try:
+        win32com = safe_import_win32com()
+        if not win32com:
+            return False
+
+        shell = win32com.Dispatch("Shell.Application")
+        folder = shell.Namespace(os.path.dirname(file_path))
+        file = folder.ParseName(os.path.basename(file_path))
+
+        rating_idx, tags_idx = _resolve_shell_indices(folder)
+
+        # Rating (scale 1-99); Windows 5-star uses 1-5, but accepting broader range
+        if rating is not None:
+            rating_val = max(0, min(99, int(rating)))
+            folder.GetDetailsOf(file, rating_idx)  # force index load
+            folder.SetDetailsOf(file, rating_idx, str(rating_val))
+
+        if tags:
+            folder.GetDetailsOf(file, tags_idx)
+            folder.SetDetailsOf(file, tags_idx, "; ".join(tags))
+        elif tags == []:
+            folder.GetDetailsOf(file, tags_idx)
+            folder.SetDetailsOf(file, tags_idx, "")
+
+        return True
+    except Exception:
+        return False
+
+
+def _resolve_shell_indices(folder) -> (int, int):
+    """
+    Tente de localiser les colonnes "Rating"/"Tags" en fonction de la locale.
+    Retourne un tuple (rating_idx, tags_idx) avec fallback sur les constantes.
+    """
+    global _CACHED_RATING_INDEX, _CACHED_TAGS_INDEX
+    if _CACHED_RATING_INDEX is not None and _CACHED_TAGS_INDEX is not None:
+        return _CACHED_RATING_INDEX, _CACHED_TAGS_INDEX
+
+    rating_idx = WINDOWS_RATING_INDEX
+    tags_idx = WINDOWS_TAGS_INDEX
+
+    try:
+        for i in range(0, 256):
+            col_name = _normalize_label(folder.GetDetailsOf(None, i))
+            if not col_name:
+                continue
+            if _CACHED_RATING_INDEX is None and any(key in col_name for key in RATING_KEYS):
+                _CACHED_RATING_INDEX = i
+            if _CACHED_TAGS_INDEX is None and any(key in col_name for key in TAG_KEYS):
+                _CACHED_TAGS_INDEX = i
+            if _CACHED_RATING_INDEX is not None and _CACHED_TAGS_INDEX is not None:
+                break
+    except Exception:
+        pass
+
+    if _CACHED_RATING_INDEX is None:
+        _CACHED_RATING_INDEX = rating_idx
+    if _CACHED_TAGS_INDEX is None:
+        _CACHED_TAGS_INDEX = tags_idx
+
+    return _CACHED_RATING_INDEX, _CACHED_TAGS_INDEX
+
+
+def apply_windows_metadata(file_path: str, rating, tags: list) -> dict:
+    """
+    Applique une note et des tags via l'API Windows en gérant les valeurs par défaut.
+    Retourne {"rating": int, "tags": List[str]}.
+    """
+    try:
+        rating_int = int(rating) if rating is not None else 0
+    except Exception:
+        rating_int = 0
+    tags_list: List[str] = []
+    if isinstance(tags, list):
+        tags_list = [str(t) for t in tags if str(t).strip()]
+    set_windows_metadata(file_path, rating_int, tags_list)
+    return {"rating": rating_int, "tags": tags_list}
+
+
+def update_metadata_with_windows(file_path: str, updates: dict) -> dict:
+    """
+    Met à jour rating/tags côté Windows + sidecar (si activé).
+    Retourne les métadonnées finales.
+    """
+    current_rating = 0
+    current_tags: List[str] = []
+    try:
+        existing = load_metadata(file_path)
+        current_rating = existing.get("rating", 0)
+        current_tags = existing.get("tags", [])
+    except Exception:
+        pass
+
+    rating = updates.get("rating", current_rating)
+    tags = updates.get("tags", current_tags)
+
+    ok_win = set_windows_metadata(file_path, rating, tags)
+    ok_exif = set_exif_metadata(file_path, rating, tags)
+
+    if ENABLE_JSON_SIDECAR:
+        save_metadata(file_path, {"rating": rating, "tags": tags})
+
+    effective_rating = rating if (ok_win or ok_exif) else current_rating
+    _META_CACHE[file_path] = (_get_mtime_safe(file_path), _CACHE_EPOCH, {"rating": effective_rating, "tags": tags})
+    return {"rating": effective_rating, "tags": tags}
+
+
+def apply_system_metadata(file_path: str, rating, tags: list) -> dict:
+    """
+    Essaie d'abord exiftool (si installé), sinon Windows Shell.
+    """
+    global _CACHE_EPOCH
+    _CACHE_EPOCH += 1
+    epoch = _CACHE_EPOCH
+    try:
+        rating_int = int(rating) if rating is not None else 0
+    except Exception:
+        rating_int = 0
+    tags_list: List[str] = []
+    if isinstance(tags, list):
+        tags_list = [str(t) for t in tags if str(t).strip()]
+
+    # exiftool (meilleure compat Windows + autres OS)
+    if set_exif_metadata(file_path, rating_int, tags_list):
+        mtime = _get_mtime_safe(file_path)
+        _META_CACHE[file_path] = (mtime, epoch, {"rating": rating_int, "tags": tags_list})
+        return {"rating": rating_int, "tags": tags_list}
+
+    # fallback shell API
+    meta = apply_windows_metadata(file_path, rating_int, tags_list)
+    mtime = _get_mtime_safe(file_path)
+    _META_CACHE[file_path] = (mtime, epoch, meta)
+    return meta
+
+
+def get_system_metadata(file_path: str) -> dict:
+    """
+    Préfère les métadonnées ExifTool (Rating/Keywords), sinon API Windows.
+    """
+    mtime = _get_mtime_safe(file_path)
+    cached = _META_CACHE.get(file_path)
+    if cached and cached[0] == mtime and cached[1] == _CACHE_EPOCH:
+        return dict(cached[2])
+
+    meta = get_exif_metadata(file_path)
+    if meta.get("rating") or meta.get("tags"):
+        _META_CACHE[file_path] = (mtime, _CACHE_EPOCH, meta)
+        return meta
+
+    meta = get_windows_metadata(file_path)
+    _META_CACHE[file_path] = (mtime, _CACHE_EPOCH, meta)
+    return meta
+
+
+def classify_ext(lower_name: str) -> str:
+    """Retourne image / video / audio / model3d / other."""
+    if lower_name.endswith(IMAGE_EXTS):
+        return "image"
+    if lower_name.endswith(VIDEO_EXTS):
+        return "video"
+    if lower_name.endswith(AUDIO_EXTS):
+        return "audio"
+    if lower_name.endswith(MODEL3D_EXTS):
+        return "model3d"
+    return "other"
+
+def ensure_dir(path):
+    if not os.path.exists(path):
+        os.makedirs(path)
+
+def is_image(fname: str):
+    f = fname.lower()
+    return f.endswith(IMAGE_EXTS)
+
+def metadata_path(image_path: str):
+    if not ENABLE_JSON_SIDECAR:
+        return None
+    return image_path + METADATA_EXT
+
+def load_metadata(image_path):
+    meta_file = metadata_path(image_path)
+    if not meta_file:
+        return {}
+    if not os.path.exists(meta_file):
+        return {}
+    try:
+        with open(meta_file, "r", encoding="utf-8") as f:
+            return json.load(f)
+    except Exception:
+        return {}
+
+def save_metadata(image_path, meta: dict):
+    meta_file = metadata_path(image_path)
+    if not meta_file:
+        return
+    with open(meta_file, "w", encoding="utf-8") as f:
+        json.dump(meta, f, indent=2, ensure_ascii=False)
diff --git a/server/version.py b/server/version.py
new file mode 100644
index 00000000..204a96fe
--- /dev/null
+++ b/server/version.py
@@ -0,0 +1 @@
+VERSION = '2.0.0'