Mass Delete Tweets — Bulk API Workflow + Free Tool Options

import os, requests

HEADERS = {"X-API-Key": os.environ["TWITTERAPI_IO_KEY"]}
BASE = "https://api.twitterapi.io"

def list_my_tweet_ids(handle: str, before_date: str = None) -> list[str]:
    """Get tweet IDs to delete via advanced_search (cheap read)."""
    ids = []
    cursor = None
    while True:
        query = f"from:{handle}"
        if before_date:
            query += f" until:{before_date}"
        params = {"query": query}
        if cursor:
            params["cursor"] = cursor
        r = requests.get(
            f"{BASE}/twitter/tweet/advanced_search",
            headers=HEADERS, params=params, timeout=15,
        )
        r.raise_for_status()
        resp = r.json()
        for t in resp.get("tweets", []):
            ids.append(t["id"])
        cursor = resp.get("next_cursor")
        if not cursor: break
    return ids

old_tweets = list_my_tweet_ids("your_handle", before_date="2024-01-01")
print(f"found {len(old_tweets)} tweets to delete")
# Cost: ~$0.00015 × len(old_tweets) — read is cheap; delete is the bill

# pip install tweepy
import tweepy, time, json, random
from datetime import datetime, timezone

client = tweepy.Client(
    consumer_key="YOUR_KEY",
    consumer_secret="YOUR_SECRET",
    access_token="USER_TOKEN",
    access_token_secret="USER_TOKEN_SECRET",
)

def bulk_delete(tweet_ids: list[str], dry_run: bool = True):
    """Delete tweets with audit log + rate-limit safety."""
    log_path = f"delete_log_{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}.jsonl"
    success, fail = 0, 0
    with open(log_path, "a") as log:
        for tid in tweet_ids:
            entry = {"tweet_id": tid, "at": datetime.now(timezone.utc).isoformat()}
            if dry_run:
                entry["action"] = "dry_run"
                log.write(json.dumps(entry) + "\n")
                continue
            try:
                client.delete_tweet(id=tid)
                entry["action"] = "deleted"
                success += 1
            except tweepy.TooManyRequests:
                entry["action"] = "rate_limited"
                log.write(json.dumps(entry) + "\n")
                time.sleep(60 + random.uniform(0, 5))
                continue
            except Exception as e:
                entry["action"] = f"failed: {e}"
                fail += 1
            log.write(json.dumps(entry) + "\n")
            time.sleep(1.0 + random.uniform(0.1, 0.5))  # safe pacing
    return {"success": success, "fail": fail, "log": log_path}

# Always run dry_run=True first to verify the target set
# result = bulk_delete(old_tweets, dry_run=False)

import os, json, time, random, requests
from datetime import datetime, timezone

BASE = "https://api.twitterapi.io"
HEADERS = {"X-API-Key": os.environ["TWITTERAPI_IO_KEY"]}
PROXY = os.environ["X_PROXY"]  # a proxy you supply: http://username:password@ip:port

def login() -> str:
    """Log the account in once ($0.10) and return login_cookies. Store and reuse it."""
    body = {
        "user_name": os.environ["X_USERNAME"],
        "email": os.environ["X_EMAIL"],
        "password": os.environ["X_PASSWORD"],
        "proxy": PROXY,
    }
    if os.environ.get("X_TOTP_SECRET"):
        body["totp_secret"] = os.environ["X_TOTP_SECRET"]
    r = requests.post(f"{BASE}/twitter/user_login_v2",
                      headers=HEADERS, json=body, timeout=120)
    r.raise_for_status()
    data = r.json()
    if data.get("status") != "success":
        raise RuntimeError(f"login failed: {data.get('message')}")
    return data["login_cookies"]

def bulk_delete_v2(tweet_ids: list[str], login_cookies: str,
                   dry_run: bool = True, max_consecutive_errors: int = 5) -> dict:
    """Delete tweets one call at a time, with a pause and a log line per tweet."""
    stamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
    log_path = f"delete_v2_log_{stamp}.jsonl"
    ok = failed = streak = 0
    with open(log_path, "a", encoding="utf-8") as log:
        for tweet_id in tweet_ids:
            entry = {"tweet_id": tweet_id, "action": "delete",
                     "at": datetime.now(timezone.utc).isoformat()}
            if dry_run:
                entry["result"] = "dry_run"
                log.write(json.dumps(entry) + "\n")
                continue
            try:
                r = requests.post(
                    f"{BASE}/twitter/delete_tweet_v2",
                    headers=HEADERS,
                    json={"login_cookies": login_cookies,
                          "tweet_id": tweet_id,
                          "proxy": PROXY},
                    timeout=60,
                )
                data = r.json()
                if r.ok and data.get("status") == "success":
                    entry["result"] = "deleted"
                    ok += 1
                    streak = 0
                else:
                    entry["result"] = f"error: http {r.status_code} {data.get('message')}"
                    failed += 1
                    streak += 1
            except (requests.RequestException, ValueError) as exc:
                entry["result"] = f"error: {exc}"
                failed += 1
                streak += 1
            log.write(json.dumps(entry) + "\n")
            log.flush()
            if streak >= max_consecutive_errors:
                print("too many errors in a row, stopping; read the log first")
                break
            time.sleep(1.0 + random.uniform(0.0, 1.0))  # 1-2 s between calls
    return {"deleted": ok, "failed": failed, "log": log_path}

# tweet_ids.json: the ID list from Step 1, saved with json.dump(old_tweets, open("tweet_ids.json", "w"))
tweet_ids = json.load(open("tweet_ids.json"))
print(f"{len(tweet_ids)} deletes would cost about ${len(tweet_ids) * 0.0006:.2f} plus one $0.10 login")

print(bulk_delete_v2(tweet_ids, login_cookies="", dry_run=True))  # writes the log only
# When the dry-run log looks right:
# cookies = os.environ.get("X_LOGIN_COOKIES") or login()
# print(bulk_delete_v2(tweet_ids, cookies, dry_run=False))

# Stage 1 — export to JSONL BEFORE running bulk_delete()
import os, json, requests

HEADERS = {"X-API-Key": os.environ["TWITTERAPI_IO_KEY"]}
BASE = "https://api.twitterapi.io"

def export_all_tweets(handle: str, out_path: str = "tweets_archive.jsonl") -> int:
    """Full-account export before mass-delete. Cost ~$0.00015 per tweet."""
    cursor = None
    count = 0
    with open(out_path, "w") as f:
        while True:
            params = {"query": f"from:{handle}"}
            if cursor:
                params["cursor"] = cursor
            r = requests.get(
                f"{BASE}/twitter/tweet/advanced_search",
                headers=HEADERS, params=params, timeout=15,
            )
            r.raise_for_status()
            resp = r.json()
            for t in resp.get("tweets", []):
                f.write(json.dumps(t) + "\n")
                count += 1
            cursor = resp.get("next_cursor")
            if not cursor:
                break
    return count

count = export_all_tweets("your_handle")
print(f"exported {count} tweets to tweets_archive.jsonl")
print(f"estimated cost: ${count * 0.00015:.4f} (read-only)")
# NOW verify tweets_archive.jsonl locally before running bulk_delete() from Step 2