patchcenter/tools/import_tour_de_garde_xlsx.py

"""Import tour de garde SecOps depuis Tour de garde secops_2026.xlsx.

Lit la feuille 'Tour de garde', UPSERT dans secops_duty.

Usage:
    python tools/import_tour_de_garde_xlsx.py [xlsx] [--truncate]
"""
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_dates(c_val):
    if not c_val:
        return None, None
    s = str(c_val).strip()
    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
    return None, None


def s(val):
    if val is None:
        return None
    t = str(val).strip()
    return t or None


def find_xlsx():
    for p in [
        ROOT / "deploy" / "Tour de garde secops_2026.xlsx",
        ROOT / "deploy" / "Tour_de_garde_secops_2026.xlsx",
    ]:
        if p.exists():
            return str(p)
    hits = glob.glob(str(ROOT / "deploy" / "*our*garde*.xlsx"))
    return hits[0] if hits else None


def parse_tour_de_garde(xlsx_path):
    wb = openpyxl.load_workbook(xlsx_path, data_only=True)
    ws_name = next((n for n in wb.sheetnames if "garde" in n.lower()), None)
    if not ws_name:
        raise SystemExit(f"[ERR] Sheet 'Tour de garde' introuvable. Sheets: {wb.sheetnames}")
    ws = wb[ws_name]

    rows = []
    for i, row in enumerate(ws.iter_rows(values_only=True)):
        if i == 0:
            continue
        week_code = row[0]
        if not week_code or not str(week_code).strip().startswith("S"):
            continue

        wc = str(week_code).strip()
        m = re.match(r"S(\d+)", wc)
        if not m:
            continue
        week_num = int(m.group(1))

        d1, d2 = parse_dates(row[2])
        year = d1.year if d1 and d1.month > 6 else (d2.year if d2 else 2026)
        if d1 and d1.month == 12 and week_num <= 1:
            year = d2.year if d2 else d1.year + 1

        rows.append({
            "year": year,
            "week_number": week_num,
            "week_code": wc,
            "week_start": d1,
            "week_end": d2,
            "absences": s(row[1]),
            "tdg_s1": s(row[3]),
            "tdg_symantec": s(row[4]),
            "tdg_m365": s(row[5]),
            "emails_dest": s(row[6]),
            "tdg_commvault": s(row[7]),
            "tdg_meteo": s(row[8]),
            "tdg_dmz": s(row[11]) if len(row) > 11 else None,
            "tdg_safenet": s(row[12]) if len(row) > 12 else None,
            "tdg_quarantaine": s(row[13]) if len(row) > 13 else None,
            "tdg_securisation": s(row[14]) if len(row) > 14 else None,
            "tdg_incident_majeur": s(row[16]) if len(row) > 16 else None,
            "tdg_incident_critique": s(row[17]) if len(row) > 17 else None,
        })
    return rows


SQL_UPSERT = text("""
    INSERT INTO secops_duty
        (year, week_number, week_code, week_start, week_end, absences,
         tdg_s1, tdg_symantec, tdg_m365, tdg_commvault, tdg_meteo,
         tdg_dmz, tdg_safenet, tdg_quarantaine, tdg_securisation,
         tdg_incident_majeur, tdg_incident_critique, emails_dest)
    VALUES
        (:year, :week_number, :week_code, :week_start, :week_end, :absences,
         :tdg_s1, :tdg_symantec, :tdg_m365, :tdg_commvault, :tdg_meteo,
         :tdg_dmz, :tdg_safenet, :tdg_quarantaine, :tdg_securisation,
         :tdg_incident_majeur, :tdg_incident_critique, :emails_dest)
    ON CONFLICT (year, week_number) DO UPDATE SET
        week_code = EXCLUDED.week_code,
        week_start = EXCLUDED.week_start,
        week_end = EXCLUDED.week_end,
        absences = EXCLUDED.absences,
        tdg_s1 = EXCLUDED.tdg_s1,
        tdg_symantec = EXCLUDED.tdg_symantec,
        tdg_m365 = EXCLUDED.tdg_m365,
        tdg_commvault = EXCLUDED.tdg_commvault,
        tdg_meteo = EXCLUDED.tdg_meteo,
        tdg_dmz = EXCLUDED.tdg_dmz,
        tdg_safenet = EXCLUDED.tdg_safenet,
        tdg_quarantaine = EXCLUDED.tdg_quarantaine,
        tdg_securisation = EXCLUDED.tdg_securisation,
        tdg_incident_majeur = EXCLUDED.tdg_incident_majeur,
        tdg_incident_critique = EXCLUDED.tdg_incident_critique,
        emails_dest = EXCLUDED.emails_dest
""")


def main():
    xlsx = sys.argv[1] if len(sys.argv) > 1 else find_xlsx()
    if not xlsx or not os.path.exists(xlsx):
        print("[ERR] Fichier Tour de garde introuvable. Place-le dans deploy/")
        sys.exit(1)

    print(f"[INFO] Fichier: {xlsx}")
    rows = parse_tour_de_garde(xlsx)
    print(f"[INFO] Semaines parsees: {len(rows)}")

    engine = create_engine(DATABASE_URL)
    print(f"[INFO] DB: {DATABASE_URL.rsplit('@', 1)[-1]}")

    truncate = "--truncate" in sys.argv
    with engine.begin() as conn:
        if truncate:
            conn.execute(text("TRUNCATE TABLE secops_duty RESTART IDENTITY"))
            print("[INFO] TRUNCATE secops_duty")
        for r in rows:
            conn.execute(SQL_UPSERT, r)

    print(f"[OK] UPSERT: {len(rows)} semaines")


if __name__ == "__main__":
    main()