"""Login-free public Meta Ad Library intake.
Uses the publicly rendered Meta Ad Library page in a clean Chromium session.
No login, cookies, session reuse, CAPTCHA bypass, or interaction is used.
Each public Library ID is the deduplication key and the existing insert_signal
pipeline remains the only persistence path.
"""
from __future__ import annotations
import asyncio
import re
import time
from datetime import datetime, timezone, timedelta
from pathlib import Path
import sys
import yaml
from playwright.async_api import async_playwright
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from signal_db import insert_signal, record_source_run # noqa: E402
from published_at_normalizer import normalize_published_at # noqa: E402
from classification.signal_classifier import score_signal # noqa: E402
CONFIG_PATH = Path(__file__).resolve().parents[1] / 'sources_multi.yaml'
BASE_URL = 'https://www.facebook.com/ads/library/'
def _lines(text: str) -> list[str]:
return [line.strip() for line in text.splitlines() if line.strip()]
def _parse_start(text: str) -> str | None:
match = re.search(r'Started running on\s+([^\n]+)', text, re.IGNORECASE)
return match.group(1).strip() if match else None
def _advertiser_and_creative(text: str) -> tuple[str, str]:
lines = _lines(text)
try:
start = next(i for i, line in enumerate(lines) if line.lower() == 'see ad details') + 1
except StopIteration:
start = 0
advertiser = ''
for line in lines[start:]:
if line.lower() in {'sponsored', 'active', 'inactive'}:
break
if not line.startswith(('Library ID:', 'Started running on')):
advertiser = line
break
try:
sponsored = next(i for i, line in enumerate(lines) if line.lower() == 'sponsored') + 1
except StopIteration:
sponsored = 0
creative = []
for line in lines[sponsored:]:
if line.startswith(('Active', 'Inactive', 'Library ID:', 'Started running on')):
break
if line not in {'Learn More', 'Learn more', 'Apply Now', 'Contact us'}:
creative.append(line)
return advertiser, ' '.join(creative)[:1800]
def _matches(advertiser: str, creative: str, terms: list[str]) -> bool:
haystack = (advertiser + ' ' + creative).lower()
return any(term.lower() in haystack for term in terms)
async def _read_ad_card(page, ad_id: str) -> dict | None:
marker = page.get_by_text(f'Library ID: {ad_id}', exact=True)
if await marker.count() == 0:
return None
ancestor = marker
card_text = ''
image_urls: list[str] = []
video_posters: list[str] = []
# The rendered card is the first ancestor containing the marker and its
# creative images. This avoids relying on Meta's private CSS class names.
for _ in range(12):
ancestor = ancestor.locator('xpath=..')
try:
text = await ancestor.inner_text()
images = await ancestor.locator('img').evaluate_all('(els)=>els.map(e=>e.src).filter(Boolean)')
posters = await ancestor.locator('video').evaluate_all('(els)=>els.map(e=>e.poster).filter(Boolean)')
except Exception:
continue
if 'Sponsored' in text and ('Active' in text or 'Inactive' in text) and images:
card_text, image_urls, video_posters = text, images, posters
break
if not card_text:
card_text = await marker.locator('xpath=../../../..').inner_text()
advertiser, creative = _advertiser_and_creative(card_text)
# Prefer a creative-sized image over the tiny advertiser avatar.
image = next((u for u in reversed(image_urls) if 's60x60' not in u), '')
poster = next((u for u in video_posters if u), '')
return {'ad_id': ad_id, 'advertiser': advertiser, 'creative': creative,
'started_raw': _parse_start(card_text), 'image_url': image,
'video_poster_url': poster,
'card_text': card_text[:5000]}
async def _run_async(config: dict) -> dict:
source = config.get('sources', {}).get('meta_ads', {})
if not source.get('enabled'):
return {'enabled': False, 'items_seen': 0, 'items_new': 0}
t0 = time.monotonic()
seen = new = duplicate = rejected = 0
ads: list[dict] = []
active_ids: set[str] = set()
advertisers = source.get('advertisers', [])
country = source.get('country', 'ES')
max_per = int(source.get('max_ads_per_advertiser', 12))
async with async_playwright() as pw:
browser = await pw.chromium.launch(headless=True, executable_path='/snap/bin/chromium', args=['--no-sandbox'])
page = await browser.new_page()
for item in advertisers:
query = item['query']
url = (f'{BASE_URL}?active_status=active&ad_type=all&country={country}'
f'&q={__import__("urllib.parse").parse.quote(query)}&search_type=keyword_unordered')
try:
await page.goto(url, wait_until='domcontentloaded', timeout=60000)
await page.wait_for_timeout(3500)
text = await page.locator('body').inner_text()
ids = list(dict.fromkeys(re.findall(r'(?:Library ID|Identificación de la biblioteca):\s*(\d+)', text)))[:max_per]
print(f'[META-ADS] {query}: {len(ids)} öffentliche IDs')
for ad_id in ids:
card = await _read_ad_card(page, ad_id)
if not card:
continue
seen += 1
if not _matches(card['advertiser'], card['creative'], item.get('advertiser_terms', [query])):
rejected += 1
continue
card.update({'config': item, 'public_url': f'{BASE_URL}?id={ad_id}&country={country}'})
card['active_status'] = 'active' if any(x.lower() == 'active' for x in _lines(card.get('card_text',''))) else 'unknown'
if card['active_status'] == 'active':
active_ids.add(ad_id)
ads.append(card)
except Exception as exc:
print(f'[META-ADS] Fehler {query}: {exc}')
await browser.close()
expires = (datetime.now(timezone.utc) + timedelta(days=30)).isoformat()
for ad in ads:
cfg = ad['config']
sd = score_signal(ad['creative'] or ad['advertiser'], language='es',
base_category=cfg.get('category', 'allgemein'),
source_url=ad['public_url'])
if sd['radar_score'] < 31:
rejected += 1
continue
image_url = ad.get('image_url') or ad.get('video_poster_url') or ''
signal = {
'source_type': 'meta_ad',
'source_platform': 'meta_ad',
'external_id': ad['ad_id'],
'observed_at': datetime.now(timezone.utc).isoformat(),
'published_at': normalize_published_at(ad.get('started_raw'), datetime.now(timezone.utc).isoformat()),
'ad_start_at': normalize_published_at(ad.get('started_raw'), datetime.now(timezone.utc).isoformat()),
'last_seen_at': datetime.now(timezone.utc).isoformat() if ad.get('active_status') == 'active' else None,
'active_status': ad.get('active_status', 'unknown'),
'source_name': ad['advertiser'] or cfg['name'],
'source_url': ad['public_url'],
'language': 'es',
'signal_category': sd['signal_category'] or cfg.get('category', 'allgemein'),
'topic': (ad['creative'] or ad['advertiser'])[:200],
'short_summary': ad['creative'][:500],
'extracted_hook': sd['extracted_hook'],
'emotional_direction': sd['emotional_direction'],
'urgency_level': sd['urgency_level'],
'mallorca_relevance': sd['mallorca_relevance'],
'risk_relevance': sd['risk_relevance'],
'estimated_noise_level': sd['estimated_noise_level'],
'suggested_case_types': sd['suggested_case_types'],
'suggested_platforms': sd['suggested_platforms'],
'suggested_cta': sd['suggested_cta'],
'confidence_score': sd['confidence_score'],
'radar_score': sd['radar_score'],
'decay_rate': 0.5,
'expires_at': expires,
'virality_level': sd['virality_level'],
'emotionality_level': sd['emotionality_level'],
'comment_potential': sd['comment_potential'],
'content_potential': sd['content_potential'],
'score_breakdown': sd['score_breakdown'],
'platform_facebook': 1,
'topic_class': 'sozial',
'image_url': image_url,
'image_source': ('meta_ad_creative' if ad.get('image_url') else 'meta_ad_video_poster') if image_url else '',
'image_found': 1 if image_url else 0,
'watch_entity_id': cfg.get('watch_entity_id', ''),
'watch_sector': cfg.get('category', ''),
'watch_priority': 'high',
}
sid = insert_signal(signal)
if sid is None:
duplicate += 1
# Refresh metadata/creative for an already-known public Ad ID.
import sqlite3
with sqlite3.connect('/var/lib/sma-data/signals.db') as sync:
sync.execute(
"UPDATE signals SET published_at=COALESCE(?,published_at), ad_start_at=COALESCE(?,ad_start_at,published_at), last_seen_at=CASE WHEN ?='active' THEN ? ELSE last_seen_at END, active_status=?, image_url=COALESCE(NULLIF(?,''),image_url), image_source=COALESCE(NULLIF(?,''),image_source), image_found=CASE WHEN ? <> '' THEN 1 ELSE image_found END, image_checked_at=? WHERE external_id=? AND source_type='meta_ad'",
(signal['published_at'], signal['ad_start_at'], signal['active_status'], signal['last_seen_at'], signal['active_status'], signal['image_url'], signal['image_source'], signal['image_url'], datetime.now(timezone.utc).isoformat(), signal['external_id']),
)
else:
new += 1
# Bekannte IDs, die in der aktiven öffentlichen Suche nicht erschienen,
# werden nicht fälschlich als inaktiv markiert: ohne belastbare Detailantwort
# bleibt der Status unknown. Aktive Treffer werden oben last_seen_at-aktualisiert.
import sqlite3
with sqlite3.connect('/var/lib/sma-data/signals.db') as sync:
known = [r[0] for r in sync.execute("SELECT external_id FROM signals WHERE source_type='meta_ad' AND external_id<>''").fetchall()]
for ad_id in known:
if ad_id not in active_ids:
sync.execute("UPDATE signals SET active_status=CASE WHEN active_status='active' THEN 'unknown' ELSE COALESCE(active_status,'unknown') END WHERE external_id=? AND source_type='meta_ad'", (ad_id,))
record_source_run('meta_ads', t0, seen, new, duplicate, rejected=rejected,
error=None, quota_usage=f'advertisers={len(advertisers)}')
return {'enabled': True, 'advertisers': len(advertisers), 'items_seen': seen,
'items_new': new, 'items_duplicate': duplicate,
'items_rejected': rejected, 'ads_matched': len(ads)}
def run_meta_ads(config: dict | None = None) -> int:
if config is None:
with CONFIG_PATH.open(encoding='utf-8') as fh:
config = yaml.safe_load(fh)
result = asyncio.run(_run_async(config))
print('[META-ADS] ' + str(result))
return int(result.get('items_new', 0))
if __name__ == '__main__':
run_meta_ads()