Vytvořil jsem bota, který třikrát denně posílá některý z mnou vybraných obrázků na sociální síť Bluesky. Podělím se s vámi, jak jsem toho dosáhl.
Bluesky je nová sociální síť Jacka Dorseyho, bývalého majitele Twitteru, kterou zveřejnil poté, co starou síť prodal Elonu Muskovi. Síť byla nejdříve přístupná pouze pro lidi s pozvánkou, už v té době ale měla dostupné API, pomocí které mohli uživatelé se sociální sítí interagovat.
Očekávaná masová migrace uživatelů ze sítě Twitter/X se nekonala ani po otevření bran veřejnosti, a tak na modrém nebi stále panuje komorní atmosféra. Rozhodl jsem se přispět svou maličkostí a napsat bota, který bude na síť X posílat obrázky s memy.
Abych se o systém nemusel starat, rozhodl jsem se pro serverless řešení s využitím Azure Functions. Kód, mimochodem psaný v Pythonu, je tak uložený a běží na cloudových serverech Microsoftu.
V tomto článku vám ukážu:
- Jak napsat Azure Function kód, který se bude pravidelně spouštět na serverech Microsoft Azure
- Jak získat data souboru uloženého na cloudovém uložišti Azure Storage
- Jak tento obrázek nahrát jako příspěvek na svůj profil pomocí Bluesky API
Funkce s časovým spínačem
Když se přihlásíme do svého Microsoft Azure účtu v editoru Visual Studio Code, můžeme v nabídce Resources vybrat volbu vytvořit novou Function App. Visual Studio nás už navede různými nastaveními, jako je jméno funkce, verze Pythonu (já volím 3.11), a následně nám vygeneruje základní potřebné soubory.
Když to máme hotové, můžeme začít napsáním funkce, která se automatický spustí v určitý čas. V souboru function_app.py jsem nalezneme naši vstupní metodu, která v mém případě vypadá takto:
import azure.functions as func
import logging
app = func.FunctionApp()
def main():
pass
@app.timer_trigger(schedule="0 0 6,13,20 * * *", arg_name="myTimer", run_on_startup=False,
use_monitor=False)
def test_timer_trigger(myTimer: func.TimerRequest):
if myTimer.past_due:
logging.info('The timer is past due!')
blob_name, success, exception_msg = main()
if not success:
raise Exception(f"Could not post {blob_name}, exception: {exception_msg}")
logging.info('Python timer trigger function executed.')Nejzásadnější je zde dekorátor @app.timer_trigger a jeho parametry. Parametr schedule formátem připomíná čásový záznam CRON, ale přesněji jde o zápis ve formátu NCRONTAB, který umožňuje specifikovat i sekundy spuštění.
Zápis 0 0 6,13,20 * * * znamená, že skript se má spustit v nultou vteřinu a nultou minutu každé šesté, třinácté a dvacáté hodiny. Jednoduše řečeno, v 6:00, 13:00 a 20:00. Ve skutečnosti je ale funkce spuštěna o dvě hodiny později, protože se řídí časem UTC. Tento čas samozřejmě nerespektuje letní čas, proto jsem musel parametr nedávno o hodinu posouvat.
Azure Functions běžící na Windows nebo na Linux Premium plánu umožňují nastavit proměnnou prostředí WEBSITE_TIME_ZONE na hodnotu Central Europe Standard Time, a zbavit se tak trablí se změnou času. Moje funkce ale zatím běží na standardním Linux plánu, takže je mi tato možnost zapovězena.
Další parametr dekorátoru časové spouště, kterému je třeba věnovat pozornost, je run_on_startup. Pokud bychom parametr nastavili jako true, naše funkce bude zavolána i při spuštění nebo restartu naší Azure function.
Abychom Azure dali vědět, že jsme to my, kdo bude funkci na Azure spouštět a kdo bude například přistupovat k na cloudu uloženým obrázkům, v souboru local.settings.json specifikujeme AZURE_CLIENT_ID, AZURE_TENANT_ID, a AZURE_CLIENT_SECRET. Hodnoty můžeme najít v administračním portálu Azure, pod naší Function App, konkrétně v sekci Settings -> Authentication. Pokud se nám nepodaří najít AZURE_TENANT_ID, budeme se muset přihlásit přes portál Microsoft Entra, jak popisuje tento návod. Tento soubor je pouze pro potřeby lokálního testování, nebudeme ho nahrávat na server, a protože obsahuje tajné klíče, raději ho přidáme i do souboru .gitignore.
{
"IsEncrypted": false,
"Values": {
"AzureWebJobsStorage": "DefaultEndpointsProtocol=https;AccountName=mystorageaccount;AccountKey=*****;EndpointSuffix=core.windows.net",
"FUNCTIONS_WORKER_RUNTIME": "python",
"AzureWebJobsFeatureFlags": "EnableWorkerIndexing",
"AZURE_CLIENT_ID": "***",
"AZURE_TENANT_ID": "***",
"AZURE_CLIENT_SECRET": "***",
"BLUESKY_HANDLE": "***",
"BLUESKY_APP_PASSWORD": "****"
}
}Pokud už máme založený účet uložiště (Storage account) na Azure. Můžeme vyplnit i hodnoty AccountName (název účtu uložiště) a AccountKey (tajný klíč), abychom naši funkci zajistili přístup k souborům v uložišti.
Můžeme také přidat ověřovací údaje pro přístup k síti Bluesky. Konkrétně svůj handle (ve formátu username.doména.tld) a heslo aplikace. Nejedná se o heslo k vašemu účtu! Tento klíč si můžeme vytvořit v nastavení účtu Bluesky v sekci App Passwords.
Naši funkci můžeme nahrát a spustit v administračním portálu na stránkách Microsoft Azure. Pokud jsme vše nastavili správně, pod záložkou Metrics se objeví údaje o spuštění aplikace. Azure pro nás také vytvoří modul Monitor, který zaznamenává logy z každého spuštění funkce. To může být kriticky důležité, pokud si nejsme jisti, proč naše aplikace na Azure serverech selhává. Pokud tato data nepotřebujeme, můžeme modul Monitor zrušit, a ušetřit za běh funkce.
Poznámka bokem: pokud bychom nechtěli, aby se funkce spouštěla pravidelně v určitý čas, ale chtěli bychom ji spouštět manuálně, můžeme místo časového spouštěče vytvořit spouštěč HTTP. Při aktivaci funkce získáme webovou adresu, přes kterou můžeme program spustit. Příklad níže je bez ověření spouštějícího uživatele (funkci by mohl opakovaně spouštět kdokoliv s odkazem), a proto se hodí pouze pro testování. Když jsme s naší aplikací spokojení, můžeme spouštěč změnit na časovou spoušť.
@app.route(route="test_http_trigger", auth_level=func.AuthLevel.ANONYMOUS)
def test_http_trigger(req: func.HttpRequest) -> func.HttpResponse:
logging.info('Python HTTP trigger function processed a request.')
blob_name, success, exception_msg = main()
if success:
return func.HttpResponse(f"Successfully posted {blob_name}")
return func.HttpResponse(f"Could not post {blob_name} ({exception_msg})")Získávání nahrávaných souborů
Jak už bylo zmíněno, na Azure je potřeba vytvořit Account Storage (účet uložiště) a následně Container, kam uložíme obrázky, které chceme později nahrát na Bluesky. Toto můžeme provést ve webové administraci našeho účtu Azure. Zde je python funkce, kterou umístíme do stejného function_app.py jako náš spouštěč, abychom mohli získat soubor z našeho uložiště.
from azure.identity import DefaultAzureCredential
from azure.storage.blob import ContainerClient
def pop_file_from_azure_storage():
blob_name=""
blob_type=""
blob_text=""
account_url = "https://mystorageaccount.blob.core.windows.net"
default_credential = DefaultAzureCredential()
containername="mycontainer"
with ContainerClient(account_url, credential=default_credential, container_name=containername) as blob_container_client:
blob_list = blob_container_client.list_blobs()
for blob in blob_list:
blob_name = blob.name
with blob_container_client.get_blob_client(blob=blob_name) as file:
blob_type = f"image/{blob_name.split('.')[-1]}"
downloader = file.download_blob(max_concurrency=1)
blob_text = downloader.content_as_bytes()
file.delete_blob()
break
return blob_name, blob_type, blob_text
K autentifikaci naší aplikace, tak aby mohla přistupovat k našemu uložišti blobů, jsme použili DefaultAzureCredential(). To vyzkouší různé metody ověření, jako první však použije třídu EnvironmentCredential, která nás přihlásí pomoci hodnot AZURE_CLIENT_ID, AZURE_TENANT_ID a AZURE_CLIENT_SECRET. Nejspíš budeme muset změnit i nastavení uložiště blobů tak, aby nás Azure účet měl přístup ke čtení i zápisu, čehož můžeme dosáhnout tím, že se nastavíme jako owner uložiště. Po nastavení v sekci Access Control (u účtu uložiště) bychom měli vidět něco podobného:
V pythonovém kódu výše jsme dále vytvořili instanci třídy ContainerClient, které dáme své přístupové údaje a specifikujeme název kontejneru (jeden účet uložiště může obsahovat více kontejnerů).
Dále jsme získali seznam blobů v uložišti pomocí metody list_blobs(), ale nám stačí zastavit se u prvního a vytvořit instanci klienta pro jeho stažení pomocí metody get_blob_client(blob=jméno blobu). Na tomto klientovi jsme zavolali metodu download_blob a stáhli tak obsah souboru do proměnné. Nakonec jsme blob smazali, abychom příště nečetli ten samý soubor.
Komunikace s Bluesky API
Nyní se už vrhneme na komunikaci s Bluesky. Tu v pythonu zatím provedeme velmi naivně s použitím knihovny requests. Nejprve musíme získat přihlašovací session. Všechen kód je potřeba vkládat do stejného souboru jako naši časovou spoušť (function_app.py):
import os
import requests
def login_to_bluesky():
BLUESKY_HANDLE = os.environ["BLUESKY_HANDLE"]
BLUESKY_APP_PASSWORD = os.environ["BLUESKY_APP_PASSWORD"]
logging.info(f"Logging in as {BLUESKY_HANDLE}...")
res = requests.post(
"https://bsky.social/xrpc/com.atproto.server.createSession",
json={"identifier": BLUESKY_HANDLE, "password": BLUESKY_APP_PASSWORD},
)
res.raise_for_status()
session = res.json()
return session
Přihlašovací údaje si zde bereme z proměnných prostředí. Na lokálním prostředí je můžeme získat po uložení do souboru local.settings.json, na serveru je musíme nastavit v nastavení naší Function App pod menu Configuration -> Application Settings.
Jako další krok nahrajeme data obrázku na server Bluesky a získáme identifikátor, který na nahraný obrázek ukazuje. Pokud budeme posílat pouze textový příspěvek, tento krok můžeme přeskočit.
def upload_image_to_bsky(session, blob_type, blob_data):
# this size limit is specified in the app.bsky.embed.images lexicon
if len(blob_data) > 1000000:
raise Exception(
f"image file size too large. 1000000 bytes maximum, got: {len(blob_data)}"
)
logging.info(f"Uploading file of type {blob_type} to bsky")
res = requests.post(
"https://bsky.social/xrpc/com.atproto.repo.uploadBlob",
headers={
"Content-Type": blob_type,
"Authorization": "Bearer " + session["accessJwt"],
},
data=blob_data,
)
res.raise_for_status()
uploaded_file_identifier = res.json()["blob"]
return uploaded_file_identifier
Teď když už máme přihlášení k Bluesky, i odkaz na soubor, který jsme na něj nahráli, můžeme poslat příspěvek na sociální síť:
from datetime import datetime, timezone
def post_image_to_bsky(session, text_content, bsky_file_identifier):
now = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
logging.info(f"Posting uploaded image to bsky...")
# Required fields that each post must include
post = {"$type": "app.bsky.feed.post",
"text": text_content,
"createdAt": now,
"embed": {
"$type": "app.bsky.embed.images",
"images": [{
"alt":"",
"image": bsky_file_identifier,
}],
}}
res = requests.post(
"https://bsky.social/xrpc/com.atproto.repo.createRecord",
headers={"Authorization": "Bearer " + session["accessJwt"]},
json={
"repo": session["did"],
"collection": "app.bsky.feed.post",
"record": post,
},
)
res.raise_for_status()
A nakonec doplníme hlavní metodu, která vše sváže dohromady, a kterou voláme z metody časové spouště:
def main():
blob_name = ""
exception_msg = ""
success = False
try:
blob_name, blob_type, blob_content = pop_file_from_azure_storage()
bluesky_session = login_to_bluesky()
bsky_uploaded_file = upload_image_to_bsky(bluesky_session, blob_type, blob_content)
# no text to accompany the image, just send empty string
post_image_to_bsky(session=bluesky_session, text_content="", bsky_file_identifier=bsky_uploaded_file)
success = True
except Exception as e:
exception_msg = str(e)
finally:
return blob_name, success, exception_msg
Podrobnější informace o tom, jak Bluesky API používat, najdete v oficiální dokumentaci.
Nyní stačí znovu nahrát funkci na server Microsoft Azure a ujistit se, že funkce běží, případně ji spustit. Pokud jste si účet Microsoft Azure teprve založili, první měsíc můžete čerpat kredit zdarma. Od následujícího měsíce za poskytovanou službu budete muset platit. V mém případě, kdy jsem pro úsporu deaktivoval modul monitoringu, běh aplikace stojí asi 2 koruny měsíčně.
Závěr
Doufám, že návod byl dostatečně srozumitelný. Pokud s Microsoft Azure pracujete poprvé, jistě bude potřeba postupovat pomalu, a v administračním portálu se řádně zorientovat. Využijte kreditu zdarma, protože vám po prvním měsíci stejně vyprší, a nebojte se experimentovat. V řešení problémů vám bude velkým pomocníkem také oficiální dokumentace od Microsoftu.
Pokud byste si chtěli nechat napsat řešení na míru, neváhejte se mi ozvat — například na můj e-mail.
