Štítky: - 1x denně pravidelný report o stavu ovzduší - minimální, maximální a průměrné hodnoty PM10 a PM2.5 za posledních 24h vč.. grafů
- v případě překročení maximální hodnoty denního průměru PM10 50ug/m3 viz ČHMI i EU 1x za hodinu varování
Skript je opět napsaný v Pythonu a využívá moji vlastní knihovnu atprotoLib a pomocnou knihovnu aqbLib, která slouží pro download dat a obrázků a pro otagování textů postu. Spouštění CRONem v pravidelných intervalech.
Celý algoritmus lze ve zkratce popsat takto:
Stáhnout AirQualityData pro všechny monitory kvality ovzduší v JSON, něco jako
{
"CeskyTesin":{
"devID":"xxx",
"pm10":"2.1/17.6/8.6",
"pm2":"2.1/16.9/8.3",
"limit":50,
"avg":8.5729352794994,
"alert":false}
,
"Prague":{
"devID":"yyy",
"pm10":"1/5.6/3.1",
"pm2":"1/5.6/3.1",
"limit":50,
"avg":3.1451248882936,
"alert":false}
}
Projít JSON, z hodnot poskládat text postů, otagovat je, stáhnout obrázky a nahrát je na servery BlueSky a pak to celé postnout.
Z pohledu API je to opět o volání com.atproto.repo.createRecord (neveřejný endpoint, takže před tím získat AUTH token) - viz minulý článek AirQualityBot - 1.díl - Repost podle hashtagů.
Před samotným POSTem je však nutno
- poskládat text, vč. přidání hashtagů, případně linků
- nahrát obrázek na servery BlueSky a získat jejich ID
Hashtagy, citace, odkazy
Bluesky pro přidávání hashtagů, citací a linků nepoužívá značkovací jazyk, ale koncept „rich text facets“, které ukazují na místa v textu - Guide zde.
V případě, že text postu obsahuje Unicode znaky (emotikony apod) nezapomeňte přidat jeden znak navíc - proto ten offset v definici hashtagů
tags = {
"#AirQmonitor": [0, 0],
"#PM10": [0, 6],
"#PM2.5": [2, 8],
"#AirQuality": [4, 10],
"#AirPollution": [4, 10],
"#CitizenScience": [4,10],
"#AirQualityBot": [4, 10]
}
Funkce pro přidání hashtagů pak zde:
def add_tags_to_data(text, location, alert, data, tags):
index = 0
tags[f"#{location}"] = [0, 0]
for tag, offset in tags.items():
if tag in text:
offs = offset[1] if alert else offset[0]
byte_start = text.find(tag) + offs
byte_end = byte_start + len(tag)
facet = {
'index': {
'byteStart': byte_start,
'byteEnd': byte_end
},
'features': [
{
'$type': 'app.bsky.richtext.facet#tag',
'tag': tag.replace("#", "")
}
]
}
data['record']['facets'].append(facet)
index += 1
return data
Pozn: Koukám, že by ta funkce šla trochu optimalizovat... někdy časem možná.
Obrázky
Pro nahrání obrázků slouží com.atproto.repo.uploadBlob, (neveřejný endpoint), který vrací JSON s ukazatelem na nahraný obrázek. Ten je pak nutno přidat k datům POSTu.
def upload_image(token, filename, tcp_delay, request_type=""):
url = "https://bsky.social/xrpc/com.atproto.repo.uploadBlob"
with open(filename, 'rb') as f:
file_data = f.read()
headers = {
"Content-Type": "image/png"
}
if token:
headers["Authorization"] = f"Bearer {token}"
files = {'file': (filename, file_data, 'image/png')}
try:
response = requests.post(url, headers=headers, data=file_data, timeout=tcp_delay)
response.raise_for_status()
response_data = response.json()
show_rate_limits(response.headers, request_type)
return response_data
except requests.RequestException as e:
print(f"Request error: {e}")
return []
Pozor, obrázek je nutno nahrávat jako binární stream s patřičným Content-Type v hlavičce! Detail:
with open(filename, 'rb') as f:
file_data = f.read()
headers = {
"Content-Type": "image/png"
}
....
response = requests.post(url, headers=headers, data=file_data, timeout=tcp_delay)
Celý kód, ze kterého jsou patrné detaily, včetně knihoven, je opět u mne na GitHubu AirQualityBot
Zpět na Blog