717 lines
28 KiB
Python
717 lines
28 KiB
Python
#!/usr/bin/env python3
|
|
"""twitter-cli 互換アダプタ (twscrape + twikit バックエンド)
|
|
|
|
なぜ存在するか:
|
|
X.com が 2026 年に Web クライアントを新バンドル (x-web / `sign.o*.js`) へ移行し、
|
|
twitter-cli が依存する x_client_transaction (1.0.2) の `ondemand.s` 解析が壊れた。
|
|
transaction-id を生成できず全 X ツールが HTTP 404 で全滅 (`exited with code 1`)。
|
|
twscrape の `XClIdGen` は新バンドルに追従済みなので、txid 生成と GraphQL 取得を
|
|
twscrape に任せ、twitter-cli が出していた YAML と同じ形を stdout に吐く。
|
|
既存の src/engine/tools/x.ts は無改修で、config の tools.x_cli_command を
|
|
このアダプタに向けるだけで動く。
|
|
|
|
サブコマンド (twitter-cli と同じ呼ばれ方):
|
|
search <query> -t <tab> --max N --yaml [--full-text] [--compact]
|
|
user-posts <username> --max N --yaml [...]
|
|
tweet <id|url> --yaml [...]
|
|
feed --max N --yaml [-t following] [...] # ホームタイムライン
|
|
--version
|
|
|
|
認証 (env 経由、x.ts が設定):
|
|
TWITTER_AUTH_TOKEN, TWITTER_CT0
|
|
|
|
出力 YAML (x.ts の parseXPostsFromYaml / downloadTweetMedia が解釈する形):
|
|
ok: true
|
|
schema_version: '1'
|
|
data:
|
|
- id, text, author{name,screenName,profileImageUrl,id},
|
|
metrics{likes,retweets,replies,views}, createdAtISO, isRetweet,
|
|
media[]{type, url, variants[]{url,bitrate,contentType}}
|
|
article{title, previewText, plainText, ...} # X 長文記事のみ (tweet サブコマンド)
|
|
# 長文記事の埋め込み画像・動画は media[] に合流する (自動 DL 対象)
|
|
|
|
注意: twscrape / twikit / httpx は遅延 import (関数内)。pure なマッピング関数だけ
|
|
なら依存なしで import でき、test_x_adapter.py がネットワーク・依存なしで回る。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import sys
|
|
from email.utils import parsedate_to_datetime
|
|
|
|
VERSION = "x-adapter 1.2.0 (twscrape backend)"
|
|
|
|
# twitter-cli の YAML スキーマバージョン (x.ts は使わないが互換のため踏襲)
|
|
SCHEMA_VERSION = "1"
|
|
|
|
|
|
# ── pure helpers (依存なし・テスト対象) ──────────────────────────────
|
|
|
|
def parse_tweet_ref(raw: str) -> str | None:
|
|
"""tweet ID もしくは status / article URL から数値 ID を取り出す。
|
|
|
|
X の長文記事 (X Articles) は `x.com/{user}/article/{id}` 形式で共有されるが、
|
|
その id はツイート ID と同じ空間なので TweetDetail でそのまま引ける。
|
|
"""
|
|
s = (raw or "").strip()
|
|
if not s:
|
|
return None
|
|
if s.isdigit():
|
|
return s
|
|
m = re.search(r"(?:x\.com|twitter\.com)/[^/]+/(?:status(?:es)?|article)/(\d+)", s)
|
|
return m.group(1) if m else None
|
|
|
|
|
|
def twitter_time_iso(created_at: str) -> str:
|
|
"""GraphQL legacy の 'Wed Oct 10 20:19:24 +0000 2018' を ISO8601 へ。"""
|
|
if not created_at:
|
|
return ""
|
|
try:
|
|
return parsedate_to_datetime(created_at).isoformat()
|
|
except Exception:
|
|
return created_at
|
|
|
|
|
|
def _iso_from_dt(dt) -> str:
|
|
if dt is None:
|
|
return ""
|
|
try:
|
|
return dt.isoformat()
|
|
except Exception:
|
|
return str(dt)
|
|
|
|
|
|
def _as_dict(obj) -> dict:
|
|
if isinstance(obj, dict):
|
|
return obj
|
|
return getattr(obj, "__dict__", {}) or {}
|
|
|
|
|
|
def _dig(d, *keys, default=None):
|
|
"""ネストした dict を安全にたどる。"""
|
|
cur = d
|
|
for k in keys:
|
|
if not isinstance(cur, dict):
|
|
return default
|
|
cur = cur.get(k)
|
|
if cur is None:
|
|
return default
|
|
return cur
|
|
|
|
|
|
# ── twscrape Tweet オブジェクト → YAML dict ───────────────────────────
|
|
|
|
def _media_from_twscrape(t) -> list:
|
|
out: list = []
|
|
media = getattr(t, "media", None)
|
|
if not media:
|
|
return out
|
|
for p in getattr(media, "photos", None) or []:
|
|
url = getattr(p, "url", None) if not isinstance(p, dict) else p.get("url")
|
|
if url:
|
|
out.append({"type": "photo", "url": url})
|
|
for v in getattr(media, "videos", None) or []:
|
|
vd = _as_dict(v)
|
|
variants = []
|
|
for var in (getattr(v, "variants", None) or vd.get("variants") or []):
|
|
d = _as_dict(var)
|
|
vu = d.get("url")
|
|
if vu:
|
|
variants.append({
|
|
"url": vu,
|
|
"bitrate": d.get("bitrate") or 0,
|
|
"contentType": d.get("contentType") or d.get("content_type") or "",
|
|
})
|
|
out.append({
|
|
"type": "video",
|
|
"url": (getattr(v, "thumbnailUrl", None) or vd.get("thumbnailUrl") or ""),
|
|
"variants": variants,
|
|
})
|
|
for g in getattr(media, "animated", None) or []:
|
|
gd = _as_dict(g)
|
|
variants = []
|
|
for var in (getattr(g, "variants", None) or gd.get("variants") or []):
|
|
d = _as_dict(var)
|
|
if d.get("url"):
|
|
variants.append({"url": d["url"], "bitrate": d.get("bitrate") or 0,
|
|
"contentType": d.get("contentType") or d.get("content_type") or ""})
|
|
out.append({
|
|
"type": "animated_gif",
|
|
"url": (getattr(g, "thumbnailUrl", None) or gd.get("thumbnailUrl") or ""),
|
|
"variants": variants,
|
|
})
|
|
return out
|
|
|
|
|
|
def twscrape_tweet_to_dict(t) -> dict:
|
|
u = getattr(t, "user", None)
|
|
return {
|
|
"id": str(getattr(t, "id_str", None) or getattr(t, "id", "") or ""),
|
|
"text": getattr(t, "rawContent", "") or "",
|
|
"author": {
|
|
"id": str(getattr(u, "id_str", "") or "") if u else "",
|
|
"name": (getattr(u, "displayname", "") or "") if u else "",
|
|
"screenName": (getattr(u, "username", "") or "") if u else "",
|
|
"profileImageUrl": (getattr(u, "profileImageUrl", "") or "") if u else "",
|
|
},
|
|
"metrics": {
|
|
"likes": getattr(t, "likeCount", 0) or 0,
|
|
"retweets": getattr(t, "retweetCount", 0) or 0,
|
|
"replies": getattr(t, "replyCount", 0) or 0,
|
|
"views": getattr(t, "viewCount", 0) or 0,
|
|
},
|
|
"createdAtISO": _iso_from_dt(getattr(t, "date", None)),
|
|
"isRetweet": getattr(t, "retweetedTweet", None) is not None,
|
|
"media": _media_from_twscrape(t),
|
|
}
|
|
|
|
|
|
# ── raw GraphQL (home timeline) result → YAML dict ────────────────────
|
|
|
|
def parse_graphql_tweet_result(result: dict) -> dict | None:
|
|
"""GraphQL の tweet_results.result を YAML dict に変換。
|
|
|
|
X は user フィールドを legacy → core へ移行中なので両方から拾う。
|
|
"""
|
|
if not isinstance(result, dict):
|
|
return None
|
|
if result.get("__typename") == "TweetWithVisibilityResults":
|
|
result = result.get("tweet", result)
|
|
legacy = result.get("legacy") or {}
|
|
if not legacy:
|
|
return None
|
|
|
|
user_result = _dig(result, "core", "user_results", "result", default={}) or {}
|
|
user_legacy = user_result.get("legacy") or {}
|
|
user_core = user_result.get("core") or {}
|
|
views = _dig(result, "views", "count", default=0)
|
|
|
|
media = []
|
|
for me in (_dig(legacy, "extended_entities", "media", default=[]) or []):
|
|
mt = me.get("type")
|
|
if mt == "photo":
|
|
media.append({"type": "photo", "url": me.get("media_url_https")})
|
|
elif mt in ("video", "animated_gif"):
|
|
variants = [
|
|
{"url": v.get("url"), "bitrate": v.get("bitrate", 0),
|
|
"contentType": v.get("content_type", "")}
|
|
for v in (_dig(me, "video_info", "variants", default=[]) or [])
|
|
if v.get("url")
|
|
]
|
|
media.append({"type": mt, "url": me.get("media_url_https"), "variants": variants})
|
|
|
|
return {
|
|
"id": str(result.get("rest_id") or legacy.get("id_str") or ""),
|
|
"text": legacy.get("full_text", "") or "",
|
|
"author": {
|
|
"id": str(user_result.get("rest_id", "") or ""),
|
|
"name": user_legacy.get("name") or user_core.get("name") or "",
|
|
"screenName": (user_legacy.get("screen_name")
|
|
or user_core.get("screen_name") or ""),
|
|
"profileImageUrl": (user_legacy.get("profile_image_url_https")
|
|
or _dig(user_result, "avatar", "image_url", default="") or ""),
|
|
},
|
|
"metrics": {
|
|
"likes": legacy.get("favorite_count", 0) or 0,
|
|
"retweets": legacy.get("retweet_count", 0) or 0,
|
|
"replies": legacy.get("reply_count", 0) or 0,
|
|
"views": int(views) if str(views).isdigit() else 0,
|
|
},
|
|
"createdAtISO": twitter_time_iso(legacy.get("created_at", "")),
|
|
"isRetweet": "retweeted_status_result" in legacy,
|
|
"media": media,
|
|
}
|
|
|
|
|
|
# ── X 長文記事 (X Articles) の抽出 ──────────────────────────────────
|
|
#
|
|
# 長文記事ポストの legacy.full_text は記事への t.co リンク 1 本だけで本文を含まない。
|
|
# 本文は TweetDetail レスポンスの tweet result 直下 `article.article_results.result`
|
|
# に入る (plain_text は fieldToggles.withArticlePlainText: true を付けた時のみ)。
|
|
|
|
# plain_text の既定上限 (文字数)。LLM コンテキスト保護。--full-text で解除。
|
|
ARTICLE_TEXT_CAP = 12000
|
|
|
|
# 記事内埋め込みメディアの件数上限。x.ts の downloadTweetMedia は media[] を
|
|
# 全件 DL する (件数制限なし・サイズ上限のみ) ので、画像だらけの記事で
|
|
# ダウンロードが暴発しないようアダプタ側で頭を抑える。
|
|
ARTICLE_MEDIA_CAP = 20
|
|
|
|
|
|
def find_tweet_result_in_detail(payload, tid: str) -> dict | None:
|
|
"""TweetDetail レスポンス全体から rest_id == tid の tweet_results.result を探す。
|
|
|
|
conversation スレッド内の別ツイート (リプライ等) も同じ形で並ぶので、
|
|
rest_id の一致で focal tweet を特定する。visibility wrapper は剥がして返す。
|
|
"""
|
|
if isinstance(payload, dict):
|
|
tr = payload.get("tweet_results")
|
|
if isinstance(tr, dict):
|
|
r = tr.get("result")
|
|
if isinstance(r, dict):
|
|
if r.get("__typename") == "TweetWithVisibilityResults":
|
|
r = r.get("tweet") if isinstance(r.get("tweet"), dict) else r
|
|
if isinstance(r, dict) and str(r.get("rest_id") or "") == str(tid):
|
|
return r
|
|
for v in payload.values():
|
|
hit = find_tweet_result_in_detail(v, tid)
|
|
if hit is not None:
|
|
return hit
|
|
elif isinstance(payload, list):
|
|
for v in payload:
|
|
hit = find_tweet_result_in_detail(v, tid)
|
|
if hit is not None:
|
|
return hit
|
|
return None
|
|
|
|
|
|
def extract_article(result, full_text: bool = False) -> dict | None:
|
|
"""tweet result の article.article_results.result から記事情報を YAML dict に。
|
|
|
|
記事でないツイート・形が崩れている場合は None (呼び出し側は無視して従来出力)。
|
|
"""
|
|
art = _dig(result if isinstance(result, dict) else {}, "article", "article_results", "result")
|
|
if not isinstance(art, dict):
|
|
return None
|
|
out: dict = {
|
|
"title": art.get("title") or "",
|
|
"previewText": art.get("preview_text") or "",
|
|
}
|
|
plain = art.get("plain_text")
|
|
if isinstance(plain, str) and plain:
|
|
if not full_text and len(plain) > ARTICLE_TEXT_CAP:
|
|
out["plainText"] = plain[:ARTICLE_TEXT_CAP]
|
|
out["plainTextTruncated"] = True
|
|
else:
|
|
out["plainText"] = plain
|
|
pub = _dig(art, "metadata", "first_published_at_secs")
|
|
if isinstance(pub, (int, float)) and pub > 0:
|
|
from datetime import datetime, timezone
|
|
out["publishedAtISO"] = datetime.fromtimestamp(int(pub), tz=timezone.utc).isoformat()
|
|
cover = _dig(art, "cover_media", "media_info", "original_img_url")
|
|
if cover:
|
|
out["coverImageUrl"] = cover
|
|
return out
|
|
|
|
|
|
def extract_article_media(result) -> list:
|
|
"""記事内の埋め込みメディア (media_entities) を row の media[] と同じ形に変換。
|
|
|
|
media_entities は fieldToggles.withArticleRichContentState: true の時のみ返る。
|
|
row["media"] に合流させることで x.ts の downloadTweetMedia がそのまま
|
|
自動 DL し localPath を付ける (記事ポストの tweet-level media は常に空なので
|
|
衝突しない)。未知の __typename (音声スペース等) は黙ってスキップ。
|
|
"""
|
|
art = _dig(result if isinstance(result, dict) else {}, "article", "article_results", "result")
|
|
if not isinstance(art, dict):
|
|
return []
|
|
out: list = []
|
|
for entity in art.get("media_entities") or []:
|
|
if len(out) >= ARTICLE_MEDIA_CAP:
|
|
break
|
|
info = _dig(_as_dict(entity), "media_info") or {}
|
|
tname = info.get("__typename")
|
|
if tname == "ApiImage":
|
|
url = info.get("original_img_url")
|
|
if url:
|
|
out.append({"type": "photo", "url": url})
|
|
elif tname in ("ApiVideo", "ApiGif"):
|
|
variants = [
|
|
{"url": v.get("url"), "bitrate": v.get("bit_rate") or 0,
|
|
"contentType": v.get("content_type") or ""}
|
|
for v in (info.get("variants") or [])
|
|
if isinstance(v, dict) and v.get("url")
|
|
]
|
|
out.append({
|
|
"type": "video" if tname == "ApiVideo" else "animated_gif",
|
|
"url": _dig(info, "preview_image", "original_img_url") or "",
|
|
"variants": variants,
|
|
})
|
|
return out
|
|
|
|
|
|
def _collect_tweet_results(node, out: list, limit: int) -> None:
|
|
"""entry / module item から tweet_results.result を拾って out に積む。
|
|
|
|
通常ツイート (content.itemContent...) と、モジュール内アイテム
|
|
(content.items[].item.itemContent... / moduleItems[]) の両方をたどる。
|
|
ツイート以外 (who-to-follow ユーザー, cursor) は parse_graphql_tweet_result が
|
|
None を返すので自然に除外される。
|
|
"""
|
|
if len(out) >= limit:
|
|
return
|
|
content = node.get("content") or node.get("item") or {}
|
|
res = _dig(content, "itemContent", "tweet_results", "result")
|
|
if res:
|
|
row = parse_graphql_tweet_result(res)
|
|
if row:
|
|
out.append(row)
|
|
return
|
|
for it in (content.get("items") or []):
|
|
_collect_tweet_results(it, out, limit)
|
|
if len(out) >= limit:
|
|
return
|
|
|
|
|
|
def parse_home_timeline(payload: dict, limit: int) -> list:
|
|
"""HomeTimeline / HomeLatestTimeline の GraphQL レスポンスを dict 配列へ。
|
|
|
|
TimelineAddEntries の entries[] と TimelineAddToModule の moduleItems[] の
|
|
両方を走査する。
|
|
"""
|
|
out: list = []
|
|
instructions = (_dig(payload, "data", "home", "home_timeline_urt", "instructions", default=[]) or [])
|
|
for ins in instructions:
|
|
itype = ins.get("type")
|
|
if itype == "TimelineAddEntries":
|
|
for entry in ins.get("entries", []) or []:
|
|
_collect_tweet_results(entry, out, limit)
|
|
if len(out) >= limit:
|
|
return out
|
|
elif itype == "TimelineAddToModule":
|
|
for it in ins.get("moduleItems", []) or []:
|
|
_collect_tweet_results(it, out, limit)
|
|
if len(out) >= limit:
|
|
return out
|
|
return out
|
|
|
|
|
|
# ── 出力 ──────────────────────────────────────────────────────────────
|
|
|
|
def emit(rows: list) -> None:
|
|
import yaml # PyYAML
|
|
doc = {"ok": True, "schema_version": SCHEMA_VERSION, "data": rows}
|
|
sys.stdout.write(yaml.safe_dump(doc, allow_unicode=True, sort_keys=False))
|
|
|
|
|
|
def fail(message: str, code: int = 1) -> None:
|
|
"""twitter-cli と同じく非ゼロ終了 + stderr。x.ts がエラー扱いにする。"""
|
|
sys.stderr.write(message.rstrip() + "\n")
|
|
sys.exit(code)
|
|
|
|
|
|
def _silence_logs() -> None:
|
|
"""twscrape は loguru で出力する。万一 stdout に乗ると x.ts の YAML.parse が
|
|
壊れるので、全 loguru sink を落とす (現状は stderr 行きで実害なしだが防御的に)。"""
|
|
try:
|
|
from loguru import logger as _loguru
|
|
_loguru.remove()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
# ── backend (twscrape / twikit、遅延 import) ──────────────────────────
|
|
|
|
def _creds() -> tuple[str, str]:
|
|
import os
|
|
auth = os.environ.get("TWITTER_AUTH_TOKEN", "").strip()
|
|
ct0 = os.environ.get("TWITTER_CT0", "").strip()
|
|
if not auth or not ct0:
|
|
fail("x-adapter: TWITTER_AUTH_TOKEN / TWITTER_CT0 are not set. "
|
|
"Configure tools.x_auth_token / tools.x_ct0 in config.yaml.")
|
|
return auth, ct0
|
|
|
|
|
|
_TMP_PREFIX = "xadapter-"
|
|
|
|
|
|
def _cleanup_stale_tmp() -> None:
|
|
"""過去の SIGKILL タイムアウト等で残った temp dir (cookie 入り sqlite) を掃除。
|
|
|
|
x.ts はタイムアウト時に SIGKILL するため finally が走らず temp dir が残りうる。
|
|
起動時に 1 時間以上前の `xadapter-*` を削除し、平文 cookie の /tmp 残留を抑える。
|
|
"""
|
|
import glob
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
import time
|
|
try:
|
|
now = time.time()
|
|
for d in glob.glob(os.path.join(tempfile.gettempdir(), _TMP_PREFIX + "*")):
|
|
try:
|
|
if os.path.isdir(d) and now - os.path.getmtime(d) > 3600:
|
|
shutil.rmtree(d, ignore_errors=True)
|
|
except Exception:
|
|
pass
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
async def _ensure_alive(api) -> None:
|
|
"""直近のリクエストでアカウントが 401/banned 扱いになっていないか確認。
|
|
|
|
twscrape は 401/429 を例外にせずログして停止し、空イテレーションを返す。空結果が
|
|
『本当に0件』か『認証切れ/レート制限』かを区別するため、停止時に無効化される
|
|
account.active を見て、無効なら nonzero で fail する (x.ts がエラー表示)。
|
|
"""
|
|
try:
|
|
accs = await api.pool.get_all()
|
|
except Exception:
|
|
return
|
|
if accs and not getattr(accs[0], "active", True):
|
|
fail("x-adapter: X rejected the session (cookies expired/banned or rate-limited). "
|
|
"Refresh tools.x_auth_token / tools.x_ct0, or wait out the rate limit.")
|
|
|
|
|
|
async def _make_api():
|
|
"""cookie だけで twscrape API を用意 (login_all は呼ばない)。"""
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
from twscrape import API
|
|
_silence_logs() # twscrape import 後に呼ぶ (import 時に loguru sink が張られるため)
|
|
auth, ct0 = _creds()
|
|
tmpdir = tempfile.mkdtemp(prefix=_TMP_PREFIX)
|
|
db = os.path.join(tmpdir, "accounts.db")
|
|
api = API(db)
|
|
await api.pool.add_account(
|
|
"x_adapter", "-", "x_adapter@local", "-",
|
|
cookies=f"auth_token={auth}; ct0={ct0}",
|
|
)
|
|
return api, tmpdir, shutil
|
|
|
|
|
|
def _search_product(tab: str) -> str:
|
|
t = (tab or "Latest").strip().lower()
|
|
return {
|
|
"latest": "Latest", "top": "Top",
|
|
"photos": "Media", "videos": "Media", "media": "Media",
|
|
}.get(t, "Latest")
|
|
|
|
|
|
async def cmd_search(args) -> None:
|
|
api, tmpdir, shutil = await _make_api()
|
|
try:
|
|
rows = []
|
|
kv = {"product": _search_product(args.tab)}
|
|
async for t in api.search(args.query, limit=args.max, kv=kv):
|
|
rows.append(twscrape_tweet_to_dict(t))
|
|
if len(rows) >= args.max:
|
|
break
|
|
if not rows:
|
|
await _ensure_alive(api) # 空が認証切れ由来なら nonzero fail
|
|
emit(rows)
|
|
finally:
|
|
shutil.rmtree(tmpdir, ignore_errors=True)
|
|
|
|
|
|
async def cmd_user_posts(args) -> None:
|
|
api, tmpdir, shutil = await _make_api()
|
|
try:
|
|
login = args.username.lstrip("@").strip()
|
|
user = await api.user_by_login(login)
|
|
if not user:
|
|
await _ensure_alive(api) # 認証切れなら not found より先に auth エラー
|
|
fail(f"x-adapter: user '{login}' not found")
|
|
rows = []
|
|
async for t in api.user_tweets(user.id, limit=args.max):
|
|
rows.append(twscrape_tweet_to_dict(t))
|
|
if len(rows) >= args.max:
|
|
break
|
|
if not rows:
|
|
await _ensure_alive(api)
|
|
emit(rows)
|
|
finally:
|
|
shutil.rmtree(tmpdir, ignore_errors=True)
|
|
|
|
|
|
async def _tweet_detail_raw_with_article(api, tid: str):
|
|
"""TweetDetail を fieldToggles.withArticlePlainText 付きで直接叩く。
|
|
|
|
twscrape の tweet_details_raw は fieldToggles を渡せない (_gql_item が
|
|
variables/features のみ) ので、同じ op / variables / QueueClient を使い
|
|
パラメータだけ足す。X 長文記事の本文 (plain_text) は withArticlePlainText、
|
|
記事内埋め込みメディア (media_entities) は withArticleRichContentState が
|
|
無いとレスポンスに含まれない (後者は content_state も同乗するが emit しない)。
|
|
"""
|
|
from twscrape.api import GQL_URL, OP_TweetDetail
|
|
from twscrape.api import GQL_FEATURES # type: ignore[attr-defined]
|
|
from twscrape.queue_client import QueueClient
|
|
from twscrape.utils import encode_params
|
|
|
|
kv = {
|
|
"focalTweetId": str(tid),
|
|
"with_rux_injections": True,
|
|
"includePromotedContent": True,
|
|
"withCommunity": True,
|
|
"withQuickPromoteEligibilityTweetFields": True,
|
|
"withBirdwatchNotes": True,
|
|
"withVoice": True,
|
|
"withV2Timeline": True,
|
|
}
|
|
params = {
|
|
"variables": kv,
|
|
"features": GQL_FEATURES,
|
|
"fieldToggles": {"withArticlePlainText": True, "withArticleRichContentState": True},
|
|
}
|
|
queue = OP_TweetDetail.split("/")[-1]
|
|
async with QueueClient(api.pool, queue, False) as client:
|
|
return await client.get(f"{GQL_URL}/{OP_TweetDetail}", params=encode_params(params))
|
|
|
|
|
|
async def cmd_tweet(args) -> None:
|
|
tid = parse_tweet_ref(args.tweet)
|
|
if not tid:
|
|
fail(f"x-adapter: could not parse tweet id from '{args.tweet}'")
|
|
api, tmpdir, shutil = await _make_api()
|
|
try:
|
|
# 記事本文トグル付きの直接呼び出しを優先。twscrape 内部 API の変化などで
|
|
# 壊れたら従来の tweet_details に落として XPostDetail 自体は生かす。
|
|
t = None
|
|
article = None
|
|
article_media: list = []
|
|
try:
|
|
from twscrape.models import parse_tweet
|
|
rep = await _tweet_detail_raw_with_article(api, tid)
|
|
if rep is not None:
|
|
t = parse_tweet(rep, int(tid))
|
|
result = find_tweet_result_in_detail(rep.json(), tid)
|
|
article = extract_article(result, full_text=bool(args.full_text))
|
|
article_media = extract_article_media(result)
|
|
except SystemExit:
|
|
raise
|
|
except Exception as e:
|
|
print(f"x-adapter: article-aware fetch failed, falling back: {type(e).__name__}: {e}",
|
|
file=sys.stderr)
|
|
if not t:
|
|
t = await api.tweet_details(int(tid))
|
|
if not t:
|
|
await _ensure_alive(api) # 認証切れなら nonzero
|
|
fail(f"x-adapter: tweet {tid} not found or inaccessible")
|
|
row = twscrape_tweet_to_dict(t)
|
|
if article:
|
|
row["article"] = article
|
|
if article_media:
|
|
# 記事ポストの tweet-level media は空なので合流しても衝突しない。
|
|
# x.ts の downloadTweetMedia がここを見て自動 DL → localPath 付与する。
|
|
row["media"] = (row.get("media") or []) + article_media
|
|
emit([row])
|
|
finally:
|
|
shutil.rmtree(tmpdir, ignore_errors=True)
|
|
|
|
|
|
async def cmd_feed(args) -> None:
|
|
"""ホームタイムライン。twscrape の txid + twikit の endpoint で raw GraphQL。"""
|
|
import httpx
|
|
from urllib.parse import urlparse
|
|
import twscrape.xclid as xc
|
|
from twikit.client.gql import Endpoint
|
|
from twikit.constants import FEATURES
|
|
_silence_logs() # twscrape import 後に呼ぶ
|
|
|
|
auth, ct0 = _creds()
|
|
following = (args.tab or "").strip().lower() == "following"
|
|
url = Endpoint.HOME_LATEST_TIMELINE if following else Endpoint.HOME_TIMELINE
|
|
path = urlparse(url).path
|
|
query_id = path.rstrip("/").split("/")[-2]
|
|
# X web app が使う公開 web bearer (guest/web 共通の定数)
|
|
bearer = ("AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs"
|
|
"%3D1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA")
|
|
|
|
gen = await xc.XClIdGen.create()
|
|
tid = gen.calc("POST", path)
|
|
variables = {
|
|
"count": args.max,
|
|
"includePromotedContent": False,
|
|
"latestControlAvailable": True,
|
|
"requestContext": "launch",
|
|
"seenTweetIds": [],
|
|
}
|
|
body = {"variables": variables, "features": FEATURES, "queryId": query_id}
|
|
headers = {
|
|
"authorization": "Bearer " + bearer,
|
|
"x-csrf-token": ct0,
|
|
"x-twitter-auth-type": "OAuth2Session",
|
|
"x-twitter-active-user": "yes",
|
|
"content-type": "application/json",
|
|
"x-client-transaction-id": tid,
|
|
"cookie": f"auth_token={auth}; ct0={ct0}",
|
|
"user-agent": ("Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
|
|
"(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36"),
|
|
}
|
|
async with httpx.AsyncClient(timeout=30) as h:
|
|
r = await h.post(url, json=body, headers=headers)
|
|
if r.status_code != 200:
|
|
# レスポンス本文はそのまま吐かない (デバッグ用に X の error code/message のみ抽出)
|
|
fail(f"x-adapter: home timeline HTTP {r.status_code}{_x_error_suffix(r)}")
|
|
try:
|
|
payload = r.json()
|
|
except Exception:
|
|
fail("x-adapter: home timeline returned non-JSON")
|
|
# 200 でも GraphQL の errors が返ることがある (認証劣化など)。entries が無く errors が
|
|
# あるなら成功偽装せず fail。
|
|
rows = parse_home_timeline(payload, args.max)
|
|
if not rows and isinstance(payload, dict) and payload.get("errors"):
|
|
msgs = "; ".join(str(e.get("message", "")) for e in payload["errors"] if isinstance(e, dict))
|
|
fail(f"x-adapter: home timeline error: {msgs[:200] or 'unknown'}")
|
|
emit(rows)
|
|
|
|
|
|
def _x_error_suffix(resp) -> str:
|
|
"""X のエラーレスポンスから errors[].message だけ安全に取り出す (cookie/header は出さない)。"""
|
|
try:
|
|
data = resp.json()
|
|
errs = data.get("errors") if isinstance(data, dict) else None
|
|
if errs:
|
|
return ": " + "; ".join(str(e.get("message", "")) for e in errs if isinstance(e, dict))[:200]
|
|
except Exception:
|
|
pass
|
|
return ""
|
|
|
|
|
|
def _clamp_max(v) -> int:
|
|
try:
|
|
n = int(v)
|
|
except Exception:
|
|
n = 20
|
|
return max(1, min(n, 50))
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
p = argparse.ArgumentParser(prog="x-adapter", add_help=True)
|
|
p.add_argument("--version", action="store_true")
|
|
sub = p.add_subparsers(dest="command")
|
|
|
|
def common(sp):
|
|
sp.add_argument("--max", type=_clamp_max, default=20)
|
|
sp.add_argument("--yaml", action="store_true")
|
|
sp.add_argument("--full-text", action="store_true")
|
|
sp.add_argument("--compact", action="store_true")
|
|
|
|
sp = sub.add_parser("search"); sp.add_argument("query"); sp.add_argument("-t", "--tab", default="Latest"); common(sp)
|
|
sp = sub.add_parser("user-posts"); sp.add_argument("username"); common(sp)
|
|
sp = sub.add_parser("tweet"); sp.add_argument("tweet"); common(sp)
|
|
sp = sub.add_parser("feed"); sp.add_argument("-t", "--tab", default="for_you"); common(sp)
|
|
return p
|
|
|
|
|
|
def main(argv=None) -> None:
|
|
import asyncio
|
|
args = build_parser().parse_args(argv)
|
|
if args.version:
|
|
print(VERSION)
|
|
return
|
|
handlers = {
|
|
"search": cmd_search,
|
|
"user-posts": cmd_user_posts,
|
|
"tweet": cmd_tweet,
|
|
"feed": cmd_feed,
|
|
}
|
|
handler = handlers.get(args.command)
|
|
if not handler:
|
|
fail("x-adapter: no subcommand. Use search|user-posts|tweet|feed.")
|
|
_cleanup_stale_tmp() # 過去の SIGKILL タイムアウトで残った cookie 入り temp を掃除
|
|
try:
|
|
asyncio.run(handler(args))
|
|
except SystemExit:
|
|
raise
|
|
except Exception as e:
|
|
fail(f"x-adapter: {type(e).__name__}: {e}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|