Twitter (X) Templates — Bulk Post via API

# pip install tweepy jinja2 pandas
import tweepy, time, random, json
from jinja2 import Template
from datetime import datetime, timezone
import pandas as pd

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

TEMPLATE = Template("📝 New post: {{ title }} — {{ url }} #{{ tag }}")

def bulk_post(rows_csv: str, dry_run: bool = True):
    df = pd.read_csv(rows_csv)
    log_path = f"post_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 i, row in df.iterrows():
            text = TEMPLATE.render(
                title=row["title"],
                url=row["url"],
                tag=row["tag"],
            )
            entry = {
                "row_index": int(i),
                "rendered": text,
                "at": datetime.now(timezone.utc).isoformat(),
            }
            if len(text) > 280:
                entry["action"] = "skip: over 280 chars"
                log.write(json.dumps(entry) + "\n")
                continue
            if dry_run:
                entry["action"] = "dry_run"
            else:
                try:
                    resp = client.create_tweet(text=text)
                    entry["action"] = "posted"
                    entry["tweet_id"] = resp.data["id"]
                    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.5 + random.uniform(0.1, 0.5))
    return {"success": success, "fail": fail, "log": log_path}

# Always dry_run first to inspect rendered output
# result = bulk_post("new_posts.csv", dry_run=False)

# pip install jinja2 pandas requests
import os, json, time, random, requests
from datetime import datetime, timezone
from jinja2 import Template
import pandas as pd

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 create_post(login_cookies: str, text: str) -> dict:
    """One post through the logged-in session. Returns the parsed response."""
    r = requests.post(
        f"{BASE}/twitter/create_tweet_v2",
        headers=HEADERS,
        json={"login_cookies": login_cookies, "tweet_text": text, "proxy": PROXY},
        timeout=60,
    )
    r.raise_for_status()
    return r.json()  # carries "status": "success" or "error"

TEMPLATE = Template("New post: {{ title }} #{{ tag }}")

def bulk_post_v2(rows_csv: str, login_cookies: str, dry_run: bool = True) -> dict:
    df = pd.read_csv(rows_csv)
    log_path = f"post_v2_log_{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}.jsonl"
    success = fail = 0
    with open(log_path, "a", encoding="utf-8") as log:
        for i, row in df.iterrows():
            text = TEMPLATE.render(title=row["title"], tag=row["tag"])
            entry = {"row_index": int(i), "rendered": text,
                     "at": datetime.now(timezone.utc).isoformat()}
            if len(text) > 280:
                entry["action"] = "skip: over 280 chars"
            elif dry_run:
                entry["action"] = "dry_run"
            else:
                try:
                    result = create_post(login_cookies, text)
                    entry["action"] = result.get("status")
                    entry["message"] = result.get("message")
                    if result.get("status") == "success":
                        success += 1
                    else:
                        fail += 1
                except requests.RequestException as exc:
                    entry["action"] = f"failed: {exc}"
                    fail += 1
                time.sleep(1.5 + random.uniform(0.1, 0.5))
            log.write(json.dumps(entry, ensure_ascii=False) + "\n")
    return {"success": success, "fail": fail, "log": log_path}

# Always dry-run first to inspect the rendered output
print(bulk_post_v2("new_posts.csv", login_cookies="", dry_run=True))
# cookies = os.environ.get("X_LOGIN_COOKIES") or login()
# print(bulk_post_v2("new_posts.csv", cookies, dry_run=False))