oto-enterprise-os-dtp/05_deliverables_mvp/seo/seo_gen.py

#!/usr/bin/env python3
"""Generateur SEO trilingue · Sprint 6 · SEO.

Roadmap Sprint 6 SEO (L60) : « Refactor mission seo_autonome/ -> 200+ mots-cles
FR/EN/ES · schema.org · hreflang ». Consomme `projets_master.json` (sortie
Publiciste, derivee de data_room/PXX) et produit un bundle de hand-off :
  - seo_keywords.json    : 200+ mots-cles FR/EN/ES, chacun sourcE ;
  - seo_schema_org.json  : graphe JSON-LD (Organization + une Residence/projet) ;
  - seo_hreflang.json    : carte hreflang (alternates + x-default) ;
  - MANIFEST.json        : compte-rendu.

Ce worker n'ecrit JAMAIS sur le VPS (#8) : l'injection des balises dans les
pages `www/`, le sitemap et la soumission Google Search Console restent cote
agent SEO / Frontend.

ANTI-INVENTION (#6) : aucun fait de projet (nom/localisation/prix) n'est fabrique
ici — tout vient de l'entree. Un mot-cle ne peut contenir que les chiffres deja
presents dans son champ projet source ; schema.org n'emet un prix que pour un
projet 'disponible' a typologie sourcee (USD · #10). 15 invariants le garantissent.

Sous-commandes :
  build    [--master M] [-o OUT]  -> ecrit les 4 fichiers de hand-off ;
  validate [--master M]           -> (re)genere en memoire, valide entree +
                                     schema de sortie + 15 invariants ; sort en
                                     erreur sinon.

Sortie deterministe (tri stable, aucun horodatage) -> diffable + re-generable.
"""

from __future__ import annotations

import argparse
import os
import sys

_HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, _HERE)

from seolib import builder, deps  # noqa: E402

_DEFAULT_OUT = os.path.join(_HERE, "out")


def _eprint(*args) -> None:
    print(*args, file=sys.stderr)


def _project_field_ok(source: str, by_code: dict) -> bool:
    """Une source `projet:<code>.<champ>` doit viser un projet + un champ reels."""
    body = source[len("projet:"):]
    code, _, field = body.partition(".")
    proj = by_code.get(code)
    return proj is not None and field in proj


