"""
AA-045-F4: Link-/Bild-Anreicherung fuer bestehende Radar-Signale.
- Google-News-Aggregator-URL -> final_article_url aufloesen (mit Cache)
- Zielartikel laden -> og:image/twitter:image
- YouTube-Thumbnails als Sicherheitsnetz
Kontrollierter Backfill: nur aktuelle Signale ohne Bild, kurze Timeouts,
kein historischer Blind-Crawl. Idempotent.
"""
import sys
import re
import time
import urllib.request
import urllib.error
from datetime import datetime, timezone
from pathlib import Path
sys.path.insert(0, "/opt/struktur/social-media-radar")
HEADERS = {
"User-Agent": ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124 Safari/537.36"),
"Accept-Language": "de,es;q=0.9,en;q=0.8",
}
TIMEOUT = 10
MAX_HTML = 65536
# Redirect-Cache: gnews_url -> final_url (Datei-basiert, ueberlaeufe hinweg)
CACHE_FILE = Path("/var/lib/sma-data/gnews_redirect_cache.txt")
def _load_cache() -> dict:
cache = {}
if CACHE_FILE.exists():
for line in CACHE_FILE.read_text().splitlines():
if "\t" in line:
k, v = line.split("\t", 1)
cache[k] = v
return cache
def _save_cache(k: str, v: str):
with CACHE_FILE.open("a") as fh:
fh.write(f"{k}\t{v}\n")
def resolve_gnews(url: str, cache: dict) -> str:
"""Loest eine Google-News-URL zum Zielartikel auf (caching, kein Loop)."""
if url in cache:
return cache[url]
final = ""
try:
req = urllib.request.Request(url, headers=HEADERS)
# Google-News-Redirects sind JS-basiert; die Zielfirma steht oft im HTML.
with urllib.request.urlopen(req, timeout=TIMEOUT) as resp:
html = resp.read(MAX_HTML).decode("utf-8", errors="replace")
if resp.url and "news.google.com" not in resp.url:
final = resp.url
if not final:
m = re.search(r'data-n-au="([^"]+)"', html) or \
re.search(r'href="(https://[^"]+)"[^>]*jslog', html) or \
re.search(r'<a[^>]+href="(https?://(?!www\.google|news\.google)[^"]+)"', html)
if m:
final = m.group(1)
except Exception:
pass
if final:
_save_cache(url, final)
cache[url] = final
return final
def fetch_og_image(url: str):
"""og:image / twitter:image aus einer Seite (klein, mit Timeout)."""
try:
req = urllib.request.Request(url, headers=HEADERS)
with urllib.request.urlopen(req, timeout=TIMEOUT) as resp:
html = resp.read(MAX_HTML).decode("utf-8", errors="replace")
except Exception:
return "", ""
for pat in (
r'<meta[^>]+property=["\']og:image["\'][^>]+content=["\']([^"\'>\s]+)',
r'<meta[^>]+content=["\']([^"\'>\s]+)["\'][^>]+property=["\']og:image["\']',
r'<meta[^>]+name=["\']twitter:image["\'][^>]+content=["\']([^"\'>\s]+)',
):
m = re.search(pat, html, re.IGNORECASE)
if m:
u = m.group(1).strip()
if u.startswith("http"):
src = "twitter:image" if "twitter" in pat else "og:image"
return u, src
return "", ""
def yt_thumb_ok(video_id: str) -> bool:
url = f"https://i.ytimg.com/vi/{video_id}/hqdefault.jpg"
try:
req = urllib.request.Request(url, method="HEAD")
with urllib.request.urlopen(req, timeout=8) as resp:
return resp.status == 200 and "image" in resp.headers.get("Content-Type", "")
except Exception:
return False
def run_backfill(days: int = 7, limit: int = 60) -> dict:
import sqlite3
conn = sqlite3.connect("/var/lib/sma-data/signals.db")
conn.row_factory = sqlite3.Row
stats = {"youtube": 0, "gnews": 0, "direct_og": 0, "errors": 0}
now = datetime.now(timezone.utc).isoformat()
cutoff = (datetime.now(timezone.utc).__class__
.fromtimestamp(__import__("time").time() - days * 86400, tz=timezone.utc).isoformat())
cache = _load_cache()
# 1. YouTube ohne Bild
rows = conn.execute("""
SELECT signal_id, source_url FROM signals
WHERE source_type='youtube' AND COALESCE(image_found,0)=0
AND created_at > ? LIMIT ?""", (cutoff, limit)).fetchall()
for r in rows:
m = re.search(r"[?&]v=([A-Za-z0-9_-]{11})", r["source_url"] or "")
if m and yt_thumb_ok(m.group(1)):
conn.execute(
"UPDATE signals SET image_url=?, image_source='youtube', image_found=1,"
" image_checked_at=? WHERE signal_id=?",
(f"https://i.ytimg.com/vi/{m.group(1)}/hqdefault.jpg", now, r["signal_id"]))
stats["youtube"] += 1
# 2. Google News: URL aufloesen + og:image
rows = conn.execute("""
SELECT signal_id, source_url FROM signals
WHERE source_name LIKE '%Google News%'
AND (COALESCE(image_found,0)=0 OR COALESCE(final_article_url,'')='')
AND created_at > ? LIMIT ?""", (cutoff, limit)).fetchall()
for r in rows:
try:
final = resolve_gnews(r["source_url"], cache)
if not final:
continue
img, src = fetch_og_image(final)
sets, vals = ["final_article_url=?"], [final]
if img:
sets += ["image_url=?", "image_source=?", "image_found=1"]
vals += [img, src, 1]
vals += [now, r["signal_id"]]
conn.execute(f"UPDATE signals SET {', '.join(sets)}, image_checked_at=? WHERE signal_id=?", vals)
stats["gnews"] += 1 if img else 0
time.sleep(0.3)
except Exception:
stats["errors"] += 1
# 3. Direkte Artikel ohne Bild: og:image nachladen
rows = conn.execute("""
SELECT signal_id, source_url FROM signals
WHERE source_type='rss' AND source_name NOT LIKE '%Google News%'
AND COALESCE(image_found,0)=0 AND created_at > ? LIMIT ?""", (cutoff, limit)).fetchall()
for r in rows:
try:
img, src = fetch_og_image(r["source_url"])
if img:
conn.execute(
"UPDATE signals SET image_url=?, image_source=?, image_found=1,"
" image_checked_at=? WHERE signal_id=?",
(img, src, now, r["signal_id"]))
stats["direct_og"] += 1
else:
conn.execute("UPDATE signals SET image_checked_at=? WHERE signal_id=?",
(now, r["signal_id"]))
time.sleep(0.3)
except Exception:
stats["errors"] += 1
conn.commit()
# Vorher/Nachher
total = conn.execute("SELECT COUNT(*) FROM signals WHERE created_at > ?", (cutoff,)).fetchone()[0]
with_img = conn.execute("SELECT COUNT(*) FROM signals WHERE image_found=1 AND created_at > ?", (cutoff,)).fetchone()[0]
conn.close()
stats["total_period"] = total
stats["with_img_after"] = with_img
return stats
if __name__ == "__main__":
import json
print(json.dumps(run_backfill(), indent=2))