publish_to_twitter.py

import os
import time

from scripts import utils as utils
from scripts import syndication as syndication
import fire
import tweepy


# A simple text file with two urls per line
DATABASE_FILE = "scripts/published_twitter.txt"
TWITTER_URL_TEMPLATE = "https://x.com/jeremyjkun/status/{id}"


def twitter_post_publisher(post: str, twitter_client=None, **kwargs):
    if not twitter_client:
        raise ValueError("twitter_client must be provided")

    response = twitter_client.create_tweet(text=post)
    return TWITTER_URL_TEMPLATE.format(id=response.data["id"])


def twitter_thread_adjuster(posts, blog_post_permalink=None, **kwargs):
    if not blog_post_permalink:
        raise ValueError("blog_post_permalink must be provided")

    # Twitter has a 280-character per post limit
    # This is a hacky method to handle this.
    limited_posts = []
    for i, post in enumerate(posts):
        max_len = 280

        if i == 0:
            prefix = "\n\nArchived at: "
            backref = f"{prefix}{blog_post_permalink}"
            post += backref
            # URL takes 23 chars regardless of length
            max_len -= len(prefix) + 23

        limited_posts.extend(utils.split_post(post, max_char_len=max_len))

    return limited_posts


def twitter_thread_publisher(posts, twitter_client=None, **kwargs):
    if not twitter_client:
        raise ValueError("twitter_client must be provided")

    last_post_id = None
    first_post_uri = None
    for i, post in enumerate(posts):
        if i == 0:
            # create the root post
            response = twitter_client.create_tweet(text=post)
            last_post_id = response.data["id"]
            first_post_uri = TWITTER_URL_TEMPLATE.format(id=last_post_id)
        else:
            assert last_post_id is not None
            response = twitter_client.create_tweet(
                text=post,
                in_reply_to_tweet_id=last_post_id,
            )
            last_post_id = response.data["id"]

        uri = TWITTER_URL_TEMPLATE.format(id=last_post_id)
        print(
            f"Successfully posted post {i} of the thread: " f"{last_post_id} -> {uri}"
        )
        time.sleep(2)

    return first_post_uri


def publish_to_twitter(since_days=1, dry_run=False):
    """Idempotently publish shortform and regular posts to twitter."""
    # File generated by scripts/login_with_twitter.py or else set in
    # environment for headless usage in GH actions.
    if dry_run:
        twitter_client = None
    else:
        access_token = os.environ["TWITTER_API_ACCESS_TOKEN"]
        access_token_secret = os.environ["TWITTER_API_ACCESS_TOKEN_SECRET"]
        bearer_token = os.environ["TWITTER_API_BEARER_TOKEN"]
        consumer_key = os.environ["TWITTER_API_CONSUMER_KEY"]
        consumer_secret = os.environ["TWITTER_API_CONSUMER_KEY_SECRET"]

        twitter_client = tweepy.Client(
            bearer_token=bearer_token,
            consumer_key=consumer_key,
            consumer_secret=consumer_secret,
            access_token=access_token,
            access_token_secret=access_token_secret,
        )

    syndication.syndicate_to_service(
        "twitter",
        database_filepath=DATABASE_FILE,
        thread_publisher=twitter_thread_publisher,
        thread_adjuster=twitter_thread_adjuster,
        post_publisher=twitter_post_publisher,
        since_days=since_days,
        dry_run=dry_run,
        twitter_client=twitter_client,
        convert_math=False,
    )


if __name__ == "__main__":
    fire.Fire(publish_to_twitter)