rpa_vision_v3/tools/backfill_anchors_r1.py

#!/usr/bin/env python3
"""
Backfill des ancres visuelles pour les workflows importés par le pont R1.

Contexte (diagnostic + plan 2026-07-03) : 62 workflows `source='learned_import'`
sont déjà en DB VWB avec le crop `_anchor_image_base64` resté dans
`Step.parameters_json` et `anchor_id = NULL` → le VWB affiche « Ancre requise »
alors que l'image est là (verdict H1 : ~69 % des clics ont le crop en base).
Corriger le flux futur (helper R1 + route HTTP) ne répare PAS l'existant.

Ce script matérialise l'ancre pour chaque step réparable, via le MÊME helper
partagé `materialize_anchor_from_b64` (zéro logique dupliquée).

Sécurité / garde-fous :
  * `--dry-run` par défaut : n'écrit RIEN. Affiche le chemin DB résolu, le
    nombre de steps réparables (par workflow), l'estimation du nombre de PNG à
    créer et l'espace disque.
  * Filtre dur : `source='learned_import'` ET `action_type ∈ NEEDS_ANCHOR` ET
    `anchor_id IS NULL` ET `_anchor_image_base64` présent dans parameters_json.
    Les workflows `source='manual'` ne sont JAMAIS touchés.
  * Idempotent : un step déjà pourvu d'`anchor_id` est ignoré ; une ré-exécution
    d'`--apply` est un no-op (le helper retire aussi la clé b64 des params).
  * Cible DB résolue depuis la config runtime VWB (env `DATABASE_URL` + logique
    `instance_path` de Flask), PAS un chemin codé en dur (réserve QG R1).

⚠️ Ce script est manuel, hors flux runtime. Il NE fait PAS de backup — l'appelant
doit sauvegarder la DB avant `--apply` (copie horodatée), cf. plan §3.3.
Ne JAMAIS lancer `--apply` sans backup ni sans avoir validé le `--dry-run`.

Usage :
    python tools/backfill_anchors_r1.py                 # dry-run (défaut)
    python tools/backfill_anchors_r1.py --dry-run       # explicite
    python tools/backfill_anchors_r1.py --db /chemin/workflows.db --dry-run
    python tools/backfill_anchors_r1.py --apply         # écrit (backup requis avant)

Auteur : Dom, Claude — 2026-07-03
"""

import argparse
import base64
import json
import os
import sys
from pathlib import Path

# --- Chemins : racine projet + backend VWB (pour db.models, services.*) ---
_ROOT = Path(__file__).resolve().parent.parent
_BACKEND = _ROOT / "visual_workflow_builder" / "backend"
for p in (str(_ROOT), str(_BACKEND)):
    if p not in sys.path:
        sys.path.insert(0, p)


def resolve_vwb_db_path(explicit: str | None = None) -> Path:
    """Résout le chemin du fichier SQLite VWB depuis la config runtime.

    Ordre de priorité :
      1. `--db` explicite (échappatoire manuelle) ;
      2. `DATABASE_URL` env (même variable que `app.py`) si sqlite:/// ;
      3. défaut `app.py` = `sqlite:///workflows.db`, résolu relativement à
         `Flask.instance_path` du backend VWB — EXACTEMENT comme l'app au runtime.

    On instancie une app Flask *sonde* légère (root_path = backend) au lieu
    d'importer `app.py` (lourd : socketio, blueprints). Flask calcule alors le
    même `instance_path` que l'app réelle → même résolution de la DB.
    """
    if explicit:
        return Path(explicit).resolve()

    uri = os.getenv("DATABASE_URL", "sqlite:///workflows.db")
    if not uri.startswith("sqlite:"):
        raise SystemExit(
            f"DATABASE_URL non-SQLite ({uri!r}) : backfill SQLite uniquement. "
            "Passez --db pour un chemin explicite."
        )

    # sqlite:////abs/path (4 slashes) → chemin absolu ; sqlite:///rel → relatif.
    tail = uri[len("sqlite://"):]
    if tail.startswith("//"):  # //// → chemin absolu
        return Path(tail[1:]).resolve()

    rel = tail.lstrip("/")  # ex. "workflows.db"
    from flask import Flask

    probe = Flask("vwb_backfill_probe", root_path=str(_BACKEND))
    return (Path(probe.instance_path) / rel).resolve()


