Adafruit_Learning_System_Guides/Pi_Bluesky_Scroller/code.py

# SPDX-FileCopyrightText: 2024 Tim Cocks
#
# SPDX-License-Identifier: MIT
"""
Bluesky_RPi_TFT_Scroller code.py
Infinitely scroll Bluesky posts on a 320x240 pixel TFT
"""
import json
import os
import sys

import requests
import webview

FEEDLINK_RETROCOMPUTING = (
    "https://bsky.app/profile/did:plc:tbo4hkau3p2itkar2vsnb3gp/feed/aaabo5oe7bzok"
)

# Un-comment a single key inside of FEED_ARGS and set it's value to the feed, list or search
# that you want to scroll.
FETCH_ARGS = {
    # "feed_share_link": FEEDLINK_RETROCOMPUTING,
    # "feed_share_link": "https://bsky.app/profile/did:plc:463touruejpokvutnn5ikxb5/lists/3lbfdtahfzt2a", # pylint: disable=line-too-long
    # "search_args": {"q": "Adafruit", "sort": "latest"}
    "search_args": {"q": "#circuitpython", "sort": "latest"}
}


def at_feed_uri_from_share_link(share_link):
    """
    Converts a share link into an AT URI for that resource.

    :param share_link: The share link to convert.
    :return str: The AT URI pointing at the resource.
    """
    at_feed_uri = share_link.replace("https://bsky.app/profile/", "at://")
    if "/feed/" in share_link:
        at_feed_uri = at_feed_uri.replace("/feed/", "/app.bsky.feed.generator/")
    if "/lists/" in share_link:
        at_feed_uri = at_feed_uri.replace("/lists/", "/app.bsky.graph.list/")
    return at_feed_uri


def fetch_data(feed_share_link=None, search_args=None):
    """
    Fetch posts from Bluesky API and write them into the local cached
    data files. After posts are written locally iterates over them
    and downloads the relevant photos from them.

    Must pass either feed_share_link or search_args.

    :param feed_share_link: The link copied from Bluesky front end to share the feed or list.
    :param search_args: A dictionary containing at minimum a ``q`` key with string value of
        the hashtag or term to search for. See bsky API docs for other supported keys.
    :return: None
    """
    # pylint: disable=too-many-statements,too-many-branches
    if feed_share_link is None and search_args is None:
        # If both inputs are None, just use retrocomputing feed.
        feed_share_link = FEEDLINK_RETROCOMPUTING

    # if a share link input was provided
    if feed_share_link is not None:
        FEED_AT = at_feed_uri_from_share_link(feed_share_link)
        # print(FEED_AT)

        # if it's a feed
        if "/app.bsky.feed.generator/" in FEED_AT:
            URL = (f"https://public.api.bsky.app/xrpc/app.bsky.feed.getFeed?"
                   f"feed={FEED_AT}&limit=30")
            headers = {"Accept-Language": "en"}
            resp = requests.get(URL, headers=headers)

        # if it's a list
        elif "/app.bsky.graph.list/" in FEED_AT:
            URL = (f"https://public.api.bsky.app/xrpc/app.bsky.feed.getListFeed?"
                   f"list={FEED_AT}&limit=30")
            headers = {"Accept-Language": "en"}
            resp = requests.get(URL, headers=headers)

        # raise error if it's an unknown type
        else:
            raise ValueError(
                "Only 'app.bsky.feed.generator' and 'app.bsky.graph.list' URIs are supported."
            )

    # if a search input was provided
    if search_args is not None:
        URL = "https://public.api.bsky.app/xrpc/app.bsky.feed.searchPosts"
        headers = {"Accept-Language": "en"}
        resp = requests.get(URL, headers=headers, params=search_args)

    with open(".data/raw_data.json", "wb") as f:
        # write raw response to cache
        f.write(resp.content)

    # Process the post data into a smaller subset
    # containing just the bits we need for showing
    # on the TFT.
    resp_json = json.loads(resp.text)
    processed_posts = {"posts": []}
    fetched_posts = None
    if "feed" in resp_json.keys():
        fetched_posts = resp_json["feed"]
    elif "posts" in resp_json.keys():
        fetched_posts = resp_json["posts"]

    for post in fetched_posts:
        cur_post = {}
        if "post" in post.keys():
            post = post["post"]
        cur_post["author"] = post["author"]["handle"]
        cur_post["text"] = post["record"]["text"]

        # image handling
        if "embed" in post.keys():
            cid = post["cid"]
            if "images" in post["embed"].keys():
                cur_post["image_url"] = post["embed"]["images"][0]["thumb"]
            elif "thumbnail" in post["embed"].keys():
                cur_post["image_url"] = post["embed"]["thumbnail"]
            elif (
                "external" in post["embed"].keys()
                and "thumb" in post["embed"]["external"].keys()
            ):
                cur_post["image_url"] = post["embed"]["external"]["thumb"]

            # if we actually have an image to show
            if "image_url" in cur_post.keys():
                # check if we already downloaded this image
                if f"{cid}.jpg" not in os.listdir("static/imgs/"):
                    print(f"downloading: {cur_post['image_url']}")

                    # download image and write to file
                    img_resp = requests.get(cur_post["image_url"])
                    with open(f"static/imgs/{cid}.jpg", "wb") as f:
                        f.write(img_resp.content)

                cur_post["image_file"] = f"{cid}.jpg"
        processed_posts["posts"].append(cur_post)

    # save the processed data to a file
    with open(".data/processed_data.json", "w", encoding="utf-8") as f:
        f.write(json.dumps(processed_posts))


def read_cached_data():
    """
    Load the cached processed data file and return
    the data from within it.

    :return: The posts data loaded from JSON
    """
    with open(".data/processed_data.json", "r") as f:
        return json.load(f)


class Api:
    """
    API object for interaction between python code here
    and JS code running inside the page.
    """

    # pylint: disable=no-self-use
    def get_posts(self):
        """
        Fetch new posts data from Bluesky API, cache and return it.
        :return: Processed data containing everything necessary to show
            posts on the TFT.
        """
        fetch_data(**FETCH_ARGS)
        return read_cached_data()

    def check_quit(self):
        """
        Allows the python code to correctly handle KeyboardInterrupt
        more quickly.

        :return: None
        """
        # pylint: disable=unnecessary-pass
        pass

    def quit(self):
        window.destroy()
        sys.exit(0)


# create a webview and load the index.html page
window = webview.create_window(
    "bsky posts", "static/index.html", js_api=Api(),
    #width=320, height=240,
    width=640, height=480,
    x=0, y=0, #frameless=True, fullscreen=True

)
webview.start()
# webview.start(debug=True)  # use this one to enable chromium dev tools to see console.log() output from the page.