prospection/cli.py

# prospection/cli.py
"""prospection/cli.py — point d'entrée unique du sous-système.

Aucune sous-commande n'envoie d'email : la sortie terminale est un fichier
.eml sur disque, relu et expédié par l'humain depuis sa propre messagerie.
"""
from __future__ import annotations

import argparse
import sys
from datetime import date
from pathlib import Path

from prospection import config, scoring
from prospection.demo import build_demo as demo_module
from prospection.enrich import registry, waterfall
from prospection.outreach import compliance, compose as compose_module, eml
from prospection.outreach.templates import email_fr
from prospection.sources import amf_registry, entreprises_api, geco_funds, holdings, signals
from prospection.store import prospects_db as db

DEMO_DIR = Path("data/demos")


def _cmd_universe_refresh(args) -> int:
    db.init_db()
    rows = amf_registry.parse_rows(amf_registry.fetch_csv())
    kept = 0
    failures = 0
    for row in rows:
        in_scope, _reason = amf_registry.is_in_scope(row)
        if not in_scope:
            continue
        try:
            company = entreprises_api.fetch_company(row["siren"]) if row.get("siren") else None
        except Exception:
            # L'API entreprises est publique et sans SLA. Une erreur dégrade la
            # couverture d'un compte ; elle n'interrompt pas un refresh de 700 lignes.
            # L'échec est compté et affiché, jamais avalé en silence.
            company = None
            failures += 1
        db.upsert_account({
            **row,
            "activites": row.get("activite"),
            "effectif": (company or {}).get("effectif"),
            "ville": (company or {}).get("ville"),
            "dirigeants": (company or {}).get("dirigeants") or [],
            "independante": int(amf_registry.is_independent(row["nom"])),
        })
        kept += 1
    print(f"{kept} SGP retenues sur {len(rows)} lignes AMF")
    if failures:
        print(f"{failures} enrichissements SIREN en échec "
              f"(effectif et dirigeants manquants)")
    return 0


def _cmd_signals_scan(args) -> int:
    today = date.today()
    added = 0
    for account in db.list_accounts():
        for row in filter(None, [signals.licence_signal(account, today),
                                 *signals.scan_careers_page(account, today)]):
            db.add_signal(row)
            added += 1
    print(f"{added} signaux enregistrés")
    return 0


def _score_inputs(account: dict) -> tuple[bool, bool, list[dict]]:
    equity = geco_funds.has_equity_funds(account)
    overlap = any(h.get("ticker") for h in db.confirmed_holdings(account["id"]))
    return equity, overlap, db.active_signals(account["id"], date.today())


def _cmd_score(args) -> int:
    accounts = db.list_accounts()
    if not args.final:
        results = []
        for account in accounts:
            equity, overlap, active = _score_inputs(account)
            results.append((account["id"], scoring.score_pass1(account, equity, overlap, active)))
        tiers = scoring.assign_provisional(results)
    else:
        results = []
        for account in accounts:
            equity, overlap, active = _score_inputs(account)
            base = scoring.score_pass1(account, equity, overlap, active)
            contacts = db.list_contacts(account["id"])
            results.append((account["id"], scoring.score_pass2(base, contacts), contacts))
        tiers = scoring.assign_final(results)

    for account in accounts:
        if account.get("demo_blocked_at") and tiers.get(account["id"]) == 1:
            # Aucun brief livrable pour ce compte : le rescore ne doit pas
            # annuler le constat de la démo. `assign_final` n'en sait rien, et
            # la séquence documentée lance `score --final` deux fois.
            tiers[account["id"]] = 2

    by_id = {r[0]: r[1] for r in results}
    for account_id, tier in tiers.items():
        result = by_id[account_id]
        db.set_tier(account_id, tier, result.total, result.reasons)
    print(f"Tier 1 : {sum(1 for t in tiers.values() if t == 1)} · "
          f"Tier 2 : {sum(1 for t in tiers.values() if t == 2)}")
    return 0


def _cmd_contacts_enrich(args) -> int:
    providers = registry.build_providers()
    total = 0
    for account in db.list_accounts(tier=1):
        for record in waterfall.run_waterfall(account, providers):
            db.upsert_contact(record.as_row(account["id"]))
            total += 1
    print(f"{total} contacts enregistrés")
    return 0


def _cmd_holdings_discover(args) -> int:
    found = 0
    for account in db.list_accounts(tier=args.tier):
        for url in holdings.discover_report_urls(account):
            print(f"{account['nom']} : {url}")
            found += 1
    print(f"{found} sources candidates — téléchargez et passez `holdings confirm`")
    return 0


def _cmd_holdings_confirm(args) -> int:
    db.confirm_holding(args.holding_id, date.today().isoformat())
    print(f"position {args.holding_id} confirmée")
    return 0


def _cmd_demo_build(args) -> int:
    built = 0
    for account in db.list_accounts(tier=args.tier):
        if demo_module.build_demo(account["id"], DEMO_DIR):
            built += 1
    print(f"{built} démos produites (les briefs non livrables sont écartés)")
    return 0


