bsky feeds about music
music-atmosphere-feed.plyr.fm/
zzstoatzz.io
bsky
feed
zig
1#!/usr/bin/env python3
2"""check feed status and lag"""
3
4import argparse
5import json
6import urllib.request
7from datetime import datetime, timezone
8
9FEED_URL = "https://zig-bsky-feed.fly.dev"
10BSKY_API = "https://public.api.bsky.app"
11
12def fetch_json(url):
13 req = urllib.request.Request(url, headers={"Accept": "application/json"})
14 with urllib.request.urlopen(req, timeout=10) as resp:
15 return json.loads(resp.read())
16
17def get_feed_posts(feed_name="music-atmosphere", limit=5):
18 feed_uri = f"at://did:plc:vs3hnzq2daqbszxlysywzy54/app.bsky.feed.generator/{feed_name}"
19 url = f"{FEED_URL}/xrpc/app.bsky.feed.getFeedSkeleton?feed={feed_uri}&limit={limit}"
20 data = fetch_json(url)
21 return [p["post"] for p in data.get("feed", [])]
22
23def get_post_details(uri):
24 url = f"{BSKY_API}/xrpc/app.bsky.feed.getPosts?uris={uri}"
25 data = fetch_json(url)
26 if data.get("posts"):
27 return data["posts"][0]
28 return None
29
30def parse_time(ts):
31 return datetime.fromisoformat(ts.replace("Z", "+00:00"))
32
33def format_ago(td):
34 secs = int(td.total_seconds())
35 if secs < 60:
36 return f"{secs}s"
37 mins = secs // 60
38 if mins < 60:
39 return f"{mins}m {secs % 60}s"
40 hours = mins // 60
41 return f"{hours}h {mins % 60}m"
42
43def cmd_status(args):
44 now = datetime.now(timezone.utc)
45 print(f"current time: {now.strftime('%H:%M:%S')} UTC")
46 print()
47
48 # get stats
49 try:
50 stats = fetch_json(f"{FEED_URL}/stats")
51 print(f"jetstream status: {stats.get('status', 'unknown')}")
52 print(f"event lag: {stats.get('lag_seconds', '?')}s")
53 print(f"messages: {stats.get('messages_total', '?'):,}")
54 print(f"matches: {stats.get('matches_total', '?')}")
55 print(f"posts in db: {stats.get('posts_in_db', '?'):,}")
56 print()
57 except Exception as e:
58 print(f"stats error: {e}")
59 print()
60
61 # get latest post
62 print("latest posts:")
63 posts = get_feed_posts(args.feed, limit=3)
64 for uri in posts:
65 post = get_post_details(uri)
66 if post:
67 created = parse_time(post["record"]["createdAt"])
68 lag = now - created
69 rkey = uri.split("/")[-1]
70 handle = post.get("author", {}).get("handle", "?")
71 print(f" {rkey} by @{handle}")
72 print(f" created: {created.strftime('%H:%M:%S')} ({format_ago(lag)} ago)")
73 else:
74 print(f" {uri} - could not fetch")
75
76 if posts:
77 post = get_post_details(posts[0])
78 if post:
79 created = parse_time(post["record"]["createdAt"])
80 lag = now - created
81 print()
82 print(f"actual feed lag: {format_ago(lag)}")
83 if lag.total_seconds() > 60:
84 print("STATUS: BEHIND")
85 else:
86 print("STATUS: LIVE")
87
88def cmd_recent(args):
89 """show recent posts with their timestamps"""
90 posts = get_feed_posts(args.feed, limit=args.limit)
91 now = datetime.now(timezone.utc)
92
93 for uri in posts:
94 post = get_post_details(uri)
95 if post:
96 created = parse_time(post["record"]["createdAt"])
97 lag = now - created
98 rkey = uri.split("/")[-1]
99 handle = post.get("author", {}).get("handle", "?")
100 text = post.get("record", {}).get("text", "")[:50].replace("\n", " ")
101 print(f"{rkey} | {format_ago(lag):>8} ago | @{handle}: {text}...")
102
103def main():
104 parser = argparse.ArgumentParser(description="feed status checker")
105 parser.add_argument("--feed", default="music-atmosphere", help="feed name")
106
107 subs = parser.add_subparsers(dest="cmd")
108
109 status_p = subs.add_parser("status", help="show feed status")
110
111 recent_p = subs.add_parser("recent", help="show recent posts")
112 recent_p.add_argument("-n", "--limit", type=int, default=10)
113
114 args = parser.parse_args()
115
116 if args.cmd == "recent":
117 cmd_recent(args)
118 else:
119 cmd_status(args)
120
121if __name__ == "__main__":
122 main()