msaid-soighiri/META-LEAD
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Adaptation de la base de données pour sql server

Browse Source
This commit is contained in:
SaidSoighiri94
2026-06-26 12:12:12 +02:00
parent b5fb954ab6
commit 5f6524f63b
4 changed files with 263 additions and 424 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.py
+137 -200
View File
@@ -1,52 +1,34 @@
"""
Synchronisation Meta Lead Gen vers MySQL local (V1).
Synchronisation Meta Lead Gen vers SQL Server local.
Récupère les formulaires actifs et leurs leads via l'API Graph Meta,
puis les stocke dans SQL Server sans créer de doublons.
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
from datetime import datetime, timezone
import mysql.connector
import pyodbc
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,
}
DB_DRIVER = os.getenv("DB_DRIVER", "SQL Server")
DB_SERVER = os.getenv("DB_SERVER", "localhost")
DB_PORT = os.getenv("DB_PORT", "1433")
DB_NAME = os.getenv("DB_NAME", "MetaLeadDb")
DB_USER = os.getenv("DB_USER", "sa")
DB_PASSWORD = os.getenv("DB_PASSWORD")
# ============================================================
# 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",
@@ -56,25 +38,22 @@ FORM_CAMPUSES = {
def get_database_connection():
"""Ouvre une connexion MySQL locale."""
"""Ouvre une connexion vers SQL Server."""
return mysql.connector.connect(**DB_CONFIG)
connection_string = (
f"DRIVER={{{DB_DRIVER}}};"
f"SERVER={DB_SERVER},{DB_PORT};"
f"DATABASE={DB_NAME};"
f"UID={DB_USER};"
f"PWD={DB_PASSWORD};"
"TrustServerCertificate=yes;"
)
return pyodbc.connect(connection_string)
def call_meta_api(path, params=None):
"""
Exécute un appel GET vers l’API 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.
"""
"""Exécute un appel GET vers l'API Graph Meta."""
url = f"https://graph.facebook.com/{META_API_VERSION}/{path}"
@@ -88,39 +67,19 @@ def call_meta_api(path, params=None):
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.
"""
"""Convertit une date Meta en datetime compatible SQL Server."""
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")
utc_date = parsed_date.astimezone(timezone.utc)
return utc_date.replace(tzinfo=None)
def normalize_field_data(field_data):
"""
Transforme field_data Meta en dictionnaire simple.
Exemple d’entrée Meta :
[
{"name": "nom", "values": ["Dupont"]},
{"name": "first_name", "values": ["Jean"]}
]
Exemple de sortie :
{
"nom": "Dupont",
"first_name": "Jean"
}
"""
"""Convertit field_data Meta en dictionnaire simple."""
normalized = {}
@@ -134,8 +93,19 @@ def normalize_field_data(field_data):
return normalized
def get_email_from_fields(fields):
"""Récupère l'email en gérant les variantes possibles du champ Meta."""
return (
fields.get("adresse_e-mail")
or fields.get("adresse-e-mail")
or fields.get("adresse_email")
or fields.get("email")
)
def fetch_active_forms():
"""Récupère les formulaires Lead Gen actifs de la Page Meta."""
"""Récupère les formulaires Lead Gen actifs autorisés."""
result = call_meta_api(
f"{META_PAGE_ID}/leadgen_forms",
@@ -149,30 +119,13 @@ def fetch_active_forms():
return [
form for form in forms
if form.get("status") == "ACTIVE" and form.get("id") in FORM_CAMPUSES
if form.get("status") == "ACTIVE"
and form.get("id") in FORM_CAMPUSES
]
def fetch_form_questions(form_id):
"""Récupère les questions/champs d’un 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 d’un formulaire Meta.
La pagination est prise en compte afin de récupérer toutes les
pages de résultats retournées par Meta.
"""
"""Récupère tous les leads d'un formulaire Meta avec pagination."""
all_leads = []
@@ -185,7 +138,6 @@ def fetch_form_leads(form_id):
)
all_leads.extend(result.get("data", []))
next_url = result.get("paging", {}).get("next")
while next_url:
@@ -194,103 +146,89 @@ def fetch_form_leads(form_id):
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."""
def upsert_leadgen_form(connection, form):
"""Insère ou met à jour un formulaire Meta dans leadgen_forms."""
query = """
INSERT INTO meta_forms (
update_query = """
UPDATE dbo.leadgen_forms
SET
page_id = ?,
name = ?,
locale = ?,
status = ?,
updated_at = SYSDATETIME()
WHERE meta_form_id = ?
"""
insert_query = """
INSERT INTO dbo.leadgen_forms (
meta_form_id,
page_id,
name,
status,
locale,
campus,
raw_payload
status
)
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 (?, ?, ?, ?, ?)
"""
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(
update_query,
(
META_PAGE_ID,
form.get("name"),
form.get("locale"),
form.get("status"),
form.get("id"),
),
)
cursor = connection.cursor()
cursor.execute(query, values)
if cursor.rowcount == 0:
cursor.execute(
insert_query,
(
form.get("id"),
META_PAGE_ID,
form.get("name"),
form.get("locale"),
form.get("status"),
),
)
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 n’est pas dupliqué grâce à la clé
unique meta_lead_id. Les données existantes sont conservées.
"""
def upsert_lead_data(connection, form, lead):
"""Insère ou met à jour un lead Meta dans lead_data."""
fields = normalize_field_data(lead.get("field_data", []))
campus = FORM_CAMPUSES.get(form.get("id"))
query = """
INSERT INTO meta_leads (
update_query = """
UPDATE dbo.lead_data
SET
meta_form_id = ?,
created_time = ?,
je_suis = ?,
nom = ?,
first_name = ?,
adresse_email = ?,
phone_number = ?,
city = ?,
campus = ?,
updated_at = SYSDATETIME()
WHERE meta_lead_id = ?
"""
insert_query = """
INSERT INTO dbo.lead_data (
meta_lead_id,
meta_form_id,
created_time,
@@ -301,75 +239,71 @@ def insert_lead(connection, form, lead):
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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, N'RECEIVED')
"""
values = (
values_for_update = (
form.get("id"),
parse_meta_datetime(lead.get("created_time")),
fields.get("je_suis"),
fields.get("nom"),
fields.get("first_name"),
get_email_from_fields(fields),
fields.get("phone_number"),
fields.get("city"),
campus,
lead.get("id"),
)
values_for_insert = (
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"),
get_email_from_fields(fields),
fields.get("phone_number"),
fields.get("city"),
FORM_CAMPUSES.get(form.get("id")),
json.dumps(lead, ensure_ascii=False),
campus,
)
cursor = connection.cursor()
cursor.execute(query, values)
cursor.execute(update_query, values_for_update)
if cursor.rowcount == 0:
cursor.execute(insert_query, values_for_insert)
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.
"""
def synchronize_meta_to_sql_server():
"""Synchronise les formulaires et les leads Meta vers SQL Server."""
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)
upsert_leadgen_form(connection, form)
total_forms += 1
for question in questions:
upsert_question(connection, form_id, question)
total_questions += 1
leads = fetch_form_leads(form_id)
leads = fetch_form_leads(form.get("id"))
for lead in leads:
insert_lead(connection, form, lead)
upsert_lead_data(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:
@@ -383,4 +317,7 @@ if __name__ == "__main__":
if not META_PAGE_ID:
raise RuntimeError("META_PAGE_ID est manquant dans le fichier .env.")
synchronize_meta_to_mysql()
if not DB_PASSWORD:
raise RuntimeError("DB_PASSWORD est manquant dans le fichier .env.")
synchronize_meta_to_sql_server()