Top kép community voting Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Add a tailnet-only Hungarian community voting flow (Nap → Hét → Hónap) to the existing Frigate picker, storing voted pics on the mainframe and sending Telegram reminders — no Google Drive.

Architecture: New pure-Python store module (top_kep_store.py, stdlib sqlite3 + urllib) holds all data logic and is unit-testable in isolation. Voting routes are added to the existing top4-web.py — which is a stdlib http.server app (class H(BaseHTTPRequestHandler) + ThreadingHTTPServer), NOT Flask (Flask is not installed on the box). Three systemd timers call a reminder CLI that sends via the existing Telegram bot token.

Tech Stack: Python 3 (stdlib only — sqlite3, urllib, zoneinfo, http.server, unittest), systemd, Frigate HTTP API, Telegram Bot API. No new dependencies.

Global Constraints

All code runs on telep-mainframe as user levander. Access: /usr/bin/ssh levander@telep-mainframe.
No git commit anywhere (user rule). Deploy = write file on box, restart service/timer.
No new dependencies — Python stdlib only.
No secrets in code or notes — bot token is read at runtime from /home/levander/nvr/frigate-notify/config.yml (mode 600).
Frigate API base: http://127.0.0.1:5000. Public link base: https://telep-mainframe.taild4189d.ts.net:8443.
Telegram target chat id: -1004475187307.
Timezone for all day/week/month boundaries: Europe/Budapest.
Identity: Tailscale-User-Login request header. /vote returns 403 when absent.
Storage roots: DB /srv/top-kep/votes.db, images /srv/top-kep/img/YYYY-MM-DD/<event_id>.jpg.
User-facing copy is Hungarian.

File Structure

Create /home/levander/top_kep_store.py — all data logic (schema, dedup, Frigate fetch, vote toggle, persist-on-first-vote, window ranking, time bounds, watermark). Pure functions, dependency-injected fetch for testing.
Create /home/levander/top-kep-remind.py — reminder CLI (--daily|--weekly|--monthly|--dry).
Create /home/levander/test_top_kep.py — stdlib unittest self-checks.
Modify /home/levander/top4-web.py — add routes /nap, /het, /honap, /vote, /pic/<event_id> + render_grid helper.
Create /etc/systemd/system/top-kep-remind-{daily,weekly,monthly}.{service,timer} (via sudo).
Modify /home/levander/home-portal/index.html — repoint the “Top képek” tile to …:8443/nap.
Disable top4-export.timer.

Pre-flight (run once before Task 1):

/usr/bin/ssh levander@telep-mainframe 'sudo install -d -o levander -g levander /srv/top-kep /srv/top-kep/img && python3 -c "import zoneinfo, sqlite3; print(\"ok\")"'

Task 1: Store module — schema, dedup, time bounds

Files:

Create: /home/levander/top_kep_store.py
Test: /home/levander/test_top_kep.py

Interfaces:

Produces: get_conn(path=DB_PATH) -> sqlite3.Connection; init_db(conn); dedup_events(events: list[dict]) -> list[dict] (keeps highest top_score per (camera, floor(start_time)), sorted by start_time); day_bounds/week_bounds/month_bounds(now=None) -> (start_ts:int, end_ts:int); tomorrow_is_first(now=None) -> bool; constants DB_PATH, IMG_ROOT, FRIGATE, TZ.
Step 1: Write the failing test

Write /home/levander/test_top_kep.py:

import unittest
from datetime import datetime
from zoneinfo import ZoneInfo
import top_kep_store as s
 
BUD = ZoneInfo("Europe/Budapest")
 
class TestDedup(unittest.TestCase):
    def test_same_second_same_cam_collapses_keeping_top_score(self):
        ev = [
            {"id": "a", "camera": "telep_cam1", "start_time": 1000.1, "top_score": 0.7},
            {"id": "b", "camera": "telep_cam1", "start_time": 1000.9, "top_score": 0.9},
            {"id": "c", "camera": "telep_cam1", "start_time": 1001.2, "top_score": 0.5},
            {"id": "d", "camera": "telep_cam2", "start_time": 1000.4, "top_score": 0.6},
        ]
        out = s.dedup_events(ev)
        ids = [e["id"] for e in out]
        self.assertEqual(ids, ["b", "d", "c"])  # sec1000 cam1->b, sec1000 cam2->d, sec1001 cam1->c; sorted by start_time
 
    def test_score_from_data_field(self):
        ev = [
            {"id": "a", "camera": "c1", "start_time": 5.0, "data": {"top_score": 0.4}},
            {"id": "b", "camera": "c1", "start_time": 5.4, "data": {"top_score": 0.8}},
        ]
        self.assertEqual([e["id"] for e in s.dedup_events(ev)], ["b"])
 
