twitter 自动化方案

domain:x.com,
domain:twitter.com,
domain:twimg.com,
domain:t.co

import os, json, tweepy
from pathlib import Path
import time

# 开发期允许http回调
os.environ["OAUTHLIB_INSECURE_TRANSPORT"] = "1"

CLIENT_ID = ""
CLIENT_SECRET = ""
REDIRECT_URI = "http://localhost:8000/callback"
TOKEN_URL = "https://api.x.com/2/oauth2/token"
SCOPES = [
    "tweet.write","tweet.read","users.read", "offline.access",
    "dm.read", "dm.write" 
]
# 身份令牌保存路径
TOKEN_FILE = Path("token.json")

# ==================== 工具函数 ====================
# 构造 OAuth2 处理器
def _new_oauth() -> tweepy.OAuth2UserHandler:
    return tweepy.OAuth2UserHandler(
        client_id=CLIENT_ID,
        client_secret=CLIENT_SECRET,   # confidential client 需要
        redirect_uri=REDIRECT_URI,
        scope=SCOPES,
    )

# 保存 token
def _save_token(token: dict) -> None:
    TOKEN_FILE.write_text(json.dumps(token, ensure_ascii=False, indent=2), encoding="utf-8")

# 加载 token
def _load_token() -> dict | None:
    if TOKEN_FILE.exists():
        return json.loads(TOKEN_FILE.read_text(encoding="utf-8"))
    return None

# 确保 token 新鲜
def _ensure_fresh_token(saved: dict) -> dict:
    """仅在将过期/已过期时刷新；返回可用 token（已自动持久化旋转后的refresh_token）"""
    now = int(time.time())
    exp_at = int(saved.get("expires_at", 0))
    # 距离过期 < 60 秒 或已过期 才刷新
    if exp_at and now < exp_at - 60:
        return saved

    oauth = _new_oauth()
    refreshed = oauth.refresh_token(
        TOKEN_URL,
        refresh_token=saved.get("refresh_token"),
        client_id=CLIENT_ID,
        client_secret=CLIENT_SECRET,   # confidential client 刷新必带
    )
    # 旋转：服务端会下发全新的 refresh_token；保存它
    _save_token(refreshed)
    return refreshed

def _client_from_token(tok: dict) -> tweepy.Client:
    """用用户 access_token 构造 v2 Client。后续调用需传 user_auth=False"""
    return tweepy.Client(bearer_token=tok["access_token"], wait_on_rate_limit=True)

def _get_client_and_token() -> tuple[tweepy.Client, dict]:
    saved = _load_token()
    if not saved:
        raise RuntimeError("未找到 token.json，请先调用 login() 完成一次授权。")
    fresh = _ensure_fresh_token(saved)
    return _client_from_token(fresh), fresh

# ==================== 对外函数 ====================
# 登录授权
def login() -> None:
    """一步式本地授权：打印URL -> 粘贴回调完整地址（含 ?code=...&state=...） -> 保存 token"""
    oauth = _new_oauth()
    url = oauth.get_authorization_url()
    print("在浏览器打开授权：", url)
    cb = input("粘贴回调后的完整URL：").strip()
    tok = oauth.fetch_token(authorization_response=cb)
    _save_token(tok)


# 获取当前用户信息
def get_me() -> tweepy.Response:
    """
    复用已保存身份获取当前用户信息（自动刷新token，避免过期）
    返回 Tweepy Response；你可 .data 取 user 对象
    """
    client, _ = _get_client_and_token()
    resp = client.get_me(user_auth=False)  # 关键：用 OAuth2 用户令牌时要禁用 OAuth1 分支
    return resp.data.data if hasattr(resp.data, "data") else resp.data

# 发帖
def post_tweet(text: str) -> tweepy.Response:
    """发一条贴文；返回推文ID"""
    client, _ = _get_client_and_token()
    resp = client.create_tweet(text=text, user_auth=False)
    return resp.data["id"]


