bsky

blusky/.direnv/bin/nix-direnv-reload

@@ -0,0 +1,19 @@
+#!/usr/bin/env bash
+set -e
+if [[ ! -d "/home/jawz/Development/Scripts/blusky" ]]; then
+  echo "Cannot find source directory; Did you move it?"
+  echo "(Looking for "/home/jawz/Development/Scripts/blusky")"
+  echo 'Cannot force reload with this script - use "direnv reload" manually and then try again'
+  exit 1
+fi
+
+# rebuild the cache forcefully
+_nix_direnv_force_reload=1 direnv exec "/home/jawz/Development/Scripts/blusky" true
+
+# Update the mtime for .envrc.
+# This will cause direnv to reload again - but without re-building.
+touch "/home/jawz/Development/Scripts/blusky/.envrc"
+
+# Also update the timestamp of whatever profile_rc we have.
+# This makes sure that we know we are up to date.
+touch -r "/home/jawz/Development/Scripts/blusky/.envrc" "/home/jawz/Development/Scripts/blusky/.direnv"/*.rc

blusky/.envrc

@@ -0,0 +1 @@
+use nix

blusky/feed-jawz.py

@@ -0,0 +1,89 @@
+#!/usr/bin/env python3
+import re
+import requests
+from typing import List, Dict, Optional
+
+config = {
+    "displayName": "JawZ Art",
+    "blocks": [
+        {
+            "id": "aaakxm4a5pbcs",
+            "type": "input",
+            "inputType": "did",
+            "did": "did:plc:3y42h6r3t4rfnxc57jtcw5lk",
+        },
+        {
+            "id": "aaakxm4a5ozyu",
+            "type": "regex",
+            "value": "#(?i)art",
+            "caseSensitive": False,
+            "invert": False,
+        },
+        {"id": "aaakxm4a5nze6", "type": "sort", "sortType": "created_at"},
+    ],
+}
+
+
+def get_api_data(endpoint: str, params: Optional[Dict] = None) -> Optional[Dict]:
+    if not endpoint:
+        return None
+    response = requests.get(endpoint, params=params)
+    if response.status_code != 200:
+        return None
+    return response.json()
+
+
+def filter_by_did(posts: List[Dict], did: str) -> List[Dict]:
+    if not posts or not did:
+        return []
+    return [post for post in posts if post.get("author", {}).get("did") == did]
+
+
+def regex_filter(posts: List[Dict], pattern: str) -> List[Dict]:
+    if not posts or not pattern:
+        return []
+    compiled_pattern = re.compile(pattern)
+    return [post for post in posts if compiled_pattern.search(post.get("content", ""))]
+
+
+def sort_posts(posts: List[Dict], key: str = "created_at") -> List[Dict]:
+    return sorted(posts, key=lambda x: x.get(key, ""), reverse=True)
+
+
+def generate_feed(endpoint: str, config: Dict) -> Optional[List[Dict]]:
+    data = get_api_data(endpoint)
+    print(data)
+    if not data:
+        return None
+
+    posts = data.get("posts", [])
+    did_block = next(
+        (block for block in config["blocks"] if block["type"] == "input"), {}
+    )
+    regex_block = next(
+        (block for block in config["blocks"] if block["type"] == "regex"), {}
+    )
+    sort_block = next(
+        (block for block in config["blocks"] if block["type"] == "sort"), {}
+    )
+
+    posts = filter_by_did(posts, did_block.get("did", "")) if did_block else posts
+    posts = regex_filter(posts, regex_block.get("value", "")) if regex_block else posts
+    posts = (
+        sort_posts(posts, key=sort_block.get("sortType", "created_at"))
+        if sort_block
+        else posts
+    )
+    return posts
+
+
+def main():
+    endpoint = "https://api.bsky.app/feeds"
+    feed = generate_feed(endpoint, config)
+    if not feed:
+        return
+    print(feed)
+
+
+if __name__ == "__main__":
+    main()

blusky/shell.nix

@@ -0,0 +1,19 @@
+{
+  pkgs ? import <nixpkgs> { },
+}:
+
+with pkgs;
+
+mkShell {
+  packages = [
+    (python3.withPackages (
+      ps: with ps; [
+        setuptools
+        requests
+      ]
+    ))
+  ];
+  buildInputs = [
+
+  ];
+}