Nicolas Fryder
dc242d1390
- flaresolverr.py: refactor stateless (create_session/destroy_session/get) L'ancienne approche warmup→cookies→requests était bloquée par Cloudflare (JA3 TLS fingerprinting). Toutes les requêtes passent maintenant via une session Chrome persistante FlareSolverr. - ma_http.py: MASession utilise les sessions FS persistantes. Ajout de _extract_json (html.unescape + extraction <pre>) et _build_url. Gestion du refresh sur 403/429 avec recréation de session. - db.py: ajout du champ data JSONB dans upsert_bands INSERT+ON CONFLICT. Avant ce fix, l'URL n'était jamais stockée → get_bands_to_enrich retournait toujours 0 résultats → enrichissement mort. - scraper_band.py: schéma band_page aligné sur l'historique DB (lineup structuré avec URLs artistes, discography_url, lyrical_themes, location, info_raw). Ajout de "Themes" dans _pick pour couvrir les deux variantes de clé HTML (MA utilise parfois "Themes", parfois "Lyrical themes"). - 005_fix_checkpoints_and_data.sql: renomme last_additions_check → last_created_check (clé morte vs clé utilisée par le code). Backfill enriched=true + colonnes top-level (formed_year, themes, status, location_text) depuis band_page JSONB pour les 85k bands de l'ancien scraper. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
123 lines
4.2 KiB
Python
123 lines
4.2 KiB
Python
"""
|
|
Parser d'une page individuelle de band Metal Archives.
|
|
|
|
Produit un dict compatible avec upsert_band_enriched() ET cohérent avec
|
|
le schéma band_page historique stocké en DB (lineup structuré, discography_url, etc.)
|
|
"""
|
|
import hashlib
|
|
import re
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from bs4 import BeautifulSoup
|
|
|
|
# Section header → clé lineup
|
|
_LINEUP_SECTIONS = {
|
|
"current": "current",
|
|
"past": "past",
|
|
"live": "live",
|
|
"session": "session",
|
|
}
|
|
|
|
|
|
def parse_band_page(html: str) -> Dict[str, Any]:
|
|
soup = BeautifulSoup(html, "lxml")
|
|
out: Dict[str, Any] = {}
|
|
|
|
h1 = soup.find("h1", class_=re.compile(r"band_name", re.I)) or soup.find("h1")
|
|
out["name"] = _text(h1)
|
|
|
|
# --- band_stats : dt/dd pairs ---
|
|
stats_div = soup.find("div", id="band_stats") or soup.find("div", id="band_info") or soup
|
|
kv: Dict[str, str] = {}
|
|
for dl in stats_div.find_all("dl"):
|
|
for dt in dl.find_all("dt"):
|
|
dd = dt.find_next_sibling("dd")
|
|
k = _text(dt).rstrip(":")
|
|
v = _text(dd) if dd else None
|
|
if k and v:
|
|
kv[k] = v
|
|
out["info_raw"] = {k.lower(): v for k, v in kv.items()}
|
|
|
|
out["status"] = _pick(kv, "Status")
|
|
out["genre"] = _pick(kv, "Genre")
|
|
out["formed_in"] = _pick(kv, "Formed in", "Formed")
|
|
out["years_active"] = _pick(kv, "Years active")
|
|
out["location"] = _pick(kv, "Location")
|
|
# MA uses either "Lyrical themes", "Lyrical Themes", or simply "Themes"
|
|
out["lyrical_themes"] = _pick(kv, "Lyrical themes", "Lyrical Themes", "Themes")
|
|
# Alias for upsert_band_enriched which reads data.get("themes")
|
|
out["themes"] = out["lyrical_themes"]
|
|
out["label"] = _pick(kv, "Current label", "Last label", "Label")
|
|
|
|
# --- Lineup (grouped by section, with artist URLs) ---
|
|
lineup: Dict[str, List[Dict]] = {"current": [], "past": [], "live": [], "session": []}
|
|
for table in soup.find_all("table", class_="lineupTable"):
|
|
section_key = _lineup_section(table)
|
|
bucket = lineup.setdefault(section_key, [])
|
|
for tr in table.find_all("tr")[1:]:
|
|
tds = tr.find_all(["td", "th"])
|
|
if not tds:
|
|
continue
|
|
a = tds[0].find("a", href=re.compile(r"/artists/"))
|
|
if not a:
|
|
continue
|
|
role = _text(tds[1]) if len(tds) > 1 else None
|
|
bucket.append({
|
|
"name": _text(a),
|
|
"url": a.get("href", ""),
|
|
"role": role,
|
|
})
|
|
out["lineup"] = lineup
|
|
|
|
# --- Discography URL ---
|
|
disc = soup.find("a", href=re.compile(r"/band/discography/"))
|
|
out["discography_url"] = disc["href"] if disc else None
|
|
|
|
# --- Liens externes ---
|
|
links_div = soup.find("div", id="band_links")
|
|
out["links"] = [
|
|
{"text": _text(a), "url": a["href"]}
|
|
for a in links_div.find_all("a", href=True)
|
|
] if links_div else []
|
|
|
|
# --- Dates audit trail ---
|
|
audit = soup.find(id="auditTrail") or soup.find("div", class_=re.compile(r"audit", re.I))
|
|
if audit:
|
|
trail = _text(audit)
|
|
m_added = re.search(r"Added on:\s*(\S+\s+\S+)", trail, re.I)
|
|
m_modified = re.search(r"Last modified on:\s*(\S+\s+\S+)", trail, re.I)
|
|
out["ma_created_at"] = m_added.group(1).strip() if m_added else None
|
|
out["ma_modified_at"] = m_modified.group(1).strip() if m_modified else None
|
|
|
|
return out
|
|
|
|
|
|
def page_hash(html: str) -> str:
|
|
return hashlib.md5(html.encode()).hexdigest()
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Helpers
|
|
# ------------------------------------------------------------------
|
|
|
|
def _text(el) -> str:
|
|
return (el.get_text(" ", strip=True) if el else "").strip()
|
|
|
|
|
|
def _pick(kv: Dict, *keys: str) -> Optional[str]:
|
|
for k in keys:
|
|
if k in kv:
|
|
return kv[k]
|
|
return None
|
|
|
|
|
|
def _lineup_section(table) -> str:
|
|
"""Déduit la section lineup depuis le h2/h3 qui précède la table."""
|
|
el = table.find_previous(["h2", "h3"])
|
|
if not el:
|
|
return "current"
|
|
text = _text(el).lower()
|
|
for kw, key in _LINEUP_SECTIONS.items():
|
|
if kw in text:
|
|
return key
|
|
return "current"
|