"""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 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] = []
# 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)')
except Exception:
continue
if 'Sponsored' in text and ('Active' in text or 'Inactive' in text) and images:
card_text, image_urls = text, images
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), '')
return {'ad_id': ad_id, 'advertiser': advertiser, 'creative': creative,
'started_raw': _parse_start(card_text), 'image_url': image,
'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] = []
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}'})
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 ''
signal = {
'source_type': 'meta_ad',
'source_platform': 'meta_ad',
'external_id': ad['ad_id'],
'observed_at': datetime.now(timezone.utc).isoformat(),
'published_at': ad.get('started_raw'),
'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 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
else:
new += 1
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()