class TestBounds(unittest.TestCase):
    def test_month_bounds_wraps_december(self):
        now = datetime(2026, 12, 15, 12, 0, tzinfo=BUD)
        start, end = s.month_bounds(now)
        self.assertEqual(datetime.fromtimestamp(start, BUD), datetime(2026, 12, 1, tzinfo=BUD))
        self.assertEqual(datetime.fromtimestamp(end, BUD), datetime(2027, 1, 1, tzinfo=BUD))
 
    def test_week_bounds_starts_monday(self):
        now = datetime(2026, 8, 12, 9, 0, tzinfo=BUD)  # Wednesday
        start, _ = s.week_bounds(now)
        self.assertEqual(datetime.fromtimestamp(start, BUD).weekday(), 0)  # Monday
 
    def test_tomorrow_is_first(self):
        self.assertTrue(s.tomorrow_is_first(datetime(2026, 4, 30, 21, 0, tzinfo=BUD)))
        self.assertTrue(s.tomorrow_is_first(datetime(2026, 2, 28, 21, 0, tzinfo=BUD)))
        self.assertFalse(s.tomorrow_is_first(datetime(2026, 2, 27, 21, 0, tzinfo=BUD)))
        self.assertTrue(s.tomorrow_is_first(datetime(2026, 7, 31, 21, 0, tzinfo=BUD)))
 
if __name__ == "__main__":
    unittest.main()

Step 2: Run test to verify it fails

Run: /usr/bin/ssh levander@telep-mainframe 'cd ~ && python3 -m unittest test_top_kep -v' Expected: FAIL — ModuleNotFoundError: No module named 'top_kep_store'

Step 3: Write minimal implementation

Write /home/levander/top_kep_store.py:

import os
import sqlite3
import json
import time
import urllib.request
import urllib.parse
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
 
DB_PATH = "/srv/top-kep/votes.db"
IMG_ROOT = "/srv/top-kep/img"
FRIGATE = "http://127.0.0.1:5000"
TZ = ZoneInfo("Europe/Budapest")
 
 
def init_db(conn):
    conn.executescript(
        """
        CREATE TABLE IF NOT EXISTS pics (
          event_id TEXT PRIMARY KEY,
          camera TEXT NOT NULL,
          captured_ts INTEGER NOT NULL,
          saved_path TEXT,
          first_voted_ts INTEGER
        );
        CREATE TABLE IF NOT EXISTS votes (
          event_id TEXT NOT NULL,
          voter TEXT NOT NULL,
          voted_ts INTEGER NOT NULL,
          PRIMARY KEY (event_id, voter)
        );
        CREATE TABLE IF NOT EXISTS reminder_state (
          kind TEXT PRIMARY KEY,
          last_event_ts INTEGER
        );
        """
    )
    conn.commit()
 
 
def get_conn(path=DB_PATH):
    conn = sqlite3.connect(path)
    conn.row_factory = sqlite3.Row
    init_db(conn)
    return conn
 
 
def _score(e):
    if e.get("top_score") is not None:
        return e["top_score"]
    return (e.get("data") or {}).get("top_score") or 0
 
 
def dedup_events(events):
    best = {}
    for e in events:
        key = (e["camera"], int(e["start_time"]))
        if key not in best or _score(e) > _score(best[key]):
            best[key] = e
    return sorted(best.values(), key=lambda e: e["start_time"])
 
 
def day_bounds(now=None):
    now = now or datetime.now(TZ)
    start = now.replace(hour=0, minute=0, second=0, microsecond=0)
    return int(start.timestamp()), int((start + timedelta(days=1)).timestamp())
 
 
def week_bounds(now=None):
    now = now or datetime.now(TZ)
    start = (now - timedelta(days=now.weekday())).replace(
        hour=0, minute=0, second=0, microsecond=0
    )
    return int(start.timestamp()), int((start + timedelta(days=7)).timestamp())
 
 
def month_bounds(now=None):
    now = now or datetime.now(TZ)
    start = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
    if start.month == 12:
        nxt = start.replace(year=start.year + 1, month=1)
    else:
        nxt = start.replace(month=start.month + 1)
    return int(start.timestamp()), int(nxt.timestamp())
 
 
def tomorrow_is_first(now=None):
    now = now or datetime.now(TZ)
    return (now + timedelta(days=1)).day == 1

Step 4: Run test to verify it passes

Run: /usr/bin/ssh levander@telep-mainframe 'cd ~ && python3 -m unittest test_top_kep -v' Expected: PASS (5 tests)

Step 5: Deploy check

Run: /usr/bin/ssh levander@telep-mainframe 'cd ~ && python3 -c "import top_kep_store as s; c=s.get_conn(\":memory:\"); print(sorted(r[0] for r in c.execute(\"select name from sqlite_master where type=\\\"table\\\"\")))"' Expected: ['pics', 'reminder_state', 'votes']

Task 2: Persist-on-first-vote + vote toggle

Files:

Modify: /home/levander/top_kep_store.py
Test: /home/levander/test_top_kep.py

Interfaces:

Consumes: get_conn, IMG_ROOT, FRIGATE from Task 1.
Produces: persist_pic(conn, event_id, camera, captured_ts, fetch=_fetch_snapshot) -> str (idempotent, returns saved_path); vote_count(conn, event_id) -> int; toggle_vote(conn, event_id, voter, camera, captured_ts, fetch=_fetch_snapshot) -> int (returns new count; persists pic on first-ever vote only).
Step 1: Write the failing test

Append to /home/levander/test_top_kep.py (before the if __name__ line):

import top_kep_store as s2  # same module; alias for clarity
 