def _validate(spec: dict, master: dict, bundle: dict) -> list[str]:
    """Schema de sortie + 15 invariants de cross-coherence / anti-invention."""
    errors = deps.validate(bundle, deps.load_json(deps.SCHEMA_PATH))

    site = spec["site"]
    langs = site["langs"]
    base = site["base_url"].rstrip("/")
    default_lang = site["default_lang"]
    targets = spec["targets"]
    projets = master["projets"]
    by_code = {p["code"]: p for p in projets}

    kws = bundle["keywords"]
    sorg = bundle["schema_org"]
    href = bundle["hreflang"]
    m = bundle["manifest"]

    # 1 · l'entree respecte le contrat Publiciste (garde-fou amont).
    for e in deps.validate_master(master):
        errors.append(f"entree projets_master non conforme : {e}")

    # 2 · volume total >= cible roadmap (200+).
    if len(kws) < targets["min_keywords_total"]:
        errors.append(
            f"mots-cles total {len(kws)} < cible {targets['min_keywords_total']} (roadmap L60)"
        )

    # 3 · les 3 langues du site sont couvertes, chacune >= min par langue.
    per_lang = {lang: [k for k in kws if k["lang"] == lang] for lang in langs}
    present = {k["lang"] for k in kws}
    if present != set(langs):
        errors.append(f"langues couvertes {sorted(present)} != site.langs {langs}")
    for lang in langs:
        n = len(per_lang.get(lang, []))
        if n < targets["min_keywords_per_lang"]:
            errors.append(
                f"langue {lang!r} : {n} mots-cles < min {targets['min_keywords_per_lang']}"
            )

    # 4 · unicite (term, lang).
    keys = [(k["term"], k["lang"]) for k in kws]
    if len(keys) != len(set(keys)):
        errors.append("mot-cle (term, lang) duplique")

    # 5 · chaque mot-cle est sourcE et resoluble (lexicon:* ou projet:<code>.<champ>).
    for k in kws:
        if not k["sources"]:
            errors.append(f"mot-cle {k['term']!r} sans source")
        for s in k["sources"]:
            if s.startswith("lexicon:"):
                continue
            if s.startswith("projet:"):
                if not _project_field_ok(s, by_code):
                    errors.append(f"mot-cle {k['term']!r} : source {s!r} non resoluble")
            else:
                errors.append(f"mot-cle {k['term']!r} : source {s!r} de forme inconnue")

    # 6 · scope coherent : projet<->code present ; global<->projet null.
    for k in kws:
        if k["scope"] == "projet":
            if k["projet"] not in by_code:
                errors.append(f"mot-cle {k['term']!r} scope projet mais code {k['projet']!r} absent")
        elif k["projet"] is not None:
            errors.append(f"mot-cle {k['term']!r} scope global mais projet={k['projet']!r}")

    # 7 · ANTI-INVENTION (#6) : un mot-cle ne peut porter que les chiffres deja
    #     presents dans son champ projet source (aucun chiffre fabrique).
    for k in kws:
        term_digits = deps.digits(k["term"])
        if not term_digits:
            continue
        allowed: set[str] = set()
        if k["projet"] in by_code:
            p = by_code[k["projet"]]
            allowed = deps.digits(p["nom"] + " " + p["localisation"])
        if not term_digits <= allowed:
            errors.append(
                f"mot-cle {k['term']!r} : chiffre(s) {sorted(term_digits - allowed)} "
                f"absent(s) de la donnee source (invention interdite #6)"
            )

    # 8 · schema.org : contexte + noeud Organization aligne au spec.
    if sorg["@context"] != spec["schema_org"]["context"]:
        errors.append("schema_org.@context != spec")
    orgs = [n for n in sorg["@graph"] if n.get("@type") == spec["schema_org"]["organization"]["type"]]
    if len(orgs) != 1:
        errors.append(f"schema_org : {len(orgs)} noeud Organization (attendu 1)")
    elif orgs[0].get("name") != spec["schema_org"]["organization"]["name"]:
        errors.append("schema_org Organization.name != spec")

    # 9 · un listing par projet ; nom + localite = donnee source (fidelite).
    listing_type = spec["schema_org"]["listing_type"]
    listings = [n for n in sorg["@graph"] if n.get("@type") == listing_type]
    if len(listings) != len(projets):
        errors.append(f"schema_org : {len(listings)} listings != {len(projets)} projets")
    listing_by_name = {n.get("name"): n for n in listings}
    for p in projets:
        node = listing_by_name.get(p["nom"])
        if node is None:
            errors.append(f"schema_org : aucun listing pour {p['nom']!r}")
            continue
        if node.get("address", {}).get("addressLocality") != p["localisation"]:
            errors.append(f"schema_org {p['nom']!r} : addressLocality != localisation source")
        if not str(node.get("url", "")).startswith(base):
            errors.append(f"schema_org {p['nom']!r} : url hors base_url")

    # 10 · offres : presentes IFF projet 'disponible' + typologie a prix numerique
    #      sourcE ; prix == valeur source ; devise USD (#10).
    for p in projets:
        node = listing_by_name.get(p["nom"], {})
        priced = [t for t in p.get("typologies", [])
                  if isinstance(t.get("prix_depuis_usd"), (int, float))
                  and not isinstance(t.get("prix_depuis_usd"), bool)]
        expect_offer = p["statut"] not in deps.STATUTS_SANS_PRIX and bool(priced)
        offers = node.get("offers", [])
        if bool(offers) != expect_offer:
            errors.append(f"schema_org {p['nom']!r} : presence d'offre {bool(offers)} != attendu {expect_offer}")
        for off in offers:
            if off.get("priceCurrency") != deps.DEVISE_PRIMAIRE:
                errors.append(f"schema_org {p['nom']!r} : devise offre != {deps.DEVISE_PRIMAIRE}")
            src_prices = {t["prix_depuis_usd"] for t in priced}
            if off.get("price") not in src_prices:
                errors.append(f"schema_org {p['nom']!r} : prix d'offre non sourcE dans le master")

    # 11 · ANTI-INVENTION (#6) : aucun prix pour un statut sans prix.
    for p in projets:
        if p["statut"] in deps.STATUTS_SANS_PRIX and listing_by_name.get(p["nom"], {}).get("offers"):
            errors.append(f"schema_org {p['nom']!r} : offre interdite (statut '{p['statut']}' sans prix · #6)")

    # 12 · hreflang : une page d'accueil + une page par projet.
    pages = href["pages"]
    page_names = [pg["page"] for pg in pages]
    expect_pages = ["home"] + [p["code"] for p in projets]
    if sorted(page_names) != sorted(expect_pages):
        errors.append(f"hreflang : pages {sorted(page_names)} != {sorted(expect_pages)}")

    # 13 · chaque page : exactement une alternate par langue + un x-default ;
    #      x-default et canonical == URL de la langue par defaut.
    for pg in pages:
        alts = {a["hreflang"]: a["href"] for a in pg["alternates"]}
        if len(pg["alternates"]) != len(alts):
            errors.append(f"hreflang page {pg['page']!r} : hreflang duplique")
        if set(alts) != set(langs) | {"x-default"}:
            errors.append(f"hreflang page {pg['page']!r} : alternates {sorted(alts)} != langs + x-default")
        if alts.get("x-default") != alts.get(default_lang):
            errors.append(f"hreflang page {pg['page']!r} : x-default != langue par defaut")
        if pg["canonical"] != alts.get(default_lang):
            errors.append(f"hreflang page {pg['page']!r} : canonical != URL langue par defaut")

    # 14 · toutes les URLs (schema.org + hreflang) sous base_url en https.
    urls = [n.get("url") for n in listings if n.get("url")]
    urls += [n.get("@id") for n in sorg["@graph"]]
    for pg in pages:
        urls.append(pg["canonical"])
        urls += [a["href"] for a in pg["alternates"]]
    for u in urls:
        if not str(u).startswith(base + "/") and str(u) != base:
            errors.append(f"URL hors base_url : {u!r}")

    # 15 · comptes du manifeste coherents.
    checks = {
        "keywords_total": len(kws),
        "schema_org_nodes": len(sorg["@graph"]),
        "listings": len(listings),
        "hreflang_pages": len(pages),
        "projects": len(projets),
        "offers": sum(len(n.get("offers", [])) for n in sorg["@graph"]),
    }
    for key, val in checks.items():
        if m["counts"].get(key) != val:
            errors.append(f"manifest.counts.{key} incoherent ({m['counts'].get(key)} != {val})")
    for lang in langs:
        if m["counts"]["keywords_per_lang"].get(lang) != len(per_lang[lang]):
            errors.append(f"manifest.counts.keywords_per_lang.{lang} incoherent")

    return errors


