#!/usr/bin/env python3 """ Aria Knowledge Pipeline (AKP) — X (Twitter) Single-URL-Fetch (KAR-82) Manual-Link-Drop Pattern: Aria bekommt Tweet/Thread-URLs via Telegram oder discover/-Folder, pullt via Twitter Syndication API (no auth needed) und schreibt JSON nach brain/akp/raw/x/.json. Usage: python3 aria-akp-x-fetch.py python3 aria-akp-x-fetch.py --discover # batch alle x-*.url aus discover/ Cheap-Filter (Aria-First Value-Content): - text-length min 100 chars (skip Witz-Tweets) - skip wenn Werbung-Pattern oder Pure-Links-only - bevorzuge Threads über Single-Tweets (Indikator: replies vom selben Author) """ from __future__ import annotations import sys as _sys _sys.path.insert(0, "/root/aria/lib") from aria_logging import get_logger as _get_logger _log = _get_logger("aria-akp-x-fetch") import json, os, re, sqlite3, subprocess, sys, time, yaml from datetime import datetime, timezone from pathlib import Path CONFIG_PATH = Path("/root/aria/brain/youtube/.config.yaml") X_RAW_DIR = Path("/root/aria/brain/akp/raw/x") DISCOVER_DIR = Path("/root/aria/brain/akp/discover") AD_PATTERNS = [ r"\bbuy now\b", r"\baffiliate\b", r"\bsponsored\b", r"\bdiscount code\b", r"\bcheck my bio\b", ] def load_config(): with CONFIG_PATH.open() as f: return yaml.safe_load(f) def db_conn(db_path): con = sqlite3.connect(db_path); con.row_factory = sqlite3.Row return con def tweet_id_from_url(url: str) -> str | None: m = re.search(r"(?:twitter|x)\.com/[^/]+/status(?:es)?/(\d+)", url) return m.group(1) if m else None def gen_syndication_token(tweet_id: int) -> str: """Generate auth token for cdn.syndication.twimg.com. Algo extracted from Twitter's embed.js — calculated from tweet_id.""" return ((tweet_id / 1e15) * 3.1415).__repr__().replace("0", "").replace(".", "") def fetch_tweet(tweet_id: str, proxy: str | None) -> dict | None: """Fetch tweet. Try Syndication-API first, fxtwitter as fallback.""" # Pass 1: Twitter Syndication API token = gen_syndication_token(int(tweet_id)) url = f"https://cdn.syndication.twimg.com/tweet-result?id={tweet_id}&token={token}" data = _curl_json(url, proxy, referer="https://platform.twitter.com/") if data and "text" in data: return data # Pass 2: fxtwitter (community frontend with JSON-API) url2 = f"https://api.fxtwitter.com/i/status/{tweet_id}" data2 = _curl_json(url2, proxy) if data2 and data2.get("tweet"): t = data2["tweet"] # Normalize fxtwitter shape to syndication-like return { "text": t.get("text", ""), "created_at": t.get("created_timestamp") and datetime.fromtimestamp(t["created_timestamp"], tz=timezone.utc).isoformat() or "", "user": { "name": t.get("author", {}).get("name", ""), "screen_name": t.get("author", {}).get("screen_name", ""), }, "favorite_count": t.get("likes"), "conversation_count": t.get("replies"), "lang": t.get("lang"), } return None def _curl_json(url, proxy, referer=None): cmd = ["curl", "-s", "--max-time", "15", "-H", "User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)", "-H", "Accept: application/json"] if referer: cmd += ["-H", f"Referer: {referer}"] if proxy: cmd += ["--proxy", proxy] cmd.append(url) try: r = subprocess.run(cmd, capture_output=True, text=True, timeout=20) except subprocess.TimeoutExpired: return None if not r.stdout.strip() or r.stdout.startswith("<"): return None try: return json.loads(r.stdout) except json.JSONDecodeError: return None def already_ingested(con, pid): return con.execute("SELECT 1 FROM ingested WHERE video_id=?", (pid,)).fetchone() is not None def cheap_filter(tweet, text) -> str | None: if len(text.strip()) < 100: return "text_too_short" text_lower = text.lower() for pat in AD_PATTERNS: if re.search(pat, text_lower): return f"ad_pattern_{pat}" # Pure-Links Skip: wenn Tweet hauptsächlich URLs ohne Text-Substanz url_chars = sum(len(m.group(0)) for m in re.finditer(r"https?://\S+", text)) if url_chars / max(len(text), 1) > 0.6: return "mostly_urls" return None def assemble_text(tweet) -> str: """Extract main tweet-text + thread continuation if available.""" parts = [] if tweet.get("text"): parts.append(tweet["text"]) # Tweet might have parent/quoted content if tweet.get("parent") and isinstance(tweet["parent"], dict): p = tweet["parent"] if p.get("text"): parts.insert(0, f"[Quoted/Reply-To @{p.get('user_screen_name','?')}]: {p['text']}") return "\n\n".join(parts) def write_raw(tid, tweet, text, url): X_RAW_DIR.mkdir(parents=True, exist_ok=True) out_path = X_RAW_DIR / f"{tid}.json" user = tweet.get("user", {}) payload = { "video_id": f"x_{tid}", "channel_slug": user.get("screen_name", "x"), "channel_handle": f"@{user.get('screen_name','x')}", "title": text.split("\n")[0][:120], "url": url, "upload_date": (tweet.get("created_at", "") or "")[:10].replace("-", ""), "published_at": tweet.get("created_at", ""), "transcript": text, "transcript_chars": len(text), "ingested_at": datetime.now(timezone.utc).isoformat(), "source": "x", "yt_meta": { "user_name": user.get("name"), "user_screen_name": user.get("screen_name"), "favorite_count": tweet.get("favorite_count"), "conversation_count": tweet.get("conversation_count"), "lang": tweet.get("lang"), }, } out_path.write_text(json.dumps(payload, indent=2, ensure_ascii=False)) return str(out_path) def store_ingested(con, tid, screen_name, title, url, raw_path, chars, pub): con.execute( "INSERT OR REPLACE INTO ingested (video_id, channel, channel_handle, title, " "duration_seconds, published_at, url, transcript_chars, raw_path, source) " "VALUES (?,?,?,?,?,?,?,?,?,'x')", (f"x_{tid}", screen_name, f"@{screen_name}", title[:300], None, pub, url, chars, raw_path), ) con.commit() def store_skipped(con, tid, title, url, reason): con.execute( "INSERT OR IGNORE INTO ingested (video_id, channel, title, url, skipped_reason, source) VALUES (?,?,?,?,?,'x')", (f"x_{tid}", "x", title[:200], url, reason), ) con.commit() def process_url(url, con, proxy): tid = tweet_id_from_url(url) if not tid: print(f" [{url}] invalid URL", file=sys.stderr); return "failed" if already_ingested(con, f"x_{tid}"): print(f" [{tid}] already ingested"); return "dup" tweet = fetch_tweet(tid, proxy) if not tweet or "text" not in tweet: print(f" [{tid}] fetch failed (private/deleted/syndication-block?)", file=sys.stderr); return "failed" text = assemble_text(tweet) title_short = text.split("\n")[0][:100] reason = cheap_filter(tweet, text) if reason: store_skipped(con, tid, title_short, url, reason) print(f" [{tid}] skipped: {reason}"); return "skipped" raw_path = write_raw(tid, tweet, text, url) user = tweet.get("user", {}) store_ingested(con, tid, user.get("screen_name","x"), title_short, url, raw_path, len(text), tweet.get("created_at","")) print(f" [{tid}] ingested by @{user.get('screen_name')}: {title_short[:50]}") return "ok" def main(argv): cfg = load_config() con = db_conn(cfg["paths"]["state_db"]) proxy = os.environ.get("YT_PROXY") or None urls = [] if "--discover" in argv: for f in DISCOVER_DIR.glob("x-*.url"): text = f.read_text(errors="replace") for m in re.finditer(r"https?://(?:twitter|x)\.com/[^/\s]+/status(?:es)?/\d+", text): urls.append(m.group(0)) try: f.unlink() except OSError: pass else: for arg in argv: if arg.startswith("http"): urls.append(arg) if not urls: print("[akp-x] no URLs. Usage: | --discover") return 1 print(f"[akp-x] processing {len(urls)} URL(s)") counts = {"ok":0,"skipped":0,"failed":0,"dup":0} for url in urls: counts[process_url(url, con, proxy)] += 1 time.sleep(0.5) print(f"[akp-x] {counts}") con.close() return 0 if __name__ == "__main__": _log.event("script_start") sys.exit(main(sys.argv[1:]))