1cf427a0f2
- Statut de prospection (colonne CSV) avec badge coloré et filtre - Import en masse de liens Facebook (streaming, dédoublonnage) - Modèle de message de contact configurable + copie en un clic - Estimation distance/carburant/péage via OpenStreetMap (Nominatim + OSRM) - Section Paramètres + config.json (non versionné)
296 lines
12 KiB
Python
296 lines
12 KiB
Python
"""Scraping des pages « À propos » Facebook avec une session Chromium persistante."""
|
||
|
||
import re
|
||
import sys
|
||
import time
|
||
from pathlib import Path
|
||
from urllib.parse import urlparse, parse_qs
|
||
|
||
from playwright.sync_api import sync_playwright
|
||
|
||
PROFIL = str(Path(__file__).parent / "fb_profile")
|
||
ATTENTE_LOGIN_S = 300
|
||
|
||
|
||
class ErreurScrape(Exception):
|
||
def __init__(self, code, message):
|
||
super().__init__(message)
|
||
self.code = code
|
||
|
||
|
||
# Liens de partage/contenu qui ne désignent pas directement une page : à résoudre en les visitant
|
||
SLUGS_A_RESOUDRE = {"share", "reel", "reels", "story.php", "photo", "photo.php", "video.php", "watch", "posts", "l.php"}
|
||
|
||
|
||
def _urls_pour_id(page_id):
|
||
# marqueur = les chiffres seuls : Facebook réécrit souvent profile.php?id=N
|
||
# en /people/Nom-De-Page/N/, où « id= » disparaît mais pas le numéro
|
||
canonique = f"https://www.facebook.com/profile.php?id={page_id}"
|
||
return {"canonique": canonique, "about": f"{canonique}&sk=about", "marqueur": str(page_id)}
|
||
|
||
|
||
def normaliser_url(url):
|
||
"""Retourne {canonique, abouts, marqueur} ou {resoudre: url} ; lève ErreurScrape sinon.
|
||
|
||
« marqueur » est un fragment qui doit rester présent dans l'URL après navigation :
|
||
il sert à détecter les redirections de Facebook (vers le fil, votre profil...).
|
||
"""
|
||
url = url.strip()
|
||
if not url.startswith("http"):
|
||
url = "https://" + url
|
||
p = urlparse(url)
|
||
if not re.search(r"(?:^|\.)(facebook|fb)\.com$", p.netloc.lower().split(":")[0]):
|
||
raise ErreurScrape("url_invalide", "Ce n'est pas un lien Facebook.")
|
||
|
||
segments = [s for s in p.path.split("/") if s]
|
||
|
||
if segments and segments[0] == "profile.php" or p.path.rstrip("/").endswith("profile.php"):
|
||
page_id = parse_qs(p.query).get("id", [""])[0]
|
||
if not page_id:
|
||
raise ErreurScrape("url_invalide", "Lien profile.php sans identifiant.")
|
||
return _urls_pour_id(page_id)
|
||
|
||
if not segments:
|
||
raise ErreurScrape("url_invalide", "Lien Facebook sans nom de page.")
|
||
|
||
premier = segments[0].lower()
|
||
|
||
if premier in SLUGS_A_RESOUDRE:
|
||
return {"resoudre": url}
|
||
|
||
# facebook.com/p/Nom-De-Page-100063123456789 (nouvelles pages)
|
||
if premier == "p" and len(segments) >= 2:
|
||
m = re.search(r"(\d{5,})/?$", segments[1])
|
||
if m:
|
||
return _urls_pour_id(m.group(1))
|
||
raise ErreurScrape("url_invalide", "Lien /p/ sans identifiant numérique reconnaissable.")
|
||
|
||
# facebook.com/people/Nom/123456789 et facebook.com/pages/Nom/123456789 (anciens formats)
|
||
if premier in ("people", "pages"):
|
||
for seg in reversed(segments):
|
||
if re.fullmatch(r"\d{5,}", seg):
|
||
return _urls_pour_id(seg)
|
||
raise ErreurScrape("url_invalide", f"Lien /{premier}/ sans identifiant numérique.")
|
||
|
||
slug = segments[0]
|
||
canonique = f"https://www.facebook.com/{slug}"
|
||
return {"canonique": canonique, "about": f"{canonique}/about", "marqueur": slug.lower()}
|
||
|
||
|
||
def _sur_la_bonne_page(page, marqueur):
|
||
return marqueur in page.url.lower()
|
||
|
||
|
||
def _refuser_cookies(page):
|
||
try:
|
||
bouton = page.get_by_role(
|
||
"button",
|
||
name=re.compile(r"refuser les cookies optionnels|decline optional cookies|autoriser tous les cookies", re.I),
|
||
).first
|
||
bouton.click(timeout=3000)
|
||
page.wait_for_timeout(1000)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _mur_de_connexion(page):
|
||
if "/login" in page.url:
|
||
return True
|
||
try:
|
||
corps = page.inner_text("body", timeout=5000)
|
||
except Exception:
|
||
return False
|
||
return bool(re.search(r"connectez-vous pour continuer|log into facebook|vous devez vous connecter", corps, re.I))
|
||
|
||
|
||
def _est_connecte(contexte):
|
||
"""Le cookie de session « c_user » n'existe que si un compte est connecté."""
|
||
try:
|
||
return any(c["name"] == "c_user" for c in contexte.cookies("https://www.facebook.com"))
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def connexion():
|
||
"""Ouvre Facebook dans la fenêtre persistante et attend que l'utilisateur se connecte.
|
||
|
||
Aucune autre action ne tourne pendant ce temps : l'utilisateur a jusqu'à
|
||
ATTENTE_LOGIN_S secondes pour se connecter tranquillement (captcha, 2FA...).
|
||
Retourne True si la session est établie.
|
||
"""
|
||
with sync_playwright() as pw:
|
||
contexte = pw.chromium.launch_persistent_context(
|
||
user_data_dir=PROFIL, headless=False, locale="fr-FR",
|
||
viewport={"width": 1280, "height": 900},
|
||
)
|
||
try:
|
||
if _est_connecte(contexte):
|
||
return True
|
||
page = contexte.pages[0] if contexte.pages else contexte.new_page()
|
||
page.goto("https://www.facebook.com/login", wait_until="domcontentloaded", timeout=30000)
|
||
page.wait_for_timeout(2000)
|
||
_refuser_cookies(page)
|
||
debut = time.time()
|
||
while time.time() - debut < ATTENTE_LOGIN_S:
|
||
if _est_connecte(contexte):
|
||
page.wait_for_timeout(2000) # laisser FB finir d'écrire la session
|
||
return True
|
||
page.wait_for_timeout(2000)
|
||
return False
|
||
finally:
|
||
contexte.close()
|
||
|
||
|
||
def _texte_page(page):
|
||
try:
|
||
return page.locator('div[role="main"]').first.inner_text(timeout=5000)
|
||
except Exception:
|
||
try:
|
||
return page.inner_text("body", timeout=5000)
|
||
except Exception:
|
||
return ""
|
||
|
||
|
||
def _scrape_avec_page(page, url):
|
||
"""Visite les pages « À propos » d'un lieu via une page déjà connectée.
|
||
|
||
Retourne {"titre", "texte", "url"}. Suppose la session Facebook déjà active :
|
||
c'est le cœur partagé par `scrape` (un lien) et `scrape_lot` (plusieurs liens).
|
||
"""
|
||
info = normaliser_url(url)
|
||
|
||
# Lien de partage (/share/, /reel/...) : le visiter pour découvrir la vraie page
|
||
if info.get("resoudre"):
|
||
page.goto(info["resoudre"], wait_until="domcontentloaded", timeout=30000)
|
||
page.wait_for_timeout(3000)
|
||
_refuser_cookies(page)
|
||
info = normaliser_url(page.url)
|
||
if info.get("resoudre"):
|
||
raise ErreurScrape(
|
||
"url_invalide",
|
||
"Ce lien de partage ne mène pas à une page de lieu : ouvrez la page du lieu "
|
||
"dans Facebook et copiez son adresse directe.",
|
||
)
|
||
|
||
# Page d'accueil du lieu : la carte « Intro » contient souvent l'adresse,
|
||
# qui n'apparaît dans aucune sous-section « À propos »
|
||
page.goto(info["canonique"], wait_until="domcontentloaded", timeout=30000)
|
||
page.wait_for_timeout(2500)
|
||
_refuser_cookies(page)
|
||
if _mur_de_connexion(page):
|
||
raise ErreurScrape(
|
||
"login_required",
|
||
"Session Facebook expirée : cliquez sur « Connexion Facebook » pour vous reconnecter.",
|
||
)
|
||
if not _sur_la_bonne_page(page, info["marqueur"]):
|
||
# profile.php?id=N est parfois réécrit vers l'adresse « vanity » de la
|
||
# page : si on a atterri sur un profil/une page plausible, adopter ce slug.
|
||
nouveau = urlparse(page.url).path.strip("/").split("/")[0].lower()
|
||
if (info["marqueur"].isdigit() and nouveau
|
||
and nouveau not in SLUGS_A_RESOUDRE | {"p", "people", "pages", "profile.php"}):
|
||
info["marqueur"] = nouveau
|
||
else:
|
||
# Redirigé ailleurs (fil d'actualité, votre profil...) :
|
||
# surtout ne pas extraire ce texte, ce serait VOS infos.
|
||
raise ErreurScrape(
|
||
"redirection",
|
||
f"Facebook a redirigé vers {page.url} au lieu de la page demandée. "
|
||
"Ouvrez la page du lieu dans Facebook et copiez son adresse directe.",
|
||
)
|
||
titre = ""
|
||
for _ in range(6): # le titre met parfois quelques secondes à refléter la page
|
||
titre = re.sub(r"\s*[|\-–]\s*(Facebook|À propos|About).*$", "", page.title()).strip()
|
||
titre = re.sub(r"^\(\d+\+?\)\s*", "", titre) # compteur de notifications « (20+) »
|
||
if titre and titre.lower() != "facebook":
|
||
break
|
||
page.wait_for_timeout(500)
|
||
corps = _texte_page(page)
|
||
if re.search(r"ce contenu n'est pas disponible|this content isn'?t available|page introuvable", corps, re.I):
|
||
raise ErreurScrape("page_introuvable", "Cette page Facebook est introuvable ou inaccessible.")
|
||
if not titre or titre.lower() == "facebook":
|
||
# repli : la première ligne du contenu est le nom du lieu
|
||
titre = next((l.strip() for l in corps.split("\n") if l.strip()), "")
|
||
textes = [corps]
|
||
|
||
# Page « À propos »
|
||
page.goto(info["about"], wait_until="domcontentloaded", timeout=30000)
|
||
page.wait_for_timeout(2500)
|
||
if _sur_la_bonne_page(page, info["marqueur"]):
|
||
textes.append(_texte_page(page))
|
||
|
||
# Sous-section « Coordonnées » : on CLIQUE l'élément dans la page plutôt que
|
||
# de deviner l'URL — l'URL directe n'existe pas pour les nouvelles pages et
|
||
# Facebook redirige alors vers VOTRE propre profil. Selon le style de page,
|
||
# c'est un onglet (nouvelles pages) ou un lien (pages classiques).
|
||
for role in ("tab", "link"):
|
||
try:
|
||
page.locator('div[role="main"]').get_by_role(
|
||
role, name=re.compile(r"coordonnées|contact and basic info", re.I)
|
||
).first.click(timeout=4000)
|
||
page.wait_for_timeout(2500)
|
||
if _sur_la_bonne_page(page, info["marqueur"]):
|
||
textes.append(_texte_page(page))
|
||
break
|
||
except Exception:
|
||
continue # pas trouvé sous ce rôle ; sinon on garde le texte À propos
|
||
|
||
return {"titre": titre, "texte": "\n".join(textes), "url": info["canonique"]}
|
||
|
||
|
||
def scrape(url):
|
||
"""Visite les pages « À propos » d'un lieu et retourne {"titre", "texte", "url"}."""
|
||
with sync_playwright() as pw:
|
||
contexte = pw.chromium.launch_persistent_context(
|
||
user_data_dir=PROFIL, headless=False, locale="fr-FR",
|
||
viewport={"width": 1280, "height": 900},
|
||
)
|
||
try:
|
||
if not _est_connecte(contexte):
|
||
raise ErreurScrape(
|
||
"login_required",
|
||
"Aucune session Facebook : cliquez d'abord sur « Connexion Facebook ».",
|
||
)
|
||
page = contexte.pages[0] if contexte.pages else contexte.new_page()
|
||
return _scrape_avec_page(page, url)
|
||
finally:
|
||
contexte.close()
|
||
|
||
|
||
def scrape_lot(urls):
|
||
"""Analyse une liste de liens en réutilisant une seule fenêtre Chromium.
|
||
|
||
Générateur : produit, pour chaque lien, un dict
|
||
{"url", "ok": True, "resultat": {...}} en cas de succès,
|
||
{"url", "ok": False, "code", "message"} en cas d'échec — un lien raté
|
||
n'interrompt pas le reste du lot.
|
||
"""
|
||
with sync_playwright() as pw:
|
||
contexte = pw.chromium.launch_persistent_context(
|
||
user_data_dir=PROFIL, headless=False, locale="fr-FR",
|
||
viewport={"width": 1280, "height": 900},
|
||
)
|
||
try:
|
||
if not _est_connecte(contexte):
|
||
for url in urls:
|
||
yield {"url": url, "ok": False, "code": "login_required",
|
||
"message": "Aucune session Facebook : cliquez d'abord sur « Connexion Facebook »."}
|
||
return
|
||
page = contexte.pages[0] if contexte.pages else contexte.new_page()
|
||
for url in urls:
|
||
try:
|
||
yield {"url": url, "ok": True, "resultat": _scrape_avec_page(page, url)}
|
||
except ErreurScrape as e:
|
||
yield {"url": url, "ok": False, "code": e.code, "message": str(e)}
|
||
except Exception as e:
|
||
yield {"url": url, "ok": False, "code": "erreur", "message": str(e)}
|
||
finally:
|
||
contexte.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
if len(sys.argv) != 2:
|
||
print("Usage : python scraper.py <url-facebook>")
|
||
sys.exit(1)
|
||
resultat = scrape(sys.argv[1])
|
||
print(f"=== TITRE ===\n{resultat['titre']}\n=== TEXTE ===\n{resultat['texte']}")
|