#!/usr/bin/env python3 """ ingest_equirs.py — Ingestion du projet pilote equirs dans le modèle Projet / Campagne / Post défini dans architecture_outil_creation_ia.md (§2). Sources (Equirs_Plan_daction/) : 01_..._Brand_Marketing_Knowledge_Base.docx -> Projet (charte + connaissance marque) 02_..._Plan_Marketing_Communication_90_Jours -> Projet (stratégie & plan d'action) 03D_..._52_fiches_finales.docx -> 52 Posts (import quasi direct) La partie Projet est saisie manuellement (curated), car ce sont des documents de cadrage courts, lus une fois — le risque d'un parsing regex fragile sur du texte libre dépasse le bénéfice. La partie Post (03D) est parsée automatiquement car elle est structurée en 52 fiches répétitives (tableaux label/valeur) : c'est là que l'automatisation apporte une vraie valeur et une vraie fiabilité. Sortie : ingestion/output/{projet,campagne,posts}_equirs.json + un rapport de validation (ingestion_report.json) signalant les posts à compléter avant génération. """ import json import re import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).parent)) from docx_lib import read_blocks, key_value_table # noqa: E402 BASE = Path(__file__).resolve().parent.parent SRC = BASE / "Equirs_Plan_daction" OUT = Path(__file__).parent / "output" FICHE_03D = SRC / "EQUIRS_03D_52_fiches_finales.docx" # --------------------------------------------------------------------------- # 1. Projet — saisi à la main à partir des docs 01 et 02 (cf. architecture §7.5) # --------------------------------------------------------------------------- PROJET_EQUIRS = { "id": "equirs", "nom": "equirs", "tagline": "Your Trusted Business Transfer Platform", "source_docs": [ "01_EQUIRS_Brand_Marketing_Knowledge_Base.docx", "02_EQUIRS_Plan_Marketing_Communication_90_Jours.docx", ], "charte_graphique": { "couleurs": { "marine": "#0f1b3d", "emeraude": "#1eb367", "violet": "#7352c7", }, "police": "Inter", "style": "Minimal, premium, finance / M&A / SaaS B2B, beaucoup d'espace blanc", "a_eviter": [ "clichés poignée de main", "esthétique startup flashy", "surcharge de texte", ], "templates_graphiques": { "EDU": "Carrousel pédagogique", "LIST": "Checklist utilitaire", "TRUST": "Confidentialité / badges / sécurité", "DATA": "Chiffre ou concept financier", "DEAL": "Annonce / opportunité", "PRODUCT": "Fonctionnalité equirs", "PARTNER": "Cabinets / experts", "VIDEO": "Reel / vidéo / démo", }, }, "base_connaissance": { "positionnement": ( "equirs est la marketplace marocaine dédiée à la cession et à la " "transmission d'entreprises ainsi qu'à la recherche d'associés opérationnels." ), "personas": [ {"code": "A", "nom": "Dirigeant cédant"}, {"code": "B", "nom": "Repreneur individuel"}, {"code": "C", "nom": "Entrepreneur cherchant un associé"}, {"code": "D", "nom": "Acheteur professionnel"}, {"code": "E", "nom": "Prescripteur"}, ], "piliers_marque": ["Confiance", "Confidentialité", "Structure", "Transparence", "Technologie"], "vocabulaire_interdit": [ "affaire à saisir", "investissement garanti", "rendement garanti", "opportunité sans risque", "gagnez X %", "gratuit jusqu'à la vente", "toutes les annonces sont vérifiées", "toutes les entreprises sont auditées", "equirs garantit la transaction", "equirs garantit la valeur de l'entreprise", ], "regles_bloquantes": [ "Ne jamais générer spontanément un message de type « Levez des fonds sur " "equirs » — verticale Levée de fonds EN PAUSE tant que non réactivée dans la KB.", "Ne jamais dire « entreprise vérifiée » pour un niveau Standard.", "Ne jamais exporter de champ confidentiel (email, téléphone, identité masquée, " "documents Data Room, contenu NDA, notes internes) vers un outil marketing.", ], }, "strategie": { "objectifs_90_jours": { "comptes_crees": 1500, "profils_acheteurs_qualifies": 600, "annonces_publiees": "60-80", "annonces_verifiees": "20+", "partenaires_actifs": 30, }, "cadence_canaux": { "LinkedIn": "4 posts / semaine", "Instagram": "4 posts/reels / semaine + stories", "Facebook": "3-4 posts / semaine", "Google Business": "1 post / semaine", }, }, "conformite": { "loi_applicable": "Loi 44-12 (protection des données personnelles, Maroc)", "niveaux_confiance": ["Standard", "Vérifié", "Expertisé", "Transaction Ready"], }, } # --------------------------------------------------------------------------- # 2. Campagne — saisie à la main à partir du doc 02/03 # --------------------------------------------------------------------------- CAMPAGNE_LANCEMENT90 = { "id": "equirs-lancement-90j", "projet_id": "equirs", "nom": "Lancement equirs — 90 jours", "periode": {"debut": "2026-08-31", "fin": "2026-11-28"}, "objectif": "Créer simultanément l'offre (cédants), la demande (repreneurs) et un réseau de prescripteurs.", "cadence_editoriale": "4 contenus maîtres / semaine + 3 stories / semaine, sur 13 semaines", "regle_calendrier": {"planifie": "70%", "reallocation_perf": "20%", "actualite": "10%"}, "source_docs": [ "03_EQUIRS_Plan_Editorial_90_Jours.docx", "03B_EQUIRS_52_Contenus_Maitres_Integrals.docx", "EQUIRS_03D_52_fiches_finales.docx", ], } # --------------------------------------------------------------------------- # 3. Posts — parsés automatiquement depuis 03D (52 fiches) # --------------------------------------------------------------------------- NETWORK_CODES = { "LI": "LinkedIn", "IG": "Instagram", "FB": "Facebook", "GB": "Google Business", "REEL": "Reel", "STORY": "Story", "STORIES": "Story", } HEADER_FIELD_MAP = { "CONTENT_ID": "content_id", "Date": "date", "Persona": "persona_cible", "Funnel": "funnel", "Objectif métier": "objectif_metier", "Pilier": "pilier", "Langue": "langue", "Réseaux": "reseaux_raw", "Format source": "format_source", "Template Canva": "template_canva", "Automation": "automation", "Compliance": "compliance", } TRACKING_FIELD_MAP = { "CTA FR": "cta_fr", "CTA Darija": "cta_darija", "Hashtags": "hashtags_raw", "Brief graphique": "brief_creatif", "URL cible": "url_cible", "UTM template": "utm_template", } # Sections texte : certaines fiches déclinent aussi en Darija (Instagram, Facebook, # Reel, Stories, Carrousel) en plus du FR — cf. mode d'emploi du doc 03D : "LinkedIn # reste majoritairement en français ; Instagram, Facebook, Reels et Stories utilisent # la Darija sur les sujets de proximité." On matche par motif plutôt que par table figée # pour capturer les deux langues sans dupliquer les règles. MASTER_RE = re.compile(r"^Master (FR|Darija) — référence$") CANAL_TEXTE_RE = re.compile( r"^(LinkedIn|Instagram|Facebook|Google Business) — (?:texte final|caption finale) \((FR|DARIJA)\)$" ) STORIES_RE = re.compile(r"^Stories — 3 frames \((FR|DARIJA)\)$") REEL_RE = re.compile(r"^Reel — script final \((FR|DARIJA)\)$") CAROUSEL_RE = re.compile(r"^Carrousel — slide par slide \((FR|DARIJA)\)$") TRACKING_SECTION = "Production & tracking" CANAL_KEY = { "LinkedIn": "linkedin", "Instagram": "instagram", "Facebook": "facebook", "Google Business": "google_business", } FICHE_TITLE_RE = re.compile(r"^\s*(\d{2})\s*—\s*(.+)$") BRACKET_RE = re.compile(r"\[[^\]]+\]") def _lang_suffix(lang): return "_darija" if lang == "DARIJA" else "" def classify_type_contenu(format_source, template_canva): fs = (format_source or "").lower() tc = (template_canva or "").upper() if tc == "VIDEO" or "reel" in fs or "vidéo" in fs or "démo" in fs or "screencast" in fs: return "vidéo" if "carrousel" in fs or "carousel" in fs: return "carousel" if "post" in fs and "carrousel" not in fs and "carousel" not in fs: return "visuel simple" # cards, checklists, pricing... restent des visuels simples par défaut return "visuel simple" def default_outil_crea(type_contenu): # Décision du 31/08/2026 : pas de Higgsfield — appel direct aux modèles IA # vidéo (Veo retenu par défaut, Sora en fallback), voir architecture §4/§7.6. return {"vidéo": "Veo", "carousel": "Canva", "visuel simple": "Canva"}.get(type_contenu, "Canva") def split_fiches(blocks): """Découpe la liste de blocs en une liste de (titre, blocs_de_la_fiche), une entrée par Heading1 dont le texte matche '## — titre'.""" fiches = [] current_title = None current_blocks = [] for b in blocks: if b["type"] == "h1" and FICHE_TITLE_RE.match(b["text"]): if current_title is not None: fiches.append((current_title, current_blocks)) current_title = b["text"] current_blocks = [] elif current_title is not None: current_blocks.append(b) if current_title is not None: fiches.append((current_title, current_blocks)) return fiches def parse_fiche(title, blocks): m = FICHE_TITLE_RE.match(title) numero, sujet = m.group(1), m.group(2) post = { "numero": int(numero), "sujet": sujet, "textes_par_canal": {}, "autres_sections": {}, } idx = 0 # Le premier bloc est le tableau d'en-tête (CONTENT_ID, Date, Persona, ...) if idx < len(blocks) and blocks[idx]["type"] == "table": kv = key_value_table(blocks[idx]["rows"]) for label, value in kv.items(): key = HEADER_FIELD_MAP.get(label) if key: post[key] = value else: post["autres_sections"][label] = value idx += 1 # Sections suivantes, découpées par titres Heading2 current_section = None buffer_paras = [] def flush(): nonlocal current_section, buffer_paras if current_section is None: buffer_paras = [] return text = "\n".join(buffer_paras).strip() m_master = MASTER_RE.match(current_section) m_canal = CANAL_TEXTE_RE.match(current_section) m_stories = STORIES_RE.match(current_section) if m_master: post["master_darija" if m_master.group(1) == "Darija" else "master_fr"] = text elif m_canal: canal_name, lang = m_canal.groups() post["textes_par_canal"][CANAL_KEY[canal_name] + _lang_suffix(lang)] = text elif m_stories: post["stories" + _lang_suffix(m_stories.group(1))] = text elif current_section == TRACKING_SECTION and text: # Cas des posts dynamiques : une note d'avertissement en paragraphe # remplace le tableau CTA/hashtags/URL habituel (ex. contenus B/C). post["note_production"] = text elif text: post["autres_sections"][current_section] = text buffer_paras = [] def parse_timed_script(rows): return [ {"timing": row[0].strip(), "texte": row[1].strip()} for row in rows if len(row) >= 2 and row[0].strip().lower() not in ("timing",) ] def parse_slides(rows): return [ {"slide": row[0].strip(), "texte": row[1].strip()} for row in rows if len(row) >= 2 and row[0].strip().lower() not in ("slide",) ] while idx < len(blocks): b = blocks[idx] if b["type"] == "h2": flush() current_section = b["text"] elif b["type"] == "p": buffer_paras.append(b["text"]) elif b["type"] == "table": m_reel = REEL_RE.match(current_section or "") m_carousel = CAROUSEL_RE.match(current_section or "") if m_reel: post["reel_script" + _lang_suffix(m_reel.group(1))] = parse_timed_script(b["rows"]) elif m_carousel: post["carousel_slides" + _lang_suffix(m_carousel.group(1))] = parse_slides(b["rows"]) elif current_section == TRACKING_SECTION: kv = key_value_table(b["rows"]) for label, value in kv.items(): key = TRACKING_FIELD_MAP.get(label) if key: post[key] = value else: post["autres_sections"][label] = value else: # tableau non reconnu : conservé brut pour ne rien perdre label = current_section or f"table_{idx}" post["autres_sections"][label] = b["rows"] idx += 1 flush() # --- Normalisations --- reseaux_raw = post.pop("reseaux_raw", "") or "" tokens = [t.strip().upper() for t in re.split(r"[+,/]", reseaux_raw) if t.strip()] canal, formats_suppl = [], [] for t in tokens: name = NETWORK_CODES.get(t, t.title()) if name in ("LinkedIn", "Instagram", "Facebook", "Google Business"): canal.append(name) else: formats_suppl.append(name) post["canal"] = canal post["formats_supplementaires"] = formats_suppl hashtags_raw = post.pop("hashtags_raw", "") or "" post["hashtags"] = re.findall(r"#\S+", hashtags_raw) post["type_contenu"] = classify_type_contenu(post.get("format_source"), post.get("template_canva")) post["outil_crea_choisi"] = default_outil_crea(post["type_contenu"]) post["resultat_crea"] = None post["valide_par"] = None post["date_validation"] = None # contenu_texte de référence = Master FR (fallback : premier texte réseau dispo) post["contenu_texte"] = post.get("master_fr") or next(iter(post["textes_par_canal"].values()), "") # Statut + notes selon le code Automation (A / B / C, cf. doc 03 §1) automation = (post.get("automation") or "").strip().upper() notes = [] if automation != "A": notes.append( f"Automation={automation or '?'} : données equirs et/ou validation externe " "requises avant génération (cf. architecture §3 étape 2 / §6)." ) if post.get("note_production"): notes.append(post["note_production"]) # Scan de tout le texte produit (y compris carrousel/stories/reel, FR + Darija) # pour repérer les champs entre crochets restants — règle anti-gabarit §6. text_fragments = [post.get("contenu_texte", ""), post.get("brief_creatif", ""), post.get("url_cible", "")] text_fragments += list(post["textes_par_canal"].values()) for key in ("reel_script", "reel_script_darija"): text_fragments += [s["texte"] for s in post.get(key, [])] for key in ("carousel_slides", "carousel_slides_darija"): text_fragments += [s["texte"] for s in post.get(key, [])] for key in ("stories", "stories_darija"): if post.get(key): text_fragments.append(post[key]) brackets = sorted(set(BRACKET_RE.findall(" ".join(text_fragments)))) if brackets: notes.append(f"Champs entre crochets à compléter avant génération : {', '.join(brackets)}") post["notes"] = notes post["statut"] = "IDÉE" return post def parse_posts(): blocks = read_blocks(FICHE_03D) fiches = split_fiches(blocks) posts = [] for title, fblocks in fiches: try: posts.append(parse_fiche(title, fblocks)) except Exception as e: # ne bloque pas tout l'import pour une fiche posts.append({ "titre_brut": title, "erreur_parsing": str(e), "statut": "ERREUR_IMPORT", }) return posts def build_report(posts): n_total = len(posts) n_ok = sum(1 for p in posts if "erreur_parsing" not in p) n_a = sum(1 for p in posts if p.get("automation") == "A") n_b_c = sum(1 for p in posts if p.get("automation") not in ("A", None) and "erreur_parsing" not in p) n_avec_notes = sum(1 for p in posts if p.get("notes")) par_type = {} for p in posts: t = p.get("type_contenu", "?") par_type[t] = par_type.get(t, 0) + 1 return { "fiches_source": str(FICHE_03D.relative_to(BASE)), "total_fiches_detectees": n_total, "fiches_parsees_ok": n_ok, "fiches_en_erreur": n_total - n_ok, "posts_automation_A": n_a, "posts_automation_B_ou_C": n_b_c, "posts_avec_anomalie_a_completer": n_avec_notes, "repartition_type_contenu": par_type, "attendu": "52 fiches (03 — Plan éditorial 90 jours)", } def main(): OUT.mkdir(parents=True, exist_ok=True) posts = parse_posts() report = build_report(posts) (OUT / "projet_equirs.json").write_text( json.dumps(PROJET_EQUIRS, ensure_ascii=False, indent=2), encoding="utf-8" ) (OUT / "campagne_equirs_lancement90.json").write_text( json.dumps(CAMPAGNE_LANCEMENT90, ensure_ascii=False, indent=2), encoding="utf-8" ) (OUT / "posts_equirs.json").write_text( json.dumps(posts, ensure_ascii=False, indent=2), encoding="utf-8" ) (OUT / "ingestion_report.json").write_text( json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8" ) print(json.dumps(report, ensure_ascii=False, indent=2)) if report["fiches_en_erreur"]: print(f"\n⚠️ {report['fiches_en_erreur']} fiche(s) en erreur — voir posts_equirs.json (statut=ERREUR_IMPORT)") if report["total_fiches_detectees"] != 52: print(f"\n⚠️ {report['total_fiches_detectees']} fiches détectées, 52 attendues — vérifier le découpage.") if __name__ == "__main__": main()