#!/usr/bin/env python3 """Contrôleur public de planning d'internes — rend PASS ou FAIL selon des règles écrites. Il ne dépend que de la bibliothèque standard de Python 3.9+ : pas d'installation, pas de réseau. Il relit le dossier d'entrée (classeur de préparation + note de service) et le planning produit, puis vérifie les règles obligatoires H-01 à H-13 du classeur. Les objectifs S-01 et S-02 sont rapportés sans faire échouer. Usage : python3 controle_planning.py --dossier DOSSIER --planning PLANNING [--json] DOSSIER contient dossier-planning-internes.xlsx et la note de service (mise-a-jour-absence.docx ou mise-a-jour-absence.md). PLANNING planning produit : classeur .xlsx (onglet « Planning ») ou .csv, colonnes Date, Créneau, Activité, Service, Interne (voir REGLES.md). Code de sortie : 0 = PASS, 1 = FAIL, 2 = entrée illisible. """ from __future__ import annotations import argparse import csv import datetime as dt import json import re import sys import unicodedata import zipfile from collections import Counter, defaultdict from pathlib import Path from xml.etree import ElementTree as ET NS = {"m": "http://schemas.openxmlformats.org/spreadsheetml/2006/main", "r": "http://schemas.openxmlformats.org/officeDocument/2006/relationships", "pr": "http://schemas.openxmlformats.org/package/2006/relationships"} JOURS = ["LUNDI", "MARDI", "MERCREDI", "JEUDI", "VENDREDI", "SAMEDI", "DIMANCHE"] class Illisible(Exception): pass def norm(texte: object) -> str: """Minuscules, sans accents ni espaces superflus : « Médecine générale » == « Medecine generale ».""" s = unicodedata.normalize("NFKD", str(texte or "").replace("\u2019", "'")).encode("ascii", "ignore").decode() return re.sub(r"\s+", " ", s).strip().lower() # ---------- lecture xlsx sans dépendance ---------- def _col(ref: str) -> int: n = 0 for ch in re.match(r"[A-Z]+", ref).group(0): n = n * 26 + ord(ch) - 64 return n - 1 def lire_xlsx(chemin: Path) -> dict[str, list[list[object]]]: try: z = zipfile.ZipFile(chemin) except (zipfile.BadZipFile, FileNotFoundError) as exc: raise Illisible(f"{chemin.name} : classeur illisible ({exc})") with z: partages = [] if "xl/sharedStrings.xml" in z.namelist(): for si in ET.fromstring(z.read("xl/sharedStrings.xml")).findall("m:si", NS): partages.append("".join(t.text or "" for t in si.iter(f"{{{NS['m']}}}t"))) classeur = ET.fromstring(z.read("xl/workbook.xml")) liens = {r.get("Id"): r.get("Target") for r in ET.fromstring(z.read("xl/_rels/workbook.xml.rels")).findall("pr:Relationship", NS)} onglets = {} for feuille in classeur.find("m:sheets", NS): cible = liens[feuille.get(f"{{{NS['r']}}}id")].lstrip("/") cible = cible if cible.startswith("xl/") else "xl/" + cible lignes: dict[int, dict[int, object]] = defaultdict(dict) for c in ET.fromstring(z.read(cible)).iter(f"{{{NS['m']}}}c"): ref, t = c.get("r"), c.get("t") v = c.find("m:v", NS) if t == "s" and v is not None: val: object = partages[int(v.text)] elif t == "inlineStr": val = "".join(x.text or "" for x in c.iter(f"{{{NS['m']}}}t")) elif v is None or v.text is None: continue elif t in ("str", "e"): val = v.text elif t == "b": val = v.text == "1" else: f = float(v.text) val = int(f) if f.is_integer() else f lignes[int(re.search(r"\d+", ref).group(0))][_col(ref)] = val largeur = max((max(l) + 1 for l in lignes.values() if l), default=0) onglets[feuille.get("name")] = [[lignes[i].get(j) for j in range(largeur)] for i in range(1, max(lignes, default=0) + 1)] return onglets def table(lignes: list[list[object]], colonnes: list[str], nom: str) -> list[dict[str, object]]: """Trouve la ligne d'en-tête (dans les 10 premières) contenant toutes les colonnes demandées.""" voulues = [norm(c) for c in colonnes] for i, ligne in enumerate(lignes[:10]): entetes = [norm(x) for x in ligne] if all(v in entetes for v in voulues): idx = {v: entetes.index(v) for v in voulues} sortie = [] for brut in lignes[i + 1:]: if not any(x not in (None, "") for x in brut): if sortie: break continue sortie.append({c: (brut[idx[v]] if idx[v] < len(brut) else None) for c, v in zip(colonnes, voulues)}) return sortie raise Illisible(f"{nom} : en-tête introuvable (colonnes attendues : {', '.join(colonnes)})") def jour_iso(v: object) -> str: if isinstance(v, (int, float)): return (dt.date(1899, 12, 30) + dt.timedelta(days=int(v))).isoformat() s = str(v).strip()[:10] try: return dt.date.fromisoformat(s).isoformat() except ValueError: m = re.match(r"(\d{2})/(\d{2})/(\d{4})", s) if m: return f"{m.group(3)}-{m.group(2)}-{m.group(1)}" raise Illisible(f"date illisible : {v!r}") # ---------- dossier d'entrée ---------- CRENEAUX = {"matin": "AM", "am": "AM", "apres-midi": "PM", "pm": "PM", "journee entiere": "JOUR", "all_day": "JOUR", "nuit": "NUIT", "night": "NUIT"} ACTIVITES = {"service": "SERVICE", "garde": "GARDE", "guard": "GARDE", "astreinte": "ASTREINTE", "on_call": "ASTREINTE", "repos post-garde": "REPOS", "repos": "REPOS", "rest": "REPOS", "demi-journee libre": "LIBRE", "off": "LIBRE"} def lire_note(dossier: Path) -> tuple[str, str, str]: docx, md = dossier / "mise-a-jour-absence.docx", dossier / "mise-a-jour-absence.md" if docx.exists(): with zipfile.ZipFile(docx) as z: texte = " ".join(n.text or "" for n in ET.fromstring(z.read("word/document.xml")).iter() if n.tag.endswith("}t")) elif md.exists(): texte = md.read_text(encoding="utf-8").replace("*", "") else: raise Illisible("note de service absente (mise-a-jour-absence.docx ou .md)") m = re.search(r"Interne\s*:\s*(INT-\d{2}).*?Date\s*:\s*(\d{4}-\d{2}-\d{2}).*?Créneau\s*:\s*(MATIN|APRÈS-MIDI|JOURNÉE ENTIÈRE)", texte, flags=re.IGNORECASE | re.DOTALL) if not m: raise Illisible("note de service : interne, date ou créneau introuvable") return m.group(1).upper(), m.group(2), CRENEAUX[norm(m.group(3))] def lire_dossier(dossier: Path) -> dict: x = lire_xlsx(dossier / "dossier-planning-internes.xlsx") periode = re.findall(r"\d{4}-\d{2}-\d{2}", " ".join(str(v) for l in x["Mode d'emploi"] for v in l if v)) if len(periode) < 2: raise Illisible("période introuvable dans l'onglet « Mode d'emploi »") debut, fin = dt.date.fromisoformat(periode[0]), dt.date.fromisoformat(periode[1]) equipe_brut = x["Équipe et compétences"] entete = next(l for l in equipe_brut[:10] if l and norm(l[0]) == "code") services = [c for c in entete[3:] if c and norm(c) != "statut"] equipe = table(equipe_brut, ["Code", entete[1], "Statut", *services], "Équipe et compétences") internes = {str(e["Code"]): {"libelle": str(e[entete[1]]), "competences": {norm(s) for s in services if norm(e[s]) == "oui"}} for e in equipe if norm(e["Statut"]) == "actif"} besoins = [(norm(b["Service"]), norm(b["Jour"]).upper(), CRENEAUX[norm(b["Créneau"])], int(b["Effectif requis"])) for b in table(x["Besoins par service"], ["Service", "Jour", "Créneau", "Effectif requis"], "Besoins")] gardes = {jour_iso(g["Date"]): str(g["Interne"]) for g in table(x["Gardes"], ["Date", "Interne"], "Gardes")} astreintes = {jour_iso(a["Date"]): str(a["Interne"]) for a in table(x["Astreintes"], ["Date", "Interne"], "Astreintes")} absences = {(str(a["Interne"]), jour_iso(a["Date"]), CRENEAUX[norm(a["Créneau"])]) for a in table(x["Absences"], ["Interne", "Date", "Créneau", "Statut"], "Absences") if norm(a["Statut"]) == "active"} note = lire_note(dossier) return {"debut": debut, "fin": fin, "internes": internes, "besoins": besoins, "gardes": gardes, "astreintes": astreintes, "absences": absences | {note}, "note": note} # ---------- planning produit ---------- def lire_planning(chemin: Path, dossier: dict) -> list[tuple[str, str, str, str, str]]: colonnes = ["Date", "Créneau", "Activité", "Service", "Interne"] if chemin.suffix.lower() == ".csv": texte = chemin.read_text(encoding="utf-8-sig") sep = ";" if texte.count(";") > texte.count(",") else "," brut = [list(r) for r in csv.reader(texte.splitlines(), delimiter=sep)] else: onglets = lire_xlsx(chemin) nom = next((n for n in onglets if norm(n) in ("planning", "planning complet")), None) if nom is None: raise Illisible(f"{chemin.name} : onglet « Planning » absent (onglets : {', '.join(onglets)})") brut = onglets[nom] par_libelle = {norm(v["libelle"]): k for k, v in dossier["internes"].items()} lignes = [] for r in table(brut, colonnes, chemin.name): interne = str(r["Interne"] or "").strip() code = interne.upper() if interne.upper() in dossier["internes"] else par_libelle.get(norm(interne)) if code is None: m = re.search(r"INT-\d{2}", interne.upper()) code = m.group(0) if m else f"INCONNU:{interne}" act = ACTIVITES.get(norm(r["Activité"])) cre = CRENEAUX.get(norm(r["Créneau"])) if act is None or cre is None: raise Illisible(f"ligne illisible : activité {r['Activité']!r}, créneau {r['Créneau']!r}") lignes.append((jour_iso(r["Date"]), cre, act, norm(r["Service"]) if r["Service"] else "", code)) return lignes # ---------- règles ---------- def controler(d: dict, lignes: list[tuple[str, str, str, str, str]]) -> dict: res: dict[str, list[str]] = {f"H-{i:02d}": [] for i in range(1, 14)} avert: dict[str, list[str]] = {"S-01": [], "S-02": []} jours = [d["debut"] + dt.timedelta(days=i) for i in range((d["fin"] - d["debut"]).days + 1)] lendemain = (d["fin"] + dt.timedelta(days=1)).isoformat() connus = set(d["internes"]) for l in lignes: if l[4] not in connus: res["H-10"].append(f"{l[0]} : interne inconnu « {l[4]} »") if not (d["debut"].isoformat() <= l[0] <= lendemain): res["H-10"].append(f"{l[0]} : ligne hors période") for l, n in Counter(lignes).items(): if n > 1: res["H-10"].append(f"{l[0]} {l[1]} {l[4]} : ligne dupliquée ×{n}") service = [l for l in lignes if l[2] == "SERVICE"] # H-01 couverture exacte attendu = Counter() for j in jours: for s, jour, cre, n in d["besoins"]: if JOURS[j.weekday()] == norm(jour).upper(): attendu[(j.isoformat(), cre, s)] += n obtenu = Counter((l[0], l[1], l[3]) for l in service) for cle in sorted(set(attendu) | set(obtenu)): if attendu[cle] != obtenu[cle]: res["H-01"].append(f"{cle[0]} {cle[1]} {cle[2]} : {obtenu[cle]} affecté(s) pour {attendu[cle]} requis") # H-02 un service par demi-journée for (j, c, i), n in Counter((l[0], l[1], l[4]) for l in service).items(): if n > 1: res["H-02"].append(f"{j} {c} {i} : {n} services") # H-03 compétences for j, c, _, s, i in service: if i in connus and s not in d["internes"][i]["competences"]: res["H-03"].append(f"{j} {c} {i} affecté en « {s} » sans la compétence") # H-04 / H-13 indisponibilités for i, j, c in sorted(d["absences"]): for l in lignes: if l[4] == i and l[0] == j and l[2] in ("SERVICE", "ASTREINTE", "GARDE") and ( c == "JOUR" or l[1] == c or (l[2] == "ASTREINTE" and c == "AM") or l[1] == "JOUR"): if l[2] == "GARDE" and c != "JOUR": continue cible = "H-13" if (i, j, c) == d["note"] else "H-04" res[cible].append(f"{j} {c} {i} : {l[2].lower()} pendant une indisponibilité") # H-05 / H-06 / H-07 gardes et repos gardes = {l[0]: l[4] for l in lignes if l[2] == "GARDE"} for j, i in sorted(d["gardes"].items()): if gardes.get(j) != i: res["H-07"].append(f"{j} : garde attendue {i}, trouvée {gardes.get(j, 'aucune')}") for j in sorted(set(gardes) - set(d["gardes"])): res["H-07"].append(f"{j} : garde non préaffectée ({gardes[j]})") if Counter(l[0] for l in lignes if l[2] == "GARDE") != Counter(d["gardes"].keys()): res["H-07"].append("nombre de gardes par date différent de une") repos = {(l[0], l[4]) for l in lignes if l[2] == "REPOS"} for j, i in d["gardes"].items(): suivant = (dt.date.fromisoformat(j) + dt.timedelta(days=1)).isoformat() if (suivant, i) not in repos: res["H-05"].append(f"{suivant} : repos post-garde de {i} absent") for j, i in repos: for l in lignes: if l[4] == i and l[0] == j and l[2] in ("SERVICE", "ASTREINTE"): res["H-06"].append(f"{j} {i} : {l[2].lower()} pendant le repos post-garde") # H-08 / H-09 astreintes du samedi liste_astr = [(l[0], l[4]) for l in lignes if l[2] == "ASTREINTE"] astr = dict(liste_astr) if Counter(liste_astr) != Counter(d["astreintes"].items()): res["H-08"].append(f"astreintes trouvées {sorted(liste_astr)}, attendues {sorted(d['astreintes'].items())}") for j, i in astr.items(): veille = (dt.date.fromisoformat(j) - dt.timedelta(days=1)).isoformat() if i in (d["gardes"].get(j), d["gardes"].get(veille)): res["H-09"].append(f"{j} : {i} d'astreinte alors qu'en garde ou en repos") # H-11 / H-12 équité des gardes tot = Counter({i: 0 for i in connus}); we = Counter({i: 0 for i in connus}) for j, i in gardes.items(): tot[i] += 1 if dt.date.fromisoformat(j).weekday() >= 5: we[i] += 1 if tot and max(tot.values()) - min(tot.values()) > 1: res["H-11"].append(f"écart de gardes {max(tot.values()) - min(tot.values())}") if we and max(we.values()) - min(we.values()) > 1: res["H-12"].append(f"écart de gardes de week-end {max(we.values()) - min(we.values())}") # Objectifs libres = Counter((i, (dt.date.fromisoformat(j) - d["debut"]).days // 7) for j, _, a, _, i in lignes if a == "LIBRE") for i in sorted(connus): for s in range(len(jours) // 7): if libres[(i, s)] == 0: avert["S-01"].append(f"{i} sans demi-journée libre en semaine {s + 1}") charge = Counter({i: 0 for i in connus}); charge.update(l[4] for l in service) if charge: avert["S-02"].append(f"services par interne : min {min(charge.values())}, max {max(charge.values())}") compte = Counter(l[2] for l in lignes) return {"verdict": "PASS" if not any(res.values()) else "FAIL", "regles": {k: ("PASS" if not v else "FAIL") for k, v in res.items()}, "violations": {k: v[:20] for k, v in res.items() if v}, "objectifs": avert, "comptes": {"lignes": len(lignes), "services": compte["SERVICE"], "gardes": compte["GARDE"], "repos": compte["REPOS"], "astreintes": compte["ASTREINTE"], "libres": compte["LIBRE"]}} def main(argv: list[str] | None = None) -> int: p = argparse.ArgumentParser(description="Contrôleur public de planning (PASS/FAIL).") p.add_argument("--dossier", type=Path, required=True) p.add_argument("--planning", type=Path, required=True) p.add_argument("--json", action="store_true", help="sortie JSON") a = p.parse_args(argv) try: dossier = lire_dossier(a.dossier) rapport = controler(dossier, lire_planning(a.planning, dossier)) except (Illisible, KeyError, StopIteration, OSError) as exc: print(json.dumps({"verdict": "ILLISIBLE", "detail": str(exc)}, ensure_ascii=False) if a.json else f"ILLISIBLE — {exc}") return 2 if a.json: print(json.dumps(rapport, ensure_ascii=False, indent=1)) else: c = rapport["comptes"] print(f"{rapport['verdict']} — {sum(v == 'PASS' for v in rapport['regles'].values())}/13 règles obligatoires respectées") print(f"{c['lignes']} lignes : {c['services']} services, {c['gardes']} gardes, {c['repos']} repos, " f"{c['astreintes']} astreintes, {c['libres']} demi-journées libres") for regle, etat in rapport["regles"].items(): print(f" {regle} {etat}") for v in rapport["violations"].get(regle, [])[:5]: print(f" - {v}") for k, v in rapport["objectifs"].items(): print(f" {k} (objectif) : {'; '.join(v[:3]) if v else 'atteint'}") print("Rappel : ce contrôle ne remplace pas la validation du planning par la personne responsable.") return 0 if rapport["verdict"] == "PASS" else 1 if __name__ == "__main__": sys.exit(main())