Ajout suivi prospection : statut, import en masse, message type, trajet+péage
- 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é)
This commit is contained in:
jerem
190
scraper.py
190
scraper.py
@@ -151,9 +151,94 @@ def _texte_page(page):
|
||||
return ""
|
||||
|
||||
|
||||
def scrape(url):
|
||||
"""Visite les pages « À propos » et retourne {"titre", "texte", "url"}."""
|
||||
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",
|
||||
@@ -166,83 +251,38 @@ def scrape(url):
|
||||
"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()
|
||||
|
||||
# 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]
|
||||
def scrape_lot(urls):
|
||||
"""Analyse une liste de liens en réutilisant une seule fenêtre Chromium.
|
||||
|
||||
# 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"):
|
||||
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:
|
||||
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"]}
|
||||
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()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user