Bsky

.github/workflows/test.yml

@@ -21,3 +21,7 @@ jobs:
 
       - name: Check format/lint
         run: ruff check
+
+      - name: Behave testing
+        working-directory: ./bluto
+        run: behave

.gitignore

@@ -8,4 +8,5 @@ flaskr.egg-info/
 .env
 venv
 *.swp
-.ruff_cache
+.ruff_cache
+session.txt

autoapp.py

@@ -1,4 +1,5 @@
 """Create an application instance."""
+
 from bluto.app import create_app
 
 app = create_app()

bluto/app.py

@@ -14,6 +14,7 @@
 # Use cdn if in production
 STATIC_URL_PATH = "/static"
 
+
 def create_app(config_object="bluto.settings"):
     """Create application factory, as explained here: http://flask.pocoo.org/docs/patterns/appfactories/.
 
@@ -34,11 +35,13 @@ def register_extensions(app):
     cache.init_app(app)
     debug_toolbar.init_app(app)
 
+
 def register_security_headers(app):
     """Register a bunch of sec."""
     if app.config["ENV"] == "production":
         Talisman(app, force_https=False)
 
+
 def register_blueprints(app):
     """Register Flask blueprints."""
     app.register_blueprint(public.views.blueprint)
@@ -63,4 +66,5 @@ def configure_logger(app):
     if not app.logger.handlers:
         app.logger.addHandler(handler)
 
+
 app = Flask(__name__, static_url_path=STATIC_URL_PATH)

bluto/extensions.py

@@ -1,5 +1,6 @@
 """Extensions module.
 Each extension is initialized in the app factory located in app.py."""
+
 from flask_caching import Cache
 from flask_debugtoolbar import DebugToolbarExtension
 

bluto/features/steps/steps.py

@@ -0,0 +1,28 @@
+import behave
+import markov as mk
+
+from pathlib import Path
+
+
+# Given
+@given("a list of tweets")
+def step_impl(context):
+    with Path.open("features/test_tweets.txt") as f:
+        content = f.readlines()
+        tweets = [x.strip() for x in content]
+    context.tweet_list = tweets
+
+
+# When
+@when("we generate a {number:d} of new {length:d} character tweets")
+def step_impl(context, number, length):
+    data = mk.make_markov_model(context.tweet_list)
+    context.new_tweet_list = [data.make_short_sentence(length) for i in range(number)]
+    assert not (None in context.new_tweet_list)
+
+
+# Then
+@then("the {new_number:d} and {new_length:d} of tweets is correct")
+def step_impl(context, new_number, new_length):
+    assert len(context.new_tweet_list) == new_number
+    assert len(max(context.new_tweet_list, key=len)) <= new_length

bluto/markov.py

@@ -2,78 +2,88 @@
 
 import os
 import re
+from pathlib import Path
 
 import markovify
-import twitter
+from atproto import Client
+from atproto import IdResolver
+from atproto import SessionEvent
 
-TWEET_LIMIT = 200
 
+def get_all_posts(username):
+    """Returns list of text entries for <username>'s last 100 posts'"""
+    # We don't need to authenticate this request
+    client = Client()
 
-def generate(text_model, size, bound):
-    """Makes 140 character tweets"""
-    return [text_model.make_short_sentence(size) for i in range(bound)]
+    user_id = IdResolver().handle.resolve(username)
 
+    posts = list(client.app.bsky.feed.post.list(user_id, limit=100).records.values())
 
-def get_all_tweets(username):
-    """Spawns api and gets last 2000 tweets"""
-    api = new_api()
-    tweets = get_tweets(api, username)
+    return [post.text for post in posts]
 
-    for _ in range(20):
-        tweets += get_tweets(api, username, since=tweets[len(tweets) - 1].id)
 
-    return tweets
+def get_avatar_url(username):
+    """Get the URL of <username>'s avatar"""
+    client = get_client()
 
+    user_id = IdResolver().handle.resolve(username)
 
-def get_profile_url(api, username):
-    """Get a big version of the profile image"""
-    user = (api.GetUser(screen_name=username),)
+    profile = client.app.bsky.actor.get_profile({"actor": user_id})
 
-    return user[0].profile_image_url.replace("normal", "400x400")
+    return profile.avatar
 
 
 def remove_twitlonger(tweet_list):
     """Removes all tweets that have a twitlonger link in them"""
     return [re.sub(r" \S*…[^']*", "", tweet) for tweet in tweet_list]
 
 
-def make_tweets(username, num_tweets):
-    """Produce an array of generated tweets"""
-    api = new_api()
-    data = remove_twitlonger([tweet.text for tweet in get_all_tweets(username)])
+def make_posts(username, num_posts):
+    """Produce an array of generated posts"""
+    data = remove_twitlonger(get_all_posts(username))
     model = make_markov_model(data)
 
     return {
         "username": username,
-        "profile_url": get_profile_url(api, username),
-        "tweets": generate(model, 140, num_tweets),
-        "long": generate(model, 240, 2),
+        "profile_url": get_avatar_url(username),
+        "tweets": [model.make_short_sentence(140) for i in range(num_posts)],
+        "long": [model.make_short_sentence(240) for i in range(2)],
     }
 
 
