140 lines
5.8 KiB
Python
140 lines
5.8 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: конкретные профили через fxtwitter (без ключей, отдаёт полный текст)
|
|
for h in (handles or [])[:6]:
|
|
for it in profile(h, limit=12):
|
|
items.append(it)
|
|
|
|
# дедуп по ссылке
|
|
uniq, seen_u = [], set()
|
|
for it in items:
|
|
if it.url in seen_u:
|
|
continue
|
|
seen_u.add(it.url)
|
|
uniq.append(it)
|
|
return uniq[:max(limit, 12) if handles else len(uniq)]
|
|
|
|
|
|
def profile(handle, limit=20, cursor=None):
|
|
"""Лента профиля X через api.fxtwitter.com — без токенов и логина.
|
|
|
|
Возвращает список Item; при ошибке сети/404 — пустой список.
|
|
"""
|
|
h = (handle or "").lstrip("@").strip()
|
|
if not h:
|
|
return []
|
|
url = f"https://api.fxtwitter.com/2/profile/{h}/statuses"
|
|
if cursor:
|
|
url += "?cursor=" + urllib.parse.quote(str(cursor))
|
|
try:
|
|
r = net.get(url, timeout=25, json_mode=True)
|
|
except Exception:
|
|
return []
|
|
if r.status != 200:
|
|
return []
|
|
data = r.json() or {}
|
|
out = []
|
|
for t in (data.get("results") or []):
|
|
if t.get("type") != "status":
|
|
continue
|
|
rt = t.get("raw_text")
|
|
txt = (rt.get("text") if isinstance(rt, dict) else rt) or t.get("text") or ""
|
|
if not isinstance(txt, str):
|
|
txt = str(txt)
|
|
if len(txt) < 15:
|
|
continue
|
|
tid = t.get("id") or ""
|
|
author = (t.get("author") or {}).get("screen_name") or h if isinstance(t.get("author"), dict) else h
|
|
likes = t.get("likes") or 0
|
|
it = Item("x", "tweet",
|
|
url=t.get("url") or f"https://x.com/{h}/status/{tid}",
|
|
title=txt.strip().splitlines()[0][:140], text=txt,
|
|
author=author, source="x.com/" + h,
|
|
published=t.get("created_at", ""),
|
|
extra={"handle": h, "lang": t.get("lang", ""), "likes": likes, "replies": t.get("replies"),
|
|
"reposts": t.get("reposts"), "views": t.get("views"),
|
|
"quote": (t.get("quote") or {}).get("url") if isinstance(t.get("quote"), dict) else None})
|
|
it.score = float(likes) ** 0.5 + float(t.get("replies") or 0) ** 0.4
|
|
out.append(it)
|
|
if len(out) >= limit:
|
|
break
|
|
return out
|
|
|
|
return items[:limit * 2] |