#!/usr/bin/env python3 from __future__ import annotations import calendar import argparse import json import re import sys import unicodedata from collections import defaultdict from datetime import datetime from pathlib import Path from typing import Any from openpyxl import load_workbook DAYS_TO_HOURS = 7.4 MONTHS = { "janvier": 1, "fevrier": 2, "février": 2, "mars": 3, "avril": 4, "mai": 5, "juin": 6, "juillet": 7, "aout": 8, "août": 8, "septembre": 9, "octobre": 10, "novembre": 11, "decembre": 12, "décembre": 12, } CANONICAL_PEOPLE = { "camille b": "Camille Bechet", "camille bechet": "Camille Bechet", "camille sechet": "Camille Sechet", "clara": "Clara Chanteloup", "clara chanteloup": "Clara Chanteloup", "denis": "Denis Charrier", "denis charier": "Denis Charrier", "denis charrier": "Denis Charrier", "elise": "Élise Brohan", "elise brohan": "Élise Brohan", "leo": "Léo Frémaux", "leo fremaux": "Léo Frémaux", "leo fremaux designer": "Léo Frémaux", "léo frémaux": "Léo Frémaux", "olivia": "Olivia Richard", "olivia richard": "Olivia Richard", "simon": "Simon Boussard", "simon boussard": "Simon Boussard", "tom": "Tom Eade", "tom eade": "Tom Eade", "vincent": "Vincent Pujos", "vincent pujos": "Vincent Pujos", } def normalize(value: Any) -> str: text = str(value or "").strip().lower() text = unicodedata.normalize("NFD", text) text = "".join(char for char in text if unicodedata.category(char) != "Mn") text = re.sub(r"[^a-z0-9]+", " ", text) return re.sub(r"\s+", " ", text).strip() def clean_text(value: Any) -> str: if value is None: return "" if isinstance(value, datetime): return value.strftime("%Y-%m-%d") return re.sub(r"\s+", " ", str(value).replace("\xa0", " ")).strip() def as_float(value: Any) -> float: if isinstance(value, (int, float)) and not isinstance(value, bool): return float(value) if isinstance(value, str): text = value.strip().replace(",", ".") if re.fullmatch(r"-?\d+(\.\d+)?", text): return float(text) return 0.0 def parse_date(value: Any, fallback_year: int | None = None) -> str | None: if isinstance(value, datetime): return value.date().isoformat() text = clean_text(value).lower() if not text: return None date_match = re.search(r"(\d{1,2})[/-](\d{1,2})(?:[/-](\d{2,4}))?", text) if date_match and fallback_year: day = int(date_match.group(1)) month = int(date_match.group(2)) year = int(date_match.group(3) or fallback_year) if year < 100: year += 2000 try: return datetime(year, month, day).date().isoformat() except ValueError: return None for label, month in MONTHS.items(): if label in text and "fin" in text and fallback_year: day = calendar.monthrange(fallback_year, month)[1] return datetime(fallback_year, month, day).date().isoformat() return None def quote_from_title(title: str) -> str: match = re.search(r"(D\d{5}|OF\d{5}|OF\d{4}|INTM)", title.upper()) if match: return match.group(1) return re.sub(r"\s+", "-", title.upper()) def role_from_name(name: str) -> str: normalized = normalize(name) if "chef de projet" in normalized: return "chef_de_projet" if "directeur" in normalized: return "directeur_de_projet" if "sous traitant" in normalized or "freelance" in normalized: return "sous_traitance" return "designer" def person_name(raw: str) -> str: value = clean_text(raw) value = re.sub(r"\s*-\s+.*$", "", value) value = value.replace("CHARRIER", "Charrier").replace("SECHET", "Sechet").replace("PUJOS", "Pujos") value = value.replace("Charier", "Charrier") normalized = normalize(value) if normalized in CANONICAL_PEOPLE: return CANONICAL_PEOPLE[normalized] return " ".join(part.capitalize() if part.isupper() else part for part in value.split()) def likely_phase_code(value: Any) -> bool: text = clean_text(value) return bool(re.fullmatch(r"[A-Z]", text) or re.fullmatch(r"\d+", text) or re.fullmatch(r"PHASE\s*\d+", text.upper())) def likely_step_code(value: Any, phase_code: str) -> bool: text = clean_text(value).upper() code = clean_text(phase_code).upper() return bool(re.fullmatch(rf"{re.escape(code)}[\.\-]?\d+", text)) def find_project_info(rows: list[list[Any]]) -> dict[str, Any]: info: dict[str, Any] = {} team_rows: list[dict[str, str]] = [] for row in rows: label = normalize(row[0] if row else "") if label == "projet": info["contract_label"] = clean_text(row[2] if len(row) > 2 and row[2] else row[1] if len(row) > 1 else "") elif label == "date de debut du projet": info["starts_at"] = parse_date(row[2] if len(row) > 2 else None) elif label == "date de fin du projet": start_year = int(info["starts_at"][:4]) if info.get("starts_at") else None info["ends_at"] = parse_date(row[2] if len(row) > 2 else None, start_year) elif label == "chef de projet sensipode": info["project_manager"] = person_name(clean_text(row[2] if len(row) > 2 else "")) elif label.startswith("equipe projet"): org = "&Si" if "&si" in clean_text(row[0]).lower() or "etsi" in normalize(row[0]) else "SENSIPODE" team_rows.append({"name": person_name(clean_text(row[2] if len(row) > 2 else "")), "organization": org}) info["team_rows"] = [item for item in team_rows if item["name"]] return info def find_tjm_people(rows: list[list[Any]]) -> dict[int, dict[str, Any]]: people: dict[int, dict[str, Any]] = {} for row in rows[:20]: key = clean_text(row[0] if row else "") match = re.fullmatch(r"TJM\s*(\d+)", key.upper()) if not match: continue idx = int(match.group(1)) name = person_name(clean_text(row[2] if len(row) > 2 else "")) if not name or "Nom" in name: continue people[idx] = { "name": name, "role": role_from_name(clean_text(row[2] if len(row) > 2 else "")), "organization": clean_text(row[3] if len(row) > 3 else "SENSIPODE") or "SENSIPODE", "daily_rate": as_float(row[1] if len(row) > 1 else 0) or None, } return people def is_known_metric_label(value: Any) -> bool: label = normalize(value) return label in { "j h", "prix ht", "p u ht", "tva 20", "prix ttc", "qte", "planning", "etat", "commentaire", "date realise rendu s", "facturation", "repartition", } def phase_header_people(row: list[Any], tjm_people: dict[int, dict[str, Any]]) -> dict[int, dict[str, Any]]: people: dict[int, dict[str, Any]] = {} known_by_first = {normalize(item["name"]).split(" ")[0]: item for item in tjm_people.values() if item.get("name")} for idx, value in enumerate(row): text = clean_text(value) if idx < 2 or not text or is_known_metric_label(text): continue tjm_match = re.fullmatch(r"TJM\s*(\d+)", text.upper()) if tjm_match and int(tjm_match.group(1)) in tjm_people: people[idx] = tjm_people[int(tjm_match.group(1))] continue key = normalize(text).split(" ")[0] if normalize(text) else "" if key in known_by_first: people[idx] = known_by_first[key] continue if text in {"Maureen", "Alexandre"}: people[idx] = {"name": text, "role": "sous_traitance", "organization": "EtSI", "daily_rate": None} return people def find_columns(row: list[Any]) -> dict[str, int | None]: cols: dict[str, int | None] = {"quantity": None, "total_days": None, "planning": None, "status": None, "comment": None} for idx, value in enumerate(row): label = normalize(value) if label == "qte": cols["quantity"] = idx elif label == "j h": cols["total_days"] = idx elif label == "planning" or label == "date realise rendu s": cols["planning"] = idx elif label == "etat": cols["status"] = idx elif label == "commentaire": cols["comment"] = idx return cols def status_code(value: Any) -> str: label = normalize(value) if "termine" in label: return "termine" if "cours" in label: return "en_cours" if "standby" in label: return "standby" return "non_realise" def task_type(name: str) -> str: label = normalize(name) if "atelier" in label or "workshop" in label: return "atelier" if "terrain" in label or "immersion" in label or "residence" in label: return "terrain" if "livrable" in label or "restitution" in label or "synthese" in label: return "livrable" if "reunion" in label or "copil" in label or "cotech" in label: return "reunion" return "tache" def parse_standard_project(rows: list[list[Any]], info: dict[str, Any], tjm_people: dict[int, dict[str, Any]]) -> dict[str, Any]: phases: list[dict[str, Any]] = [] team_totals: dict[str, float] = defaultdict(float) team_meta: dict[str, dict[str, Any]] = {} current_phase: dict[str, Any] | None = None current_step: dict[str, Any] | None = None people_cols: dict[int, dict[str, Any]] = {} cols: dict[str, int | None] = {} task_position = 1 step_position = 1 start_year = int((info.get("starts_at") or "2026")[:4]) def ensure_step(code: str, name: str) -> dict[str, Any]: nonlocal current_step, step_position normalized_code = normalize(code) for step in current_phase["steps"]: if normalize(step["code"]) == normalized_code: current_step = step return step current_step = { "code": code, "name": name or code, "planned_days": 0, "planned_starts_at": None, "planned_ends_at": None, "position": step_position, } current_phase["steps"].append(current_step) step_position += 1 return current_step for row in rows: if likely_phase_code(row[0] if row else None) and clean_text(row[1] if len(row) > 1 else ""): header_people = phase_header_people(row, tjm_people) if header_people: current_phase = { "code": clean_text(row[0]), "name": clean_text(row[1]), "starts_at": info.get("starts_at"), "ends_at": info.get("ends_at") or info.get("starts_at"), "planned_starts_at": None, "planned_ends_at": None, "sold_days": 0, "deliverables": None, "tasks": [], "steps": [], } phases.append(current_phase) current_step = None people_cols = header_people cols = find_columns(row) task_position = 1 step_position = 1 continue if current_phase is None: continue first = clean_text(row[0] if row else "") second = clean_text(row[1] if len(row) > 1 else "") third = clean_text(row[2] if len(row) > 2 else "") if normalize(first) == "livrables": current_phase["deliverables"] = second if cols.get("total_days") is not None: current_phase["sold_days"] = as_float(row[cols["total_days"]]) continue is_step = likely_step_code(first, current_phase["code"]) if is_step: current_step = ensure_step(first, second or first) if not is_step and not third and not second: continue task_name = third if third else (second if is_step else first) if not task_name or normalize(task_name) in {"total", "total ht"}: continue if current_step is None: current_step = ensure_step(f"{current_phase['code']}{step_position}", second or task_name) quantity = as_float(row[cols["quantity"]]) if cols.get("quantity") is not None else 1.0 if quantity <= 0: quantity = 1.0 assignments = [] planned_days_from_people = 0.0 for col, person in people_cols.items(): days = as_float(row[col] if col < len(row) else 0) * quantity if days <= 0: continue planned_days_from_people += days name = person["name"] team_totals[name] += days team_meta[name] = person assignments.append({ "name": name, "organization": person.get("organization") or "SENSIPODE", "role": person.get("role") or "designer", "daily_rate": person.get("daily_rate"), "planned_days": round(days, 3), }) total_days = as_float(row[cols["total_days"]]) if cols.get("total_days") is not None else planned_days_from_people planned_at = parse_date(row[cols["planning"]] if cols.get("planning") is not None else None, start_year) state = status_code(row[cols["status"]] if cols.get("status") is not None else "") comment = clean_text(row[cols["comment"]] if cols.get("comment") is not None else "") if planned_days_from_people <= 0 and total_days <= 0: continue task = { "name": task_name, "type": task_type(" ".join([second, task_name])), "step_code": current_step["code"] if current_step else None, "starts_at": None, "ends_at": None, "planned_starts_at": planned_at, "planned_ends_at": planned_at, "planned_days": round(total_days or planned_days_from_people, 3), "status": state, "position": task_position, "comment": comment or None, "assignments": assignments, } current_phase["tasks"].append(task) if current_step: current_step["planned_days"] = round(current_step["planned_days"] + task["planned_days"], 3) task_position += 1 for phase in phases: if not phase["sold_days"]: phase["sold_days"] = round(sum(task["planned_days"] for task in phase["tasks"]), 3) phase["milestone_at"] = phase["ends_at"] team = [] for name, days in sorted(team_totals.items()): meta = team_meta.get(name, {}) team.append({ "name": name, "organization": meta.get("organization") or "SENSIPODE", "role": "chef_de_projet" if normalize(name) == normalize(info.get("project_manager")) else meta.get("role", "designer"), "daily_rate": meta.get("daily_rate"), "sold_days": round(days, 3), }) for row in info.get("team_rows", []): if not any(normalize(member["name"]) == normalize(row["name"]) for member in team): team.append({"name": row["name"], "organization": row.get("organization", "SENSIPODE"), "role": "designer", "daily_rate": None, "sold_days": 0}) return {"phases": phases, "team": team, "sold_days": round(sum(phase["sold_days"] for phase in phases), 3)} def parse_regie_project(rows: list[list[Any]], info: dict[str, Any], tjm_people: dict[int, dict[str, Any]]) -> dict[str, Any]: team = [] for person in tjm_people.values(): team.append({**person, "sold_days": 0}) billing = [] for row in rows: first = clean_text(row[0] if row else "") if normalize(first) in {normalize(month) for month in MONTHS}: continue if len(row) >= 7 and isinstance(row[6], datetime): billing.append({ "name": f"Facturation {row[6].strftime('%B %Y')}", "planned_invoice_at": row[6].date().isoformat(), "amount_excluding_tax": as_float(row[3] if len(row) > 3 else 0) or None, "trigger": "date_fixe", }) sold_days = sum(as_float(row[1] if len(row) > 1 else 0) for row in rows if len(row) > 2 and as_float(row[1]) and isinstance(row[0], str) and row[0].startswith("SEM ")) if team and sold_days: per_member = sold_days / len(team) for member in team: member["sold_days"] = round(per_member, 3) phase = { "code": "REGIE", "name": "Régie", "starts_at": info.get("starts_at"), "ends_at": info.get("ends_at"), "planned_starts_at": None, "planned_ends_at": None, "milestone_at": info.get("ends_at"), "sold_days": round(sold_days, 3), "deliverables": None, "steps": [{ "code": "REGIE1", "name": "Régie", "planned_days": round(sold_days, 3), "planned_starts_at": None, "planned_ends_at": None, "position": 1, }], "tasks": [{ "name": "Régie", "type": "tache", "step_code": "REGIE1", "starts_at": info.get("starts_at"), "ends_at": info.get("ends_at"), "planned_starts_at": None, "planned_ends_at": None, "planned_days": round(sold_days, 3), "status": "en_cours", "position": 1, "comment": None, "assignments": [ { "name": member["name"], "organization": member.get("organization") or "SENSIPODE", "role": member.get("role", "designer"), "daily_rate": member.get("daily_rate"), "planned_days": member["sold_days"], } for member in team ], }], } return {"phases": [phase], "team": team, "sold_days": round(sold_days, 3), "billing": billing} def parse_sheet(ws) -> dict[str, Any]: rows = [[cell.value for cell in row] for row in ws.iter_rows()] client_name = clean_text(rows[0][0] if rows and rows[0] else ws.title) if normalize(client_name).startswith("urm"): client_name = client_name.replace("URM", "UMR", 1) project_name = clean_text(rows[0][3] if rows and len(rows[0]) > 3 else None) or clean_text(rows[0][1] if rows and len(rows[0]) > 1 else ws.title) info = find_project_info(rows) tjm_people = find_tjm_people(rows) contract_label = normalize(info.get("contract_label", "forfait")) parsed = parse_regie_project(rows, info, tjm_people) if "regie" in contract_label else parse_standard_project(rows, info, tjm_people) return { "source_sheet": ws.title, "client_name": client_name, "client_short_name": client_name.split(" - ")[0][:80], "project_name": project_name, "project_short_name": project_name[:120], "quote_number": quote_from_title(ws.title), "contract_type": "regie" if "regie" in contract_label else "forfait", "starts_at": info.get("starts_at"), "ends_at": info.get("ends_at"), "project_manager": info.get("project_manager"), "sold_days": parsed["sold_days"], "team": parsed["team"], "phases": parsed["phases"], "billing_milestones": parsed.get("billing", []), } def project_diagnostic(project: dict[str, Any]) -> dict[str, Any]: phases = project.get("phases", []) team = project.get("team", []) tasks = [task for phase in phases for task in phase.get("tasks", [])] steps = [step for phase in phases for step in phase.get("steps", [])] team_names = {normalize(member.get("name")) for member in team} step_codes = { (normalize(phase.get("code")), normalize(step.get("code"))) for phase in phases for step in phase.get("steps", []) } signals: list[str] = [] if not project.get("starts_at"): signals.append("date_contractuelle_debut_absente") if not project.get("ends_at"): signals.append("date_contractuelle_fin_absente") if not team: signals.append("aucun_collaborateur_identifie") if any(re.fullmatch(r"tjm\s*\d+", normalize(member.get("name"))) for member in team): signals.append("tjm_non_mappe") phase_total = round(sum(as_float(phase.get("sold_days")) for phase in phases), 3) if abs(phase_total - as_float(project.get("sold_days"))) > 0.02: signals.append("incoherence_charge_projet_vs_phases") unmapped_people = set() unallocated_tasks = [] tasks_without_step = [] phases_without_deliverables = [] invalid_dates = [] for phase in phases: if not clean_text(phase.get("deliverables")): phases_without_deliverables.append(clean_text(phase.get("code")) or clean_text(phase.get("name"))) task_total = round(sum(as_float(task.get("planned_days")) for task in phase.get("tasks", [])), 3) if abs(task_total - as_float(phase.get("sold_days"))) > 0.02: signals.append(f"incoherence_charge_phase_{clean_text(phase.get('code'))}") for task in phase.get("tasks", []): step_code = clean_text(task.get("step_code")) if not step_code or (normalize(phase.get("code")), normalize(step_code)) not in step_codes: tasks_without_step.append(clean_text(task.get("name"))) assignment_total = round(sum(as_float(assignment.get("planned_days")) for assignment in task.get("assignments", [])), 3) if abs(assignment_total - as_float(task.get("planned_days"))) > 0.02: unallocated_tasks.append(clean_text(task.get("name"))) for assignment in task.get("assignments", []): assignment_name = normalize(assignment.get("name")) if re.fullmatch(r"tjm\s*\d+", assignment_name): unmapped_people.add(clean_text(assignment.get("name"))) if assignment_name and assignment_name not in team_names: unmapped_people.add(clean_text(assignment.get("name"))) for date_key in ("starts_at", "ends_at", "planned_starts_at", "planned_ends_at"): value = task.get(date_key) if value and not re.fullmatch(r"\d{4}-\d{2}-\d{2}", str(value)): invalid_dates.append(f"{clean_text(task.get('name'))}:{date_key}") if unmapped_people: signals.append("collaborateur_affecte_hors_equipe") if unallocated_tasks: signals.append("charge_tache_non_ventilee") if tasks_without_step: signals.append("tache_sans_etape") if phases_without_deliverables: signals.append("phase_sans_livrables") if invalid_dates: signals.append("date_invalide") return { "project": f"{project.get('client_name')} — {project.get('project_name')}", "quote_number": project.get("quote_number"), "source_sheet": project.get("source_sheet"), "counts": { "phases": len(phases), "steps": len(steps), "tasks": len(tasks), "collaborators": len(team), }, "sold_days": as_float(project.get("sold_days")), "sold_hours": round(as_float(project.get("sold_days")) * DAYS_TO_HOURS, 1), "phases": [ { "code": phase.get("code"), "name": phase.get("name"), "steps": len(phase.get("steps", [])), "tasks": len(phase.get("tasks", [])), "sold_days": as_float(phase.get("sold_days")), "has_deliverables": bool(clean_text(phase.get("deliverables"))), } for phase in phases ], "signals": sorted(set(signals)), "details": { "unmapped_people": sorted(unmapped_people), "unallocated_tasks": unallocated_tasks[:20], "tasks_without_step": tasks_without_step[:20], "phases_without_deliverables": phases_without_deliverables[:20], "invalid_dates": invalid_dates[:20], }, } def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Extrait les projets d'un classeur Excel vers le JSON d'import SensiTools.", ) parser.add_argument("source", help="Fichier Excel source.") parser.add_argument("output", help="Fichier JSON de sortie.") parser.add_argument( "--exclude-sheet-containing", action="append", default=[], metavar="TEXT", help="Ignore les onglets dont le nom contient ce texte (insensible à la casse). Option répétable.", ) parser.add_argument( "--skip-invalid-project-names", action="store_true", help="Ignore les projets dont le nom extrait est vide ou correspond à une erreur Excel.", ) return parser.parse_args() def should_include_sheet(title: str, excluded_fragments: list[str]) -> bool: normalized_title = normalize(title) return not any(normalize(fragment) in normalized_title for fragment in excluded_fragments if normalize(fragment)) def has_valid_project_name(project: dict[str, Any]) -> bool: name = clean_text(project.get("project_name")) return name != "" and not name.startswith("#") def main() -> None: args = parse_args() source = Path(args.source) output = Path(args.output) excluded_fragments = list(args.exclude_sheet_containing) wb = load_workbook(source, data_only=True, read_only=True) included_sheets = [ws for ws in wb.worksheets if should_include_sheet(ws.title, excluded_fragments)] excluded_sheets = [ws.title for ws in wb.worksheets if not should_include_sheet(ws.title, excluded_fragments)] parsed_projects = [parse_sheet(ws) for ws in included_sheets] if args.skip_invalid_project_names: projects = [project for project in parsed_projects if has_valid_project_name(project)] invalid_project_sheets = [ clean_text(project.get("source_sheet")) for project in parsed_projects if not has_valid_project_name(project) ] else: projects = parsed_projects invalid_project_sheets = [] payload = { "source_file": source.name, "generated_from": str(source), "days_to_hours": DAYS_TO_HOURS, "excluded_sheet_fragments": excluded_fragments, "excluded_sheets": excluded_sheets, "invalid_project_sheets": invalid_project_sheets, "projects": projects, "diagnostics": [project_diagnostic(project) for project in projects], } output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") print(f"Wrote {len(projects)} projects to {output}") if excluded_sheets: print(f"Excluded {len(excluded_sheets)} sheets: {', '.join(excluded_sheets)}") if invalid_project_sheets: print(f"Skipped {len(invalid_project_sheets)} invalid projects: {', '.join(invalid_project_sheets)}") if __name__ == "__main__": main()