106 lines
4.7 KiB
Python
106 lines
4.7 KiB
Python
"""X/Twitter без платного API.
|
|
|
|
Три пути:
|
|
1. X_BEARER есть -> официальный /2/tweets/search/recent.
|
|
2. Иначе -> ищем ссылки на твиты через поисковики, текст достаём из publish.twitter.com/oembed (без ключей).
|
|
3. Профиль целиком -> syndication.twitter.com/srv/timeline-profile (отдаёт JSON в HTML, без ключей).
|
|
"""
|
|
import json
|
|
import re
|
|
import urllib.parse
|
|
|
|
from .. import config, net
|
|
from ..model import Item
|
|
|
|
TWEET_RE = re.compile(r"https?://(?:www\.)?(?:twitter\.com|x\.com)/([A-Za-z0-9_]{1,15})/status/(\d+)")
|
|
UA = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/141.0.0.0 Safari/537.36"}
|
|
|
|
|
|
def available(name="x"):
|
|
if config.KEYS.get("X_BEARER"):
|
|
return True, "официальный API v2 (X_BEARER)"
|
|
return True, "без ключей: oembed + syndication timeline"
|
|
|
|
|
|
def _oembed(url):
|
|
api = "https://publish.twitter.com/oembed?omit_script=1&dnt=1&url=" + urllib.parse.quote(url, safe="")
|
|
r = net.get(api, headers=UA, json_mode=True)
|
|
if r.status != 200:
|
|
return None
|
|
d = r.json() or {}
|
|
html = d.get("html") or ""
|
|
text = re.sub(r"<[^>]+>", " ", html)
|
|
text = re.sub(r"\s+", " ", text).replace("—", "-").strip()
|
|
return {"author": d.get("author_name") or "", "text": text,
|
|
"url": d.get("url") or url, "handle": (d.get("author_url") or "").rstrip("/").split("/")[-1]}
|
|
|
|
|
|
def search_x(query, limit=15, handles=None):
|
|
items = []
|
|
bearer = config.KEYS.get("X_BEARER")
|
|
if bearer:
|
|
q = urllib.parse.quote(f"{query} -is:retweet lang:ru OR lang:en")
|
|
url = (f"https://api.x.com/2/tweets/search/recent?query={q}"
|
|
f"&max_results={min(max(limit, 10), 100)}&tweet.fields=created_at,public_metrics,lang")
|
|
r = net.get(url, headers={"Authorization": f"Bearer {bearer}"}, json_mode=True)
|
|
for t in ((r.json() or {}).get("data") or []):
|
|
m = t.get("public_metrics") or {}
|
|
it = Item("x", "tweet", url=f"https://x.com/i/status/{t['id']}", title=t["text"][:120],
|
|
text=t["text"], author="", published=t.get("created_at", ""),
|
|
extra={"likes": m.get("like_count"), "rt": m.get("retweet_count")})
|
|
it.score = (m.get("like_count") or 0) ** 0.5
|
|
items.append(it)
|
|
else:
|
|
# путь 2: находим ссылки поисковиками, текст тянем oembed-ом
|
|
from . import searchers
|
|
found = []
|
|
for eng in ("ddg", "mojeek", "bing", "tavily"):
|
|
try:
|
|
res = getattr(searchers, "search_" + eng)(f"{query} site:x.com OR site:twitter.com", limit=limit)
|
|
except Exception:
|
|
res = []
|
|
for r0 in res:
|
|
for m in TWEET_RE.finditer(r0.url or ""):
|
|
found.append(m.group(0))
|
|
if len(found) >= limit:
|
|
break
|
|
seen = set()
|
|
for u in found[:limit * 2]:
|
|
if u in seen:
|
|
continue
|
|
seen.add(u)
|
|
d = _oembed(u)
|
|
if not d or len(d["text"]) < 20:
|
|
continue
|
|
it = Item("x", "tweet", url=d["url"], title=d["text"][:120], text=d["text"],
|
|
author=d["author"], source="x.com/" + (d["handle"] or ""),
|
|
extra={"handle": d["handle"]})
|
|
items.append(it)
|
|
|
|
# путь 3: конкретные профили
|
|
for h in (handles or [])[:4]:
|
|
r = net.get(f"https://syndication.twitter.com/srv/timeline-profile/screen-name/{h}",
|
|
headers=UA)
|
|
if r.status != 200:
|
|
continue
|
|
m = re.search(r'<script id="__NEXT_DATA__" type="application/json">(.*?)</script>', r.text, re.S)
|
|
if not m:
|
|
continue
|
|
try:
|
|
data = json.loads(m.group(1))
|
|
except Exception:
|
|
continue
|
|
entries = (((data.get("props") or {}).get("pageProps") or {}).get("timeline") or {}).get("entries") or []
|
|
for e in entries[:12]:
|
|
tw = ((e.get("content") or {}).get("tweet") or {})
|
|
txt = tw.get("full_text") or tw.get("text") or ""
|
|
if not txt:
|
|
continue
|
|
tid = tw.get("id_str") or ""
|
|
it = Item("x", "tweet", url=f"https://x.com/{h}/status/{tid}" if tid else f"https://x.com/{h}",
|
|
title=txt[:120], text=txt, author=h, source="x.com/" + h,
|
|
published=tw.get("created_at", ""),
|
|
extra={"likes": tw.get("favorite_count"), "rt": tw.get("retweet_count")})
|
|
it.score = float(tw.get("favorite_count") or 0) ** 0.5
|
|
items.append(it)
|
|
return items[:limit * 2] |