Twitter Monitor

[
  "sama",
  "elonmusk",
  "https://x.com/OpenAI"
]


TWITTER_API_KEY=replace_with_twitterapi_io_key


# Feishu Output

Only use Feishu when the user asks to write or sync results to Feishu/Lark Bitable.

Preferred workflow:

1. Run `twitter_monitor.py` first and produce JSON or CSV.
2. Check whether `feishu-cli` is installed.
3. If it is missing, ask whether to install and configure it.
4. Use the user's existing Feishu CLI auth/session and target app/table details.

Recommended Bitable fields:

| Field | Type |
| --- | --- |
| 推文内容 | Text |
| 日期 | Date/time |
| 推文链接 | URL |
| 推文ID | Text |
| 作者 | Text |
| 作者ID | Text |
| 阅读量 | Number |
| 点赞数 | Number |
| 转发数 | Number |
| 评论数 | Number |
| 收藏数 | Number |
| 是否回复 | Checkbox |
| 抓取时间 | Date/time |

Deduplicate by `推文ID` before appending records when the target table already contains prior fetches.

Do not ask users for Feishu app secrets unless `feishu-cli` cannot support their target operation and the user explicitly chooses API-based setup.


# Scheduling

Offer scheduling only after a one-shot fetch works or when the user explicitly asks for recurring monitoring.

Options:

- **OpenClaw**: Prefer this when the user already works in OpenClaw or asks for agent-run recurring jobs. Create a repeatable command that exports `TWITTER_API_KEY`, runs `scripts/twitter_monitor.py`, then performs any requested Feishu sync.
- **Codex automation**: Use when the current environment supports reminders/automations and the user wants the agent thread to wake up and run the job.
- **cron**: Use for simple Linux/macOS shell schedules.
- **launchd**: Use for persistent macOS user jobs when cron is not appropriate.

Before creating any schedule, confirm:

1. Account ids.
2. API key source, preferably an environment variable or secret manager.
3. Page count and time window.
4. Output destination.
5. Feishu target, if applicable.

Keep recurring commands idempotent by deduplicating on `推文ID` before appending to a durable destination.


# twitterapi.io Setup

Use `twitterapi.io` advanced search for one-shot X/Twitter account fetches.

Required environment variable:

```bash
export TWITTER_API_KEY="..."
```

Default command:

```bash
python scripts/twitter_monitor.py --accounts sama --pages 1 --format json
```

Use `--pages` to cap cursor pagination. Keep the default at `1` unless the user asks for a deeper pull or confirms an API-cost tradeoff.

Useful variants:

```bash
python scripts/twitter_monitor.py --accounts sama,elonmusk --pages 2 --format csv
python scripts/twitter_monitor.py --accounts-file assets/accounts.example.json --since-hours 24
python scripts/twitter_monitor.py --accounts sama --since-date 2026-04-27 --until-date 2026-04-27
```

The script accepts handles with or without `@`, comma-separated lists, and `x.com` or `twitter.com` profile URLs.


#!/usr/bin/env python3
"""Fetch recent X/Twitter posts through twitterapi.io and write JSON or CSV."""

from __future__ import annotations

import argparse
import csv
import json
import os
import re
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timedelta, timezone
from email.utils import parsedate_to_datetime
from pathlib import Path
from typing import Any


API_BASE = "https://api.twitterapi.io/twitter/tweet/advanced_search"
UTC = timezone.utc

FIELDNAMES = [
    "推文内容",
    "日期",
    "推文链接",
    "推文ID",
    "作者",
    "作者ID",
    "阅读量",
    "点赞数",
    "转发数",
    "评论数",
    "收藏数",
    "是否回复",
    "抓取时间",
]


def normalize_account(value: str) -> str:
    value = value.strip()
    if not value:
        raise ValueError("empty account value")
    match = re.search(r"(?:x\.com|twitter\.com)/@?([A-Za-z0-9_]{1,15})", value)
    if match:
        return match.group(1)
    value = value.removeprefix("@").strip().strip("/")
    if not re.fullmatch(r"[A-Za-z0-9_]{1,15}", value):
        raise ValueError(f"invalid X account id: {value!r}")
    return value


def parse_accounts(args: arg