Automation for comment System using Bluesky

[pubmania]

Hi,

I have managed to get Bluesky comments on my blog and fine tuned it so the manual steps of the current process are reduced. I have explained the steps I took in detail on my blogpost . However, it still requires manual steps and I am trying to work out a way that allows me to create a post on bluesky each time a new blogpost is published.

Perhaps best explained in following diagrams:

Current Flow (already live):

What I want to achieve:

Now I have worked out the code to post to bluesky using theor API in python and it works quite well. Problem I have is like this:

In order to post the link on bluesky along with social card, when using API I have to use the following function:

from bs4 import BeautifulSoup

def fetch_embed_url_card(access_token: str, url: str) -> Dict:

    # the required fields for every embed card
    card = {
        "uri": url,
        "title": "",
        "description": "",
    }

    # fetch the HTML
    resp = requests.get(url)
    resp.raise_for_status()
    soup = BeautifulSoup(resp.text, "html.parser")

    # parse out the "og:title" and "og:description" HTML meta tags
    title_tag = soup.find("meta", property="og:title")
    if title_tag:
        card["title"] = title_tag["content"]
    description_tag = soup.find("meta", property="og:description")
    if description_tag:
        card["description"] = description_tag["content"]

    # if there is an "og:image" HTML meta tag, fetch and upload that image
    image_tag = soup.find("meta", property="og:image")
    if image_tag:
        img_url = image_tag["content"]
        # naively turn a "relative" URL (just a path) into a full URL, if needed
        if "://" not in img_url:
            img_url = url + img_url
        resp = requests.get(img_url)
        resp.raise_for_status()

        blob_resp = requests.post(
            "https://bsky.social/xrpc/com.atproto.repo.uploadBlob",
            headers={
                "Content-Type": "image/png",
                "Authorization": "Bearer " + access_token,
            },
            data=resp.content,
        )
        blob_resp.raise_for_status()
        card["thumb"] = blob_resp.json()["blob"]

    return {
        "$type": "app.bsky.embed.external",
        "external": card,
    }

Then I first get the access token using code below:

import requests
resp = requests.post(
    "https://bsky.social/xrpc/com.atproto.server.createSession",
    json={"identifier": BLUESKY_HANDLE, "password": BLUESKY_APP_PASSWORD},
)
resp.raise_for_status()
session = resp.json()
access_token = session["accessJwt"]
print(access_token)

Finally I call the function in first code block and it creates a bluesky post with social card:

embed_post = fetch_embed_url_card(access_token, "https://mgw.dumatics.com/2023/02/26/git-basics.html")
text = 'Example post with embed external resource (link card from website)'
post_with_link_card_from_website = at_client.send_post(text=text, embed=embed_post)

Now the key thing is Bluesky will need the social card to be uploaded as blob as can be seen in the function above and this is where I am struglling to come up with a strategy.

I thought perhaps it can be done through a github action that is triggered after the gh-pages deployment is completed successfully. I am not sure what I have been messing up but the action approach just wasn't even getting to a point where I could get handle on new post files. I am thinking maybe I can try to create a hook that runs post-build but am now entering a very unknown territory and was hoping more knowledgeable people here can give me some guidance.

As a side note, please know that I am not exactly a developer so if my question or subsequent responses sound naive then it is very likely because I am that. I just tinker with this stuff as a way of learning but it is not something I have ever formally been trained on or worked on.

[pubmania]

ok looks like I have managed to get this working.... I created the following python script and saved in the root of my test repo

# post_deploy.py
import re
import os
import yaml
import datetime
from atproto import Client, IdResolver, models
import requests

def get_yaml_frontmatter(path,access_token,at_client,image_directory,site_url):
    # Regex to match YAML front matter
    yaml_regex = re.compile(r'^(---\n.*?\n---\n)', re.DOTALL)

    # Check if the path is a directory or a file
    if os.path.isdir(path):
        # If it's a directory, process all .md files
        for filename in os.listdir(path):
            if filename.endswith('.md'):
                file_path = os.path.join(path, filename)
                process_file_yaml(file_path, yaml_regex,access_token,at_client,image_directory,site_url)
    elif os.path.isfile(path) and path.endswith('.md'):
        # If it's a single .md file, process it
        process_file_yaml(path, yaml_regex,access_token,at_client,image_directory,site_url)
    else:
        print("Provided path is neither a valid directory nor a .md file.")

