214 lines
7.2 KiB
Python
214 lines
7.2 KiB
Python
#!/usr/bin/env python3
|
||
import html
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import argparse
|
||
import urllib.request
|
||
import urllib.error
|
||
from datetime import datetime, timezone, timedelta
|
||
|
||
TIPS = [
|
||
"Hashtagy fungují jako klíčová slova – používej je a ostatní tě snáz najdou.",
|
||
"Pomocí seznamů si můžeš organizovat sledované účty do tematických skupin.",
|
||
"Příspěvky s viditelností \"Pouze sledující\" vidí jen tvoji sledující, ne celý fediverse.",
|
||
"Filtrovat nežádoucí obsah lze přes Nastavení → Filtry – hodí se na citlivá témata.",
|
||
"Zmínit někoho funguje i napříč instancemi – stačí napsat @uzivatel@instance.tld.",
|
||
]
|
||
|
||
def load_env(path=".env"):
|
||
env = {}
|
||
try:
|
||
with open(path) as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if not line or line.startswith("#") or "=" not in line:
|
||
continue
|
||
key, _, val = line.partition("=")
|
||
env[key.strip()] = val.strip().strip('"').strip("'")
|
||
except FileNotFoundError:
|
||
pass
|
||
return env
|
||
|
||
def api_get(url, token):
|
||
req = urllib.request.Request(url, headers={"Authorization": f"Bearer {token}"})
|
||
try:
|
||
with urllib.request.urlopen(req) as resp:
|
||
return json.loads(resp.read().decode())
|
||
except urllib.error.HTTPError as e:
|
||
print(f"HTTP {e.code} při volání {url}: {e.read().decode()}", file=sys.stderr)
|
||
raise
|
||
except urllib.error.URLError as e:
|
||
print(f"Chyba sítě při volání {url}: {e.reason}", file=sys.stderr)
|
||
raise
|
||
|
||
def api_post(url, token, data):
|
||
body = json.dumps(data).encode()
|
||
req = urllib.request.Request(
|
||
url,
|
||
data=body,
|
||
headers={
|
||
"Authorization": f"Bearer {token}",
|
||
"Content-Type": "application/json",
|
||
},
|
||
method="POST",
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(req) as resp:
|
||
return json.loads(resp.read().decode())
|
||
except urllib.error.HTTPError as e:
|
||
print(f"HTTP {e.code} při odesílání tootu: {e.read().decode()}", file=sys.stderr)
|
||
raise
|
||
except urllib.error.URLError as e:
|
||
print(f"Chyba sítě při odesílání tootu: {e.reason}", file=sys.stderr)
|
||
raise
|
||
|
||
def get_measures(base_url, admin_token, date_from, date_to):
|
||
url = f"{base_url}/api/v1/admin/measures"
|
||
payload = {
|
||
"keys": ["new_users", "active_users", "interactions"],
|
||
"start_at": date_from.isoformat(),
|
||
"end_at": date_to.isoformat(),
|
||
}
|
||
body = json.dumps(payload).encode()
|
||
req = urllib.request.Request(
|
||
url,
|
||
data=body,
|
||
headers={
|
||
"Authorization": f"Bearer {admin_token}",
|
||
"Content-Type": "application/json",
|
||
},
|
||
method="POST",
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(req) as resp:
|
||
return json.loads(resp.read().decode())
|
||
except urllib.error.HTTPError as e:
|
||
print(f"HTTP {e.code} při volání measures API: {e.read().decode()}", file=sys.stderr)
|
||
raise
|
||
except urllib.error.URLError as e:
|
||
print(f"Chyba sítě při volání measures API: {e.reason}", file=sys.stderr)
|
||
raise
|
||
|
||
def truncate(text, max_chars=100):
|
||
text = re.sub(r'<a\b[^>]*class="[^"]*hashtag[^"]*"[^>]*>.*?</a>', "",text, flags=re.IGNORECASE)
|
||
text = re.sub(r"<[^>]+>", " ", text)
|
||
text = html.unescape(text)
|
||
text = re.sub(r"\s+", " ", text).strip()
|
||
if len(text) <= max_chars:
|
||
return text
|
||
cut = text[:max_chars].rsplit(" ", 1)[0].rstrip(".,!?;:")
|
||
return cut + "…"
|
||
|
||
def format_date_cs(dt):
|
||
months = [
|
||
"ledna", "února", "března", "dubna", "května", "června",
|
||
"července", "srpna", "září", "října", "listopadu", "prosince",
|
||
]
|
||
return f"{dt.day}. {months[dt.month - 1]}"
|
||
|
||
def build_toot(measures_data, tags, trend_status, date_from, date_to, week_number):
|
||
stats = {m["key"]: int(m["total"]) for m in measures_data}
|
||
new_users = stats.get("new_users", 0)
|
||
active_users = stats.get("active_users", 0)
|
||
interactions = stats.get("interactions", 0)
|
||
|
||
hashtags = " ".join(f"#{t['name']}" for t in tags[:3]) if tags else "(žádné)"
|
||
|
||
tip = TIPS[week_number % len(TIPS)]
|
||
|
||
if trend_status:
|
||
acct = trend_status.get("account", {}).get("acct", "?")
|
||
content = truncate(trend_status.get("content", ""), 100)
|
||
boosts = trend_status.get("reblogs_count", 0)
|
||
favs = trend_status.get("favourites_count", 0)
|
||
toot_tyden = f"🌟 Toot týdne od @{acct}:\n\"{content}\"\n🔁 {boosts} ⭐ {favs}\n\n"
|
||
else:
|
||
toot_tyden = ""
|
||
|
||
date_from_str = format_date_cs(date_from)
|
||
date_to_str = format_date_cs(date_to)
|
||
year = date_to.year
|
||
|
||
return (
|
||
f"🐘 Týdenní přehled Mamutovo.cz\n"
|
||
f"📅 {date_from_str} – {date_to_str} {year}\n"
|
||
f"\n"
|
||
f"👥 Noví uživatelé: {new_users}\n"
|
||
f"✅ Aktivní uživatelé: {active_users}\n"
|
||
f"📝 Interakce: {interactions}\n"
|
||
f"\n"
|
||
f"🔥 Populární hashtagy:\n"
|
||
f"{hashtags}\n"
|
||
f"\n"
|
||
f"{toot_tyden}"
|
||
f"💡 Tip týdne: {tip}\n"
|
||
f"\n"
|
||
f"🔗 fedi.mamutovo.cz"
|
||
)
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="Týdenní statistiky Mamutovo.cz")
|
||
parser.add_argument("--dry-run", action="store_true", help="Pouze vypíše toot, neodešle")
|
||
args = parser.parse_args()
|
||
|
||
env = {**load_env(), **os.environ}
|
||
|
||
for var in ("NOVINKY_TOKEN", "INSTANCE_URL", "STATS_TOKEN"):
|
||
if not env.get(var):
|
||
print(f"Chybí proměnná prostředí: {var}", file=sys.stderr)
|
||
sys.exit(1)
|
||
|
||
novinky_token = env["NOVINKY_TOKEN"]
|
||
admin_token = env["STATS_TOKEN"]
|
||
base_url = env["INSTANCE_URL"].rstrip("/")
|
||
|
||
now = datetime.now(timezone.utc)
|
||
date_to = now.replace(hour=0, minute=0, second=0, microsecond=0)
|
||
date_from = date_to - timedelta(days=7)
|
||
week_number = now.isocalendar()[1]
|
||
|
||
try:
|
||
measures_data = get_measures(base_url, admin_token, date_from, date_to)
|
||
except Exception:
|
||
sys.exit(1)
|
||
|
||
try:
|
||
tags = api_get(f"{base_url}/api/v1/trends/tags?limit=3", admin_token)
|
||
except Exception:
|
||
tags = []
|
||
|
||
try:
|
||
statuses = api_get(f"{base_url}/api/v1/trends/statuses?limit=5", admin_token)
|
||
trend_status = None
|
||
for s in statuses:
|
||
clean = re.sub(r'<a\b[^>]*class="[^"]*hashtag[^"]*"[^>]*>.*?</a>', "",s.get("content", ""), flags=re.IGNORECASE)
|
||
clean = re.sub(r"<[^>]+>", " ", clean)
|
||
clean = html.unescape(clean)
|
||
clean = re.sub(r"\s+", " ", clean).strip()
|
||
if len(clean) >= 20:
|
||
trend_status = s
|
||
break
|
||
except Exception:
|
||
trend_status = None
|
||
|
||
toot = build_toot(measures_data, tags, trend_status, date_from, date_to, week_number)
|
||
|
||
if args.dry_run:
|
||
print(toot)
|
||
return
|
||
|
||
try:
|
||
result = api_post(
|
||
f"{base_url}/api/v1/statuses",
|
||
novinky_token,
|
||
{"status": toot, "visibility": "public"},
|
||
)
|
||
print(f"Toot odeslán: {result.get('url', '(bez URL)')}")
|
||
except Exception:
|
||
sys.exit(1)
|
||
|
||
if __name__ == "__main__":
|
||
main()
|