Blog » AirQualityBot - 1.díl - Repost podle hashtagů

Kdysi v dávných dobách, na jedné síti s modrým ptáčkem, jsem měl napsáno několik automatů a robotů, které mi posílaly různá upozornění. Tito roboti využívali veřejně dostupné API.
Pak přišla změna, ptáka udusilo černé X, API bylo vypnuto a celé to tam šlo do kopru.
Naštěstí je tady BlueSky s jejich AT protokolem a HTTP API, pomocí kterého je tvorba takovýchto robotů velmi jednoduchá.

První z mých robotů na BlueSky je AirQualityBot (@airqbot.klosko.net) a je zaměřen na informace týkající se kvality ovzduší.

Robot je z praktických důvodů rozdělen na dvě samostatné části - skripty.

AirQualityRepostBot
AirQualityBot, který postuje reporty a grafy ze dvou mých monitorů kvality ovzduší

Oba jsou napsané v Pythonu, oba využívají moji vlastní knihovnu atprotoLib
Pro vytvoření bota však můžete využít jakýkoliv programovací jazyk, který umí poslat HTTP request (GET, POST), přijmout odpověď (json) a zpracovat ji.
Taky existuje mnoho SDK (Typescript, Python), využít lze i standardní linuxové utilitky (wget, cURL)

AirQualityRepostBot
První část robota v pravidelných intervalech projde profily účtů, které profil AirQualityBota sledují (Opt-IN), a pokud jejich posty obsahují definovaná klíčová slova - hashtagy - tak tento post repostne.
Skript je spouštěný přes CRONa, interval je t.č. 4 hodiny.

Celý algoritmus lze ve zkratce popsat takto:

Načíst seznam folowerů (app.bsky.graph.getFollowers, veřejný endpoint)


def get_followers(token, bsky_user, tcp_delay, public_ep=False):
    url = f"{'https://public.api.bsky.app' if public_ep else 'https://bsky.social'}/xrpc/app.bsky.graph.getFollowers?actor={bsky_user}"
    response = get_request(url, token, tcp_delay, "getFollowers", public_ep)
    return response.get("followers", []) if response else []

Pro jednotlivé followery načíst seznam jejich postů (app.bsky.feed.getAuthorFeed, veřejný endpoint)


def get_posts(token, actor_did, limit=50, tcp_delay=5000, public_ep=False):
    url = f"{'https://public.api.bsky.app' if public_ep else 'https://bsky.social'}/xrpc/app.bsky.feed.getAuthorFeed?actor={actor_did}&limit={limit}&filter=posts_with_replies"
    response = get_request(url, token, tcp_delay, "getAuthorFeed", public_ep)
    return response.get("feed", []) if response else []

Pokud post obsahuje některý z hashtagů v seznamu, t.č.:


hashtag_arr = [
    "AirQuality",
    "AirQualityBot",
    "Feinstaub",
    "AirPollution",
    "KvalitaOvzdusi",
    "KvalitaVzduchu",
    "@airqbot.klosko.net"
]

tak repost (com.atproto.repo.createRecord, neveřejný endpoint, takže před tím získat AUTH token)


def repost_post(token, bsky_user, post_uri, cid, tcp_delay):
    url = "https://bsky.social/xrpc/com.atproto.repo.createRecord"
    data = {
        "repo": bsky_user,
        "collection": "app.bsky.feed.repost",
        "record": {
            "subject": {"uri": post_uri, "cid": cid},
            "createdAt": datetime.datetime.utcnow().isoformat() + "Z"
        }
    }
    return post_request(url, data, tcp_delay, "createRecord", token)

Využití neveřejných endpointů vyžaduje autorizaci, Bearer token, který je pak součástí HTTP hlavičky požadavku.
Ten lze získat přihlášením pomocí com.atproto.server.createSession.
Pro přihlášení budete potřebovat "login" a "heslo aplikace". To si vytvoříte na https://bsky.app/settings/app-passwords
Doporučuju pro každého bota vygenerovat samostatné heslo.

Všechny příklady, které jsem našel, pracují přihlášením vždy při každém požadavku na některý z neveřejných endpointů.
To však při kombinaci velkého množství postů a omezení API (Rate-Limits) může vést k dočasnému zablokování bota na dobu definovanou v hlavičce RateLimit-Reset, která je součástí odpovědi. T.č. je to jeden den od posledního požadavku (viz. RateLimit-Policy)


RateLimit-Limit => 10
RateLimit-Remaining => 9
RateLimit-Reset => 2025-05-DDTHH:MM:SS
RateLimit-Policy => 10;w=86400

Takže jsem autorizaci vyřešil trošku jinak - uložením tokenu lokálně a kombinací com.atproto.server.getSession, com.atproto.server.refreshSession.
Samozřejmě, že com.atproto.server.createSession tam je taky, jako poslední instance při expiraci tokenů.


def login(bsky_user, bsky_pass, tcp_delay, token_file=None):
    if token_file and os.path.exists(token_file):
        with open(token_file, "r") as file:
            tokens = json.load(file)
        if get_session(tokens["accessJwt"], tcp_delay) == bsky_user:
            print("getSession OK")
            return tokens["accessJwt"]
        else:
            print("refreshToken")
            new_token = refresh_session(tokens["refreshJwt"], tcp_delay, token_file)
            if new_token:
                return new_token
    print("Go to login")
    url = "https://bsky.social/xrpc/com.atproto.server.createSession"
    data = json.dumps({
        "identifier": bsky_user,
        "password": bsky_pass
    })
    response = requests.post(url, data=data, headers={"Content-Type": "application/json"})
    if response.status_code == 200:
        token_data = response.json()
        if token_file:
            with open(token_file, "w") as file:
                json.dump(token_data, file)
        return token_data.get("accessJwt")
    return None

Celý kód ze kterého jsou patrné detaily, včetně knihoven, je u mne na GitHubu AirQualityBot

AirQualitytBot
Popis druhé části část robota, AirQualityBot - Reporty kvality ovzduší je téma na samostatný článek.
Takže příště.

Jméno
E-Mail

Zadejte šestou čísici z čísla

Pokud očekáváte odpověď na Vámi vložený komentář, zadejte Váš e-mail. Vložením komentáře souhlasíte s pravidly.

AirQualityBot - 1.díl - Repost podle hashtagů

První část BlueSky robota AirQualityBot v pravidelných intervalech projde profily účtů, které tento profil sledují (Opt-IN), a pokud jejich posty obsahují definovaná klíčová slova - hashtagy - tak tento post repostne.

První část BlueSky robota AirQualityBot v pravidelných intervalech projde profily účtů, které tento profil sledují (Opt-IN), a pokud jejich posty obsahují definovaná klíčová slova - hashtagy - tak tento post repostne.