def _iter_repairable_steps(session, NEEDS_ANCHOR):
    """Génère les tuples (workflow, step, params) réparables.

    Réparable = source learned_import, action_type ∈ NEEDS_ANCHOR,
    anchor_id NULL, et `_anchor_image_base64` présent (non vide) dans params.
    """
    from db.models import Workflow, Step

    q = (
        session.query(Workflow, Step)
        .join(Step, Step.workflow_id == Workflow.id)
        .filter(Workflow.source == "learned_import")
        .filter(Step.anchor_id.is_(None))
        .filter(Step.action_type.in_(tuple(NEEDS_ANCHOR)))
        .filter(Step.parameters_json.like("%_anchor_image_base64%"))
    )
    for wf, step in q.all():
        params = step.parameters  # dict décodé
        b64 = params.get("_anchor_image_base64")
        if b64:
            yield wf, step, params


def _estimate_png_bytes(b64: str) -> int:
    """Taille approximative du PNG décodé (le fichier écrit ≈ crop optimisé)."""
    try:
        clean = b64.split(",", 1)[1] if "," in b64 else b64
        return len(base64.b64decode(clean))
    except Exception:
        return 0


def run(db_path: Path, apply: bool) -> int:
    from flask import Flask
    from db.models import db
    from services.anchor_action_types import NEEDS_ANCHOR
    from services.anchor_image_service import materialize_anchor_from_b64

    if not db_path.exists():
        raise SystemExit(f"DB introuvable : {db_path}")

    app = Flask("vwb_backfill")
    app.config["SQLALCHEMY_DATABASE_URI"] = f"sqlite:///{db_path}"
    app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
    db.init_app(app)

    mode = "APPLY (écriture)" if apply else "DRY-RUN (lecture seule)"
    print(f"=== Backfill ancres R1 — {mode} ===")
    print(f"DB résolue : {db_path}")
    print(f"action_type ∈ NEEDS_ANCHOR ({len(NEEDS_ANCHOR)}) : "
          f"{', '.join(sorted(NEEDS_ANCHOR))}")
    print()

    with app.app_context():
        session = db.session
        per_workflow: dict = {}
        total_repairable = 0
        total_est_bytes = 0

        for wf, step, params in _iter_repairable_steps(session, NEEDS_ANCHOR):
            total_repairable += 1
            total_est_bytes += _estimate_png_bytes(params["_anchor_image_base64"])
            bucket = per_workflow.setdefault(
                wf.id, {"name": wf.name, "count": 0}
            )
            bucket["count"] += 1

            if apply:
                # Le helper retire la clé b64 de params, crée VisualAnchor +
                # fichier, pose step.anchor_id. On réécrit params (b64 retiré).
                anchor_id = materialize_anchor_from_b64(
                    step, params, db_session=session, source="learned_import"
                )
                if anchor_id:
                    step.parameters = params  # persiste params sans le b64
                    session.add(step)

        if apply:
            session.commit()

        # --- Rapport ---
        print(f"Workflows learned_import concernés : {len(per_workflow)}")
        for wf_id, info in sorted(per_workflow.items(),
                                  key=lambda kv: -kv[1]["count"]):
            print(f"  - {wf_id}  ({info['name'][:48]}) : "
                  f"{info['count']} step(s) réparable(s)")
        print()
        print(f"Total steps réparables : {total_repairable}")

        if apply:
            print(f"→ {total_repairable} ancres matérialisées (VisualAnchor + PNG).")
            print("Ré-exécuter ce script = no-op (idempotent).")
        else:
            est_png_mb = total_est_bytes / (1024 * 1024)
            # Thumbnails JPEG q80 ≈ 30 % du PNG en plus (majoration prudente).
            est_total_mb = est_png_mb * 1.3
            print(f"→ PNG à créer : {total_repairable}")
            print(f"→ Espace disque estimé : ~{est_png_mb:.2f} Mo (PNG) "
                  f"+ thumbnails ≈ ~{est_total_mb:.2f} Mo total")
            print()
            print("DRY-RUN : aucune écriture. Vérifiez le chemin DB ci-dessus,")
            print("SAUVEGARDEZ la DB, puis relancez avec --apply.")

    return 0


def main(argv=None) -> int:
    parser = argparse.ArgumentParser(
        description="Backfill idempotent des ancres visuelles des workflows R1."
    )
    group = parser.add_mutually_exclusive_group()
    group.add_argument(
        "--dry-run", action="store_true", default=True,
        help="Lecture seule (défaut) : rapport sans écriture.",
    )
    group.add_argument(
        "--apply", action="store_true", default=False,
        help="Écrit les ancres (backup DB requis AVANT).",
    )
    parser.add_argument(
        "--db", default=None,
        help="Chemin explicite du fichier SQLite VWB (sinon résolu depuis config).",
    )
    args = parser.parse_args(argv)

    apply = bool(args.apply)
    db_path = resolve_vwb_db_path(args.db)
    return run(db_path, apply=apply)


if __name__ == "__main__":
    raise SystemExit(main())