# 给指定用户ID发私信
def send_dm_to_user_id(user: str, text: str) -> str:
    """给指定用户发私信支持用户名(不带@)和id；返回 dm_event_id"""

    client, _ = _get_client_and_token()

    # 如果用户不是数字，则先获取用户id
    if not user.isdigit():
        u = client.get_user(username=user, user_auth=False)
        user_id = str(u.data.id)
    else:
        user_id = user

    try:
        resp = client.create_direct_message(participant_id=user_id, text=text, user_auth=False)
        data = getattr(resp, "data", resp)
        if isinstance(data, dict):
            ev = data.get("event") or data.get("dm_event") or {}
            return ev.get("id") or data.get("id", "OK-未返回id")  # 中文注释：返回事件ID
        return str(data)  # 中文注释：兜底
    except tweepy.TweepyException as e:
        code = getattr(getattr(e, "response", None), "status_code", None)
        if code == 403:
            # 平台拒绝：最常见就是对方不接受你的DM（未关注且未开启“任何人可私信”），或你被屏蔽/权限不匹配
            return f"403：对方不允许接收你的私信（未关注你且未开启“任何人可私信”），或权限限制。原始信息：{body}"  # 中文注释
        return f"错误码：{code}，错误信息：{e}"  # 中文注释：其他错误原样返回
    
# 列出最近的 DM 会话
def list_dm_conversations(
    mode: str = "all",
    max_results: int = 100
) -> list[dict]:
    """
    列出最近的 DM 会话，按“全部/已回/未回”筛选。
    规则：
      - “未回”：该会话最新一条消息的 sender 不是我（即对方最后说话）
      - “已回”：该会话最新一条消息是我发的
    返回每个会话的概要：dm_conversation_id、latest_text、latest_time、partner_ids、last_sender_is_me
    """
    client, _ = _get_client_and_token()

    me = client.get_me(user_auth=False).data
    my_id = str(me.id)

    resp = client.get_direct_message_events(
        dm_event_fields=[
            "id","event_type","text","created_at","dm_conversation_id","sender_id","participant_ids"
        ],
        expansions=["participant_ids","sender_id"],  # 便于拿 includes.users
        user_fields=["username","name"],
        event_types=["MessageCreate"],               # 只看消息创建
        max_results=max(1, min(100, max_results)),
        user_auth=False
    )

    # 组装用户字典（可显示用户名）
    id2user = {}
    inc = getattr(resp, "includes", None)
    if inc and hasattr(inc, "users") and inc.users:
        for u in inc.users:
            id2user[str(u.id)] = {"id": str(u.id), "username": getattr(u, "username", None), "name": getattr(u, "name", None)}

    # 聚合到会话维度：每个会话只保留最新一条
    conv_latest: dict[str, dict] = {}
    events = resp.data or []
    for ev in events:
        # Tweepy v2 DmEvent 对象/或 dict 兼容
        e = ev.data if hasattr(ev, "data") else ev
        conv_id = e.get("dm_conversation_id")
        created_at = e.get("created_at")
        sender_id = str(e.get("sender_id"))
        text = e.get("text","")

        old = conv_latest.get(conv_id)
        if not old or (created_at and created_at > old["latest_time"]):
            partner_ids = [pid for pid in (e.get("participant_ids") or []) if str(pid) != my_id]
            conv_latest[conv_id] = {
                "dm_conversation_id": conv_id,
                "latest_text": text,
                "latest_time": created_at,
                "last_sender_is_me": (sender_id == my_id),
                "partner_ids": [str(p) for p in partner_ids],
                "partner_profiles": [id2user.get(str(p), {"id": str(p)}) for p in partner_ids],
            }

    # 按模式筛选
    rows = list(conv_latest.values())
    if mode == "replied":
        rows = [r for r in rows if r["last_sender_is_me"]]
    elif mode == "unreplied":
        rows = [r for r in rows if not r["last_sender_is_me"]]

    # 最新在前
    rows.sort(key=lambda r: (r["latest_time"] or ""), reverse=True)
    return rows

# 在既有会话里回复一条私信
def reply_dm_by_conversation(dm_conversation_id: str, text: str) -> str:
    """在既有会话里回复一条私信；返回 dm_event_id"""
    client, _ = _get_client_and_token()
    resp = client.create_direct_message(dm_conversation_id=dm_conversation_id, text=text, user_auth=False)
    data = getattr(resp, "data", resp)
    if isinstance(data, dict):
        ev = data.get("event") or data.get("dm_event") or {}
        return ev.get("id") or data.get("id", "")
    return str(data)

# ==================== 示例入口 ====================
if __name__ == "__main__":
    # # 首次授权：先运行一次 login()，后续就可以注释掉
    # login()

    # # # 取用户信息
    # me = get_me()
    # print("当前授权用户：", me)

    # # 发帖
    # resp = post_tweet("Hello X")
    # print("发帖成功：", resp)

    # # 列出所有私聊
    # print(list_dm_conversations())

    # # 主动给人发送私聊
    print(send_dm_to_user_id("caominli", "Hello 海风"))