Add tools/import_planning_xlsx.py : import patch_planning depuis xlsx Ayoub

tools/import_planning_xlsx.py

@@ -0,0 +1,213 @@
+"""Import planning annuel patching depuis Planning Patching 2026_ayoub.xlsx feuille Planning.
+
+Mapping colonnes feuille Planning :
+  A : domaine+env (ex Infrastructure HPROD, Peage PROD, FL Prod)
+  B : Patch N marker (cycle) OU semaine NN (ligne data)
+  C : plage dates DD/MM/YYYY ... DD/MM/YYYY OU Gel
+  D : ferie (datetime) OU Gel OU texte
+
+Structure cible table patch_planning :
+  year, week_number, week_code, week_start, week_end, cycle,
+  domain_code (FK domains), env_scope, status, note
+
+Usage :
+  python tools/import_planning_xlsx.py [chemin_fichier.xlsx]
+"""
+import os
+import sys
+import re
+import glob
+from pathlib import Path
+from datetime import date, datetime, timedelta
+
+import openpyxl
+from sqlalchemy import create_engine, text
+
+ROOT = Path(__file__).resolve().parent.parent
+DATABASE_URL = (os.getenv("DATABASE_URL_DEMO")
+                or os.getenv("DATABASE_URL")
+                or "postgresql://patchcenter:PatchCenter2026!@localhost:5432/patchcenter_db")
+
+
+def parse_label(a):
+    """Retourne liste (domain_code, env_scope) pour le libelle col A.
+
+    Un libelle peut mapper sur plusieurs domaines (ex BI + Gestion) ou
+    un scope combine (prod_pilot pour Peage HPROD / PROD Pilote).
+    """
+    if not a:
+        return []
+    lo = a.lower()
+
+    if "bi" in lo and "gestion" in lo:
+        return [("BI", "all"), ("GESTION", "all")]
+    if "peage" in lo or "p\xe9age" in lo:
+        if "pilot" in lo:
+            return [("PEA", "prod_pilot")]
+        if "hprod" in lo:
+            return [("PEA", "hprod")]
+        if "prod" in lo:
+            return [("PEA", "prod")]
+    if "infrastructure" in lo:
+        if "hprod" in lo:
+            return [("INFRASTRUC", "hprod")]
+        return [("INFRASTRUC", "prod")]
+    if "trafic" in lo:
+        if "hprod" in lo:
+            return [("trafic", "hprod")]
+        return [("trafic", "prod")]
+    if lo.startswith("fl"):
+        if "pre-prod" in lo or "pr\xe9-prod" in lo or "preprod" in lo or "pr\xe9prod" in lo:
+            return [("FL", "pilot")]
+        if "prod" in lo and "pre" not in lo and "pr\xe9" not in lo:
+            return [("FL", "prod")]
+        return [("FL", "hprod")]
+    return []
+
+
+def parse_dates(c_val, year):
+    """Parse col C. Retourne (week_start, week_end, is_freeze)."""
+    if not c_val:
+        return None, None, False
+    s = str(c_val).strip()
+    if s.lower() == "gel":
+        return None, None, True
+    m = re.search(r"(\d{2})/(\d{2})/(\d{4}).*?(\d{2})/(\d{2})/(\d{4})", s)
+    if m:
+        d1 = date(int(m.group(3)), int(m.group(2)), int(m.group(1)))
+        d2 = date(int(m.group(6)), int(m.group(5)), int(m.group(4)))
+        return d1, d2, False
+    return None, None, False
+
+
+def iso_week_dates(year, week):
+    """Fallback : dates debut/fin semaine ISO depuis year+week."""
+    jan4 = date(year, 1, 4)
+    start = jan4 - timedelta(days=jan4.isoweekday() - 1) + timedelta(weeks=week - 1)
+    return start, start + timedelta(days=6)
+
+
+def parse_note(d_val):
+    if d_val is None:
+        return None
+    if isinstance(d_val, (datetime, date)):
+        dd = d_val.date() if isinstance(d_val, datetime) else d_val
+        return f"Ferie : {dd.strftime('%d/%m/%Y')}"
+    s = str(d_val).strip()
+    if not s or s.lower() == "gel":
+        return None
+    return s
+
+
+def parse_planning(xlsx_path, year_default=2026):
+    wb = openpyxl.load_workbook(xlsx_path, data_only=True)
+    if "Planning" not in wb.sheetnames:
+        raise SystemExit(f"[ERR] Sheet Planning introuvable. Sheets: {wb.sheetnames}")
+    ws = wb["Planning"]
+
+    rows = []
+    current_cycle = None
+
+    for row in ws.iter_rows(values_only=True):
+        a = row[0] if len(row) > 0 else None
+        b = row[1] if len(row) > 1 else None
+        c = row[2] if len(row) > 2 else None
+        d = row[3] if len(row) > 3 else None
+
+        if b and re.match(r"^\s*Patch\s+\d+\s*$", str(b), re.I):
+            m = re.search(r"\d+", str(b))
+            current_cycle = int(m.group(0)) if m else None
+            continue
+
+        if not b:
+            continue
+        m = re.match(r"^\s*semaine\s+(\d+)\s*$", str(b), re.I)
+        if not m:
+            continue
+        week_number = int(m.group(1))
+        year = year_default
+        week_code = f"S{week_number:02d}"
+
+        d1, d2, is_freeze = parse_dates(c, year)
+        if not d1:
+            d1, d2 = iso_week_dates(year, week_number)
+
+        note = parse_note(d)
+        if is_freeze and note is None:
+            note = "Gel"
+
+        if is_freeze:
+            status = "freeze"
+        else:
+            status = "open" if a else "empty"
+
+        targets = parse_label(a)
+        if not targets:
+            targets = [(None, "all")]
+
+        for dom, env in targets:
+            rows.append({
+                "year":        year,
+                "week_number": week_number,
+                "week_code":   week_code,
+                "week_start":  d1,
+                "week_end":    d2,
+                "cycle":       current_cycle,
+                "domain_code": dom,
+                "env_scope":   env,
+                "status":      status,
+                "note":        note,
+            })
+    return rows
+
+
+SQL_INSERT = text("""
+    INSERT INTO patch_planning
+        (year, week_number, week_code, week_start, week_end, cycle,
+         domain_code, env_scope, status, note)
+    VALUES
+        (:year, :week_number, :week_code, :week_start, :week_end, :cycle,
+         :domain_code, :env_scope, :status, :note)
+""")
+
+
+def main():
+    if len(sys.argv) > 1:
+        xlsx = sys.argv[1]
+    else:
+        xlsx = None
+        for p in [
+            ROOT / "deploy" / "Planning Patching 2026_ayoub.xlsx",
+            ROOT / "deploy" / "Planning_Patching_2026_ayoub.xlsx",
+        ]:
+            if p.exists():
+                xlsx = str(p)
+                break
+        if not xlsx:
+            candidates = glob.glob(str(ROOT / "deploy" / "*Planning*ayoub*.xlsx"))
+            xlsx = candidates[0] if candidates else None
+
+    if not xlsx or not os.path.exists(xlsx):
+        print("[ERR] Fichier Planning introuvable. Place-le dans deploy/ (ex: deploy/Planning Patching 2026_ayoub.xlsx)")
+        sys.exit(1)
+
+    print(f"[INFO] Fichier: {xlsx}")
+    rows = parse_planning(xlsx)
+    print(f"[INFO] Lignes parses: {len(rows)}")
+
+    engine = create_engine(DATABASE_URL)
+    print(f"[INFO] DB: {DATABASE_URL.rsplit('@', 1)[-1]}")
+
+    inserted = 0
+    with engine.begin() as conn:
+        for r in rows:
+            conn.execute(SQL_INSERT, r)
+            inserted += 1
+
+    print(f"[OK] Termine - INSERT: {inserted}")
+    print("[INFO] Verifs :")
+    print("    SELECT week_code, domain_code, env_scope, status FROM patch_planning ORDER BY year, week_number, domain_code;")
+
+
+if __name__ == "__main__":
+    main()