fedi_welcome_bot/welcome_bot.py

#!/usr/bin/env python3

import argparse
import json
import os
import urllib.request
import urllib.parse
from datetime import datetime, timedelta, timezone
from pathlib import Path

# --- Načtení .env ---

env = {}
env_path = Path(__file__).parent / ".env"
with open(env_path) as f:
    for line in f:
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, value = line.partition("=")
        env[key.strip()] = value.strip()

BOT_TOKEN = env["BOT_TOKEN"]
ADMIN_TOKEN = env["ADMIN_TOKEN"]
INSTANCE = env["MASTODON_INSTANCE"].rstrip("/")

WELCOMED_FILE = Path(__file__).parent / "welcomed.json"

WELCOME_TEMPLATE = """\
@{username} 👋 Vítej na Mamutovo!

Tady nejsou algoritmy – feed si tvoříš tím, koho sleduješ.

👉 Začni tady: https://fedi.mamutovo.cz
Najdeš tam aktivní CZ/SK účty. Stačí klikat na „sledovat".

⚡ Chceš to rychle? Stáhni CSV a importuj ho:
Nastavení → Import a export → Import → Sledovaní

🐘 Nezapomeň vyplnit profil – fotku a bio. Ostatní tak lépe poznají kdo jsi.

✍️ První post? Napiš krátké představení s hashtagy: #cesky #novacek

💡 Odpovídej lidem a reaguj – tak si tě ostatní najdou."""


# --- Pomocné funkce ---

def api_get(path, token):
    req = urllib.request.Request(
        f"{INSTANCE}{path}",
        headers={"Authorization": f"Bearer {token}"},
    )
    with urllib.request.urlopen(req) as resp:
        return json.loads(resp.read().decode())


def api_post(path, token, data):
    body = json.dumps(data).encode()
    req = urllib.request.Request(
        f"{INSTANCE}{path}",
        data=body,
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
        },
        method="POST",
    )
    with urllib.request.urlopen(req) as resp:
        return json.loads(resp.read().decode())


def load_welcomed():
    if WELCOMED_FILE.exists():
        with open(WELCOMED_FILE) as f:
            return set(json.load(f))
    return set()


def save_welcomed(welcomed):
    with open(WELCOMED_FILE, "w") as f:
        json.dump(sorted(welcomed), f)


# --- Argumenty ---

parser = argparse.ArgumentParser()
parser.add_argument("--dry-run", action="store_true", help="Pouze vypíše co by se stalo, nic neodesílá")
parser.add_argument("--init", action="store_true", help="Stáhne všechny existující účty a uloží jejich ID do welcomed.json bez odesílání tootů")
args = parser.parse_args()

# --- --init: označit všechny existující účty jako uvítané ---

if args.init:
    all_ids = set()
    max_id = None
    while True:
        params = "limit=80"
        if max_id is not None:
            params += f"&max_id={max_id}"
        page = api_get(f"/api/v1/admin/accounts?{params}", ADMIN_TOKEN)
        if not page:
            break
        for account in page:
            all_ids.add(account["id"])
        max_id = page[-1]["id"]
        print(f"[INIT] Načteno celkem {len(all_ids)} účtů...", end="\r")
    print()
    save_welcomed(all_ids)
    print(f"[INIT] Hotovo – uloženo {len(all_ids)} ID do welcomed.json")
    raise SystemExit(0)

# --- Hlavní logika ---

accounts = api_get("/api/v1/admin/accounts?order=newest&limit=20", ADMIN_TOKEN)
welcomed = load_welcomed()
newly_welcomed = set()

cutoff = datetime.now(timezone.utc) - timedelta(minutes=10)

for account in accounts:
    account_id = account["id"]
    username = account["username"]

    if account_id in welcomed:
        continue

    created_at = datetime.fromisoformat(account["created_at"].replace("Z", "+00:00"))
    if created_at < cutoff:
        continue

    text = WELCOME_TEMPLATE.format(username=username)
    if args.dry_run:
        print(f"[DRY-RUN] Bylo by odesláno uvítání pro @{username} (id={account_id})")
    else:
        api_post("/api/v1/statuses", BOT_TOKEN, {
            "status": text,
            "visibility": "unlisted",
        })
        print(f"[OK] Uvítání odesláno: @{username} (id={account_id})")
        newly_welcomed.add(account_id)

if newly_welcomed:
    save_welcomed(welcomed | newly_welcomed)
    print(f"[INFO] Uvítáno {len(newly_welcomed)} nových uživatelů.")