feat: měsíční report, odstraněno new_followers
This commit is contained in:
+140
-23
@@ -39,8 +39,9 @@ def load_env(path=".env"):
|
||||
pass
|
||||
return env
|
||||
|
||||
def api_get(url, token):
|
||||
req = urllib.request.Request(url, headers={"Authorization": f"Bearer {token}"})
|
||||
def api_get(url, token=None):
|
||||
headers = {"Authorization": f"Bearer {token}"} if token else {}
|
||||
req = urllib.request.Request(url, headers=headers)
|
||||
try:
|
||||
with urllib.request.urlopen(req) as resp:
|
||||
return json.loads(resp.read().decode())
|
||||
@@ -72,10 +73,12 @@ def api_post(url, token, data):
|
||||
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):
|
||||
def get_measures(base_url, admin_token, date_from, date_to, keys=None):
|
||||
if keys is None:
|
||||
keys = ["new_users", "active_users", "interactions"]
|
||||
url = f"{base_url}/api/v1/admin/measures"
|
||||
payload = {
|
||||
"keys": ["new_users", "active_users", "interactions"],
|
||||
"keys": keys,
|
||||
"start_at": date_from.isoformat(),
|
||||
"end_at": date_to.isoformat(),
|
||||
}
|
||||
@@ -116,6 +119,57 @@ def format_date_cs(dt):
|
||||
]
|
||||
return f"{dt.day}. {months[dt.month - 1]}"
|
||||
|
||||
def format_month_cs(dt):
|
||||
months = [
|
||||
"leden", "únor", "březen", "duben", "květen", "červen",
|
||||
"červenec", "srpen", "září", "říjen", "listopad", "prosinec",
|
||||
]
|
||||
return f"{months[dt.month - 1]} {dt.year}"
|
||||
|
||||
def build_monthly_toot(measures_data, tags, top_tooty, date_to, prev_stats, instance_info):
|
||||
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)
|
||||
def fmt_diff(current, key, long=False):
|
||||
if not prev_stats or key not in prev_stats:
|
||||
return ""
|
||||
d = current - prev_stats[key]
|
||||
sign = "+" if d >= 0 else ""
|
||||
suffix = " oproti minulému měsíci" if long else ""
|
||||
return f" ({sign}{d}{suffix})"
|
||||
|
||||
hashtags = " ".join(f"#{t['name']}" for t in tags[:5]) if tags else "(žádné)"
|
||||
|
||||
inst_stats = instance_info.get("stats", {}) if instance_info else {}
|
||||
user_count = inst_stats.get("user_count", "?")
|
||||
domain_count = inst_stats.get("domain_count", "?")
|
||||
|
||||
if top_tooty:
|
||||
blocks = "\n\n".join(
|
||||
f"👤 @{s['acct']}\n\"{truncate(s['text'], 80).replace(chr(10), ' ')}\"\n🔁 {s['reblogs']} ⭐ {s['favourites']}"
|
||||
for s in top_tooty
|
||||
)
|
||||
tooty_sekce = f"\n🌟 Tooty měsíce:\n\n{blocks}"
|
||||
else:
|
||||
tooty_sekce = ""
|
||||
|
||||
return (
|
||||
f"🐘 Měsíční přehled Mamutovo.cz\n"
|
||||
f"📅 {format_month_cs(date_to)}\n"
|
||||
f"\n"
|
||||
f"👥 Noví uživatelé: {new_users}{fmt_diff(new_users, 'new_users', long=True)}\n"
|
||||
f"✅ Aktivní uživatelé: {active_users}{fmt_diff(active_users, 'active_users')}\n"
|
||||
f"📝 Interakce: {interactions}{fmt_diff(interactions, 'interactions')}\n"
|
||||
f"\n"
|
||||
f"📊 Celkem uživatelů: {user_count}\n"
|
||||
f"🌐 Federovaných instancí: {domain_count}\n"
|
||||
f"\n"
|
||||
f"🔥 Top hashtagy měsíce:\n"
|
||||
f"{hashtags}"
|
||||
f"{tooty_sekce}"
|
||||
)
|
||||
|
||||
def build_toot(measures_data, tags, top_tooty, date_from, date_to, week_number):
|
||||
stats = {m["key"]: int(m["total"]) for m in measures_data}
|
||||
new_users = stats.get("new_users", 0)
|
||||
@@ -154,9 +208,29 @@ def build_toot(measures_data, tags, top_tooty, date_from, date_to, week_number):
|
||||
f"💡 Tip týdne: {tip}"
|
||||
)
|
||||
|
||||
def load_tooty_from_data(date_to, days):
|
||||
seen = set()
|
||||
all_tooty = []
|
||||
for i in range(days):
|
||||
day = (date_to - timedelta(days=i)).strftime("%Y-%m-%d")
|
||||
path = os.path.join("data", f"{day}.json")
|
||||
try:
|
||||
with open(path, encoding="utf-8") as f:
|
||||
file_data = json.load(f)
|
||||
for s in file_data.get("top", []):
|
||||
key = (s.get("acct", ""), s.get("text", ""))
|
||||
if key not in seen:
|
||||
seen.add(key)
|
||||
all_tooty.append(s)
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
all_tooty.sort(key=lambda s: s.get("score", 0), reverse=True)
|
||||
return all_tooty[:3]
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Týdenní statistiky Mamutovo.cz")
|
||||
parser = argparse.ArgumentParser(description="Statistiky Mamutovo.cz")
|
||||
parser.add_argument("--dry-run", action="store_true", help="Pouze vypíše toot, neodešle")
|
||||
parser.add_argument("--monthly", action="store_true", help="Měsíční přehled místo týdenního")
|
||||
args = parser.parse_args()
|
||||
|
||||
env = {**load_env(), **os.environ}
|
||||
@@ -172,6 +246,66 @@ def main():
|
||||
|
||||
now = datetime.now(timezone.utc)
|
||||
date_to = now.replace(hour=0, minute=0, second=0, microsecond=0)
|
||||
|
||||
if args.monthly:
|
||||
date_from = date_to - timedelta(days=30)
|
||||
|
||||
try:
|
||||
measures_data = get_measures(
|
||||
base_url, admin_token, date_from, date_to,
|
||||
keys=["new_users", "active_users", "interactions"],
|
||||
)
|
||||
except Exception:
|
||||
sys.exit(1)
|
||||
|
||||
try:
|
||||
tags = api_get(f"{base_url}/api/v1/trends/tags?limit=5", admin_token)
|
||||
except Exception:
|
||||
tags = []
|
||||
|
||||
try:
|
||||
instance_info = api_get(f"{base_url}/api/v1/instance")
|
||||
except Exception:
|
||||
instance_info = {}
|
||||
|
||||
monthly_stats_path = os.path.join("data", "monthly_stats.json")
|
||||
prev_stats = None
|
||||
try:
|
||||
with open(monthly_stats_path, encoding="utf-8") as f:
|
||||
prev_stats = json.load(f)
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
|
||||
top_tooty = load_tooty_from_data(date_to, 30)
|
||||
toot = build_monthly_toot(measures_data, tags, top_tooty, date_to, prev_stats, instance_info)
|
||||
|
||||
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)
|
||||
|
||||
cur_stats = {m["key"]: int(m["total"]) for m in measures_data}
|
||||
os.makedirs("data", exist_ok=True)
|
||||
with open(monthly_stats_path, "w", encoding="utf-8") as f:
|
||||
json.dump({
|
||||
"date": date_to.strftime("%Y-%m-%d"),
|
||||
"new_users": cur_stats.get("new_users", 0),
|
||||
"active_users": cur_stats.get("active_users", 0),
|
||||
"interactions": cur_stats.get("interactions", 0),
|
||||
}, f, ensure_ascii=False, indent=2)
|
||||
print("Měsíční statistiky uloženy.")
|
||||
return
|
||||
|
||||
# Týdenní přehled
|
||||
date_from = date_to - timedelta(days=7)
|
||||
week_number = now.isocalendar()[1]
|
||||
|
||||
@@ -185,24 +319,7 @@ def main():
|
||||
except Exception:
|
||||
tags = []
|
||||
|
||||
seen = set()
|
||||
all_tooty = []
|
||||
for i in range(7):
|
||||
day = (date_to - timedelta(days=i)).strftime("%Y-%m-%d")
|
||||
path = os.path.join("data", f"{day}.json")
|
||||
try:
|
||||
with open(path, encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
for s in data.get("top", []):
|
||||
key = (s.get("acct", ""), s.get("text", ""))
|
||||
if key not in seen:
|
||||
seen.add(key)
|
||||
all_tooty.append(s)
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
all_tooty.sort(key=lambda s: s.get("score", 0), reverse=True)
|
||||
top_tooty = all_tooty[:3]
|
||||
|
||||
top_tooty = load_tooty_from_data(date_to, 7)
|
||||
toot = build_toot(measures_data, tags, top_tooty, date_from, date_to, week_number)
|
||||
|
||||
if args.dry_run:
|
||||
|
||||
Reference in New Issue
Block a user