class TestVoting(unittest.TestCase):
    def setUp(self):
        self.conn = s.get_conn(":memory:")
        self.calls = []
        self.fetch = lambda eid: (self.calls.append(eid) or b"JPEGBYTES")
        s.IMG_ROOT = "/tmp/top-kep-test-img"
 
    def test_persist_once_then_no_refetch(self):
        p1 = s.persist_pic(self.conn, "e1", "cam1", 1000, fetch=self.fetch)
        p2 = s.persist_pic(self.conn, "e1", "cam1", 1000, fetch=self.fetch)
        self.assertEqual(p1, p2)
        self.assertEqual(self.calls, ["e1"])  # fetched exactly once
 
    def test_toggle_nets_to_zero(self):
        c1 = s.toggle_vote(self.conn, "e1", "alice", "cam1", 1000, fetch=self.fetch)
        self.assertEqual(c1, 1)
        c2 = s.toggle_vote(self.conn, "e1", "alice", "cam1", 1000, fetch=self.fetch)
        self.assertEqual(c2, 0)
 
    def test_distinct_voters_count_independently(self):
        s.toggle_vote(self.conn, "e1", "alice", "cam1", 1000, fetch=self.fetch)
        c = s.toggle_vote(self.conn, "e1", "bob", "cam1", 1000, fetch=self.fetch)
        self.assertEqual(c, 2)
        self.assertEqual(self.calls, ["e1"])  # persisted once, on alice's first vote

Step 2: Run test to verify it fails

Run: /usr/bin/ssh levander@telep-mainframe 'cd ~ && python3 -m unittest test_top_kep.TestVoting -v' Expected: FAIL — AttributeError: module 'top_kep_store' has no attribute 'persist_pic'

Step 3: Write minimal implementation

Append to /home/levander/top_kep_store.py:

def _fetch_snapshot(event_id, base=FRIGATE):
    url = f"{base}/api/events/{event_id}/snapshot.jpg?bbox=0&quality=100"
    with urllib.request.urlopen(url, timeout=30) as r:
        return r.read()
 
 
def persist_pic(conn, event_id, camera, captured_ts, fetch=_fetch_snapshot):
    row = conn.execute(
        "SELECT saved_path FROM pics WHERE event_id=?", (event_id,)
    ).fetchone()
    if row and row["saved_path"] and os.path.exists(row["saved_path"]):
        return row["saved_path"]
    day = time.strftime("%Y-%m-%d", time.localtime(captured_ts))
    d = os.path.join(IMG_ROOT, day)
    os.makedirs(d, exist_ok=True)
    path = os.path.join(d, f"{event_id}.jpg")
    with open(path, "wb") as f:
        f.write(fetch(event_id))
    now = int(time.time())
    conn.execute(
        "INSERT INTO pics(event_id,camera,captured_ts,saved_path,first_voted_ts) "
        "VALUES(?,?,?,?,?) ON CONFLICT(event_id) DO UPDATE SET "
        "saved_path=excluded.saved_path, "
        "first_voted_ts=COALESCE(pics.first_voted_ts, excluded.first_voted_ts)",
        (event_id, camera, captured_ts, path, now),
    )
    conn.commit()
    return path
 
 
def vote_count(conn, event_id):
    return conn.execute(
        "SELECT COUNT(*) c FROM votes WHERE event_id=?", (event_id,)
    ).fetchone()["c"]
 
 
def toggle_vote(conn, event_id, voter, camera, captured_ts, fetch=_fetch_snapshot):
    existing = conn.execute(
        "SELECT 1 FROM votes WHERE event_id=? AND voter=?", (event_id, voter)
    ).fetchone()
    if existing:
        conn.execute(
            "DELETE FROM votes WHERE event_id=? AND voter=?", (event_id, voter)
        )
        conn.commit()
        return vote_count(conn, event_id)
    persist_pic(conn, event_id, camera, captured_ts, fetch)
    conn.execute(
        "INSERT INTO votes(event_id,voter,voted_ts) VALUES(?,?,?)",
        (event_id, voter, int(time.time())),
    )
    conn.commit()
    return vote_count(conn, event_id)

Step 4: Run test to verify it passes

Run: /usr/bin/ssh levander@telep-mainframe 'cd ~ && python3 -m unittest test_top_kep -v' Expected: PASS (all tests)

Task 3: Candidate feed, window ranking, watermark

Files:

Modify: /home/levander/top_kep_store.py
Test: /home/levander/test_top_kep.py

Interfaces:

Consumes: dedup_events, FRIGATE, get_conn from Tasks 1–2.
Produces: day_candidates(day_start, day_end, base=FRIGATE) -> list[dict]; newest_candidate_ts(day_start, day_end, base=FRIGATE) -> int|None; votes_for_window(conn, start_ts, end_ts) -> list[dict] (each {event_id,camera,captured_ts,saved_path,votes}, ordered votes desc then captured_ts desc); get_daily_watermark(conn) -> int|None; set_daily_watermark(conn, ts).
Step 1: Write the failing test

Append to /home/levander/test_top_kep.py:

