semantic bufo search
find-bufo.com
bufo
add cooldown to prevent reposting same bufo within 2 hours
checks our recent posts before posting to avoid repetition.
stateless - fetches from bluesky API, survives restarts.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
+68
-3
+1
bot/.env.example
+1
bot/.env.example
+1
bot/README.md
+1
bot/README.md
···
22
22
| `BSKY_APP_PASSWORD` | required | app password from bsky settings |
23
23
| `MIN_PHRASE_WORDS` | `4` | minimum words in phrase to match |
24
24
| `POSTING_ENABLED` | `false` | must be `true` to actually post |
25
+
| `COOLDOWN_MINUTES` | `120` | don't repost same bufo within this time |
25
26
| `JETSTREAM_ENDPOINT` | `jetstream2.us-east.bsky.network` | jetstream server |
26
27
27
28
## local dev
+3
bot/src/bufo_bot/config.py
+3
bot/src/bufo_bot/config.py
+63
-3
bot/src/bufo_bot/main.py
+63
-3
bot/src/bufo_bot/main.py
···
1
1
import logging
2
+
from datetime import datetime, timezone
2
3
3
4
import httpx
4
5
from atproto import Client, models
···
12
13
format="%(asctime)s | %(levelname)s | %(message)s",
13
14
)
14
15
logger = logging.getLogger(__name__)
16
+
17
+
18
+
def get_recent_bufo_names(client: Client, minutes: int) -> set[str]:
19
+
"""fetch bufo names we've posted recently (from alt text)"""
20
+
try:
21
+
# fetch our recent posts
22
+
feed = client.app.bsky.feed.get_author_feed(
23
+
{"actor": settings.bsky_handle, "limit": 50}
24
+
)
25
+
cutoff = datetime.now(timezone.utc).timestamp() - (minutes * 60)
26
+
recent = set()
27
+
28
+
for item in feed.feed:
29
+
post = item.post
30
+
# parse created_at timestamp
31
+
created = post.record.created_at
32
+
if isinstance(created, str):
33
+
# parse ISO format
34
+
try:
35
+
ts = datetime.fromisoformat(created.replace("Z", "+00:00"))
36
+
if ts.timestamp() < cutoff:
37
+
break # posts are sorted newest first
38
+
except ValueError:
39
+
continue
40
+
41
+
# extract alt text from embedded image
42
+
embed = post.record.embed
43
+
if embed and hasattr(embed, "media"):
44
+
media = embed.media
45
+
if hasattr(media, "images"):
46
+
for img in media.images:
47
+
if img.alt:
48
+
# convert alt text back to filename format
49
+
recent.add(img.alt.replace(" ", "-") + ".png")
50
+
recent.add(img.alt.replace(" ", "-") + ".gif")
51
+
52
+
logger.info(f"found {len(recent)} bufos posted in last {minutes} minutes")
53
+
return recent
54
+
except Exception as e:
55
+
logger.warning(f"failed to fetch recent posts: {e}")
56
+
return set()
15
57
16
58
17
59
def load_bufos() -> list[Bufo]:
···
110
152
# connect to jetstream
111
153
jetstream = JetstreamClient(settings.jetstream_endpoint)
112
154
155
+
# track recently posted bufos (refreshed periodically)
156
+
recent_bufos: set[str] = set()
157
+
last_refresh = 0.0
158
+
113
159
# process posts
114
160
for post in jetstream.stream_posts():
115
161
match = matcher.find_match(post.text)
116
162
if match:
117
163
logger.info(f"match: '{match.phrase}' -> {match.name}")
118
-
if settings.posting_enabled:
119
-
quote_post_with_bufo(client, post, match)
120
-
else:
164
+
165
+
if not settings.posting_enabled:
121
166
logger.info("posting disabled, skipping")
167
+
continue
168
+
169
+
# refresh cooldown cache every 5 minutes
170
+
now = datetime.now(timezone.utc).timestamp()
171
+
if now - last_refresh > 300:
172
+
recent_bufos = get_recent_bufo_names(client, settings.cooldown_minutes)
173
+
last_refresh = now
174
+
175
+
# check cooldown
176
+
if match.name in recent_bufos:
177
+
logger.info(f"cooldown: {match.name} posted recently, skipping")
178
+
continue
179
+
180
+
quote_post_with_bufo(client, post, match)
181
+
recent_bufos.add(match.name) # add to local cache immediately
122
182
123
183
124
184
def main():