# Ordre d'interlocuteur de la spec. Dans une boutique, le gérant fondateur
# décide seul ; le directeur de la gestion est l'utilisateur réel ; le COO tient
# le budget mais n'est pas un point d'entrée ; le RCCI ne peut que dire non.
_CONTACT_PRIORITY = (
    (0, ("fondateur", "gérant", "gerant", "cio", "chief investment",
         "président", "president", "directeur général", "directeur general")),
    (1, ("head of research", "directeur de la gestion", "directrice de la gestion",
         "responsable de la recherche", "analyste")),
    (2, ("coo", "chief operating", "secrétaire général", "secretaire general",
         "directeur administratif")),
    (3, ("rcci", "conformité", "conformite", "compliance")),
)


def _contact_rank(contact: dict) -> tuple[int, int]:
    """Rang d'interlocuteur, puis qualité de l'adresse. Plus petit est meilleur."""
    haystack = " ".join(
        filter(None, (contact.get("titre"), contact.get("seniorite")))
    ).lower()
    rank = 9
    for level, markers in _CONTACT_PRIORITY:
        if any(marker in haystack for marker in markers):
            rank = level
            break
    reach = {"verified": 0, "probable": 1}.get(contact.get("email_status"), 2)
    return rank, reach


def _cmd_campaign_draft(args) -> int:
    try:
        # Une seule vérification, avant la boucle. Si l'identité de l'expéditeur
        # est incomplète, CHAQUE compte échouera de la même façon : mieux vaut
        # refuser la commande que rendre quarante fois la même erreur et laisser
        # une campagne à moitié rédigée.
        compliance.render_rgpd_block()
    except compliance.IncompleteSenderIdentity as exc:
        print(f"rédaction impossible : {exc}")
        return 1

    drafted, refused = 0, 0
    for account in db.list_accounts(tier=args.tier):
        level = holdings.evidence_level(account["id"])
        confirmed = db.confirmed_holdings(account["id"])
        # `evidence_level` renvoie `holdings_verified` dès qu'une position est
        # confirmée : quand il dit `universe_inferred`, il n'y en a aucune, et
        # passer `confirmed[0]` donnerait `None`. La position à citer est alors
        # la meilleure NON confirmée — c'est exactement ce que ce niveau désigne.
        holding = confirmed[0] if confirmed else None
        if level == "universe_inferred" and holding is None:
            holding = db.best_unconfirmed_holding(account["id"])

        demo_row = db.latest_demo(account["id"])
        if not (demo_row and demo_row.get("pdf_path")):
            # Le gabarit affirme que le brief est joint. Sans PDF produit, le
            # message affirmerait quelque chose de faux avant même d'être relu.
            print(f"écarté — {account['nom']} : aucune démo livrable")
            refused += 1
            continue

        named = sorted(
            (c for c in db.list_contacts(account["id"]) if not c.get("is_generic")),
            key=_contact_rank,
        )
        if not named:
            continue
        # Un seul destinataire par maison. Deux emails simultanés dans une
        # boutique de quinze personnes, ils s'en parlent, et ça devient un mailing.
        contact = named[0]

        bundle = compose_module.EvidenceBundle(
            account=account, contact=contact, evidence=level,
            holding=holding,
            demo=demo_row,
            signals=db.active_signals(account["id"], date.today()),
        )
        try:
            compliance.assert_sendable(contact)
            body = compose_module.compose(bundle)
        except (compliance.SuppressedContactError, compose_module.ComposeRejected) as exc:
            print(f"écarté — {account['nom']} : {exc}")
            refused += 1
            continue
        subject = email_fr.subject_for(level, {
            "ligne": (bundle.holding or {}).get("raw_label", ""),
            "fonds": (bundle.holding or {}).get("fund_name", ""),
        })
        db.upsert_touch({"contact_id": contact["id"], "channel": "email",
                         "template_id": "email_1", "subject": subject, "body": body,
                         "evidence_refs": [(bundle.holding or {}).get("source_url")],
                         "status": "draft"})
        drafted += 1
    print(f"{drafted} brouillons · {refused} écartés")
    return 0


def _cmd_campaign_review(args) -> int:
    for touch in db.list_touches("draft"):
        print(f"\n--- touche {touch['id']} · {touch['subject']} ---\n{touch['body']}")
    print("\nApprouvez avec : campaign approve --id <n>")
    return 0


def _cmd_campaign_approve(args) -> int:
    with db.connect() as conn:
        conn.execute("UPDATE touches SET status='approved' WHERE id=?", (args.id,))
    print(f"touche {args.id} approuvée")
    return 0


