Spaces:
Sleeping
Sleeping
"""
Connexion Supabase, matching, scoring et valorisation.
Portage clean de temp_old/database.py (sans Streamlit).
"""
from __future__ import annotations
import csv
import math
import os
import re
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any
import numpy as np
from supabase import Client, create_client
from backend.env import load_backend_env
from backend.mistral_compat import build_mistral_client, create_embeddings
from backend.config import (
ACTIVITY_THRESHOLDS,
CA_MULTIPLE_SOFTWARE_AI_BONUS,
CA_RANGES,
MATCH_CANDIDATES_COUNT,
MATCH_SIMILARITY_THRESHOLD,
MIN_CA_MULTIPLE_VALUATION,
MIN_EBITDA_MULTIPLE_VALUATION,
MAX_CA_MULTIPLE_VALUATION,
MAX_EBITDA_MULTIPLE_VALUATION,
MIN_SEMANTIC_MATCH,
MISTRAL_EMBED_MODEL,
SIZE_TOLERANCE,
SOFTWARE_AI_BONUS_KEYWORDS,
TOP_COMPARABLES_DISPLAY,
TRANSACTION_TYPES_COMPARABLE,
VALUATION_MIN_COUNT,
VALUATION_MIN_SEMANTIC,
WEIGHT_CA_NO_NAF,
WEIGHT_CA_PROXIMITY,
WEIGHT_NAF_PROXIMITY,
WEIGHT_SEMANTIC_NO_NAF,
WEIGHT_SEMANTIC_SIMILARITY,
)
# ============================================================================
# ETAT GLOBAL / INITIALISATION
# ---------------------------------------------------------------------------
# - nom de table de suivi d'usage
# - etat memoire des quotas (utile pour API / debug)
# - chargement .env des l'import module
# ============================================================================
USAGE_LOGS_TABLE = "usage_logs_m_a_transaction_app"
DEBUG_LOGS_DIR = Path(__file__).resolve().parent.parent / "debug_logs"
_usage_policy_state = {"blocked_reason": None, "minute_count": 0, "day_count": 0}
# Charge .env au demarrage module pour tous les appels backend.
load_backend_env()
# ============================================================================
# VARIABLES D'ENVIRONNEMENT
# ---------------------------------------------------------------------------
# Helpers robustes pour lire des secrets:
# - nettoyage quotes/commentaires inline
# - erreurs homogenes si variable manquante
# ============================================================================
def _sanitize_env_value(raw: str) -> str:
# Nettoie les valeurs .env: espaces, comments inline, quotes.
value = (raw or "").strip()
# Tolere les .env avec commentaires inline: KEY="value" # comment
value = re.sub(r"\s+#.*$", "", value).strip()
value = value.strip("'\"")
return value
def _get_env(key: str, required: bool = True, default: str = "") -> str:
# Point unique de lecture env (messages d'erreur coherents).
value = _sanitize_env_value(os.getenv(key) or default or "")
if required and not value:
raise ValueError(f"{key} doit etre configure en variable d'environnement")
return value
# ============================================================================
# CLIENTS EXTERNES
# ---------------------------------------------------------------------------
# Fabriques de clients API (Supabase + Mistral).
# ============================================================================
def get_supabase_client() -> Client:
# Fallback de cle pour faciliter les differents environnements.
key = _get_env(
"SUPABASE_KEY",
required=False,
default=_sanitize_env_value(os.getenv("SUPABASE_ANON_KEY") or os.getenv("SUPABASE_SERVICE_ROLE_KEY") or ""),
)
return create_client(_get_env("SUPABASE_URL"), _get_env("SUPABASE_KEY", default=key))
def get_mistral_client() -> Any:
# Client embeddings/LLM partage, initialise a la demande.
return build_mistral_client(_get_env("MISTRAL_API_KEY"))
def _extract_total_tokens(usage_obj) -> int:
# Compatibilite defensive: SDK Mistral peut renvoyer objet ou dict.
if usage_obj is None:
return 0
if hasattr(usage_obj, "total_tokens"):
return int(getattr(usage_obj, "total_tokens") or 0)
if isinstance(usage_obj, dict):
return int(usage_obj.get("total_tokens") or 0)
return 0
# ============================================================================
# QUOTAS / LOGS D'USAGE
# ---------------------------------------------------------------------------
# Ce bloc:
# 1) verifie (ou cree) la table technique d'usage
# 2) controle les fenetres minute/jour
# 3) enregistre puis complete la ligne de suivi par requete
# ============================================================================
def _ensure_usage_logs_table(supabase: Client) -> bool:
# Verifie la table quota/logs; tente creation via RPC SQL si absente.
try:
supabase.table(USAGE_LOGS_TABLE).select("id", count="exact").limit(1).execute()
return True
except Exception as exc:
err = str(exc).lower()
missing_table = "does not exist" in err or "relation" in err or "42p01" in err
if not missing_table:
print(f"[SECURITY] Verification table usage_logs impossible : {exc}")
return False
create_sql = f"""
CREATE TABLE IF NOT EXISTS public.{USAGE_LOGS_TABLE} (
id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
user_lang TEXT,
target_url TEXT,
ca_saisi NUMERIC,
ebitda_saisi NUMERIC,
siren TEXT,
mots_cles_activites TEXT,
tokens_llm BIGINT,
tokens_embed BIGINT
);
"""
rpc_candidates = (
("exec_sql", {"sql": create_sql}),
("exec_sql", {"query": create_sql}),
("execute_sql", {"sql": create_sql}),
("run_sql", {"sql": create_sql}),
)
for rpc_name, payload in rpc_candidates:
try:
supabase.rpc(rpc_name, payload).execute()
supabase.table(USAGE_LOGS_TABLE).select("id", count="exact").limit(1).execute()
return True
except Exception:
continue
return False
def get_usage_policy_state() -> dict:
# Expose un snapshot lisible pour debug frontend/ops.
return dict(_usage_policy_state)
def check_usage_policy(limit_per_minute: int, limit_per_day: int) -> bool:
# Controle anti-abus calcule sur fenetres glissantes (60s / 24h).
supabase = get_supabase_client()
_usage_policy_state["blocked_reason"] = None
_usage_policy_state["minute_count"] = 0
_usage_policy_state["day_count"] = 0
if not _ensure_usage_logs_table(supabase):
_usage_policy_state["blocked_reason"] = "error"
return False
now = datetime.utcnow()
# Fenetres glissantes: 60 secondes et 24 heures.
minute_cutoff = (now - timedelta(seconds=60)).isoformat()
day_cutoff = (now - timedelta(hours=24)).isoformat()
try:
minute_result = (
supabase.table(USAGE_LOGS_TABLE)
.select("id", count="exact")
.gte("created_at", minute_cutoff)
.execute()
)
day_result = (
supabase.table(USAGE_LOGS_TABLE)
.select("id", count="exact")
.gte("created_at", day_cutoff)
.execute()
)
minute_count = minute_result.count if minute_result.count is not None else 0
day_count = day_result.count if day_result.count is not None else 0
_usage_policy_state["minute_count"] = minute_count
_usage_policy_state["day_count"] = day_count
if minute_count >= limit_per_minute:
_usage_policy_state["blocked_reason"] = "minute"
return False
if day_count >= limit_per_day:
_usage_policy_state["blocked_reason"] = "day"
return False
return True
except Exception as exc:
print(f"[SECURITY] Erreur controle quota : {exc}")
_usage_policy_state["blocked_reason"] = "error"
return False
def register_usage_request(
limit_per_day: int,
user_lang: str | None = None,
target_url: str | None = None,
ca_saisi: float | None = None,
ebitda_saisi: float | None = None,
) -> int | None:
# Cree une ligne de log des l'entree de pipeline (avant IA/scraping).
# Cette ligne est ensuite enrichie en fin de traitement via update_usage_tokens().
supabase = get_supabase_client()
if not _ensure_usage_logs_table(supabase):
return None
try:
insert_result = (
supabase.table(USAGE_LOGS_TABLE)
.insert(
{
"created_at": datetime.utcnow().isoformat(),
"user_lang": user_lang,
"target_url": target_url,
"ca_saisi": ca_saisi,
"ebitda_saisi": ebitda_saisi,
"siren": None,
"mots_cles_activites": None,
"tokens_llm": None,
"tokens_embed": None,
}
)
.execute()
)
row_id = None
if insert_result.data and len(insert_result.data) > 0:
row_id = insert_result.data[0].get("id")
if row_id is None:
return None
next_day_count = int(_usage_policy_state.get("day_count", 0)) + 1
print(f"[SECURITY] Requete autorisee | Etat du quota : {next_day_count}/{limit_per_day}")
return row_id
except Exception as exc:
print(f"[SECURITY] Erreur enregistrement quota : {exc}")
return None
def update_usage_tokens(
log_id: int | None,
tokens_llm: int | None = None,
tokens_embed: int | None = None,
siren: str | None = None,
mots_cles_activites: str | None = None,
) -> None:
# Finalise la ligne de log avec metadonnees/tokens reels.
if not log_id:
return
payload = {
"tokens_llm": tokens_llm if tokens_llm is not None else 0,
"tokens_embed": tokens_embed if tokens_embed is not None else 0,
"siren": siren,
"mots_cles_activites": mots_cles_activites,
}
try:
get_supabase_client().table(USAGE_LOGS_TABLE).update(payload).eq("id", log_id).execute()
except Exception as exc:
print(f"[SECURITY] Erreur update tokens quota (id={log_id}) : {exc}")
# ============================================================================
# EMBEDDING + RECUPERATION CANDIDATS
# ---------------------------------------------------------------------------
# Generation embedding cible puis recherche de transactions candidates
# (RPC pgvector nominale, fallback Python en secours).
# ============================================================================
def generate_query_embedding(keywords: str) -> tuple[list, int]:
# Embedding unique de la cible pour rechercher des comparables.
if not keywords or not keywords.strip():
keywords = "entreprise non renseignee"
client = get_mistral_client()
response = create_embeddings(client, model=MISTRAL_EMBED_MODEL, inputs=[keywords])
if hasattr(response, "data") and response.data:
tokens_embed = _extract_total_tokens(getattr(response, "usage", None))
return response.data[0].embedding, tokens_embed
if isinstance(response, dict):
data = response.get("data") or []
if data and isinstance(data[0], dict) and data[0].get("embedding"):
tokens_embed = _extract_total_tokens(response.get("usage"))
return data[0]["embedding"], tokens_embed
raise RuntimeError("Reponse embeddings Mistral invalide.")
def filter_by_transaction_type(candidates: list) -> list:
# Limite l'univers aux types d'operations juges comparables.
if not TRANSACTION_TYPES_COMPARABLE:
return candidates
allowed = [t.lower() for t in TRANSACTION_TYPES_COMPARABLE]
return [c for c in candidates if (c.get("transaction_type") or "").strip().lower() in allowed]
def fetch_all_transactions_fallback(query_embedding: list, limit: int | None = None) -> list:
# Plan B si RPC indisponible: similarite cosinus calculee en Python.
try:
result = (
get_supabase_client()
.table("db_transactions")
.select(
"id_target, date_transaction, transaction_type, secteur, "
"code_naf_2, mots_cles_activites, ca_range, ca_midpoint, "
"multiple_ca, multiple_ebitda, geographie, embedding"
)
.execute()
)
if not result.data:
return []
# Similarite cosinus locale quand la RPC pgvector n'est pas accessible.
query_vec = np.array(query_embedding)
query_norm = np.linalg.norm(query_vec)
candidates = []
for row in result.data:
if row.get("embedding"):
emb_vec = np.array(row["embedding"])
emb_norm = np.linalg.norm(emb_vec)
similarity = float(np.dot(query_vec, emb_vec) / (query_norm * emb_norm)) if query_norm > 0 and emb_norm > 0 else 0.0
row["similarity"] = similarity
del row["embedding"]
candidates.append(row)
candidates.sort(key=lambda x: x["similarity"], reverse=True)
cap = limit if limit is not None else MATCH_CANDIDATES_COUNT
return candidates[:cap]
except Exception as exc:
print(f"Erreur fallback : {exc}")
return []
def fetch_candidates(query_embedding: list, match_count: int | None = None, apply_type_filter: bool = True) -> list:
# Chemin nominal: RPC pgvector; fallback Python si erreur.
if match_count is None:
match_count = MATCH_CANDIDATES_COUNT
supabase = get_supabase_client()
def _has_valid_ebitda_multiple(row: dict) -> bool:
value = as_number(row.get("multiple_ebitda"))
return value is not None and value > 0
try:
result = (
supabase.rpc(
"match_transactions",
{"query_embedding": query_embedding, "match_count": match_count, "match_threshold": MATCH_SIMILARITY_THRESHOLD},
).execute()
)
candidates = result.data if result.data else []
except Exception:
# Degradation gracieuse pour eviter un hard fail cote utilisateur.
candidates = fetch_all_transactions_fallback(query_embedding, limit=match_count)
# Priorite multiples EBITDA:
# - conserve la base semantique principale
# - complete avec des candidats moins semantiques mais munis d'un multiple EBITDA
# pour augmenter la probabilite de valorisation calculable.
primary_ids = {str(c.get("id_target") or id(c)) for c in candidates}
ebitda_count = sum(1 for c in candidates if _has_valid_ebitda_multiple(c))
target_ebitda_count = max(VALUATION_MIN_COUNT * 4, 24)
if ebitda_count < target_ebitda_count:
expanded_pool = fetch_all_transactions_fallback(query_embedding, limit=max(match_count * 3, 6000))
supplemental: list[dict] = []
for row in expanded_pool:
key = str(row.get("id_target") or id(row))
if key in primary_ids:
continue
if _has_valid_ebitda_multiple(row):
supplemental.append(row)
primary_ids.add(key)
if ebitda_count + len(supplemental) >= target_ebitda_count:
break
candidates.extend(supplemental)
# Tri final: priorite aux lignes avec multiple EBITDA puis semanticite.
candidates.sort(
key=lambda c: (
1 if _has_valid_ebitda_multiple(c) else 0,
float(c.get("similarity", 0.0) or 0.0),
),
reverse=True,
)
if apply_type_filter:
return filter_by_transaction_type(candidates)
return candidates
# ============================================================================
# SCORING DE COMPARABILITE
# ---------------------------------------------------------------------------
# Sous-scores puis score total pondere:
# - proximite CA
# - proximite NAF
# - similarite semantique
# ============================================================================
def ca_proximity_score(target_ca_meur: float, transaction_midpoint: float) -> float:
# Score logarithmique: penalise ecarts importants sans annuler completement.
if not target_ca_meur or target_ca_meur <= 0:
return 0.3
if not transaction_midpoint or transaction_midpoint <= 0:
return 0.3
ratio = max(target_ca_meur, transaction_midpoint) / min(target_ca_meur, transaction_midpoint)
return max(0.0, 1.0 - math.log10(ratio) / 1.5)
def naf_proximity_score(target_naf_2: str, target_naf_full: str, transaction_naf_2: str, transaction_naf_full: str) -> float:
# Priorite: match 3 digits > 2 digits > 1 digit > mismatch.
if not target_naf_2 or not transaction_naf_2:
return 0.5
t_naf = str(target_naf_full).replace(".", "").replace(" ", "") if target_naf_full else ""
tx_naf = str(transaction_naf_full).replace(".", "").replace(" ", "") if transaction_naf_full else ""
if len(t_naf) >= 3 and len(tx_naf) >= 3 and t_naf[:3] == tx_naf[:3]:
return 1.0
if target_naf_2 == transaction_naf_2:
return 0.7
if target_naf_2[0] == transaction_naf_2[0]:
return 0.4
return 0.1
def compute_size_delta_percent(target_ca_meur: float | None, transaction_midpoint: float | None) -> float | None:
# Delta relatif de taille utilise pour le hard stop (None = taille inconnue).
if not target_ca_meur or target_ca_meur <= 0:
return None
if not transaction_midpoint or transaction_midpoint <= 0:
return None
return abs(float(transaction_midpoint) - float(target_ca_meur)) / float(target_ca_meur)
def build_quality_label(weighted_final_score: float, size_delta_percent: float | None) -> str:
# Qualite metier du match exposee au frontend/audit.
if size_delta_percent is None:
return "Match Taille Inconnue"
if weighted_final_score > 0.9:
return "Or"
if weighted_final_score > 0.8:
return "Argent"
if weighted_final_score > 0.75:
return "Bronze"
return "Non classe"
def _append_rejection_reason(reasons: list[str], reason: str) -> None:
if reason not in reasons:
reasons.append(reason)
def as_number(value: object) -> float | None:
try:
number = float(value)
except (TypeError, ValueError):
return None
if math.isnan(number) or math.isinf(number):
return None
return number
def _sanitize_target_name_for_filename(target_name: str | None) -> str:
# Nettoyage defensif pour noms de fichiers cross-platform (Windows/Linux).
raw = (target_name or "").strip().lower()
if not raw:
return "target"
cleaned = re.sub(r"[\\/:*?\"<>|]+", "_", raw)
cleaned = re.sub(r"\s+", "_", cleaned)
cleaned = re.sub(r"[^a-zA-Z0-9_-]+", "_", cleaned)
cleaned = re.sub(r"_+", "_", cleaned).strip("_-")
if not cleaned:
return "target"
# Limite preventive pour eviter les chemins trop longs.
return cleaned[:80]
def _rotate_debug_csv_logs(max_files_after_create: int = 2) -> None:
# Rotation appelee juste avant creation d'un nouveau fichier.
# On garde (max - 1) fichiers existants pour que le nouveau reste dans la limite.
if max_files_after_create < 1:
return
DEBUG_LOGS_DIR.mkdir(parents=True, exist_ok=True)
csv_files = [path for path in DEBUG_LOGS_DIR.glob("*.csv") if path.is_file()]
csv_files.sort(key=lambda p: p.stat().st_ctime, reverse=True)
keep_existing = max(max_files_after_create - 1, 0)
for old_file in csv_files[keep_existing:]:
try:
old_file.unlink()
except Exception as exc:
print(f"[AUDIT] Impossible de supprimer {old_file}: {exc}")
def write_matching_audit_csv(rows: list[dict], target_name: str | None = None) -> str | None:
# Trace complete retention/rejet pour observabilite metier.
if not rows:
return None
DEBUG_LOGS_DIR.mkdir(parents=True, exist_ok=True)
_rotate_debug_csv_logs(max_files_after_create=2)
stamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S_%f")
safe_target_name = _sanitize_target_name_for_filename(target_name)
output_path = DEBUG_LOGS_DIR / f"audit_matching_{stamp}_{safe_target_name}_.csv"
display_target_name = (target_name or "target").strip() or "target"
fieldnames = [
"target_name",
"list_origin",
"status",
"in_display_track",
"in_valuation_track",
"rejection_code",
"rejection_reason",
"id_target",
"transaction_type",
"code_naf_2",
"ca_midpoint",
"multiple_ca",
"multiple_ebitda",
"vector_score",
"size_delta_percent",
"size_score",
"naf_match",
"naf_score",
"weighted_final_score",
"quality_label",
]
with output_path.open("w", encoding="utf-8", newline="") as csv_file:
writer = csv.DictWriter(csv_file, fieldnames=fieldnames)
writer.writeheader()
for row in rows:
writer.writerow({name: (display_target_name if name == "target_name" else row.get(name)) for name in fieldnames})
return str(output_path)
def compute_weighted_score(ca_score: float, semantic_score: float, naf_score: float, has_naf: bool = True) -> float:
# Deux jeux de poids: avec NAF disponible, ou sans NAF.
if has_naf:
return WEIGHT_CA_PROXIMITY * ca_score + WEIGHT_SEMANTIC_SIMILARITY * semantic_score + WEIGHT_NAF_PROXIMITY * naf_score
return WEIGHT_CA_NO_NAF * ca_score + WEIGHT_SEMANTIC_NO_NAF * semantic_score
def has_software_ai_bonus_keywords(target_keywords: str) -> bool:
# Detection mot-entier pour eviter les faux positifs sur sous-chaines.
if not target_keywords:
return False
text = str(target_keywords).lower()
patterns = [rf"\b{re.escape(keyword.lower())}\b" for keyword in SOFTWARE_AI_BONUS_KEYWORDS]
return any(re.search(p, text, flags=re.IGNORECASE) for p in patterns)
# ============================================================================
# ORCHESTRATION MATCHING / VALORISATION
# ---------------------------------------------------------------------------
# Pipeline principal:
# 1) embedding + candidats
# 2) scoring
# 3) relaxation progressive
# 4) calcul des multiples et indicateurs secteur
# ============================================================================
def find_comparable_transactions(
target_keywords: str,
target_ca_eur: float,
target_naf_code: str | None = None,
target_naf_2: str | None = None,
target_geo: str | None = None,
target_name: str | None = None,
) -> dict:
# Orchestrateur principal dual-track:
# - Track A (list_display): comparables business pour la page transactions
# - Track B (list_valuation): base elargie pour extraire assez de multiples
# Unite interne: CA en MEUR pour alignement avec les tranches historiques.
target_ca_meur = target_ca_eur / 1_000_000 if target_ca_eur else None
has_naf = bool(target_naf_2)
ca_bonus = CA_MULTIPLE_SOFTWARE_AI_BONUS if has_software_ai_bonus_keywords(target_keywords) else 0.0
query_embedding, tokens_embed = generate_query_embedding(target_keywords)
# Universe "large": sans filtre de type pour pouvoir forcer un quota minimal.
candidates_all = fetch_candidates(query_embedding, apply_type_filter=False)
if not candidates_all:
return {
"success": False,
"message": "Aucune transaction trouvee.",
"comparables": [],
"multiples": {},
"sector_activity": {},
"tokens_embed": tokens_embed,
}
allowed_types = {t.strip().lower() for t in TRANSACTION_TYPES_COMPARABLE}
DISPLAY_WEIGHT_SEMANTIC = 0.95
DISPLAY_WEIGHT_SIZE = 0.05
ELASTIC_SEMANTIC_STEP = 0.05
ELASTIC_MIN_SEMANTIC = 0.20
VALUATION_SIZE_START = 1.00
VALUATION_SIZE_STEP = 0.10
VALUATION_SIZE_MAX = 2.00
VALUATION_NAF_MIN = 0.10 if has_naf else 0.0
scored_candidates: list[dict] = []
audit_base_rows: list[dict] = []
for candidate in candidates_all:
audit_key = str(candidate.get("id_target") or id(candidate))
semantic_score = float(candidate.get("similarity", 0.0) or 0.0)
tx_type = str(candidate.get("transaction_type") or "").strip().lower()
is_comparable_type = tx_type in allowed_types if allowed_types else True
naf_code_candidate = candidate.get("code_naf_2")
naf_match = bool(target_naf_2 and naf_code_candidate and str(target_naf_2) == str(naf_code_candidate))
# Le score NAF reste un score de proximite pur (pas de bonus additionnel).
naf_score = naf_proximity_score(target_naf_2, target_naf_code, naf_code_candidate, naf_code_candidate)
size_delta_percent = compute_size_delta_percent(target_ca_meur, as_number(candidate.get("ca_midpoint")))
size_known = size_delta_percent is not None
size_score = ca_proximity_score(target_ca_meur, candidate.get("ca_midpoint")) if size_known else 0.0
if has_naf:
if size_known:
weighted_final_score = compute_weighted_score(size_score, semantic_score, naf_score, has_naf=True)
else:
# Taille inconnue: normalisation sur semantique + NAF uniquement.
denom = WEIGHT_SEMANTIC_SIMILARITY + WEIGHT_NAF_PROXIMITY
weighted_final_score = ((WEIGHT_SEMANTIC_SIMILARITY * semantic_score) + (WEIGHT_NAF_PROXIMITY * naf_score)) / denom if denom > 0 else semantic_score
else:
if size_known:
weighted_final_score = compute_weighted_score(size_score, semantic_score, naf_score, has_naf=False)
else:
weighted_final_score = semantic_score
# Score dedie a l'affichage metier: tres forte priorite a la semantique.
if size_known:
display_score = (DISPLAY_WEIGHT_SEMANTIC * semantic_score) + (DISPLAY_WEIGHT_SIZE * size_score)
else:
display_score = semantic_score
rejection_reasons: list[str] = []
rejection_codes: list[str] = []
if semantic_score < MIN_SEMANTIC_MATCH:
_append_rejection_reason(rejection_reasons, f"Semantic score {semantic_score:.2f} < {MIN_SEMANTIC_MATCH:.2f}")
rejection_codes.append("SEMANTIC_HARD_STOP")
quality_label = build_quality_label(weighted_final_score, size_delta_percent)
candidate["vector_score"] = round(semantic_score, 4)
candidate["size_delta_percent"] = round(size_delta_percent, 4) if size_delta_percent is not None else None
candidate["naf_match"] = naf_match
candidate["naf_score"] = round(naf_score, 4)
candidate["size_score"] = round(size_score, 4) if size_known else None
candidate["weighted_final_score"] = round(weighted_final_score, 4)
candidate["display_score"] = round(display_score, 4)
candidate["quality_label"] = quality_label
candidate["total_score"] = candidate["weighted_final_score"]
candidate["is_comparable_type"] = is_comparable_type
candidate["has_ca_multiple"] = as_number(candidate.get("multiple_ca")) is not None and as_number(candidate.get("multiple_ca")) > 0
candidate["has_ebitda_multiple"] = as_number(candidate.get("multiple_ebitda")) is not None and as_number(candidate.get("multiple_ebitda")) > 0
in_display = semantic_score >= MIN_SEMANTIC_MATCH and is_comparable_type
has_ca_multiple = as_number(candidate.get("multiple_ca")) is not None and as_number(candidate.get("multiple_ca")) > 0
has_ebitda_multiple = as_number(candidate.get("multiple_ebitda")) is not None and as_number(candidate.get("multiple_ebitda")) > 0
has_any_multiple = has_ca_multiple or has_ebitda_multiple
naf_ok = (not has_naf) or (naf_score >= VALUATION_NAF_MIN)
in_valuation = (
semantic_score >= VALUATION_MIN_SEMANTIC
and has_any_multiple
and (not size_known or size_delta_percent <= VALUATION_SIZE_START)
and naf_ok
)
candidate["match_status"] = "REJECTED"
scored_candidates.append(candidate)
audit_base_rows.append(
{
"id_target": candidate.get("id_target"),
"_audit_key": audit_key,
"transaction_type": candidate.get("transaction_type"),
"code_naf_2": candidate.get("code_naf_2"),
"ca_midpoint": candidate.get("ca_midpoint"),
"multiple_ca": candidate.get("multiple_ca"),
"multiple_ebitda": candidate.get("multiple_ebitda"),
"vector_score": candidate.get("vector_score"),
"size_delta_percent": candidate.get("size_delta_percent"),
"size_score": candidate.get("size_score"),
"naf_match": candidate.get("naf_match"),
"naf_score": candidate.get("naf_score"),
"weighted_final_score": candidate.get("weighted_final_score"),
"quality_label": candidate.get("quality_label"),
"in_display_track": in_display,
"in_valuation_track": in_valuation,
"rejection_code": " | ".join(rejection_codes),
"rejection_reason": " | ".join(rejection_reasons),
}
)
scored_candidates.sort(key=lambda x: x.get("weighted_final_score", 0), reverse=True)
display_ranked_candidates = sorted(scored_candidates, key=lambda x: float(x.get("display_score", 0.0) or 0.0), reverse=True)
valuation_ranked_candidates = sorted(scored_candidates, key=lambda x: float(x.get("weighted_final_score", 0.0) or 0.0), reverse=True)
def _candidate_id(c: dict) -> str:
return str(c.get("id_target") or id(c))
def _display_eligible(c: dict, semantic_threshold: float, require_comparable_type: bool) -> bool:
if require_comparable_type and not c.get("is_comparable_type"):
return False
if float(c.get("vector_score", 0.0) or 0.0) < semantic_threshold:
return False
return True
def _valuation_eligible(c: dict, semantic_threshold: float, size_threshold: float, naf_threshold: float) -> bool:
has_any_multiple = bool(c.get("has_ca_multiple")) or bool(c.get("has_ebitda_multiple"))
if not has_any_multiple:
return False
if float(c.get("vector_score", 0.0) or 0.0) < semantic_threshold:
return False
if has_naf and float(c.get("naf_score", 0.0) or 0.0) < naf_threshold:
return False
size_delta = c.get("size_delta_percent")
if size_delta is not None and float(size_delta) > size_threshold:
return False
return True
def _elastic_display_selection() -> list[dict]:
sem = MIN_SEMANTIC_MATCH
while True:
selected = [c for c in display_ranked_candidates if _display_eligible(c, sem, require_comparable_type=True)]
if len(selected) >= TOP_COMPARABLES_DISPLAY:
return selected
if sem <= ELASTIC_MIN_SEMANTIC:
break
sem = max(ELASTIC_MIN_SEMANTIC, round(sem - ELASTIC_SEMANTIC_STEP, 2))
return [c for c in display_ranked_candidates if _display_eligible(c, ELASTIC_MIN_SEMANTIC, require_comparable_type=True)]
def _elastic_valuation_selection() -> list[dict]:
sem = VALUATION_MIN_SEMANTIC
naf_min = VALUATION_NAF_MIN
# 1) Relaxation prioritaire de la taille (+10% par palier) a semantique constante.
size = VALUATION_SIZE_START
while True:
eligible = [c for c in valuation_ranked_candidates if _valuation_eligible(c, sem, size, naf_min)]
top_ca = [c for c in eligible if c.get("has_ca_multiple")][:VALUATION_MIN_COUNT]
top_ebitda = [c for c in eligible if c.get("has_ebitda_multiple")][:VALUATION_MIN_COUNT]
selected_map: dict[str, dict] = {}
for c in top_ca + top_ebitda:
selected_map[_candidate_id(c)] = c
if len(top_ca) >= VALUATION_MIN_COUNT and len(top_ebitda) >= VALUATION_MIN_COUNT:
return list(selected_map.values())
if size >= VALUATION_SIZE_MAX:
break
size = min(VALUATION_SIZE_MAX, round(size + VALUATION_SIZE_STEP, 2))
# 2) Puis relaxation de la semantique, en rebalayant la taille a chaque palier.
while sem > ELASTIC_MIN_SEMANTIC:
sem = max(ELASTIC_MIN_SEMANTIC, round(sem - ELASTIC_SEMANTIC_STEP, 2))
naf_min = max(0.0, round(naf_min - 0.05, 2))
size = VALUATION_SIZE_START
while True:
eligible = [c for c in valuation_ranked_candidates if _valuation_eligible(c, sem, size, naf_min)]
top_ca = [c for c in eligible if c.get("has_ca_multiple")][:VALUATION_MIN_COUNT]
top_ebitda = [c for c in eligible if c.get("has_ebitda_multiple")][:VALUATION_MIN_COUNT]
selected_map: dict[str, dict] = {}
for c in top_ca + top_ebitda:
selected_map[_candidate_id(c)] = c
if len(top_ca) >= VALUATION_MIN_COUNT and len(top_ebitda) >= VALUATION_MIN_COUNT:
return list(selected_map.values())
if size >= VALUATION_SIZE_MAX:
break
size = min(VALUATION_SIZE_MAX, round(size + VALUATION_SIZE_STEP, 2))
# Retourne le meilleur pool disponible meme si incomplet.
eligible = [c for c in valuation_ranked_candidates if _valuation_eligible(c, ELASTIC_MIN_SEMANTIC, VALUATION_SIZE_MAX, 0.0)]
top_ca = [c for c in eligible if c.get("has_ca_multiple")][:VALUATION_MIN_COUNT]
top_ebitda = [c for c in eligible if c.get("has_ebitda_multiple")][:VALUATION_MIN_COUNT]
selected_map: dict[str, dict] = {}
for c in top_ca + top_ebitda:
selected_map[_candidate_id(c)] = c
return list(selected_map.values())
# Baseline (non relaxe) pour tracer les ajouts "quota force".
base_display_ids = {
_candidate_id(c)
for c in display_ranked_candidates
if _display_eligible(c, MIN_SEMANTIC_MATCH, require_comparable_type=True)
}
base_valuation_ids = {
_candidate_id(c)
for c in valuation_ranked_candidates
if _valuation_eligible(c, VALUATION_MIN_SEMANTIC, VALUATION_SIZE_START, VALUATION_NAF_MIN)
}
list_display = _elastic_display_selection()
list_valuation = _elastic_valuation_selection()
for c in list_display:
cid = _candidate_id(c)
if cid not in base_display_ids:
c["quality_label"] = "Match Élargi (Quota Forcé)"
c["match_status"] = "RETAINED_DISPLAY"
for c in list_valuation:
cid = _candidate_id(c)
if cid not in base_valuation_ids:
c["quality_label"] = "Match Élargi (Quota Forcé)"
if c.get("match_status") != "RETAINED_DISPLAY":
c["match_status"] = "RETAINED_VALUATION_POOL"
# Construction finale des lignes d'audit avec statut enrichi.
status_by_id: dict[str, str] = {}
for c in list_display:
status_by_id[str(c.get("id_target") or id(c))] = "RETAINED_DISPLAY"
for c in list_valuation:
key = str(c.get("id_target") or id(c))
prev = status_by_id.get(key)
if prev == "RETAINED_DISPLAY":
status_by_id[key] = "RETAINED_DISPLAY+VALUATION_POOL"
elif prev is None:
status_by_id[key] = "RETAINED_VALUATION_POOL"
audit_rows: list[dict] = []
for row in audit_base_rows:
key = str(row.get("_audit_key"))
status = status_by_id.get(key, "REJECTED")
audit_rows.append(
{
"status": status,
"list_origin": (
"display+valuation"
if status == "RETAINED_DISPLAY+VALUATION_POOL"
else "display"
if status == "RETAINED_DISPLAY"
else "valuation"
if status == "RETAINED_VALUATION_POOL"
else "rejected"
),
"rejection_code": row.get("rejection_code", "") if status == "REJECTED" else "",
"rejection_reason": row.get("rejection_reason", "") if status == "REJECTED" else "",
**{k: v for k, v in row.items() if k not in {"rejection_code", "rejection_reason", "_audit_key"}},
}
)
audit_file = write_matching_audit_csv(audit_rows, target_name=target_name)
if not list_display and not list_valuation:
return {
"success": False,
"message": "Aucun deal ne respecte les criteres de matching ni de valorisation.",
"comparables": [],
"multiples": {},
"sector_activity": {},
"tokens_embed": tokens_embed,
"audit_csv": audit_file,
}
if ca_bonus > 0:
# Premium CA pour profils software/AI detectes dans les mots-cles.
bonus_targets = {str(c.get("id_target") or id(c)): c for c in list_display + list_valuation}
for c in bonus_targets.values():
if c.get("multiple_ca") is not None and c["multiple_ca"] > 0:
c["multiple_ca"] = round(float(c["multiple_ca"]) + ca_bonus, 2)
# Valorisation dual-track:
# - base primaire: list_display
# - complement: list_valuation (seuil semantique elargi, multiples presents).
multiples = compute_multiples(
display_transactions=list_display,
valuation_transactions=list_valuation,
target_ca_eur=target_ca_eur,
target_ca_meur=target_ca_meur,
ca_bonus=ca_bonus,
)
sector_activity = count_sector_activity(target_naf_2)
return {
"success": True,
"comparables": list_display,
"top_comparables": list_display[:TOP_COMPARABLES_DISPLAY],
"multiples": multiples,
"sector_activity": sector_activity,
"total_candidates": len(scored_candidates),
"filtered_count": len(list_display),
"retained_strict_count": len(list_display),
"list_display_count": len(list_display),
"list_valuation_count": len(list_valuation),
"valuation_scope": multiples.get("valuation_scope", "listing stricte"),
"relaxation_level": "Dual-Track Retrieval",
"tokens_embed": tokens_embed,
"ca_bonus_applied": ca_bonus > 0,
"audit_csv": audit_file,
"matching_mode": "dual_track",
}
# ============================================================================
# CALCULS FINANCIERS
# ---------------------------------------------------------------------------
# Fonctions de calcul de multiples et de valorisation derivee.
# ============================================================================
def compute_multiples(
display_transactions: list,
valuation_transactions: list,
target_ca_eur: float,
target_ca_meur: float,
ca_bonus: float = 0.0,
) -> dict:
# Dual-track valuation:
# 1) prend les multiples de la list_display (track principal)
# 2) complete avec la list_valuation (track elargi) si < VALUATION_MIN_COUNT
# 3) retourne un contrat "zero null" + flag de scope pour le frontend.
_ = target_ca_meur # parametre conserve pour compatibilite signature/metier.
insufficient_reason = "Données insuffisantes même après élargissement"
def _sorted_unique(transactions: list) -> list:
seen: set[str] = set()
ordered = sorted(transactions, key=lambda x: float(x.get("weighted_final_score", 0.0) or 0.0), reverse=True)
out: list[dict] = []
for t in ordered:
key = str(t.get("id_target") or id(t))
if key in seen:
continue
seen.add(key)
out.append(t)
return out
def _is_valid_ca_multiple(t: dict) -> bool:
v = as_number(t.get("multiple_ca"))
return v is not None and v >= MIN_CA_MULTIPLE_VALUATION and v <= MAX_CA_MULTIPLE_VALUATION + ca_bonus
def _is_valid_ebitda_multiple(t: dict) -> bool:
v = as_number(t.get("multiple_ebitda"))
return v is not None and v >= MIN_EBITDA_MULTIPLE_VALUATION and v <= MAX_EBITDA_MULTIPLE_VALUATION
display_sorted = _sorted_unique(display_transactions)
valuation_sorted = _sorted_unique(valuation_transactions)
def _build_axis_values(is_valid_fn, field_name: str):
strict_values = [float(t[field_name]) for t in display_sorted if is_valid_fn(t)]
merged_values = list(strict_values)
strict_ids = {str(t.get("id_target") or id(t)) for t in display_sorted if is_valid_fn(t)}
for t in valuation_sorted:
key = str(t.get("id_target") or id(t))
if key in strict_ids:
continue
if is_valid_fn(t):
merged_values.append(float(t[field_name]))
return strict_values, merged_values
ca_values_strict, ca_values_final = _build_axis_values(_is_valid_ca_multiple, "multiple_ca")
ebitda_values_strict, ebitda_values_final = _build_axis_values(_is_valid_ebitda_multiple, "multiple_ebitda")
ca_scope_widened = len(ca_values_final) > len(ca_values_strict)
ebitda_scope_widened = len(ebitda_values_final) > len(ebitda_values_strict)
valuation_scope = "listing élargie" if (ca_scope_widened or ebitda_scope_widened) else "listing stricte"
def _strict_share_percent(strict_count: int, total_count: int) -> float:
if total_count <= 0:
return 0.0
return round((strict_count / total_count) * 100.0, 1)
def _calculate_stats(values_list: list[float]) -> dict:
# Stats backend de reference pour le frontend football field.
# Regle metier: si echantillon vide ou insuffisant, tout repasse a 0.
if len(values_list) < VALUATION_MIN_COUNT:
return {"min": 0.0, "max": 0.0, "median": 0.0, "mean": 0.0}
arr = np.array(values_list, dtype=float)
return {
"min": float(np.min(arr)),
"max": float(np.max(arr)),
"median": float(np.median(arr)),
"mean": float(np.mean(arr)),
}
ca_strict_share = _strict_share_percent(len(ca_values_strict), len(ca_values_final))
ebitda_strict_share = _strict_share_percent(len(ebitda_values_strict), len(ebitda_values_final))
ca_relaxed_share = round(100.0 - ca_strict_share, 1) if ca_values_final else 0.0
ebitda_relaxed_share = round(100.0 - ebitda_strict_share, 1) if ebitda_values_final else 0.0
relaxed_shares: list[float] = []
if ca_values_final:
relaxed_shares.append(ca_relaxed_share)
if ebitda_values_final:
relaxed_shares.append(ebitda_relaxed_share)
avg_relaxed_share = round(float(np.mean(np.array(relaxed_shares))), 1) if relaxed_shares else 100.0
if avg_relaxed_share <= 25.0:
valuation_data_quality = "high"
elif avg_relaxed_share <= 60.0:
valuation_data_quality = "medium"
else:
valuation_data_quality = "low"
ca_stats = _calculate_stats(ca_values_final)
ebitda_stats = _calculate_stats(ebitda_values_final)
result = {
"valuation_scope": valuation_scope,
"ca_multiples_count": len(ca_values_final),
"ebitda_multiples_count": len(ebitda_values_final),
"ca_multiples_count_strict": len(ca_values_strict),
"ebitda_multiples_count_strict": len(ebitda_values_strict),
"ca_strict_share_percent": ca_strict_share,
"ca_relaxed_share_percent": ca_relaxed_share,
"ebitda_strict_share_percent": ebitda_strict_share,
"ebitda_relaxed_share_percent": ebitda_relaxed_share,
"ca_relaxed_for_valuation": ca_scope_widened,
"ebitda_relaxed_for_valuation": ebitda_scope_widened,
"valuation_data_quality": valuation_data_quality,
"ca_min": 0,
"ca_max": 0,
"ca_median": 0,
"ca_mean": 0,
"ebitda_min": 0,
"ebitda_max": 0,
"ebitda_median": 0,
"ebitda_mean": 0,
"valuation_ca_median": 0,
"valuation_ca_mean": 0,
"valuation_ebitda_median": 0,
"valuation_ebitda_mean": 0,
"valuation_ca": {
"value": 0,
"is_calculable": False,
"reason": insufficient_reason,
},
"valuation_ebitda": {
"value": 0,
"is_calculable": False,
"reason": insufficient_reason,
},
}
# Toujours exposer les stats backend de reference (0 si echantillon insuffisant).
result["ca_min"] = round(ca_stats["min"], 2)
result["ca_max"] = round(ca_stats["max"], 2)
result["ca_median"] = round(ca_stats["median"], 2)
result["ca_mean"] = round(ca_stats["mean"], 2)
result["ebitda_min"] = round(ebitda_stats["min"], 1)
result["ebitda_max"] = round(ebitda_stats["max"], 1)
result["ebitda_median"] = round(ebitda_stats["median"], 1)
result["ebitda_mean"] = round(ebitda_stats["mean"], 1)
if ca_stats["median"] > 0:
if target_ca_eur and target_ca_eur > 0:
result["valuation_ca_median"] = int(round(target_ca_eur * result["ca_median"]))
result["valuation_ca_mean"] = int(round(target_ca_eur * result["ca_mean"]))
result["valuation_ca"] = {
"value": int(result["valuation_ca_median"]),
"is_calculable": True,
"reason": None,
}
else:
result["valuation_ca"] = {
"value": 0,
"is_calculable": False,
"reason": "CA cible manquant ou nul.",
}
if ebitda_stats["median"] > 0:
# Le calcul du montant final depend de l'EBITDA saisi par l'utilisateur
# (compute_ebitda_valuation dans main.py), on prepare ici le statut.
result["valuation_ebitda"] = {
"value": 0,
"is_calculable": True,
"reason": None,
}
return result
def compute_ebitda_valuation(ebitda_eur: float, multiples: dict) -> dict:
# Valorisation EBITDA standardisee:
# - jamais de montant null
# - contrat explicite is_calculable + reason
if not multiples:
return {"value": 0, "is_calculable": False, "reason": "Multiples indisponibles."}
if not ebitda_eur or ebitda_eur <= 0:
return {"value": 0, "is_calculable": False, "reason": "EBITDA cible manquant ou nul."}
ebitda_median = as_number(multiples.get("ebitda_median")) or 0
if ebitda_median <= 0:
return {
"value": 0,
"is_calculable": False,
"reason": "Données insuffisantes même après élargissement",
}
return {
"value": int(round(ebitda_eur * ebitda_median)),
"is_calculable": True,
"reason": None,
}
def count_sector_activity(naf_2: str) -> dict:
# Mesure l'intensite M&A du secteur (24 derniers mois).
if not naf_2:
return {"count": 0, "level": "indetermine", "emoji": "?", "message": "Secteur NAF non identifie."}
cutoff_date = (datetime.now() - timedelta(days=730)).strftime("%Y-%m-%d")
try:
result = (
get_supabase_client()
.table("db_transactions")
.select("id_target", count="exact")
.eq("code_naf_2", naf_2)
.gte("date_transaction", cutoff_date)
.execute()
)
count = result.count if result.count is not None else 0
except Exception:
# On degrade a 0 plutot que d'interrompre l'analyse utilisateur.
count = 0
level_info = None
for key in ["tres_eleve", "eleve", "moyen", "faible"]:
threshold = ACTIVITY_THRESHOLDS[key]
if count >= threshold["min"]:
level_info = threshold
break
if not level_info:
level_info = ACTIVITY_THRESHOLDS["faible"]
return {
"count": count,
"level": level_info["label"],
"emoji": level_info["emoji"],
"message": f"{level_info['emoji']} {level_info['label']} — {count} operation(s) en 24 mois.",
}
# ============================================================================
# UTILITAIRES D'AFFICHAGE
# ---------------------------------------------------------------------------
# Helpers de presentation (conversion tranche CA et format monnaie).
# ============================================================================
def ca_to_range(ca_meur: float) -> str | None:
# Conversion CA reel -> bucket normalise (utilitaire d'affichage/debug).
if ca_meur is None or ca_meur < 0:
return None
for label, (low, high, _) in CA_RANGES.items():
if label == "[0-20]":
continue
if low <= ca_meur < high:
return label
if ca_meur >= 1000:
return ">1000"
return "[0-5]"
def format_euros(amount: float | None) -> str:
# Format lisible FR pour affichage humain (pas pour calculs).
if amount is None:
return "N/A"
if abs(amount) >= 1_000_000:
return f"{amount / 1_000_000:,.1f} M€".replace(",", " ")
if abs(amount) >= 1_000:
return f"{amount / 1_000:,.0f} k€".replace(",", " ")
return f"{amount:,.0f} €".replace(",", " ")
"""
Connexion Supabase, matching, scoring et valorisation.
Portage clean de temp_old/database.py (sans Streamlit).
"""
from __future__ import annotations
import csv
import math
import os
import re
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any
import numpy as np
from supabase import Client, create_client
from backend.env import load_backend_env
from backend.mistral_compat import build_mistral_client, create_embeddings
from backend.config import (
ACTIVITY_THRESHOLDS,
CA_MULTIPLE_SOFTWARE_AI_BONUS,
CA_RANGES,
MATCH_CANDIDATES_COUNT,
MATCH_SIMILARITY_THRESHOLD,
MIN_CA_MULTIPLE_VALUATION,
MIN_EBITDA_MULTIPLE_VALUATION,
MAX_CA_MULTIPLE_VALUATION,
MAX_EBITDA_MULTIPLE_VALUATION,
MIN_SEMANTIC_MATCH,
MISTRAL_EMBED_MODEL,
SIZE_TOLERANCE,
SOFTWARE_AI_BONUS_KEYWORDS,
TOP_COMPARABLES_DISPLAY,
TRANSACTION_TYPES_COMPARABLE,
VALUATION_MIN_COUNT,
VALUATION_MIN_SEMANTIC,
WEIGHT_CA_NO_NAF,
WEIGHT_CA_PROXIMITY,
WEIGHT_NAF_PROXIMITY,
WEIGHT_SEMANTIC_NO_NAF,
WEIGHT_SEMANTIC_SIMILARITY,
)
# ============================================================================
# ETAT GLOBAL / INITIALISATION
# ---------------------------------------------------------------------------
# - nom de table de suivi d'usage
# - etat memoire des quotas (utile pour API / debug)
# - chargement .env des l'import module
# ============================================================================
USAGE_LOGS_TABLE = "usage_logs_m_a_transaction_app"
DEBUG_LOGS_DIR = Path(__file__).resolve().parent.parent / "debug_logs"
_usage_policy_state = {"blocked_reason": None, "minute_count": 0, "day_count": 0}
# Charge .env au demarrage module pour tous les appels backend.
load_backend_env()
# ============================================================================
# VARIABLES D'ENVIRONNEMENT
# ---------------------------------------------------------------------------
# Helpers robustes pour lire des secrets:
# - nettoyage quotes/commentaires inline
# - erreurs homogenes si variable manquante
# ============================================================================
def _sanitize_env_value(raw: str) -> str:
# Nettoie les valeurs .env: espaces, comments inline, quotes.
value = (raw or "").strip()
# Tolere les .env avec commentaires inline: KEY="value" # comment
value = re.sub(r"\s+#.*$", "", value).strip()
value = value.strip("'\"")
return value
def _get_env(key: str, required: bool = True, default: str = "") -> str:
# Point unique de lecture env (messages d'erreur coherents).
value = _sanitize_env_value(os.getenv(key) or default or "")
if required and not value:
raise ValueError(f"{key} doit etre configure en variable d'environnement")
return value
# ============================================================================
# CLIENTS EXTERNES
# ---------------------------------------------------------------------------
# Fabriques de clients API (Supabase + Mistral).
# ============================================================================
def get_supabase_client() -> Client:
# Fallback de cle pour faciliter les differents environnements.
key = _get_env(
"SUPABASE_KEY",
required=False,
default=_sanitize_env_value(os.getenv("SUPABASE_ANON_KEY") or os.getenv("SUPABASE_SERVICE_ROLE_KEY") or ""),
)
return create_client(_get_env("SUPABASE_URL"), _get_env("SUPABASE_KEY", default=key))