mamutovo-stats-bot/daily_top.py

#!/usr/bin/env python3
import html
import json
import os
import re
import sys
import urllib.request
import urllib.error
from datetime import datetime, timezone
from zoneinfo import ZoneInfo

_TZ_PRAGUE = ZoneInfo("Europe/Prague")

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 clean_content(content):
    text = re.sub(r"<[^>]+>", " ", content)
    text = html.unescape(text)
    text = re.sub(r"#\s+(\w)", r"#\1", text)
    return re.sub(r"\s+", " ", text).strip()

def fetch_daily_timeline(base_url, token):
    cutoff = datetime.now(timezone.utc).timestamp() - 86400
    all_toots = []
    max_id = None
    for _ in range(10):
        url = f"{base_url}/api/v1/timelines/public?local=true&limit=40"
        if max_id:
            url += f"&max_id={max_id}"
        toots = api_get(url, token)
        if not toots:
            break
        done = False
        for toot in toots:
            created_at = datetime.fromisoformat(toot["created_at"].replace("Z", "+00:00"))
            if created_at.timestamp() < cutoff:
                done = True
                break
            all_toots.append(toot)
        if done:
            break
        max_id = toots[-1]["id"]
    return all_toots

def main():
    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)

    token = env["NOVINKY_TOKEN"]
    stats_token = env["STATS_TOKEN"]
    base_url = env["INSTANCE_URL"].rstrip("/")

    try:
        timeline = fetch_daily_timeline(base_url, token)
    except Exception:
        sys.exit(1)

    tag_counts = {}
    candidates = []
    media_count = {"image": 0, "video": 0, "gifv": 0, "audio": 0, "total": 0}
    hourly_count = {str(h): 0 for h in range(24)}
    for toot in timeline:
        created_at = datetime.fromisoformat(toot["created_at"].replace("Z", "+00:00"))
        local_hour = str(created_at.astimezone(_TZ_PRAGUE).hour)
        hourly_count[local_hour] += 1
        for att in toot.get("media_attachments", []):
            att_type = att.get("type", "")
            media_count["total"] += 1
            if att_type in media_count:
                media_count[att_type] += 1
        if toot.get("language") != "cs":
            continue
        if "@" in toot.get("account", {}).get("acct", ""):
            continue
        text = clean_content(toot.get("content", ""))
        if len(text) < 10:
            continue
        for tag in toot.get("tags", []):
            name = tag["name"]
            tag_counts[name] = tag_counts.get(name, 0) + 1
        reblogs = toot.get("reblogs_count", 0)
        favourites = toot.get("favourites_count", 0)
        candidates.append({
            "acct": toot["account"]["acct"],
            "text": text,
            "url": toot.get("url", ""),
            "reblogs": reblogs,
            "favourites": favourites,
            "score": reblogs + favourites,
        })

    candidates.sort(key=lambda x: x["score"], reverse=True)
    top = candidates[:3]
    tags = [t for t, _ in sorted(tag_counts.items(), key=lambda x: x[1], reverse=True)[:5]]

    authors_count = {}
    for c in candidates:
        authors_count[c["acct"]] = authors_count.get(c["acct"], 0) + 1

    newest_account = None
    try:
        accounts = api_get(
            f"{base_url}/api/v1/admin/accounts?order=newest&limit=1", stats_token
        )
        if accounts and accounts[0].get("account", {}).get("discoverable"):
            newest_account = {
                "acct": accounts[0]["account"]["acct"],
                "created_at": accounts[0]["created_at"],
            }
    except Exception:
        pass

    top_links = []
    try:
        links_data = api_get(f"{base_url}/api/v1/trends/links?limit=3", token)
        for link in links_data[:3]:
            top_links.append({
                "url": link.get("url", ""),
                "title": link.get("title", ""),
                "description": link.get("description", ""),
                "provider_name": link.get("provider_name", ""),
            })
    except Exception:
        pass

    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    os.makedirs("data", exist_ok=True)
    out_path = os.path.join("data", f"{today}.json")

    with open(out_path, "w", encoding="utf-8") as f:
        json.dump({
            "date": today,
            "total_count": len(timeline),
            "authors_count": authors_count,
            "newest_account": newest_account,
            "top": top,
            "tags": tags,
            "top_links": top_links,
            "media_count": media_count,
            "hourly_count": hourly_count,
        }, f, ensure_ascii=False, indent=2)

    print(f"Uloženo: {out_path} ({len(timeline)} tootů načteno, {len(top)} top, {len(tags)} hashtagů)")

if __name__ == "__main__":
    main()