"""
AA-045-F5: Watch-Entity Intake — Branchen-/Wettbewerber-Beobachtung.
Liest watch_entities.yaml, pollt je aktiver verifizierter Entity:
- Website/Blog/RSS (kontrollierte Abfrage, og:image)
- YouTube Channel-RSS (falls channel_id)
- Facebook gezielt loginfrei via Bing-RSS: site:facebook.com "<Name>"
Rotation: entities_per_run pro Lauf; high zuerst. Idempotent über URL-Dedup.
Signale: source_type='watch_entity', source_platform=plattform.
"""
import sys
import re
import time
import json
import base64
import urllib.parse
import urllib.request
import xml.etree.ElementTree as ET
from datetime import datetime, timezone, timedelta
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
try:
import yaml
except ImportError:
yaml = None
from signal_db import insert_signal, record_source_run
CONFIG_PATH = Path(__file__).parent.parent / "watch_entities.yaml"
HEADERS = {
"User-Agent": ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 Chrome/124 Safari/537.36"),
"Accept-Language": "de,es;q=0.9,en;q=0.8",
}
FB_SKIP = re.compile(r"/login|register|m\.me|l\.facebook|sharer|plugins|/facebook/?$")
GNEWS_SKIP = re.compile(r"login|register|/facebook/?$")
def load_entities() -> list[dict]:
with open(CONFIG_PATH, encoding="utf-8") as fh:
data = yaml.safe_load(fh)
ents = [e for e in data.get("watch_entities", [])
if e.get("active") and e.get("verification", {}).get("status") == "verified"]
# Rotation: high priorisiert, stabile Reihenfolge über Laufnummer
settings = data.get("watch_settings", {})
per_run = int(settings.get("entities_per_run", 8))
run_no = int(datetime.now(timezone.utc).strftime("%Y%m%d%H")) % max(len(ents), 1)
ents.sort(key=lambda e: (0 if e.get("priority") == "high" else 1, e["entity_id"]))
rotated = ents[run_no:] + ents[:run_no]
return rotated[:per_run], len(ents)
def _get(url: str, timeout: int = 15) -> bytes | None:
try:
req = urllib.request.Request(url, headers=HEADERS)
with urllib.request.urlopen(req, timeout=timeout) as resp:
return resp.read()
except Exception as e:
print(f"[WATCH] HTTP-Fehler {url[:70]}: {e}")
return None
def _og_image(html_bytes: bytes) -> tuple[str, str]:
if not html_bytes:
return "", ""
h = html_bytes.decode("utf-8", errors="ignore")
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, h, 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 _store(entity: dict, platform: str, url: str, title: str, summary: str,
image_url: str = "") -> str | None:
"""Speichert einen Watch-Treffer als Signal (idempotent über URL-Dedup)."""
expires = (datetime.now(timezone.utc) + timedelta(hours=336)).isoformat()
signal = {
"source_type": "watch_entity",
"source_platform": platform,
"external_id": "",
"observed_at": datetime.now(timezone.utc).isoformat(),
"source_name": f"WATCH: {entity['canonical_name']} ({entity['sector']})",
"source_url": url,
"language": (entity.get("languages") or ["de"])[0],
"signal_category": "branche_watch",
"topic": (title or "(ohne Titel)")[:200],
"short_summary": (summary or title or "")[:500],
"extracted_hook": title[:150],
"emotional_direction": "neutral",
"urgency_level": 0,
"mallorca_relevance": 25 if entity.get("geography") == "mallorca" else 0,
"risk_relevance": 1,
"estimated_noise_level": 1,
"suggested_case_types": "branche_watch",
"suggested_platforms": "cockpit",
"suggested_cta": "beobachten",
"confidence_score": 0.7,
"radar_score": 40.0,
"decay_rate": 0.8,
"expires_at": expires,
"virality_level": 1,
"emotionality_level": 1,
"comment_potential": 2,
"content_potential": 2,
"score_breakdown": json.dumps({"watch_entity": entity["entity_id"],
"sector": entity["sector"],
"priority": entity.get("priority", "normal")}),
"platform_facebook": 1,
"platform_linkedin": 1,
"topic_class": "sozial",
"image_url": image_url,
"image_source": ("youtube" if platform == "youtube" and image_url else
("og:image" if image_url else "")),
"image_found": 1 if image_url else 0,
}
return insert_signal(signal)
# ── YouTube ──────────────────────────────────────────────────────────
def poll_youtube(entity: dict) -> int:
cid = entity.get("youtube_channel_id")
if not cid:
return 0
saved = 0
raw = _get(f"https://www.youtube.com/feeds/videos.xml?channel_id={cid}")
if not raw:
return 0
try:
root = ET.fromstring(raw)
except ET.ParseError:
return 0
ns = {"a": "http://www.w3.org/2005/Atom",
"yt": "http://www.youtube.com/xml/schemas/2015"}
for e in root.findall("a:entry", ns)[:5]:
vid = e.findtext("yt:videoId", "", ns)
if not vid:
continue
title = (e.findtext("a:title", "", ns) or "").strip()
thumb = f"https://i.ytimg.com/vi/{vid}/hqdefault.jpg"
sid = _store(entity, "youtube", f"https://www.youtube.com/watch?v={vid}",
title, title, thumb)
if sid:
saved += 1
return saved
# ── Facebook (loginfrei via Bing-RSS, firmenspezifisch) ──────────────
def poll_facebook(entity: dict) -> int:
name = entity["canonical_name"]
q = urllib.parse.quote(f'site:facebook.com "{name}"')
url = f"https://www.bing.com/search?q={q}&format=rss&count=20"
raw = _get(url)
if not raw:
return 0
try:
root = ET.fromstring(raw)
except ET.ParseError:
return 0
saved = 0
for item in root.findall(".//item"):
link = (item.findtext("link", "") or "").strip()
title = (item.findtext("title", "") or "").strip()
desc = (item.findtext("description", "") or "").strip()
if not link or "facebook.com" not in link:
continue
if FB_SKIP.search(link):
continue
# Keine bloße Firmenprofilseite als Post zählen (§29): Titel muss
# mehr enthalten als nur den Firmennamen
if title.lower().replace(" ", "") == name.lower().replace(" ", ""):
continue
sid = _store(entity, "facebook", link, title or f"{name} auf Facebook", desc)
if sid:
saved += 1
return saved
# ── Website / Blog / RSS ────────────────────────────────────────────
def poll_website(entity: dict) -> int:
saved = 0
blog = entity.get("blog_url")
if not blog:
return 0
# RSS-Versuch an üblichen Pfaden
rss_candidates = [
entity.get("rss_url"),
blog.rstrip("/") + "/feed/",
blog.rstrip("/") + "/rss",
blog.rstrip("/") + "?format=rss",
]
items = []
feed_url_used = None
for cand in filter(None, rss_candidates):
raw = _get(cand, timeout=12)
if raw:
try:
root = ET.fromstring(raw)
items = root.findall(".//item")[:5]
feed_url_used = cand
break
except ET.ParseError:
continue
if items:
for it in items:
title = (it.findtext("title", "") or "").strip()
link = (it.findtext("link", "") or "").strip()
desc = re.sub(r"<[^>]+>", " ", it.findtext("description", "") or "")
desc = re.sub(r"\s+", " ", desc).strip()
img = ""
enc = it.find("enclosure")
if enc is not None and (enc.get("type") or "").startswith("image"):
img = enc.get("url", "")
if not img:
img, _ = _og_image(_get(link, timeout=10))
if link.startswith("http"):
sid = _store(entity, "web", link, title, desc[:400], img)
if sid:
saved += 1
else:
# Blog-Seite ohne RSS: Links + og:image der Seite
raw = _get(blog, timeout=12)
if raw:
h = raw.decode("utf-8", errors="ignore")
links = re.findall(r'href="(https?://[^"]+' + re.escape(
urllib.parse.urlparse(blog).netloc.split(".")[-2] if "." in urllib.parse.urlparse(blog).netloc else ""
) + r'[^"]*)"', h)
seen = set()
for u in links[:8]:
if u in seen or u.rstrip("/") == blog.rstrip("/"):
continue
seen.add(u)
page = _get(u, timeout=10)
img, _src = _og_image(page)
tmatch = re.search(rb"<title>([^<]{5,150})", page or b"")
title = tmatch.group(1).decode("utf-8", "ignore").strip() if tmatch else u.rsplit("/", 1)[-1]
sid = _store(entity, "web", u, title, title, img)
if sid:
saved += 1
time.sleep(0.2)
return saved
# ── Orchestrator ─────────────────────────────────────────────────────
def run_all(limit_entities: int | None = None) -> int:
if yaml is None:
print("[WATCH] PyYAML fehlt")
return 0
ents, total_ents = load_entities()
if limit_entities:
ents = ents[:limit_entities]
print(f"[WATCH] {len(ents)}/{total_ents} Entities dieser Lauf "
f"(Rotation, verified+active)")
t0 = time.time()
stats = {"seen": 0, "new": 0}
for e in ents:
eid = e["entity_id"]
try:
n_web = poll_website(e)
n_fb = poll_facebook(e)
n_yt = poll_youtube(e)
new = n_web + n_fb + n_yt
stats["new"] += new
stats["seen"] += 1
record_source_run(f"watch:{eid}", t0, 3, new, 0, 0, 0, None)
print(f"[WATCH] {eid}: web={n_web} fb={n_fb} yt={n_yt}")
except Exception as exc:
print(f"[WATCH] Fehler bei {eid}: {exc}")
record_source_run(f"watch:{eid}", t0, 0, 0, 0, 0, 0, error=str(exc))
print(f"[WATCH] Gesamt neu: {stats['new']}")
return stats["new"]
if __name__ == "__main__":
import argparse
ap = argparse.ArgumentParser()
ap.add_argument("--limit", type=int, default=None)
args = ap.parse_args()
run_all(args.limit)