Source code for my personal quote bot project.
crashkeys.dev
1#!/usr/bin/env python3
2
3import time
4import random
5import os
6import logging
7import hashlib
8
9from glob import iglob
10from typing import Optional
11
12import atproto
13import atproto.exceptions
14import schedule
15import typer
16
17def auth_client() -> atproto.Client:
18 handle, password, = os.getenv('BLUESKY_USERNAME'), os.getenv('BLUESKY_PASSWORD')
19
20 client: atproto.Client = atproto.Client()
21
22 try:
23 _profile: atproto.models.AppBskyActorDefs.ProfileViewDetailed = client.login(handle, password)
24 except atproto.exceptions.AtProtocolError as e:
25 logging.error("Could not log into profile `%s`: %s", hashlib.sha256(handle.encode()).hexdigest(), e)
26 raise
27
28 logging.info("Logged into profile `%s` successfully.", hashlib.sha256(handle.encode()).hexdigest())
29 return client
30
31def sample_quote(client: Optional[atproto.Client], simulation_mode: bool = False) -> None:
32 global quotes, indices
33 path = quotes[indices.pop()]
34
35 with open(path, 'r') as f:
36 post_text: str = f.read().strip()
37
38 if simulation_mode:
39 print(post_text, '\n')
40 return
41
42 try:
43 client.send_post(post_text)
44 except atproto.exceptions.AtProtocolError as e:
45 logging.error("Error occurred when attempting to post `%s`: %s", path, e)
46 raise
47
48def main(simulation_mode: bool = True, log_level: str = "INFO") -> None:
49 global quotes, indices
50
51 logging.getLogger().setLevel(os.getenv("LOGGING") or log_level)
52
53 quotes = tuple(iglob('src/**/*.txt', recursive=True))
54 indices = []
55
56 logging.debug("Initialized `quotes` tuple with the following values: %s", quotes)
57
58 if simulation_mode:
59 schedule.every(10).seconds.do(sample_quote, None, simulation_mode=True)
60 else:
61 client: atproto.Client = auth_client()
62 schedule.every().hour.at(':00').do(sample_quote, client)
63 schedule.every().hour.at(':30').do(sample_quote, client)
64
65 while True:
66 if len(indices) == 0:
67 indices = random.sample(range(len(quotes)), k=len(quotes))
68
69 if logging.getLogger().isEnabledFor(logging.DEBUG):
70 shuffled_quotes = tuple(quotes[i] for i in indices)
71 logging.debug("Shuffled `quotes` tuple as follows: %s", shuffled_quotes)
72
73 schedule.run_pending()
74 time.sleep(1)
75
76if __name__ == '__main__':
77 random.seed(time.monotonic())
78 typer.run(main)