214 lines
6.6 KiB
Python
214 lines
6.6 KiB
Python
"""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()
|