class TestWindowAndWatermark(unittest.TestCase):
    def setUp(self):
        self.conn = s.get_conn(":memory:")
        s.IMG_ROOT = "/tmp/top-kep-test-img"
        self.fetch = lambda eid: b"J"
 
    def test_window_ranks_by_votes_and_excludes_unvoted(self):
        s.toggle_vote(self.conn, "e1", "a", "cam1", 2000, fetch=self.fetch)
        s.toggle_vote(self.conn, "e2", "a", "cam1", 2100, fetch=self.fetch)
        s.toggle_vote(self.conn, "e2", "b", "cam1", 2100, fetch=self.fetch)
        # e3 persisted but never voted -> must not appear
        s.persist_pic(self.conn, "e3", "cam1", 2200, fetch=self.fetch)
        out = s.votes_for_window(self.conn, 1000, 3000)
        self.assertEqual([r["event_id"] for r in out], ["e2", "e1"])
        self.assertEqual(out[0]["votes"], 2)
 
    def test_window_excludes_out_of_range(self):
        s.toggle_vote(self.conn, "old", "a", "cam1", 500, fetch=self.fetch)
        out = s.votes_for_window(self.conn, 1000, 3000)
        self.assertEqual(out, [])
 
    def test_watermark_roundtrip(self):
        self.assertIsNone(s.get_daily_watermark(self.conn))
        s.set_daily_watermark(self.conn, 12345)
        self.assertEqual(s.get_daily_watermark(self.conn), 12345)
        s.set_daily_watermark(self.conn, 99999)
        self.assertEqual(s.get_daily_watermark(self.conn), 99999)

Step 2: Run test to verify it fails

Run: /usr/bin/ssh levander@telep-mainframe 'cd ~ && python3 -m unittest test_top_kep.TestWindowAndWatermark -v' Expected: FAIL — AttributeError: ... 'votes_for_window'

Step 3: Write minimal implementation

Append to /home/levander/top_kep_store.py:

def _frigate_events(after, before, base=FRIGATE):
    q = urllib.parse.urlencode(
        {"label": "person", "after": after, "before": before,
         "has_snapshot": 1, "limit": 10000}
    )
    with urllib.request.urlopen(f"{base}/api/events?{q}", timeout=30) as r:
        return json.load(r)
 
 
def day_candidates(day_start, day_end, base=FRIGATE):
    return dedup_events(_frigate_events(day_start, day_end, base))
 
 
def newest_candidate_ts(day_start, day_end, base=FRIGATE):
    c = day_candidates(day_start, day_end, base)
    return int(c[-1]["start_time"]) if c else None
 
 
def votes_for_window(conn, start_ts, end_ts):
    rows = conn.execute(
        """
        SELECT p.event_id, p.camera, p.captured_ts, p.saved_path,
               COUNT(v.voter) AS votes
        FROM pics p JOIN votes v ON v.event_id = p.event_id
        WHERE p.captured_ts >= ? AND p.captured_ts < ?
        GROUP BY p.event_id
        HAVING votes > 0
        ORDER BY votes DESC, p.captured_ts DESC
        """,
        (start_ts, end_ts),
    ).fetchall()
    return [dict(r) for r in rows]
 
 
def get_daily_watermark(conn):
    r = conn.execute(
        "SELECT last_event_ts FROM reminder_state WHERE kind='daily'"
    ).fetchone()
    return r["last_event_ts"] if r else None
 
 
def set_daily_watermark(conn, ts):
    conn.execute(
        "INSERT INTO reminder_state(kind,last_event_ts) VALUES('daily',?) "
        "ON CONFLICT(kind) DO UPDATE SET last_event_ts=excluded.last_event_ts",
        (ts,),
    )
    conn.commit()

Step 4: Run test to verify it passes

Run: /usr/bin/ssh levander@telep-mainframe 'cd ~ && python3 -m unittest test_top_kep -v' Expected: PASS (all tests)

Step 5: Live Frigate sanity check

Run: /usr/bin/ssh levander@telep-mainframe 'cd ~ && python3 -c "import top_kep_store as s; a,b=s.day_bounds(); c=s.day_candidates(a,b); print(\"candidates today:\", len(c))"' Expected: prints a small-to-moderate integer (dozens–low hundreds). If it errors on the label param, note the working Frigate query param and fix _frigate_events (Frigate 0.17 accepts label; verify).

Task 4: Voting routes in top4-web.py (stdlib http.server)

Files:

Modify: /home/levander/top4-web.py — add module-level templates + render_grid + import top_kep_store as tks, handler methods on class H, and dispatch branches in do_GET/do_POST
Depends on: top_kep_store (Tasks 1–3)

Context — the existing app (verified on box): top4-web.py is stdlib http.server, NOT Flask. Idioms to match:

class H(BaseHTTPRequestHandler) (~line 363).
self._send(code, ctype, body_bytes, extra=None) — body MUST be bytes.
self._frigate(path, timeout=12) — returns bytes from f"{FRIGATE}{path}"; module-level FRIGATE already defined.
do_GET dispatches with if/elif self.path.startswith(...), ending else: self._send(404, ...).
do_POST dispatches with early if self.path.startswith(...): self._method(); return.
Query: urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query). POST body: n = int(self.headers.get("Content-Length", 0) or 0); self.rfile.read(n). Headers: self.headers.get("Header-Name").
json, urllib.parse, urllib.request, time already imported.