def _cmd_campaign_export(args) -> int:
    out_dir = Path(args.out)
    out_dir.mkdir(parents=True, exist_ok=True)
    exported = 0
    for touch in db.list_touches("approved"):
        with db.connect() as conn:
            contact = dict(conn.execute("SELECT * FROM contacts WHERE id=?",
                                        (touch["contact_id"],)).fetchone())
        try:
            compliance.assert_sendable(contact)
        except compliance.SuppressedContactError as exc:
            print(f"non exporté : {exc}")
            continue
        demo_row = db.latest_demo(contact["account_id"])
        attachment = (DEMO_DIR / demo_row["pdf_path"]) if demo_row and demo_row["pdf_path"] else None
        try:
            payload = eml.render_eml(touch["subject"], touch["body"],
                                     contact["email"], attachment)
        except (ValueError, OSError) as exc:
            # Adresse manquante, ou pièce jointe disparue du disque. Une touche
            # qui échoue ne doit pas emporter toutes les suivantes.
            print(f"non exporté (touche {touch['id']}) : {exc}")
            continue
        (out_dir / f"{touch['id']}.eml").write_bytes(payload)
        with db.connect() as conn:
            conn.execute("UPDATE touches SET status='exported' WHERE id=?", (touch["id"],))
        exported += 1
    print(f"{exported} fichiers .eml écrits dans {out_dir}")
    return 0


def _cmd_report(args) -> int:
    accounts = db.list_accounts()
    counts = {tier: sum(1 for a in accounts if a.get("tier") == tier) for tier in (1, 2, 0)}
    print(f"Comptes : {len(accounts)}")
    print(f"  Tier 1 : {counts[1]} (plafond {config.TIER1_FINAL_CAP})")
    print(f"  Tier 2 : {counts[2]}")
    print(f"  Exclus : {counts[0]}")
    for status in ("draft", "approved", "exported", "replied"):
        print(f"  Touches {status} : {len(db.list_touches(status))}")
    return 0


def _cmd_optout_add(args) -> int:
    compliance.record_optout(args.email)
    print(f"opposition enregistrée pour {args.email} (haché, adresse effacée)")
    return 0


def _cmd_purge(args) -> int:
    print(compliance.purge(date.today()))
    return 0


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="prospection")
    sub = parser.add_subparsers(dest="group", required=True)

    universe = sub.add_parser("universe").add_subparsers(dest="action", required=True)
    universe.add_parser("refresh").set_defaults(func=_cmd_universe_refresh)

    signals_p = sub.add_parser("signals").add_subparsers(dest="action", required=True)
    signals_p.add_parser("scan").set_defaults(func=_cmd_signals_scan)

    score_p = sub.add_parser("score")
    score_p.add_argument("--final", action="store_true")
    score_p.set_defaults(func=_cmd_score)

    contacts = sub.add_parser("contacts").add_subparsers(dest="action", required=True)
    contacts.add_parser("enrich").set_defaults(func=_cmd_contacts_enrich)

    holdings_p = sub.add_parser("holdings").add_subparsers(dest="action", required=True)
    discover = holdings_p.add_parser("discover")
    # Amorçage : le Tier 1 exige un recouvrement lu dans les positions
    # confirmées (`_score_inputs`), mais on ne confirme que ce qu'on a d'abord
    # découvert. Viser le Tier 1 par défaut ne trouverait jamais rien sur une
    # base vierge — personne n'y est jamais promu. Le Tier 2 ne demande qu'une
    # poche actions déclarée : il se peuple dès la première passe et rompt le
    # cercle. Voir « Amorçage » dans prospection/README.md.
    discover.add_argument("--tier", type=int, default=2)
    discover.set_defaults(func=_cmd_holdings_discover)
    confirm = holdings_p.add_parser("confirm")
    confirm.add_argument("holding_id", type=int)
    confirm.set_defaults(func=_cmd_holdings_confirm)

    demo_p = sub.add_parser("demo").add_subparsers(dest="action", required=True)
    build = demo_p.add_parser("build")
    build.add_argument("--tier", type=int, default=1)
    build.set_defaults(func=_cmd_demo_build)

    campaign = sub.add_parser("campaign").add_subparsers(dest="action", required=True)
    draft = campaign.add_parser("draft")
    draft.add_argument("--tier", type=int, default=1)
    draft.set_defaults(func=_cmd_campaign_draft)
    campaign.add_parser("review").set_defaults(func=_cmd_campaign_review)
    approve = campaign.add_parser("approve")
    approve.add_argument("--id", type=int, required=True)
    approve.set_defaults(func=_cmd_campaign_approve)
    export = campaign.add_parser("export")
    export.add_argument("--out", default="data/outbox")
    export.set_defaults(func=_cmd_campaign_export)

    sub.add_parser("report").set_defaults(func=_cmd_report)

    optout = sub.add_parser("optout").add_subparsers(dest="action", required=True)
    add = optout.add_parser("add")
    add.add_argument("email")
    add.set_defaults(func=_cmd_optout_add)

    sub.add_parser("purge").set_defaults(func=_cmd_purge)
    return parser


def main(argv: list[str] | None = None) -> int:
    args = _build_parser().parse_args(argv)
    return args.func(args)


if __name__ == "__main__":
    sys.exit(main())