def _build(master_path: str | None):
    spec = deps.load_spec()
    master = deps.load_master(master_path)
    bundle = builder.build_bundle(spec, master)
    return spec, master, bundle


def _write_json(path: str, data) -> None:
    import json
    with open(path, "w", encoding="utf-8") as fh:
        json.dump(data, fh, ensure_ascii=False, indent=2)
        fh.write("\n")


def cmd_build(args: argparse.Namespace) -> int:
    spec, master, bundle = _build(args.master)
    errors = _validate(spec, master, bundle)
    if errors:
        _eprint("❌ Bundle SEO invalide — generation refusee (anti-regression) :")
        for e in errors:
            _eprint(f"   - {e}")
        return 1

    out = os.path.abspath(args.out)
    os.makedirs(out, exist_ok=True)
    _write_json(os.path.join(out, "seo_keywords.json"), bundle["keywords"])
    _write_json(os.path.join(out, "seo_schema_org.json"), bundle["schema_org"])
    _write_json(os.path.join(out, "seo_hreflang.json"), bundle["hreflang"])
    _write_json(os.path.join(out, "MANIFEST.json"), bundle["manifest"])

    c = bundle["manifest"]["counts"]
    print(f"✅ Bundle SEO genere dans {out}")
    print(f"   mots-cles : {c['keywords_total']} "
          f"({' · '.join(f'{l}={n}' for l, n in c['keywords_per_lang'].items())})")
    print(f"   schema.org : {c['schema_org_nodes']} noeuds ({c['listings']} listings · "
          f"{c['offers']} offres) · hreflang : {c['hreflang_pages']} pages")
    print("   ⚠ Injection balises + sitemap + Google Search Console cote agent SEO/Frontend (VPS · #8).")
    return 0


def cmd_validate(args: argparse.Namespace) -> int:
    spec, master, bundle = _build(args.master)
    errors = _validate(spec, master, bundle)
    if errors:
        _eprint("❌ Validation KO :")
        for e in errors:
            _eprint(f"   - {e}")
        return 1
    c = bundle["manifest"]["counts"]
    print(f"✅ Validation OK — {c['keywords_total']} mots-cles FR/EN/ES, "
          f"{c['listings']} listings schema.org, {c['hreflang_pages']} pages hreflang, "
          f"schema + 15 invariants verts.")
    return 0


def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser(description="Generateur SEO trilingue (Sprint 6 · SEO).")
    sub = p.add_subparsers(dest="cmd", required=True)

    pb = sub.add_parser("build", help="genere seo_keywords/schema_org/hreflang + MANIFEST")
    pb.add_argument("--master", default=None, help="projets_master.json (defaut: fixtures/)")
    pb.add_argument("-o", "--out", default=_DEFAULT_OUT, help="dossier de sortie (defaut: ./out)")
    pb.set_defaults(func=cmd_build)

    pv = sub.add_parser("validate", help="valide le bundle (schema + 15 invariants) sans ecrire")
    pv.add_argument("--master", default=None, help="projets_master.json (defaut: fixtures/)")
    pv.set_defaults(func=cmd_validate)

    args = p.parse_args(argv)
    return args.func(args)


if __name__ == "__main__":
    raise SystemExit(main())