Interfaces:

Consumes: day_bounds, week_bounds, month_bounds, day_candidates, vote_count, votes_for_window, toggle_vote, get_conn, FRIGATE from top_kep_store.
Produces: GET /nap /het /honap (HTML grid), GET /pic/<event_id> (image), POST /vote (form event_id,camera,captured_ts) → JSON {"votes": int} or 403.
Step 1: Confirm imports and locate the dispatch points

Run: /usr/bin/ssh levander@telep-mainframe 'grep -n "^import os\|^import json\|^import urllib\|class H\|def do_GET\|def do_POST\|else:" ~/top4-web.py | head -30' Confirm whether import os is present (add it in Step 2 if not). Note the else: line inside do_GET (the 404 branch) and the first if inside do_POST — the new branches go just before/at those.

Step 2: Add module-level templates + render_grid + store import

Add near the top of /home/levander/top4-web.py (after the existing imports; include import os here if it was missing):

import os
import top_kep_store as tks
 
GRID_HEAD = (
    "<!doctype html><meta charset=utf-8>"
    "<meta name=viewport content='width=device-width,initial-scale=1'>"
    "<title>{title}</title>"
    "<style>body{{font-family:sans-serif;margin:0;background:#111;color:#eee}}"
    "h1{{font-size:1.1rem;padding:.6rem .8rem;margin:0;position:sticky;top:0;background:#000}}"
    "nav{{display:flex;gap:.5rem;padding:.5rem .8rem;background:#000}}"
    "nav a{{color:#8cf;text-decoration:none;font-size:.95rem}}"
    ".g{{display:grid;grid-template-columns:repeat(auto-fill,minmax(150px,1fr));gap:6px;padding:6px}}"
    ".c{{position:relative}} .c img{{width:100%;display:block;border-radius:6px}}"
    ".v{{position:absolute;bottom:6px;left:6px;background:rgba(0,0,0,.6);border:0;color:#fff;"
    "font-size:1rem;padding:.25rem .5rem;border-radius:20px;cursor:pointer}}"
    ".v.on{{background:#c0392b}}</style>"
    "<h1>{title}</h1>"
    "<nav><a href=/nap>Nap</a><a href=/het>Hét</a><a href=/honap>Hónap</a></nav>"
    "<div class=g>"
)
GRID_CARD = (
    "<div class=c><img loading=lazy src='/pic/{event_id}'>"
    "<button class=v data-id='{event_id}' data-cam='{camera}' data-ts='{captured_ts}' "
    "onclick='vote(this)'>❤ <span>{votes}</span></button></div>"
)
GRID_TAIL = (
    "</div><script>"
    "async function vote(b){{const r=await fetch('/vote',{{method:'POST',"
    "headers:{{'Content-Type':'application/x-www-form-urlencoded'}},"
    "body:new URLSearchParams({{event_id:b.dataset.id,camera:b.dataset.cam,"
    "captured_ts:b.dataset.ts}})}});"
    "if(r.status==403){{alert('Csak tailnet eszközről lehet szavazni.');return;}}"
    "const j=await r.json();b.querySelector('span').textContent=j.votes;"
    "b.classList.toggle('on');}}"
    "</script>"
)
 
 
def render_grid(title, items):
    cards = "".join(
        GRID_CARD.format(
            event_id=it["event_id"], camera=it["camera"],
            captured_ts=it["captured_ts"], votes=it["votes"],
        )
        for it in items
    )
    return GRID_HEAD.format(title=title) + cards + GRID_TAIL

Step 3: Add handler methods to class H (e.g. just after _frigate)

    def _read_form(self):
        n = int(self.headers.get("Content-Length", 0) or 0)
        return urllib.parse.parse_qs(self.rfile.read(n).decode("utf-8"))
 
    def _nap(self):
        conn = tks.get_conn()
        a, b = tks.day_bounds()
        items = [
            {"event_id": e["id"], "camera": e["camera"],
             "captured_ts": int(e["start_time"]),
             "votes": tks.vote_count(conn, e["id"])}
            for e in tks.day_candidates(a, b)
        ]
        body = render_grid("Napi képek — szavazz a kedvenceidre", items)
        self._send(200, "text/html; charset=utf-8", body.encode())
 
    def _het(self):
        conn = tks.get_conn()
        a, b = tks.week_bounds()
        body = render_grid("Heti szavazás", tks.votes_for_window(conn, a, b))
        self._send(200, "text/html; charset=utf-8", body.encode())
 
    def _honap(self):
        conn = tks.get_conn()
        a, b = tks.month_bounds()
        body = render_grid("Havi szavazás", tks.votes_for_window(conn, a, b))
        self._send(200, "text/html; charset=utf-8", body.encode())
 
    def _pic(self, eid):
        conn = tks.get_conn()
        row = conn.execute(
            "SELECT saved_path FROM pics WHERE event_id=?", (eid,)
        ).fetchone()
        if row and row["saved_path"] and os.path.exists(row["saved_path"]):
            with open(row["saved_path"], "rb") as f:
                self._send(200, "image/jpeg", f.read())
        else:
            self._send(200, "image/jpeg",
                       self._frigate(f"/api/events/{eid}/snapshot.jpg?bbox=0&quality=70"))
 
    def _vote(self):
        voter = self.headers.get("Tailscale-User-Login")
        if not voter:
            self._send(403, "application/json",
                       b'{"error":"tailnet identity required"}')
            return
        form = self._read_form()
        eid = form.get("event_id", [""])[0]
        camera = form.get("camera", ["?"])[0]
        try:
            captured_ts = int(form.get("captured_ts", ["0"])[0])
        except ValueError:
            captured_ts = 0
        conn = tks.get_conn()
        n = tks.toggle_vote(conn, eid, voter, camera, captured_ts)
        self._send(200, "application/json", json.dumps({"votes": n}).encode())

