90 lines
2.4 KiB
Python
90 lines
2.4 KiB
Python
#!/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()
|