-# Utility
-def new_api():
-    """Wrapper around spawning twitter api"""
-    return twitter.Api(
-        consumer_key=os.environ["TWITTER_API_KEY"],
-        consumer_secret=os.environ["TWITTER_API_SECRET"],
-        access_token_key=os.environ["TWITTER_ACCESS_TOKEN"],
-        access_token_secret=os.environ["TWITTER_ACCESS_SECRET"],
-    )
-
-
-def get_tweets(api, username, since=None):
-    """Wrapper around api request"""
-    return api.GetUserTimeline(
-        screen_name=username,
-        count=TWEET_LIMIT,
-        include_rts=False,
-        trim_user=False,
-        exclude_replies=True,
-        max_id=since,
-    )
-
-
-def make_markov_model(tweets):
-    """Wrapper around making Markov Chain"""
-    return markovify.Text(" ".join(tweets))
+# Useful for Behave testing
+def make_markov_model(data):
+    """Wrapper around Markovify call"""
+    return markovify.Text(" ".join(data))
+
+
+# Client handling
+def get_session():
+    try:
+        with Path.open("session.txt") as f:
+            return f.read()
+    except FileNotFoundError:
+        return None
+
+
+def save_session(session_string):
+    with Path.open("session.txt", "w") as f:
+        f.write(session_string)
+
+
+def on_session_change(event, session):
+    if event in (SessionEvent.CREATE, SessionEvent.REFRESH):
+        save_session(session.export())
+
+
+def get_client():
+    client = Client()
+    client.on_session_change(on_session_change)
+
+    session_string = get_session()
+    if session_string:
+        client.login(session_string=session_string)
+    else:
+        client.login(os.getenv("BLUESKY_USERNAME"), os.getenv("BLUESKY_PASSWORD"))
+
+    return client

bluto/public/__init__.py

@@ -1,2 +1,3 @@
 """The public module, including the homepage and user auth."""
+
 from . import views  # noqa: F401

bluto/public/views.py

@@ -1,4 +1,5 @@
 """Public section, including homepage and signup."""
+
 from flask import Blueprint
 from flask import jsonify
 from flask import render_template
@@ -8,30 +9,35 @@
 
 blueprint = Blueprint("public", __name__, static_folder="../../static")
 
+
 @blueprint.route("/")
 def index():
     """Main page"""
     return render_template("public/landing.html")
 
+
 @blueprint.route("/tweets", methods=["GET"])
 def get_tweets():
     """Makes tweets for requested user and return rendered template"""
     twitter_handle = request.args["twitter_handle"]
-    tweets = mkv.make_tweets(twitter_handle, 30)
+    tweets = mkv.make_posts(twitter_handle, 30)
 
     return render_template(
         "public/results.html",
         username=twitter_handle,
         tweets=tweets["tweets"],
         long_tweets=tweets["long"],
-        profile_url=tweets["profile_url"])
+        profile_url=tweets["profile_url"],
+    )
+
 
 @blueprint.route("/api/<twitter_handle>", methods=["GET"])
 def get_api_tweets(twitter_handle):
     """Makes tweets for requested user and return as json"""
-    tweets = mkv.make_tweets(twitter_handle, 30)
+    tweets = mkv.make_posts(twitter_handle, 30)
     return jsonify(tweets)
 
+
 @blueprint.route("/api/ping", methods=["GET"])
 def ping():
     """Simple health check"""

bluto/settings.py

@@ -5,6 +5,7 @@
 For local development, use a .env file to set
 environment variables.
 """
+
 from environs import Env
 
 env = Env()
@@ -16,4 +17,4 @@
 SEND_FILE_MAX_AGE_DEFAULT = env.int("SEND_FILE_MAX_AGE_DEFAULT", default=1000000)
 DEBUG_TB_ENABLED = DEBUG
 DEBUG_TB_INTERCEPT_REDIRECTS = False
-CACHE_TYPE = ("flask_caching.backends.SimpleCache")
+CACHE_TYPE = "flask_caching.backends.SimpleCache"

bluto/templates/401.html

@@ -0,0 +1,14 @@
+{% extends "layout.html" %}
+
+{% block page_title %}Unauthorized{% endblock %}
+{% block header %}Unauthorized{% endblock %}
+
+{% block content %}
+<div class="jumbotron">
+    <div class="text-center">
+        <h1>404</h1>
+        <p>Sorry, this request is unauthorized.</p>
+        <p>Want to <a href="{{ url_for('public.index') }}">go home</a> instead?</p>
+    </div>
+</div>
+{% endblock %}

bluto/templates/500.html

@@ -0,0 +1,14 @@
+{% extends "layout.html" %}
+
+{% block page_title %}Internal Service Error{% endblock %}
+{% block header %}Internal Service Error{% endblock %}
+
+{% block content %}
+<div class="jumbotron">
+    <div class="text-center">
+        <h1>500</h1>
+        <p>Sorry, there has been an error and the request could not be completed.</p>
+        <p>Want to <a href="{{ url_for('public.index') }}">go home</a> instead?</p>
+    </div>
+</div>
+{% endblock %}

bluto/templates/public/results.html

@@ -1,5 +1,5 @@
 {% extends "layout.html" %}
-{% block body %}
+{% block content %}
 <!-- <div class="content hide-on-load"> -->
     <div class="landing hide-on-load">
         <div class="column">

pyproject.toml

@@ -10,7 +10,7 @@ python_files = [
 target-version = "py312"
 # Exclude a variety of commonly ignored directories.
 extend-exclude = [
-    "features/*"
+    "bluto/features/*"
 ]
 
 [tool.ruff.lint]

requirements/prod.txt

@@ -20,4 +20,4 @@ environs==14.1.1
 
 # Markov
 markovify==0.7.1
-python-twitter==3.4.1
+atproto==0.0.59