Step 4: Wire the dispatch branches

In do_GET, add just before the final else: (404 branch):

            elif self.path.startswith("/nap"):
                self._nap()
            elif self.path.startswith("/honap"):
                self._honap()
            elif self.path.startswith("/het"):
                self._het()
            elif self.path.startswith("/pic/"):
                eid = self.path.split("/pic/")[1].split("?")[0]
                self._pic(eid)

In do_POST, add as the FIRST dispatch line:

        if self.path.startswith("/vote"):
            self._vote()
            return

Step 5: Restart the service

Run: /usr/bin/ssh levander@telep-mainframe 'sudo systemctl restart top4-web.service && sleep 1 && systemctl is-active top4-web.service' Expected: active. If failed: journalctl -u top4-web.service -n 30 --no-pager — most likely an indentation slip inside class H or a missing import os.

Step 6: Integration-verify routes + identity gate

Run:

/usr/bin/ssh levander@telep-mainframe 'curl -s -o /dev/null -w "%{http_code}\n" http://127.0.0.1:8090/nap; \
  curl -s -o /dev/null -w "%{http_code}\n" -X POST http://127.0.0.1:8090/vote -d "event_id=x&camera=c&captured_ts=1"'

Expected: first line 200 (grid renders); second line 403 (no Tailscale header → identity gate works).

Step 7: End-to-end vote with an identity header

Run:

/usr/bin/ssh levander@telep-mainframe 'ID=$(python3 -c "import top_kep_store as s;a,b=s.day_bounds();c=s.day_candidates(a,b);print(c[0][\"id\"] if c else \"\")"); \
  echo "event=$ID"; curl -s -X POST http://127.0.0.1:8090/vote -H "Tailscale-User-Login: test@local" \
  -d "event_id=$ID&camera=cam1&captured_ts=$(date +%s)"; echo; ls -R /srv/top-kep/img | head'

Expected: JSON {"votes":1} and a saved .jpg under /srv/top-kep/img/<today>/. Then clean the test vote: /usr/bin/ssh levander@telep-mainframe 'python3 -c "import top_kep_store as s; c=s.get_conn(); c.execute(\"DELETE FROM votes WHERE voter=?\", (\"test@local\",)); c.commit()"'

Task 5: Reminder CLI

Files:

Create: /home/levander/top-kep-remind.py
Test: /home/levander/test_top_kep.py

Interfaces:

Consumes: day_bounds, newest_candidate_ts, get_conn, get_daily_watermark, set_daily_watermark, tomorrow_is_first from top_kep_store.
Produces: CLI python3 top-kep-remind.py [--daily|--weekly|--monthly] [--dry]. --dry prints the message instead of sending. Module-level MESSAGES dict for test.
Step 1: Write the failing test

Append to /home/levander/test_top_kep.py:

class TestReminderMessages(unittest.TestCase):
    def test_messages_are_hungarian_and_link_correct_tier(self):
        import importlib
        r = importlib.import_module("top_kep_remind")
        self.assertIn("/nap", r.MESSAGES["daily"])
        self.assertIn("/het", r.MESSAGES["weekly"])
        self.assertIn("/honap", r.MESSAGES["monthly"])
        self.assertIn("szavazz", r.MESSAGES["daily"].lower())

Note: the test imports top_kep_remind (underscore), so create the file with an underscore name and a hyphen symlink for systemd readability — see Step 3.

Step 2: Run test to verify it fails

Run: /usr/bin/ssh levander@telep-mainframe 'cd ~ && python3 -m unittest test_top_kep.TestReminderMessages -v' Expected: FAIL — ModuleNotFoundError: No module named 'top_kep_remind'

Step 3: Write minimal implementation

Write /home/levander/top_kep_remind.py:

#!/usr/bin/env python3
import sys
import re
import urllib.request
import urllib.parse
import top_kep_store as s
 
CHAT_ID = "-1004475187307"
BASE_URL = "https://telep-mainframe.taild4189d.ts.net:8443"
TOKEN_FILE = "/home/levander/nvr/frigate-notify/config.yml"
 
