V1 : synchronisation Meta Lead Gen vers MySQL local

- main.py : recuperation formulaires, questions et leads via Meta Graph API
- meta_lead_db.sql : schema des tables MySQL
- gestion anti-doublons et conservation des payloads bruts
This commit is contained in:
SaidSoighiri94
2026-06-24 15:22:37 +02:00
commit b5fb954ab6
4 changed files with 619 additions and 0 deletions
+8
View File
@@ -0,0 +1,8 @@
.env
venv/
__pycache__/
*.pyc
# Fichiers de travail internes (non versionnés)
CLAUDE.md
context.md
+386
View File
@@ -0,0 +1,386 @@
"""
Synchronisation Meta Lead Gen vers MySQL local (V1).
Récupère les formulaires actifs, leurs questions et leurs leads via l'API
Graph Meta, puis les stocke dans la base locale sans créer de doublons.
L'injection vers OscarCRM n'est pas gérée à ce stade.
"""
import json
import os
from datetime import datetime
import mysql.connector
import requests
from dotenv import load_dotenv
# ============================================================
# Chargement de la configuration
# ============================================================
# Les informations sensibles sont lues depuis le fichier .env.
# Cela évite de stocker le token Meta ou le mot de passe MySQL
# directement dans le code source.
# ============================================================
load_dotenv()
META_ACCESS_TOKEN = os.getenv("META_ACCESS_TOKEN")
META_API_VERSION = os.getenv("META_API_VERSION", "v25.0")
META_PAGE_ID = os.getenv("META_PAGE_ID")
DB_CONFIG = {
"host": os.getenv("DB_HOST", "localhost"),
"port": int(os.getenv("DB_PORT", "3306")),
"database": os.getenv("DB_NAME", "meta_lead_db"),
"user": os.getenv("DB_USER", "root"),
"password": os.getenv("DB_PASSWORD", ""),
"charset": "utf8mb4",
"use_unicode": True,
}
# ============================================================
# Formulaires autorisés pour le test
# ============================================================
# On limite volontairement la synchronisation aux formulaires
# actifs connus afin d’éviter de traiter des formulaires archivés.
# ============================================================
FORM_CAMPUSES = {
"1016111097572602": "Marseille",
"1032481262797041": "Nantes",
"941574848893640": "Cergy",
"1012012785055468": "Cergy",
}
def get_database_connection():
"""Ouvre une connexion MySQL locale."""
return mysql.connector.connect(**DB_CONFIG)
def call_meta_api(path, params=None):
"""
Exécute un appel GET vers lAPI Graph Meta.
Args:
path: Chemin Graph API, par exemple 'PAGE_ID/leadgen_forms'.
params: Paramètres complémentaires de la requête.
Returns:
Réponse JSON retournée par Meta.
Raises:
requests.HTTPError: si Meta retourne une erreur HTTP.
"""
url = f"https://graph.facebook.com/{META_API_VERSION}/{path}"
request_params = params.copy() if params else {}
request_params["access_token"] = META_ACCESS_TOKEN
response = requests.get(url, params=request_params, timeout=30)
response.raise_for_status()
return response.json()
def parse_meta_datetime(value):
"""
Convertit une date Meta en format compatible MySQL.
Exemple Meta :
2026-06-23T06:37:24+0000
Returns:
Date au format 'YYYY-MM-DD HH:MM:SS' ou None.
"""
if not value:
return None
parsed_date = datetime.strptime(value, "%Y-%m-%dT%H:%M:%S%z")
return parsed_date.strftime("%Y-%m-%d %H:%M:%S")
def normalize_field_data(field_data):
"""
Transforme field_data Meta en dictionnaire simple.
Exemple dentrée Meta :
[
{"name": "nom", "values": ["Dupont"]},
{"name": "first_name", "values": ["Jean"]}
]
Exemple de sortie :
{
"nom": "Dupont",
"first_name": "Jean"
}
"""
normalized = {}
for field in field_data or []:
name = field.get("name")
values = field.get("values", [])
if name:
normalized[name] = values[0] if values else None
return normalized
def fetch_active_forms():
"""Récupère les formulaires Lead Gen actifs de la Page Meta."""
result = call_meta_api(
f"{META_PAGE_ID}/leadgen_forms",
{
"fields": "id,name,status,locale",
"limit": 100,
},
)
forms = result.get("data", [])
return [
form for form in forms
if form.get("status") == "ACTIVE" and form.get("id") in FORM_CAMPUSES
]
def fetch_form_questions(form_id):
"""Récupère les questions/champs dun formulaire Meta."""
result = call_meta_api(
form_id,
{
"fields": "id,name,status,locale,questions",
},
)
return result.get("questions", []), result
def fetch_form_leads(form_id):
"""
Récupère les leads dun formulaire Meta.
La pagination est prise en compte afin de récupérer toutes les
pages de résultats retournées par Meta.
"""
all_leads = []
result = call_meta_api(
f"{form_id}/leads",
{
"fields": "id,created_time,field_data",
"limit": 100,
},
)
all_leads.extend(result.get("data", []))
next_url = result.get("paging", {}).get("next")
while next_url:
response = requests.get(next_url, timeout=30)
response.raise_for_status()
result = response.json()
all_leads.extend(result.get("data", []))
next_url = result.get("paging", {}).get("next")
return all_leads
def upsert_form(connection, form, raw_payload):
"""Insère ou met à jour un formulaire Meta en base."""
query = """
INSERT INTO meta_forms (
meta_form_id,
page_id,
name,
status,
locale,
campus,
raw_payload
)
VALUES (%s, %s, %s, %s, %s, %s, %s)
ON DUPLICATE KEY UPDATE
name = VALUES(name),
status = VALUES(status),
locale = VALUES(locale),
campus = VALUES(campus),
raw_payload = VALUES(raw_payload),
updated_at = CURRENT_TIMESTAMP
"""
values = (
form.get("id"),
META_PAGE_ID,
form.get("name"),
form.get("status"),
form.get("locale"),
FORM_CAMPUSES.get(form.get("id")),
json.dumps(raw_payload, ensure_ascii=False),
)
cursor = connection.cursor()
cursor.execute(query, values)
connection.commit()
cursor.close()
def upsert_question(connection, form_id, question):
"""Insère ou met à jour une question de formulaire Meta."""
query = """
INSERT INTO meta_form_questions (
meta_form_id,
question_id,
field_key,
label,
question_type,
question_options,
raw_payload
)
VALUES (%s, %s, %s, %s, %s, %s, %s)
ON DUPLICATE KEY UPDATE
question_id = VALUES(question_id),
label = VALUES(label),
question_type = VALUES(question_type),
question_options = VALUES(question_options),
raw_payload = VALUES(raw_payload),
updated_at = CURRENT_TIMESTAMP
"""
values = (
form_id,
question.get("id"),
question.get("key"),
question.get("label"),
question.get("type"),
json.dumps(question.get("options"), ensure_ascii=False)
if question.get("options") is not None else None,
json.dumps(question, ensure_ascii=False),
)
cursor = connection.cursor()
cursor.execute(query, values)
connection.commit()
cursor.close()
def insert_lead(connection, form, lead):
"""
Insère un lead Meta en base.
Si le lead existe déjà, il nest pas dupliqué grâce à la clé
unique meta_lead_id. Les données existantes sont conservées.
"""
fields = normalize_field_data(lead.get("field_data", []))
query = """
INSERT INTO meta_leads (
meta_lead_id,
meta_form_id,
created_time,
je_suis,
nom,
first_name,
adresse_email,
phone_number,
city,
campus,
raw_payload,
status
)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 'RECEIVED')
ON DUPLICATE KEY UPDATE
updated_at = CURRENT_TIMESTAMP
"""
values = (
lead.get("id"),
form.get("id"),
parse_meta_datetime(lead.get("created_time")),
fields.get("je_suis"),
fields.get("nom"),
fields.get("first_name"),
fields.get("adresse_e-mail"),
fields.get("phone_number"),
fields.get("city"),
FORM_CAMPUSES.get(form.get("id")),
json.dumps(lead, ensure_ascii=False),
)
cursor = connection.cursor()
cursor.execute(query, values)
connection.commit()
cursor.close()
def synchronize_meta_to_mysql():
"""
Lance la synchronisation Meta vers MySQL.
Étapes :
1. Récupération des formulaires actifs.
2. Stockage des formulaires.
3. Récupération et stockage des questions.
4. Récupération et stockage des leads.
"""
connection = get_database_connection()
total_forms = 0
total_questions = 0
total_leads = 0
try:
forms = fetch_active_forms()
for form in forms:
form_id = form.get("id")
questions, form_details = fetch_form_questions(form_id)
upsert_form(connection, form, form_details)
total_forms += 1
for question in questions:
upsert_question(connection, form_id, question)
total_questions += 1
leads = fetch_form_leads(form_id)
for lead in leads:
insert_lead(connection, form, lead)
total_leads += 1
print("Synchronisation terminée.")
print(f"Formulaires traités : {total_forms}")
print(f"Questions traitées : {total_questions}")
print(f"Leads traités : {total_leads}")
finally:
connection.close()
if __name__ == "__main__":
if not META_ACCESS_TOKEN:
raise RuntimeError("META_ACCESS_TOKEN est manquant dans le fichier .env.")
if not META_PAGE_ID:
raise RuntimeError("META_PAGE_ID est manquant dans le fichier .env.")
synchronize_meta_to_mysql()
+222
View File
@@ -0,0 +1,222 @@
-- ============================================================
-- Script : meta_lead_db.sql
-- Base : meta_lead_db
--
-- Objectif :
-- Créer une base intermédiaire permettant de stocker les
-- formulaires et leads Meta Lead Gen avant leur injection
-- dans OscarCRM.
--
-- Remarques techniques :
-- - Les IDs Meta sont stockés en VARCHAR car ils sont fournis
-- sous forme de chaînes numériques.
-- - Les données brutes Meta sont conservées en JSON pour garder
-- une traçabilité complète.
-- - Les colonnes indexées sont limitées pour rester compatibles
-- avec les anciennes limites MySQL/InnoDB en utf8mb4.
-- ============================================================
DROP DATABASE IF EXISTS meta_lead_db;
CREATE DATABASE meta_lead_db
CHARACTER SET utf8mb4
COLLATE utf8mb4_unicode_ci;
USE meta_lead_db;
-- ============================================================
-- Table : meta_forms
--
-- Rôle :
-- Stocke les formulaires Lead Gen récupérés depuis Meta.
-- Chaque formulaire peut être lié à un campus ou à une campagne.
-- ============================================================
CREATE TABLE meta_forms (
id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
meta_form_id VARCHAR(64) NOT NULL,
page_id VARCHAR(64) NOT NULL,
name VARCHAR(255) NOT NULL,
status VARCHAR(50) NOT NULL,
locale VARCHAR(20) NULL,
campus VARCHAR(100) NULL,
raw_payload JSON NULL,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME NULL DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP,
CONSTRAINT uq_meta_forms_meta_form_id
UNIQUE (meta_form_id),
INDEX idx_meta_forms_status (status),
INDEX idx_meta_forms_campus (campus)
) ENGINE=InnoDB
DEFAULT CHARSET=utf8mb4
COLLATE=utf8mb4_unicode_ci
COMMENT='Formulaires Meta Lead Gen récupérés depuis l API Meta.';
-- ============================================================
-- Table : meta_form_questions
--
-- Rôle :
-- Stocke les questions/champs associés à chaque formulaire Meta.
-- Cette table permet de connaître la structure exacte des
-- formulaires utilisés pour générer les leads.
--
-- Note :
-- field_key est limité à 100 caractères afin d éviter les erreurs
-- d index avec utf8mb4 sur certaines versions MySQL/InnoDB.
-- ============================================================
CREATE TABLE meta_form_questions (
id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
meta_form_id VARCHAR(64) NOT NULL,
question_id VARCHAR(64) NULL,
field_key VARCHAR(100) NOT NULL,
label VARCHAR(255) NULL,
question_type VARCHAR(100) NULL,
question_options JSON NULL,
raw_payload JSON NULL,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME NULL DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP,
CONSTRAINT uq_meta_form_questions_form_field
UNIQUE (meta_form_id, field_key),
CONSTRAINT fk_meta_form_questions_form
FOREIGN KEY (meta_form_id)
REFERENCES meta_forms(meta_form_id)
ON UPDATE CASCADE
ON DELETE CASCADE,
INDEX idx_meta_form_questions_form (meta_form_id)
) ENGINE=InnoDB
DEFAULT CHARSET=utf8mb4
COLLATE=utf8mb4_unicode_ci
COMMENT='Questions et champs des formulaires Meta Lead Gen.';
-- ============================================================
-- Table : meta_leads
--
-- Rôle :
-- Stocke les leads récupérés depuis les formulaires Meta.
-- Les colonnes principales reprennent les champs du formulaire
-- Meta et correspondent au format attendu pour OscarCRM.
--
-- Statuts :
-- RECEIVED : lead récupéré depuis Meta
-- READY_TO_SEND : lead prêt à être envoyé vers OscarCRM
-- SENT : lead envoyé avec succès vers OscarCRM
-- ERROR : erreur pendant le traitement ou l envoi
-- INVALID : données obligatoires manquantes ou invalides
-- ============================================================
CREATE TABLE meta_leads (
id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
meta_lead_id VARCHAR(64) NOT NULL,
meta_form_id VARCHAR(64) NOT NULL,
meta_ad_id VARCHAR(64) NULL,
meta_campaign_id VARCHAR(64) NULL,
created_time DATETIME NULL,
je_suis VARCHAR(255) NULL,
nom VARCHAR(255) NULL,
first_name VARCHAR(255) NULL,
adresse_email VARCHAR(191) NULL,
phone_number VARCHAR(50) NULL,
city VARCHAR(255) NULL,
source VARCHAR(100) NOT NULL DEFAULT 'META_LEAD_GEN',
campus VARCHAR(100) NULL,
raw_payload JSON NOT NULL,
oscar_payload JSON NULL,
oscar_response JSON NULL,
status VARCHAR(50) NOT NULL DEFAULT 'RECEIVED',
error_message TEXT NULL,
synced_at DATETIME NULL,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME NULL DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP,
CONSTRAINT uq_meta_leads_meta_lead_id
UNIQUE (meta_lead_id),
CONSTRAINT fk_meta_leads_form
FOREIGN KEY (meta_form_id)
REFERENCES meta_forms(meta_form_id)
ON UPDATE CASCADE
ON DELETE RESTRICT,
CONSTRAINT chk_meta_leads_status
CHECK (status IN ('RECEIVED', 'READY_TO_SEND', 'SENT', 'ERROR', 'INVALID')),
INDEX idx_meta_leads_status (status),
INDEX idx_meta_leads_meta_form_id (meta_form_id),
INDEX idx_meta_leads_created_time (created_time),
INDEX idx_meta_leads_adresse_email (adresse_email),
INDEX idx_meta_leads_synced_at (synced_at)
) ENGINE=InnoDB
DEFAULT CHARSET=utf8mb4
COLLATE=utf8mb4_unicode_ci
COMMENT='Leads Meta Lead Gen stockés avant injection vers OscarCRM.';
-- ============================================================
-- Table : oscar_sync_logs
--
-- Rôle :
-- Trace chaque tentative d injection d un lead vers OscarCRM.
-- Cette table permet de conserver l historique des succès,
-- erreurs et réponses API sans écraser les données du lead.
--
-- Statuts :
-- SENT : injection réussie
-- ERROR : erreur lors de l injection
-- INVALID : payload non valide pour OscarCRM
-- ============================================================
CREATE TABLE oscar_sync_logs (
id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
lead_id BIGINT UNSIGNED NOT NULL,
oscar_payload JSON NOT NULL,
oscar_response JSON NULL,
status VARCHAR(50) NOT NULL,
error_message TEXT NULL,
attempted_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
CONSTRAINT fk_oscar_sync_logs_lead
FOREIGN KEY (lead_id)
REFERENCES meta_leads(id)
ON UPDATE CASCADE
ON DELETE CASCADE,
CONSTRAINT chk_oscar_sync_logs_status
CHECK (status IN ('SENT', 'ERROR', 'INVALID')),
INDEX idx_oscar_sync_logs_lead_id (lead_id),
INDEX idx_oscar_sync_logs_status (status),
INDEX idx_oscar_sync_logs_attempted_at (attempted_at)
) ENGINE=InnoDB
DEFAULT CHARSET=utf8mb4
COLLATE=utf8mb4_unicode_ci
COMMENT='Historique des tentatives d injection des leads vers OscarCRM.';
+3
View File
@@ -0,0 +1,3 @@
requests
mysql-connector-python
python-dotenv