Spaces:
Running on Zero
Running on Zero
| """The analyzer's brain. Pure functions over plain data (no database, no Telegram), so every number | |
| is unit-tested. The loading of the data lives in app/db/repos/analytics.py. | |
| Honesty rules built in: small samples are labelled, a correlation is never called a cause, and every | |
| insight shows the numbers it is based on.""" | |
| import html | |
| import re | |
| from collections import Counter | |
| from dataclasses import dataclass, field | |
| from datetime import date, datetime, timedelta, timezone | |
| from statistics import median | |
| from typing import Optional | |
| ET = timezone(timedelta(hours=3)) | |
| SPARSE = 3 # a course with fewer files than this is "short of material" | |
| # ------------------------------------------------------------------ plain data | |
| class UserRow: | |
| id: int | |
| created: datetime # UTC | |
| joined: date # Ethiopian date of joining | |
| source: str | |
| verified: bool | |
| onboarded: bool | |
| active_days: int | |
| stream: Optional[str] = None | |
| missing: Optional[str] = None # join-gate places last missing, e.g. "dbucentralforum" | |
| class EventRow: | |
| kind: str # download | search | inline | find | |
| ref: str | |
| n: int | |
| day: date | |
| user_id: int | |
| class MaterialRow: | |
| id: int | |
| course_id: Optional[int] | |
| title: str | |
| downloads: int | |
| status: str | |
| kind: str | |
| class SubRow: | |
| status: str | |
| user_id: int | |
| created: datetime | |
| reviewed: Optional[datetime] | |
| class PostRow: | |
| message_id: int | |
| posted: datetime # UTC | |
| preview: str | |
| tags: list[str] | |
| reactions: Optional[int] | |
| class Snapshot: | |
| now: datetime | |
| users: list[UserRow] | |
| checkins: dict[int, set[date]] | |
| events: list[EventRow] | |
| materials: list[MaterialRow] | |
| courses: dict[int, str] | |
| subs: list[SubRow] | |
| posts: list[PostRow] = field(default_factory=list) | |
| def today(self) -> date: | |
| return self.now.astimezone(ET).date() | |
| # ------------------------------------------------------------------ small helpers | |
| def pct(part: int, whole: int) -> int: | |
| return round(100 * part / whole) if whole else 0 | |
| def bar(percent: float, width: int = 10) -> str: | |
| filled = max(0, min(width, round(percent / 100 * width))) | |
| return "▓" * filled + "░" * (width - filled) | |
| def esc(text: str) -> str: | |
| return html.escape(text or "", quote=False) | |
| def extract_tags(text: str) -> list[str]: | |
| """?start= tags inside a post: 'ad1', 'dorm5' (numbers are personal invite links, ignored).""" | |
| tags = re.findall(r"[?&]start=([A-Za-z0-9_]{1,20})", text or "") | |
| return sorted({t.lower() for t in tags if not t.isdigit()}) | |
| # ------------------------------------------------------------------ growth + funnel | |
| def per_day(users: list[UserRow], today: date, days: int = 14) -> list[tuple[date, int]]: | |
| counts = Counter(u.joined for u in users) | |
| return [(today - timedelta(days=i), counts.get(today - timedelta(days=i), 0)) for i in range(days - 1, -1, -1)] | |
| def funnel(users: list[UserRow]) -> dict: | |
| n = len(users) | |
| verified = sum(u.verified for u in users) | |
| onboarded = sum(u.onboarded for u in users) | |
| active3 = sum(u.active_days >= 3 for u in users) | |
| return dict(joined=n, verified=verified, onboarded=onboarded, active3=active3) | |
| def biggest_leak(f: dict) -> Optional[tuple[str, str, int, int]]: | |
| """The stage with the largest loss between consecutive steps: (from, to, people lost, % lost).""" | |
| steps = [("joined", "in channel + forum", f["joined"], f["verified"]), | |
| ("in channel + forum", "finished setup", f["verified"], f["onboarded"])] | |
| steps = [(a, b, x - y, pct(x - y, x)) for a, b, x, y in steps if x > 0] | |
| return max(steps, key=lambda s: s[2]) if steps and max(s[2] for s in steps) > 0 else None | |
| def source_table(users: list[UserRow]) -> list[dict]: | |
| groups: dict[str, list[UserRow]] = {} | |
| for u in users: | |
| groups.setdefault(u.source, []).append(u) | |
| rows = [] | |
| for source, members in groups.items(): | |
| f = funnel(members) | |
| rows.append(dict(source=source, **f, setup_pct=pct(f["onboarded"], f["joined"]))) | |
| return sorted(rows, key=lambda r: -r["joined"]) | |
| def stuck_by_missing(users: list[UserRow]) -> list[tuple[str, int]]: | |
| c = Counter((u.missing or "not recorded yet") for u in users if not u.verified) | |
| return c.most_common() | |
| # ------------------------------------------------------------------ activity + retention | |
| def dau_wau(checkins: dict[int, set[date]], today: date) -> dict: | |
| week = [today - timedelta(days=i) for i in range(6, -1, -1)] | |
| dau = [(d, sum(1 for days in checkins.values() if d in days)) for d in week] | |
| wau = sum(1 for days in checkins.values() if any(d in days for d in week)) | |
| last_full = dau[-2][1] if len(dau) >= 2 else 0 # yesterday is complete, today is not | |
| return dict(dau=dau, wau=wau, stickiness=pct(last_full, wau)) | |
| def retention(users: list[UserRow], checkins: dict[int, set[date]], today: date, cohorts: int = 7) -> dict: | |
| """Share of each day's new students who were active exactly N days later. Check-ins only exist from the day | |
| activity tracking started, so earlier cohorts are 'n/a' instead of a misleading 0. A day that is not over | |
| yet (today) is never counted.""" | |
| all_days = [d for days in checkins.values() for d in days] | |
| start = min(all_days) if all_days else None | |
| rows = [] | |
| for offset in range(cohorts, 0, -1): | |
| d = today - timedelta(days=offset) | |
| members = [u for u in users if u.joined == d] | |
| if not members: | |
| continue | |
| cells = {} | |
| for n in (1, 3, 7): | |
| target = d + timedelta(days=n) | |
| if target >= today or start is None or target < start: | |
| cells[n] = None # not reached yet, or before we tracked activity | |
| else: | |
| cells[n] = pct(sum(1 for u in members if target in checkins.get(u.id, set())), len(members)) | |
| rows.append(dict(day=d, size=len(members), **{f"d{n}": v for n, v in cells.items()})) | |
| eligible = [u for u in users if u.joined < today] | |
| came_back = sum(1 for u in eligible if any(day > u.joined for day in checkins.get(u.id, set()))) | |
| return dict(rows=rows, tracking_start=start, came_back=came_back, eligible=len(eligible)) | |
| # ------------------------------------------------------------------ library + searches | |
| def library(snap: Snapshot) -> dict: | |
| ok = [m for m in snap.materials if m.status == "ok"] | |
| per_course = Counter(m.course_id for m in ok if m.course_id) | |
| empty = [name for cid, name in snap.courses.items() if per_course.get(cid, 0) == 0] | |
| sparse = [name for cid, name in snap.courses.items() if 0 < per_course.get(cid, 0) < SPARSE] | |
| week_start = snap.today - timedelta(days=6) | |
| downloads_week = [e for e in snap.events if e.kind == "download" and e.day >= week_start] | |
| since = snap.today - timedelta(days=13) | |
| searches = [e for e in snap.events if e.kind in ("search", "inline", "find") and e.day >= since and len(e.ref) >= 2] | |
| missed = Counter(e.ref for e in searches if e.n == 0) | |
| found = Counter(e.ref for e in searches if e.n > 0) | |
| top_files = sorted(ok, key=lambda m: -m.downloads)[:5] | |
| return dict( | |
| files=len(ok), kinds=Counter(m.kind for m in ok), courses=len(snap.courses), empty=empty, sparse=sparse, | |
| downloads_total=sum(m.downloads for m in ok), downloads_week=len(downloads_week), | |
| downloaders_week=len({e.user_id for e in downloads_week}), top_files=top_files, | |
| by_course=[(snap.courses.get(cid, "?"), n) for cid, n in per_course.most_common(5)], | |
| searches=len(searches), missed_total=sum(missed.values()), missed=missed.most_common(8), found=found.most_common(5), | |
| ) | |
| # ------------------------------------------------------------------ contributions + channel posts | |
| def contributions(snap: Snapshot) -> dict: | |
| by = Counter(s.status for s in snap.subs) | |
| done = [s for s in snap.subs if s.reviewed] | |
| hours = [(s.reviewed - s.created).total_seconds() / 3600 for s in done] | |
| pending = [s for s in snap.subs if s.status == "pending"] | |
| oldest = max(((snap.now - s.created).total_seconds() / 3600 for s in pending), default=0.0) | |
| top = Counter(s.user_id for s in snap.subs if s.status == "approved").most_common(3) | |
| return dict( | |
| pending=by["pending"], approved=by["approved"], rejected=by["rejected"], | |
| approval=pct(by["approved"], by["approved"] + by["rejected"]), | |
| avg_hours=round(sum(hours) / len(hours), 1) if hours else None, oldest_pending_hours=round(oldest, 1), top=top, | |
| ) | |
| def channel_posts(snap: Snapshot, window_hours: int = 6) -> list[dict]: | |
| """For each post: students who joined the bot in the hours after it, and students who came through a tag | |
| written in the post. NOTE: a correlation, not proof; close posts share the same joiners.""" | |
| rows = [] | |
| for p in snap.posts: | |
| end = p.posted + timedelta(hours=window_hours) | |
| soon = sum(1 for u in snap.users if p.posted <= u.created < end) | |
| tagged = sum(1 for u in snap.users if u.source in p.tags) | |
| rows.append(dict(message_id=p.message_id, posted=p.posted, preview=p.preview, tags=p.tags, | |
| reactions=p.reactions, soon=soon, tagged=tagged)) | |
| return sorted(rows, key=lambda r: (-r["tagged"], -r["soon"])) | |
| # ------------------------------------------------------------------ insights (rule based, numbers shown) | |
| class Insight: | |
| priority: int # 1 = most important | |
| title: str | |
| evidence: str | |
| action: str | |
| def insights(snap: Snapshot) -> list[Insight]: | |
| out: list[Insight] = [] | |
| users, today = snap.users, snap.today | |
| if not users: | |
| return out | |
| f = funnel(users) | |
| leak = biggest_leak(f) | |
| stuck = [u for u in users if not u.verified] | |
| if f["joined"] and pct(len(stuck), f["joined"]) >= 15: | |
| top = [(m, n) for m, n in stuck_by_missing(users)][:3] | |
| detail = ", ".join(f"{n} missing {m}" if m != "not recorded yet" else f"{n} not recorded yet" for m, n in top) | |
| out.append(Insight(1, "Students stop at the join step", | |
| f"{len(stuck)} of {f['joined']} ({pct(len(stuck), f['joined'])}%) never got through: {detail}.", | |
| "Check /gatecheck. If most miss the forum, consider making it optional or send one reminder.")) | |
| elif leak and leak[3] >= 10: | |
| out.append(Insight(2, f"Biggest drop: {leak[0]} → {leak[1]}", f"{leak[2]} people ({leak[3]}%) are lost here.", | |
| "Make this step shorter or explain why it matters.")) | |
| verified_not_setup = f["verified"] - f["onboarded"] | |
| if f["verified"] and pct(verified_not_setup, f["verified"]) >= 10: | |
| out.append(Insight(2, "Some verified students did not finish setup", | |
| f"{verified_not_setup} of {f['verified']} ({pct(verified_not_setup, f['verified'])}%) stopped at stream/year.", | |
| "Check that the two setup questions are clear.")) | |
| ret = retention(users, snap.checkins, today) | |
| d1 = [(r["d1"], r["size"]) for r in ret["rows"] if r["d1"] is not None] | |
| if d1 and sum(s for _, s in d1) >= 20: | |
| total = sum(s for _, s in d1) | |
| avg = round(sum(v * s for v, s in d1) / total) | |
| if avg < 20: | |
| out.append(Insight(1, "Few students come back the next day", f"Day-1 return is {avg}% (over {total} students).", | |
| "Give a reason to return: new files weekly, a daily XP nudge, exam-week reminders.")) | |
| elif ret["tracking_start"] is None or snap.checkins and not d1: | |
| since = ret["tracking_start"].isoformat() if ret["tracking_start"] else "now" | |
| out.append(Insight(4, "Not enough days of data for retention yet", | |
| f"Activity has been tracked since {since}.", "Check again in a few days.")) | |
| lib = library(snap) | |
| if lib["missed"] and lib["missed"][0][1] >= 2: | |
| q, n = lib["missed"][0] | |
| out.append(Insight(1, "Students search for things you do not have", | |
| f"{lib['missed_total']} searches in 14 days found nothing; top: “{q}” ({n} times).", | |
| "Add material for these words (see Library & searches), then they find it next time.")) | |
| if snap.courses and (len(lib["empty"]) / len(snap.courses)) >= 0.4: | |
| out.append(Insight(2, "Many courses have no file", f"{len(lib['empty'])} of {len(snap.courses)} courses are empty.", | |
| "Use the 🎯 wanted-courses prompt in announcements; ask seniors for past exams.")) | |
| con = contributions(snap) | |
| if con["pending"] and con["oldest_pending_hours"] >= 24: | |
| out.append(Insight(2, "Student files are waiting", f"{con['pending']} waiting; the oldest {con['oldest_pending_hours']:.0f} h.", | |
| "Run /queue. Slow reviews discourage contributors.")) | |
| days = [n for _, n in per_day(users, today, 14)] | |
| nonzero = [n for n in days if n] | |
| if len(nonzero) >= 3 and max(days) >= 3 * max(1, median(days)): | |
| out.append(Insight(3, "Growth comes from single pushes", | |
| f"Best day {max(days)} new students vs a typical day of {median(days):g}.", | |
| "Keep a steady posting rhythm; one-off pushes fade within 2-3 days.")) | |
| sources = [r for r in source_table(users) if r["joined"] >= 10 and r["source"] != "before tracking"] | |
| if len(sources) >= 2: | |
| best, worst = max(sources, key=lambda r: r["setup_pct"]), min(sources, key=lambda r: r["setup_pct"]) | |
| if best["setup_pct"] - worst["setup_pct"] >= 15: | |
| out.append(Insight(3, "Sources differ in quality", | |
| f"“{best['source']}”: {best['setup_pct']}% finish setup vs “{worst['source']}”: {worst['setup_pct']}% " | |
| f"({best['joined']} and {worst['joined']} students, still small samples).", | |
| "Repeat what works in the better source.")) | |
| referral = sum(1 for u in users if u.source == "referral") | |
| if len(users) >= 50 and pct(referral, len(users)) < 5: | |
| out.append(Insight(3, "Few students join by invite", f"{referral} of {len(users)} ({pct(referral, len(users))}%) came from an invite.", | |
| "Mention “Invite friends” in channel posts; the share button is one tap.")) | |
| posts = channel_posts(snap) | |
| if posts and (posts[0]["tagged"] or posts[0]["soon"]): | |
| p = posts[0] | |
| out.append(Insight(4, "Your best channel post so far", | |
| f"“{esc(p['preview'][:50])}” → {p['tagged']} via its tag, {p['soon']} new students within 6 h " | |
| "(a correlation, not proof).", "Write more posts like this one.")) | |
| return sorted(out, key=lambda i: i.priority) | |
| # ------------------------------------------------------------------ text for each screen | |
| def _fmt_day(d: date) -> str: | |
| return d.strftime("%m-%d") | |
| def overview(snap: Snapshot) -> str: | |
| users, today = snap.users, snap.today | |
| f = funnel(users) | |
| new_today = sum(1 for u in users if u.joined == today) | |
| wau = dau_wau(snap.checkins, today) | |
| ret = retention(users, snap.checkins, today) | |
| lib = library(snap) | |
| con = contributions(snap) | |
| lines = [ | |
| f"📊 <b>Analytics</b> · {snap.now.astimezone(ET).strftime('%d %b %H:%M')} (Ethiopian time)", | |
| f"👥 <b>{f['joined']}</b> students (+{new_today} today) · in channel+forum <b>{f['verified']}</b> ({pct(f['verified'], f['joined'])}%) " | |
| f"· finished setup <b>{f['onboarded']}</b> ({pct(f['onboarded'], f['joined'])}%)", | |
| f"🔁 Active this week <b>{wau['wau']}</b> · came back at least once <b>{ret['came_back']}</b> of {ret['eligible']} " | |
| f"({pct(ret['came_back'], ret['eligible'])}%)", | |
| f"📚 <b>{lib['files']}</b> files · {lib['downloads_total']} downloads ({lib['downloads_week']} this week) · " | |
| f"{len(lib['empty'])} courses empty", | |
| f"🔍 {lib['searches']} searches in 14 days · found nothing: {lib['missed_total']} ({pct(lib['missed_total'], lib['searches'])}%)", | |
| f"📤 Student files: {con['pending']} waiting · {con['approved']} approved · {con['rejected']} rejected", | |
| ] | |
| ins = insights(snap)[:3] | |
| if ins: | |
| lines.append("\n💡 <b>What matters most now</b>") | |
| lines += [f"{i}. <b>{esc(x.title)}</b>: {x.evidence}" for i, x in enumerate(ins, 1)] | |
| lines.append("\nTap a button for details. Small samples (under ~30 people) are only hints.") | |
| return "\n".join(lines) | |
| def section_growth(snap: Snapshot) -> str: | |
| days = per_day(snap.users, snap.today, 14) | |
| top = max((n for _, n in days), default=0) or 1 | |
| lines = ["👥 <b>Growth and sources</b>", "<b>New students per day</b>"] | |
| lines += [f"{_fmt_day(d)} {bar(100 * n / top, 12)} {n}" for d, n in days] | |
| f = funnel(snap.users) | |
| lines.append(f"\n<b>Funnel</b> (everyone ever)\njoined <b>{f['joined']}</b> → in channel + forum <b>{f['verified']}</b> → " | |
| f"finished setup <b>{f['onboarded']}</b> → used the bot on 3+ days <b>{f['active3']}</b>") | |
| lines.append("\n<b>By source</b> (joined → in channel+forum → finished setup → 3+ days)") | |
| for r in source_table(snap.users)[:10]: | |
| small = " ·small sample" if r["joined"] < 10 else "" | |
| lines.append(f"• <b>{esc(r['source'])}</b>: {r['joined']} → {r['verified']} → {r['onboarded']} → {r['active3']} " | |
| f"({r['setup_pct']}% setup){small}") | |
| stuck = stuck_by_missing(snap.users) | |
| if stuck: | |
| lines.append("\n<b>Stuck at the join step</b>") | |
| lines += [f"• {n}: {('missing ' + m) if m != 'not recorded yet' else m}" for m, n in stuck[:5]] | |
| return "\n".join(lines) | |
| def section_retention(snap: Snapshot) -> str: | |
| ret = retention(snap.users, snap.checkins, snap.today) | |
| w = dau_wau(snap.checkins, snap.today) | |
| lines = ["🔁 <b>Activity and retention</b>", "<b>Active students per day</b> (opened the bot)"] | |
| peak = max((n for _, n in w["dau"]), default=0) or 1 | |
| lines += [f"{_fmt_day(d)} {bar(100 * n / peak, 12)} {n}" for d, n in w["dau"]] | |
| lines.append(f"\nWeekly active: <b>{w['wau']}</b> · yesterday's active ÷ weekly active: <b>{w['stickiness']}%</b>") | |
| lines.append(f"Came back at least once: <b>{ret['came_back']}</b> of {ret['eligible']} ({pct(ret['came_back'], ret['eligible'])}%)") | |
| lines.append("\n<b>Return rate by signup day</b> (share active again on day 1 / 3 / 7)") | |
| cell = lambda v: "n/a" if v is None else f"{v}%" # noqa: E731 | |
| if ret["rows"]: | |
| lines += [f"{_fmt_day(r['day'])} (n={r['size']}): {cell(r['d1'])} / {cell(r['d3'])} / {cell(r['d7'])}" for r in ret["rows"]] | |
| else: | |
| lines.append("no signups in the last 7 days") | |
| since = ret["tracking_start"].isoformat() if ret["tracking_start"] else "not yet" | |
| lines.append(f"\n“n/a” = that day has not come yet, or it is before activity tracking began ({since}).") | |
| return "\n".join(lines) | |
| def section_library(snap: Snapshot) -> str: | |
| lib = library(snap) | |
| lines = ["📚 <b>Library and searches</b>", | |
| f"{lib['files']} files in {lib['courses']} courses · kinds: " + ", ".join(f"{k} {n}" for k, n in lib["kinds"].most_common()), | |
| f"Downloads: {lib['downloads_total']} all time · {lib['downloads_week']} this week by {lib['downloaders_week']} students"] | |
| if lib["top_files"]: | |
| lines.append("\n<b>Most downloaded</b>") | |
| lines += [f"• {esc(m.title[:45])} ({m.downloads})" for m in lib["top_files"] if m.downloads] | |
| if lib["by_course"]: | |
| lines.append("\n<b>Courses with the most files</b>: " + ", ".join(f"{esc(n[:25])} {c}" for n, c in lib["by_course"])) | |
| lines.append(f"\n<b>No file at all</b> ({len(lib['empty'])}): " + (", ".join(esc(n[:28]) for n in lib["empty"][:8]) or "none") | |
| + ("…" if len(lib["empty"]) > 8 else "")) | |
| if lib["sparse"]: | |
| lines.append(f"<b>Short of files</b> (under {SPARSE}): " + ", ".join(esc(n[:28]) for n in lib["sparse"][:8])) | |
| lines.append(f"\n🔍 <b>Searches (14 days)</b>: {lib['searches']} · found nothing {lib['missed_total']} ({pct(lib['missed_total'], lib['searches'])}%)") | |
| if lib["missed"]: | |
| lines.append("<b>Searched but NOT found</b> (add these!)") | |
| lines += [f"• “{esc(q)}” × {n}" for q, n in lib["missed"]] | |
| if lib["found"]: | |
| lines.append("<b>Most searched and found</b>: " + ", ".join(f"“{esc(q)}” {n}" for q, n in lib["found"])) | |
| return "\n".join(lines) | |
| def section_contrib(snap: Snapshot) -> str: | |
| c = contributions(snap) | |
| lines = ["📤 <b>Student contributions</b>", | |
| f"Waiting <b>{c['pending']}</b> · approved <b>{c['approved']}</b> · rejected <b>{c['rejected']}</b> · approval rate <b>{c['approval']}%</b>"] | |
| if c["avg_hours"] is not None: | |
| lines.append(f"Average time to review: {c['avg_hours']} h") | |
| if c["pending"]: | |
| lines.append(f"Oldest waiting: {c['oldest_pending_hours']:.0f} h") | |
| if c["top"]: | |
| lines.append("Top contributors (user id · approved): " + ", ".join(f"{u} · {n}" for u, n in c["top"])) | |
| if not snap.subs: | |
| lines.append("No student files yet.") | |
| return "\n".join(lines) | |
| def section_channel(snap: Snapshot) -> str: | |
| rows = channel_posts(snap) | |
| lines = ["📢 <b>Channel posts</b>"] | |
| if not rows: | |
| return "\n".join(lines + [ | |
| "No posts recorded yet. The bot records new posts of the channel by itself. To add an older post, " | |
| "FORWARD it (text posts) to this bot as admin."]) | |
| lines.append("Sorted by students who came through the post's own link tag, then by new students within 6 hours.") | |
| for r in rows[:8]: | |
| react = "?" if r["reactions"] is None else r["reactions"] | |
| tag = f" · tag {', '.join(r['tags'])}: <b>{r['tagged']}</b>" if r["tags"] else "" | |
| lines.append(f"• {r['posted'].astimezone(ET).strftime('%m-%d %H:%M')} “{esc(r['preview'][:40])}”\n" | |
| f" 6h: <b>{r['soon']}</b> new{tag} · reactions: {react}") | |
| lines.append("\nA post and a sign-up close in time is a hint, not proof. Reactions are only known for posts seen live.") | |
| return "\n".join(lines) | |
| def section_insights(snap: Snapshot) -> str: | |
| ins = insights(snap) | |
| if not ins: | |
| return "💡 <b>Insights</b>\nNothing stands out yet; there is not enough data." | |
| lines = ["💡 <b>Insights and what to do</b> (from your numbers, most important first)"] | |
| for i, x in enumerate(ins, 1): | |
| lines.append(f"\n<b>{i}. {esc(x.title)}</b>\n{x.evidence}\n➡️ {esc(x.action)}") | |
| return "\n".join(lines) | |