def process_file_yaml(file_path, yaml_regex,access_token,at_client,image_directory,site_url):
    with open(file_path, 'r', encoding='utf-8') as file:
        content = file.read()
    description_value = ""
    url = ""
    title_value = ""
    
    # Find YAML front matter
    match = yaml_regex.search(content)
    if match:
        frontmatter = match.group(1)
        # Parse the existing YAML front matter
        frontmatter_content = frontmatter.split('---')[1].strip()
        frontmatter_dict = yaml.safe_load(frontmatter_content)
        for key, value in frontmatter_dict.items():
            if key == 'date':
                created_date = value['created']
            if key == 'slug':
                slug_value = value
            if key == 'title':
                title_value = value
            if key == 'description':
                description_value = value
        print(f"created_date: {created_date} and slug_value: {slug_value}")
        yyyy = created_date.year
        mm = f"{created_date.month:02}"
        dd = f"{created_date.day:02}"
        #print(f"url: https://mgw.dumatics.com/{yyyy}/{mm}/{dd}/{slug_value}.html")
        print(f"url: {site_url}/{yyyy}/{mm}/{dd}/{slug_value}.html")
        print(f"img_path: {image_directory}/{file_path.split('/')[-1].split('.')[0]}.png")
        url = f"{site_url}/{yyyy}/{mm}/{dd}/{slug_value}.html"
        image_path = f"{image_directory}/{file_path.split('/')[-1].split('.')[0]}.png"
        
        ####################################################################
        #### skip posting if created date is more than 5 days old###########
        ####################################################################
        created_date_str = f"{created_date}"
        # Convert the created_date string to a datetime object
        created_date = datetime.datetime.fromisoformat(created_date_str)
        # Get the current date
        current_date = datetime.datetime.now()
        # Calculate the difference in days
        difference = (current_date - created_date).days
        if difference <= 5:
            
            #####################################################################################
            ################### skip posting if url is already posted on bluesky#################
            #####################################################################################
            
            search_params = models.app.bsky.feed.search_posts.Params(
                    q= url,
                    author=at_client.me.did,
                    limit=1,
                    sort='oldest'
                )
            
            response = at_client.app.bsky.feed.search_posts(params=search_params)
            if response.posts:
                print("BSKY POST ALREADY EXISTS, NO ACTION NEEDED")
            else:
                # Open the image file in binary mode
                with open(image_path, 'rb') as img_file:
                    # Read the content of the image file
                    img_data = img_file.read()
                
                blob_resp = requests.post(
                    "https://bsky.social/xrpc/com.atproto.repo.uploadBlob",
                    headers={
                        "Content-Type": "image/png",
                        "Authorization": "Bearer " + access_token,
                    },
                    data=img_data,
                )
                blob_resp.raise_for_status()
                card = {
                "uri": url,
                "title": title_value,
                "description": description_value,
                "thumb": blob_resp.json()["blob"]
                }
                
                embed_post = {
                "$type": "app.bsky.embed.external",
                "external": card,
                }
                
                #text = 'Check out a new post on my blog.'
                text = 'Testing automated Bsky post creation'
                post_with_link_card_from_website = at_client.send_post(text=text, embed=embed_post)
                print(post_with_link_card_from_website.uri)
    else:
        print(f"No YAML front matter found in: {file_path}")

def main():
    BLUESKY_HANDLE = os.environ.get('BSKY_HANDLE')
    BLUESKY_APP_PASSWORD = os.environ.get('BSKY_APP_PWD')
    # Make sure the environment variables are set
    if not BLUESKY_HANDLE or not BLUESKY_APP_PASSWORD:
        raise ValueError("Environment variables BLUESKY_HANDLE and BLUESKY_APP_PASSWORD must be set.")
    else:
        at_client = Client()
        at_client.login(BLUESKY_HANDLE, BLUESKY_APP_PASSWORD)
        resp = requests.post(
            "https://bsky.social/xrpc/com.atproto.server.createSession",
            json={"identifier": BLUESKY_HANDLE, "password": BLUESKY_APP_PASSWORD},
        )
        resp.raise_for_status()
        session = resp.json()
        access_token = session["accessJwt"]
        path = 'docs/posts'
        image_directory = os.path.join(os.environ['GITHUB_WORKSPACE'], 'site','assets','images','social','posts')
        site_url = os.environ['SITE_URL']
        get_yaml_frontmatter(path,access_token, at_client,image_directory,site_url)

if __name__ == "__main__":
    main()

Then in requirement.txt I added PyYAML and atproto and added following to the github action right at the end:

      - name: Run Post-Deployment Script
        run: python post_deploy.py
        env:
          BSKY_HANDLE: ${{ secrets.BSKY_HANDLE }}
          BSKY_APP_PWD: ${{ secrets.BSKY_APP_PWD }}
          DID: ${{ secrets.DID }}
          GITHUB_WORKSPACE: ${{ github.workspace }}
          SITE_URL: ${{ vars.SITE_URL }}       

Am sure there will be a better and cleaner way to do all this and I will appreciate any feedback.

The repo I used for testing is on this link.

Found the environment variable. :) There is some hard coding in here. I guess site_url can be obtained from environment variable but for image_directory I checked where the action saved the site directory after build and used that. Is there an environment variable that can be used for that too. I am thinking what if github actions start saving in different location and so on....