feat: welcome_bot.py – polling, dry-run, filtr created_at
This commit is contained in:
+125
@@ -0,0 +1,125 @@
|
|||||||
|
#!/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á")
|
||||||
|
args = parser.parse_args()
|
||||||
|
|
||||||
|
# --- 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ů.")
|
||||||
|
else:
|
||||||
|
print("[INFO] Žádní noví uživatelé.")
|
||||||
@@ -0,0 +1 @@
|
|||||||
|
["116099549727519820", "116127273263382313", "116128177820418097", "116141792826154864", "116144286217194293", "116154657073269480", "116158614490312678", "116170256054539646", "116170614642317484", "116178404922095929", "116193071329347528", "116198302059185321", "116205281310353001", "116227624836913979", "116237145399664481", "116288673489103920", "116289664908714786", "116299265563853066", "116329847494946855", "116342628895905045"]
|
||||||
Reference in New Issue
Block a user