MESSAGES = {
    "daily": f"📸 Vannak új képek a mai napról — szavazz a kedvenceidre: {BASE_URL}/nap",
    "weekly": f"🗓️ Heti szavazás — válaszd ki a hét legjobb képeit: {BASE_URL}/het",
    "monthly": f"📅 Havi szavazás — a hónap legjobb képei: {BASE_URL}/honap",
}
 
 
def bot_token():
    txt = open(TOKEN_FILE).read()
    m = re.search(r'token:\s*["\']?([0-9]{6,}:[A-Za-z0-9_-]{30,})', txt)
    if not m:
        raise SystemExit("bot token not found in " + TOKEN_FILE)
    return m.group(1)
 
 
def send(text, dry=False):
    if dry:
        print(text)
        return
    data = urllib.parse.urlencode(
        {"chat_id": CHAT_ID, "text": text, "parse_mode": "HTML"}
    ).encode()
    url = f"https://api.telegram.org/bot{bot_token()}/sendMessage"
    with urllib.request.urlopen(urllib.request.Request(url, data=data), timeout=30) as r:
        r.read()
 
 
def daily(dry=False):
    a, b = s.day_bounds()
    newest = s.newest_candidate_ts(a, b)
    if newest is None:
        return
    conn = s.get_conn()
    wm = s.get_daily_watermark(conn)
    if wm is not None and newest <= wm:
        return
    s.set_daily_watermark(conn, newest)
    send(MESSAGES["daily"], dry)
 
 
def weekly(dry=False):
    send(MESSAGES["weekly"], dry)
 
 
def monthly(dry=False):
    if not s.tomorrow_is_first():
        return
    send(MESSAGES["monthly"], dry)
 
 
if __name__ == "__main__":
    dry = "--dry" in sys.argv
    fn = {"--daily": daily, "--weekly": weekly, "--monthly": monthly}
    for a in sys.argv[1:]:
        if a in fn:
            fn[a](dry)
            break
    else:
        raise SystemExit("usage: top_kep_remind.py --daily|--weekly|--monthly [--dry]")

Create the hyphen alias systemd will call:

/usr/bin/ssh levander@telep-mainframe 'ln -sf ~/top_kep_remind.py ~/top-kep-remind.py'

Step 4: Run test to verify it passes

Run: /usr/bin/ssh levander@telep-mainframe 'cd ~ && python3 -m unittest test_top_kep -v' Expected: PASS (all tests)

Step 5: Verify token parse + dry sends (no real message)

Run:

/usr/bin/ssh levander@telep-mainframe 'cd ~ && python3 -c "import top_kep_remind as r; print(bool(r.bot_token()))" && \
  python3 top_kep_remind.py --weekly --dry && python3 top_kep_remind.py --daily --dry'

Expected: True, then the weekly Hungarian line, then the daily line (or nothing for daily if no candidates today). If bot_token() fails, inspect the telegram:/token: shape in the config and adjust the regex.

Step 6: One real weekly send to confirm the group wiring

Run: /usr/bin/ssh levander@telep-mainframe 'cd ~ && python3 top_kep_remind.py --weekly' Expected: the message appears in the 『Telephely biztonsági riasztások』 group. (One-time confirmation; the timers take over after.)

Task 6: systemd timers + disable Drive export

Files:

Create (sudo): /etc/systemd/system/top-kep-remind-daily.service + .timer, top-kep-remind-weekly.service + .timer, top-kep-remind-monthly.service + .timer
Disable: top4-export.timer

Interfaces:

Consumes: /home/levander/top_kep_remind.py (Task 5).
Step 1: Write the units

Create the three service files (identical except the flag). top-kep-remind-daily.service:

[Unit]
Description=Top kép daily voting reminder
After=network-online.target top4-web.service
 
[Service]
Type=oneshot
User=levander
ExecStart=/usr/bin/python3 /home/levander/top_kep_remind.py --daily

top-kep-remind-weekly.service and top-kep-remind-monthly.service: same, with --weekly / --monthly and matching Description.

Timers — top-kep-remind-daily.timer:

[Unit]
Description=Top kép daily reminder at 20:00
[Timer]
OnCalendar=*-*-* 20:00:00
Persistent=true
[Install]
WantedBy=timers.target

top-kep-remind-weekly.timer:

[Unit]
Description=Top kép weekly reminder Sun 19:30
[Timer]
OnCalendar=Sun *-*-* 19:30:00
Persistent=true
[Install]
WantedBy=timers.target

top-kep-remind-monthly.timer (fires daily 21:00; the script’s tomorrow_is_first guard makes it act monthly):

[Unit]
Description=Top kép monthly reminder (guarded, month-end 21:00)
[Timer]
OnCalendar=*-*-* 21:00:00
Persistent=true
[Install]
WantedBy=timers.target

Write them, e.g.:

/usr/bin/ssh levander@telep-mainframe 'sudo tee /etc/systemd/system/top-kep-remind-daily.timer >/dev/null <<'"'"'EOF'"'"'
[Unit]
Description=Top kép daily reminder at 20:00
[Timer]
OnCalendar=*-*-* 20:00:00
Persistent=true
[Install]
WantedBy=timers.target
EOF'

(Repeat for each service + timer.)

Step 2: Enable timers, disable Drive export

Run:

/usr/bin/ssh levander@telep-mainframe 'sudo systemctl daemon-reload && \
  sudo systemctl enable --now top-kep-remind-daily.timer top-kep-remind-weekly.timer top-kep-remind-monthly.timer && \
  sudo systemctl disable --now top4-export.timer'

