"""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()