j2kun · October 30, 2024 17:49 · Oct 30, 2024
diff --git a/publish_to_twitter.py b/publish_to_twitter.py
@@ -0,0 +1,109 @@
+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)
No results found