Expected: three timers enabled; top4-export.timer disabled (ignore “not found” only if it truly doesn’t exist — confirm with Step 3).

Step 3: Verify the timer schedule

Run: /usr/bin/ssh levander@telep-mainframe 'systemctl list-timers "top-kep-*" top4-export.timer --all --no-pager' Expected: three top-kep-remind-* timers with sane NEXT times (20:00, Sun 19:30, 21:00); top4-export.timer absent from active list (disabled).

Task 7: Repoint the home-portal tile

Files:

Modify: /home/levander/home-portal/index.html
Step 1: Back up and inspect the current tile

Run: /usr/bin/ssh levander@telep-mainframe 'cp ~/home-portal/index.html ~/home-portal/index.html.bak-$(date +%s) 2>/dev/null || true; grep -n "Top képek\|8443\|top4\|top-kép" ~/home-portal/index.html' Note the exact href of the existing “Top képek” tile.

Step 2: Repoint the tile href to /nap

Edit /home/levander/home-portal/index.html: change the “Top képek” tile’s href to https://telep-mainframe.taild4189d.ts.net:8443/nap. Leave label as “Top képek” (or “Top képek – szavazás” if desired).

Step 3: Verify

Run: /usr/bin/ssh levander@telep-mainframe 'grep -n "8443/nap" ~/home-portal/index.html && curl -s -o /dev/null -w "%{http_code}\n" http://127.0.0.1:8093/' Expected: the grep matches the new href; portal returns 200. (Static page served by python -m http.server needs no restart.)

Self-Review

Spec coverage:

Extend existing stdlib http.server picker, tailnet — Task 4. ✓
Tailscale-User-Login identity, 403 when absent — Task 4 (/vote), verified Task 4 Step 4. ✓
Daily votes roll up; Nap/Hét/Hónap tiers — Tasks 3–4 (day_candidates, votes_for_window). ✓
Candidate set = person events deduped one/second/camera — Task 1 (dedup_events) + Task 3 (day_candidates). ✓
Save on first vote, native-res, local path scheme — Task 2 (persist_pic, toggle_vote). ✓
SQLite schema exactly as specced — Task 1 (init_db). ✓
Reminders daily(watermark)/weekly(Sun 19:30)/monthly(month-end 21:00), Hungarian, existing bot+group — Tasks 5–6. ✓
Daily reminder only if new since last — Task 5 (daily + watermark), Task 3 (watermark helpers). ✓
Month-end guard covers 28/30/31 — Task 1 (tomorrow_is_first), tested. ✓
Disable top4-export.timer; leave old Drive endpoints/4-slot flow untouched — Task 6 Step 2; no task modifies them. ✓
Home-portal tile repoint — Task 7. ✓
One runnable self-check covering dedup/toggle/persist-once/month-guard/window/watermark — test_top_kep.py across Tasks 1–3, 5. ✓

Placeholder scan: No TBD/TODO; all code blocks concrete. The only runtime-verify points are flagged as explicit checks (Frigate label param — Task 3 Step 5; Flask instance name — Task 4 Step 1; token regex — Task 5 Step 5), each with a stated fallback.

Type consistency: event_id/camera/captured_ts/votes keys are consistent across votes_for_window (Task 3), render_grid/routes (Task 4), and toggle_vote (Task 2). Frigate raw events use id/start_time, mapped to event_id/captured_ts at the route boundary (Task 4 /nap). fetch injection signature fetch(event_id) -> bytes consistent in Tasks 2–3 tests and persist_pic.

Open / deferred (from spec)

Daily reminder time 20:00 — adjustable in the timer.
Full Drive rip-out (endpoints, helpers, config) — deferred.
Print/export off the local store — deferred (“stored until further instruction”).
Optional: mirror new files into the wowjeeez/telep-infra snapshot repo — deferred (no auto-commit).

2026-08-10-top-kep-community-voting-spec — the design spec this implements
2026-08-10-top-kepek-drive-curation-print-pipeline · telep-mainframe · telep-mainframe-handover

Levandor

Explorer

Top kép community voting — implementation plan

Top kép community voting Implementation Plan

Global Constraints

File Structure

Task 1: Store module — schema, dedup, time bounds

Task 2: Persist-on-first-vote + vote toggle

Task 3: Candidate feed, window ranking, watermark

Task 4: Voting routes in top4-web.py (stdlib http.server)

Task 5: Reminder CLI

Task 6: systemd timers + disable Drive export

Task 7: Repoint the home-portal tile

Self-Review

Open / deferred (from spec)

Graph View

Table of Contents

Backlinks

Levandor

Explorer

Top kép community voting — implementation plan

Top kép community voting Implementation Plan

Global Constraints

File Structure

Task 1: Store module — schema, dedup, time bounds

Task 2: Persist-on-first-vote + vote toggle

Task 3: Candidate feed, window ranking, watermark

Task 4: Voting routes in top4-web.py (stdlib http.server)

Task 5: Reminder CLI

Task 6: systemd timers + disable Drive export

Task 7: Repoint the home-portal tile

Self-Review

Open / deferred (from spec)

Related

Graph View

Table of Contents

Backlinks