From 098e375d6fee5e2aca584d7f62d52f8ee84a7f13 Mon Sep 17 00:00:00 2001 From: Constantin Gemmingen <49587165+constgemm@users.noreply.github.com> Date: Tue, 7 Jul 2026 23:42:12 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20Phase=204=20=E2=80=94=20Telegram=20inte?= =?UTF-8?q?ractive=20approval?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Moves the propose-then-approve gate from the terminal (--commit) to the phone. For each NEW candidate not contradicted by Dawarich, telegram.py sends a card showing every field (missing ones flagged) with an Approve/Skip inline button; on Approve it writes the flight to AirTrail and replies with the new total ("AirTrail now has N flights"). Skip or timeout writes nothing. Uses long-poll getUpdates, so the container needs no inbound port. callback_data is namespaced (lay:ok:) and decisions are buffered by index, so stray or out-of-order taps are handled. All HTTP + AirTrail post/fetch are injectable. - populate.py: opt-in approval step (env TELEGRAM_APPROVE or --telegram-approve). - Dockerfile copies telegram.py; .env.example + README document it (token is a SECRET — .env only). - 10 tests: card rendering, eligibility filter, approve/skip/timeout/failure paths, enable gating. Full suite: 39 green. Co-Authored-By: Claude Opus 4.8 --- .env.example | 8 ++ Dockerfile | 2 +- README.md | 6 ++ populate.py | 17 +++ telegram.py | 236 +++++++++++++++++++++++++++++++++++++++++ tests/test_telegram.py | 155 +++++++++++++++++++++++++++ 6 files changed, 423 insertions(+), 1 deletion(-) create mode 100644 telegram.py create mode 100644 tests/test_telegram.py diff --git a/.env.example b/.env.example index 89f041d..1baf3d6 100644 --- a/.env.example +++ b/.env.example @@ -24,3 +24,11 @@ RUN_ON_START=true # also run once when the container starts # Writes only high-confidence, non-rebooking NEW flights straight to AirTrail; # anything uncertain or flagged still waits for `populate.py --commit`. # AUTO_WRITE=false + +# --- Telegram approval (Phase 4, optional, opt-in) --- +# For each NEW flight (not contradicted by Dawarich), sends a card showing every +# field with an Approve/Skip button; on Approve it writes to AirTrail and replies +# with the new total. Long-poll, so no inbound port. The token is a SECRET. +# TELEGRAM_APPROVE=false +# TELEGRAM_BOT_TOKEN=REPLACE_ME # from @BotFather +# TELEGRAM_CHAT_ID=REPLACE_ME # your chat/user id diff --git a/Dockerfile b/Dockerfile index bbdebad..10bb827 100644 --- a/Dockerfile +++ b/Dockerfile @@ -13,7 +13,7 @@ ENV PYTHONUNBUFFERED=1 \ WORKDIR /app # Copy only the code (no secrets, no output — see .dockerignore). -COPY layover.py flightparse.py airdata.py airtrail.py notify.py populate.py scheduler.py ./ +COPY layover.py flightparse.py airdata.py airtrail.py notify.py telegram.py populate.py scheduler.py ./ # Run unprivileged; /data (state.json watermark + candidates) is a persisted volume. RUN useradd --system --uid 10001 --home-dir /app layover \ diff --git a/README.md b/README.md index 82d8b80..f9860b2 100644 --- a/README.md +++ b/README.md @@ -268,6 +268,12 @@ Phase 1 (above) is deterministic templates + digest, no LLM in the loop. Where t - **Phase 3 — location-history validation:** cross-check candidates against [Dawarich](https://github.com/Freika/dawarich) point history to auto-confirm a flight or flag a cancelled/rebooked one (the signal a phone app gets, self-hosted). +- **Phase 4 — Telegram approval** (shipped, opt-in): the propose-then-approve gate, moved from the + terminal to your phone. For each new, non-contradicted flight it sends a card with every field + (missing ones flagged) and an **Approve / Skip** button; on Approve it writes to AirTrail and + replies with the new total. Long-poll getUpdates, so the container needs no inbound port. Enable + with `TELEGRAM_APPROVE=1` + `TELEGRAM_BOT_TOKEN`/`TELEGRAM_CHAT_ID` or `populate.py + --telegram-approve`. - **Miles & More** stays a periodic manual cross-check — highest-precision source, used as a completeness checksum, not the engine. diff --git a/populate.py b/populate.py index b1cb839..1545d1a 100644 --- a/populate.py +++ b/populate.py @@ -42,6 +42,7 @@ import airtrail import flightparse import notify as notify_mod +import telegram def pull(accounts_ini, out_dir): @@ -248,6 +249,11 @@ def main(): ap.add_argument("--webhook-url", help="POST a JSON digest here when new flights " "are found (or set WEBHOOK_URL)") ap.add_argument("--user-id", help="AirTrail user id to assign seats to on writes") + ap.add_argument("--telegram-approve", dest="telegram", action="store_true", + default=None, help="ask per-flight approval over Telegram and " + "write the approved ones (also via TELEGRAM_APPROVE=1)") + ap.add_argument("--no-telegram-approve", dest="telegram", action="store_false", + help="skip Telegram approval even if configured") args = ap.parse_args() if args.pull: @@ -273,6 +279,17 @@ def main(): notify_cmd=os.environ.get("NOTIFY_CMD"), ) + if args.telegram if args.telegram is not None else telegram.enabled(): + url, key = airtrail.load_config(args.url, args.api_key) + token, chat = telegram.load_config() + if token and chat: + n = telegram.run_approvals(cands, url, key, token, chat, + user_id=args.user_id) + print(f"[telegram] {n} flight(s) approved & written", file=sys.stderr) + else: + print("[telegram] TELEGRAM_BOT_TOKEN / TELEGRAM_CHAT_ID not set", + file=sys.stderr) + if args.commit: commit(cands, args) elif args.auto_write or os.environ.get("AUTO_WRITE", "").strip().lower() in ( diff --git a/telegram.py b/telegram.py new file mode 100644 index 0000000..638c596 --- /dev/null +++ b/telegram.py @@ -0,0 +1,236 @@ +#!/usr/bin/env python3 +""" +telegram — interactive per-flight approval over a Telegram bot (Phase 4). + +The last gate before a write. For each NEW candidate that Dawarich didn't +contradict, it sends you a message showing every field (missing ones flagged) with +an inline "✅ Approve / ⏭ Skip" keyboard, then waits for your tap. On Approve it +writes the flight to AirTrail and replies with the new total ("AirTrail now has N +flights"). On Skip (or timeout) it writes nothing. + +This is the human in propose-then-approve, moved from the terminal (`--commit`) to +your phone. It uses long-poll getUpdates, so the container needs no inbound port. + +Config (env, or populate.py flags): + TELEGRAM_BOT_TOKEN bot token from @BotFather (SECRET — .env, never committed) + TELEGRAM_CHAT_ID chat/user id to message and accept taps from + TELEGRAM_APPROVE truthy to enable the approval step + +Zero dependencies (urllib + json). All HTTP is injectable for testing. +""" + +import json +import os +import time +import urllib.request + +import airtrail + +API = "https://api.telegram.org" +PREFIX = "lay" # callback_data namespace so we ignore other bots' taps + + +# --------------------------------------------------------------------------- +# config +# --------------------------------------------------------------------------- + +def load_config(token=None, chat_id=None): + return (token or os.environ.get("TELEGRAM_BOT_TOKEN"), + chat_id or os.environ.get("TELEGRAM_CHAT_ID")) + + +def enabled(token=None, chat_id=None): + on = os.environ.get("TELEGRAM_APPROVE", "").strip().lower() in ( + "1", "true", "yes", "on") + t, c = load_config(token, chat_id) + return bool(on and t and c) + + +# --------------------------------------------------------------------------- +# transport (Telegram Bot API) — thin, and injectable via `api` +# --------------------------------------------------------------------------- + +def api_call(token, method, params=None, timeout=35): + url = f"{API}/bot{token}/{method}" + data = json.dumps(params or {}).encode() + req = urllib.request.Request(url, data=data, method="POST", + headers={"Content-Type": "application/json"}) + with urllib.request.urlopen(req, timeout=timeout) as resp: + return json.loads(resp.read().decode()) + + +def send_message(token, chat_id, text, reply_markup=None, api=api_call): + params = {"chat_id": chat_id, "text": text} + if reply_markup: + params["reply_markup"] = reply_markup + return api(token, "sendMessage", params) + + +def get_updates(token, offset=None, timeout=25, api=api_call): + params = {"timeout": timeout, "allowed_updates": ["callback_query"]} + if offset is not None: + params["offset"] = offset + return api(token, "getUpdates", params, timeout=timeout + 10).get("result", []) + + +def answer_callback(token, cb_id, text=None, api=api_call): + params = {"callback_query_id": cb_id} + if text: + params["text"] = text + return api(token, "answerCallbackQuery", params) + + +# --------------------------------------------------------------------------- +# message rendering +# --------------------------------------------------------------------------- + +def _val(v): + return str(v) if v not in (None, "", []) else "⚠ missing" + + +def format_candidate(c): + """A human-readable card of every field, missing ones flagged.""" + airline = c.get("airline") + if airline and c.get("airline_iata"): + airline = f"{airline} ({c['airline_iata']})" + frm = c.get("from") + if frm and c.get("from_iata"): + frm = f"{frm} ({c['from_iata']})" + to = c.get("to") + if to and c.get("to_iata"): + to = f"{to} ({c['to_iata']})" + seat = c.get("seatNumber") + if seat and c.get("seat"): + seat = f"{seat} ({c['seat']})" + loc = {"confirmed": "📍 confirmed by Dawarich", + "contradicted": "📍 NOT seen near airport (!)", + "unknown": "location unverified"}.get(c.get("location"), "not checked") + + lines = [ + "✈️ New flight — approve?", + f"Flight: {_val(c.get('flightNumber'))}", + f"Airline: {_val(airline)}", + f"From: {_val(frm)}", + f"To: {_val(to)}", + f"Date: {_val(c.get('date'))}", + f"Departure: {_val(c.get('departure'))}", + f"Arrival: {_val(c.get('arrival'))}", + f"Seat: {_val(seat)}", + f"Class: {_val(c.get('seatClass'))}", + f"Location: {loc}", + f"Source: {_val(c.get('source_file'))} [{c.get('extractor')}]", + ] + for issue in c.get("issues", []): + lines.append(f"⚠ {issue}") + return "\n".join(lines) + + +def approval_keyboard(idx): + return {"inline_keyboard": [[ + {"text": "✅ Approve", "callback_data": f"{PREFIX}:ok:{idx}"}, + {"text": "⏭ Skip", "callback_data": f"{PREFIX}:no:{idx}"}, + ]]} + + +# --------------------------------------------------------------------------- +# approval session +# --------------------------------------------------------------------------- + +def eligible(cands): + """NEW candidates worth asking about: not a Dawarich contradiction.""" + return [c for c in cands + if c.get("status") == "new" and c.get("location") != "contradicted"] + + +class ApprovalBot: + """Sends approval cards and collects button taps via long-poll getUpdates. + + Buttons carry `lay:ok:` / `lay:no:`, so taps are matched to the exact + card and stray taps from other bots in the chat are ignored. Decisions are + buffered by index, so an out-of-order tap is never lost.""" + + def __init__(self, token, chat_id, api=api_call): + self.token = token + self.chat_id = chat_id + self.api = api + self.offset = None + self.decisions = {} + + def send(self, text, keyboard=None): + return send_message(self.token, self.chat_id, text, keyboard, api=self.api) + + def prime(self): + """Drain any pending updates so an old tap doesn't leak into this run.""" + for u in get_updates(self.token, None, 0, api=self.api): + self.offset = u["update_id"] + 1 + + def _pump(self, timeout): + for u in get_updates(self.token, self.offset, timeout, api=self.api): + self.offset = u["update_id"] + 1 + cq = u.get("callback_query") + if not cq: + continue + parts = (cq.get("data") or "").split(":") + if len(parts) == 3 and parts[0] == PREFIX: + verb, idx = parts[1], parts[2] + self.decisions[idx] = "approve" if verb == "ok" else "skip" + answer_callback(self.token, cq["id"], + "Recorded ✅" if verb == "ok" else "Skipped", + api=self.api) + + def wait(self, idx, timeout_s=600, poll=25): + """Block until the card `idx` is tapped, or timeout. Returns + 'approve' / 'skip' / None (timeout).""" + idx = str(idx) + waited = 0 + while idx not in self.decisions and waited < timeout_s: + self._pump(poll) + waited += poll + return self.decisions.get(idx) + + +def _flight_count(fetch_flights, url, key): + try: + return len(fetch_flights(url, key)) + except Exception: # noqa: BLE001 - count is a nicety, never fail the write on it + return None + + +def run_approvals(cands, airtrail_url, airtrail_key, token, chat_id, user_id=None, + timeout_s=600, poll=25, api=api_call, + post_flight=airtrail.post_flight, + fetch_flights=airtrail.fetch_flights, + build_payload=airtrail.build_payload): + """Ask about each eligible NEW flight and write the approved ones. Returns the + number written. Sends one card at a time and waits for its tap.""" + items = eligible(cands) + if not items: + return 0 + if not (airtrail_url and airtrail_key): + send_message(token, chat_id, + "Layover: AirTrail URL/key missing — cannot write approvals.", + api=api) + return 0 + bot = ApprovalBot(token, chat_id, api=api) + bot.prime() + bot.send(f"Layover found {len(items)} new flight(s) to review.") + written = 0 + for i, c in enumerate(items): + bot.send(format_candidate(c), approval_keyboard(i)) + decision = bot.wait(i, timeout_s, poll) + if decision == "approve": + try: + status, _ = post_flight(airtrail_url, airtrail_key, + build_payload(c, user_id=user_id)) + written += 1 + n = _flight_count(fetch_flights, airtrail_url, airtrail_key) + tail = f" AirTrail now has {n} flights." if n is not None else "" + bot.send(f"✅ Wrote {c.get('flightNumber')} " + f"{c.get('from')}->{c.get('to')} (HTTP {status}).{tail}") + except Exception as exc: # noqa: BLE001 + bot.send(f"⚠ Failed to write {c.get('flightNumber')}: {exc}") + elif decision == "skip": + bot.send(f"⏭ Skipped {c.get('flightNumber')}.") + else: + bot.send(f"⌛ No response for {c.get('flightNumber')} — left for later.") + return written diff --git a/tests/test_telegram.py b/tests/test_telegram.py new file mode 100644 index 0000000..6b1239b --- /dev/null +++ b/tests/test_telegram.py @@ -0,0 +1,155 @@ +#!/usr/bin/env python3 +""" +Unit tests for telegram.py — Phase 4 interactive approval. + +No network: the Telegram Bot API is a scripted fake injected as `api`, and AirTrail +post/fetch are injected too. Contract under test: cards render every field (missing +flagged), Approve writes + confirms with the new count, Skip/timeout write nothing, +and only eligible (new, not Dawarich-contradicted) flights are ever offered. +""" + +import os +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +import telegram # noqa: E402 + + +def cand(**kw): + base = {"status": "new", "location": "confirmed", "flightNumber": "LX1246", + "airline": "SWR", "airline_iata": "LX", "from": "LSZH", "to": "ESSA", + "from_iata": "ZRH", "to_iata": "ARN", "date": "2022-03-17", + "departure": "2022-03-17T08:00", "arrival": None, "seatNumber": "12A", + "seat": "window", "seatClass": None, "source_file": "acct/x.txt", + "extractor": "jsonld", "issues": []} + base.update(kw) + return base + + +class FakeAPI: + """Scripted Telegram transport. `update_batches` are returned by successive + getUpdates calls (after the first priming call).""" + + def __init__(self, update_batches=None): + self.sent = [] + self.answered = [] + self._batches = list(update_batches or []) + self._primed = False + + def __call__(self, token, method, params=None, timeout=35): + if method == "sendMessage": + self.sent.append(params) + return {"ok": True, "result": {"message_id": len(self.sent)}} + if method == "getUpdates": + if not self._primed: # first call = prime (drain) + self._primed = True + return {"result": []} + batch = self._batches.pop(0) if self._batches else [] + return {"result": batch} + if method == "answerCallbackQuery": + self.answered.append(params) + return {"ok": True} + return {"ok": True, "result": {}} + + +def cb(update_id, idx, verb="ok"): + return {"update_id": update_id, + "callback_query": {"id": f"q{update_id}", + "data": f"{telegram.PREFIX}:{verb}:{idx}"}} + + +class TestRender(unittest.TestCase): + def test_missing_fields_flagged(self): + text = telegram.format_candidate(cand(arrival=None, seatClass=None)) + self.assertIn("Arrival: ⚠ missing", text) + self.assertIn("Class: ⚠ missing", text) + self.assertIn("LX1246", text) + self.assertIn("ZRH", text) # from_iata shown alongside ICAO + + def test_keyboard_data(self): + kb = telegram.approval_keyboard(3) + row = kb["inline_keyboard"][0] + self.assertEqual(row[0]["callback_data"], "lay:ok:3") + self.assertEqual(row[1]["callback_data"], "lay:no:3") + + +class TestEligible(unittest.TestCase): + def test_filters_to_new_non_contradicted(self): + cands = [cand(), cand(status="duplicate"), cand(status="uncertain"), + cand(location="contradicted")] + elig = telegram.eligible(cands) + self.assertEqual(len(elig), 1) + self.assertEqual(elig[0]["status"], "new") + + +class TestRunApprovals(unittest.TestCase): + def _run(self, batches, **kw): + api = FakeAPI(batches) + writes = [] + post = kw.get("post", lambda u, k, p: (writes.append(p) or (200, "ok"))) + fetch = kw.get("fetch", lambda u, k: [0] * 42) + n = telegram.run_approvals( + [cand()], "http://at", "key", "TOKEN", "CHAT", + timeout_s=5, poll=1, api=api, post_flight=post, fetch_flights=fetch, + build_payload=lambda c, user_id=None: {"flightNumber": c["flightNumber"]}) + return n, api, writes + + def test_approve_writes_and_confirms_count(self): + n, api, writes = self._run([[cb(10, 0, "ok")]]) + self.assertEqual(n, 1) + self.assertEqual(len(writes), 1) + # a confirmation mentioning the new total was sent + self.assertTrue(any("42 flights" in s.get("text", "") for s in api.sent)) + self.assertTrue(api.answered) # the tap was acknowledged + + def test_skip_writes_nothing(self): + n, api, writes = self._run([[cb(11, 0, "no")]]) + self.assertEqual(n, 0) + self.assertEqual(writes, []) + self.assertTrue(any("Skipped" in s.get("text", "") for s in api.sent)) + + def test_timeout_writes_nothing(self): + n, api, writes = self._run([]) # no taps ever arrive + self.assertEqual(n, 0) + self.assertEqual(writes, []) + self.assertTrue(any("No response" in s.get("text", "") for s in api.sent)) + + def test_write_failure_reported_not_raised(self): + def boom(u, k, p): + raise RuntimeError("airtrail 500") + n, api, _ = self._run([[cb(12, 0, "ok")]], post=boom) + self.assertEqual(n, 0) + self.assertTrue(any("Failed" in s.get("text", "") for s in api.sent)) + + def test_nothing_eligible_is_noop(self): + api = FakeAPI() + n = telegram.run_approvals([cand(status="duplicate")], "u", "k", "T", "C", + api=api) + self.assertEqual(n, 0) + self.assertEqual(api.sent, []) # no messages at all + + +class TestEnabledGating(unittest.TestCase): + def setUp(self): + self._env = {k: os.environ.get(k) for k in + ("TELEGRAM_APPROVE", "TELEGRAM_BOT_TOKEN", "TELEGRAM_CHAT_ID")} + + def tearDown(self): + for k, v in self._env.items(): + os.environ.pop(k, None) if v is None else os.environ.__setitem__(k, v) + + def test_off_by_default(self): + for k in ("TELEGRAM_APPROVE", "TELEGRAM_BOT_TOKEN", "TELEGRAM_CHAT_ID"): + os.environ.pop(k, None) + self.assertFalse(telegram.enabled()) + + def test_on_when_configured(self): + os.environ.update({"TELEGRAM_APPROVE": "1", "TELEGRAM_BOT_TOKEN": "t", + "TELEGRAM_CHAT_ID": "c"}) + self.assertTrue(telegram.enabled()) + + +if __name__ == "__main__": + unittest.main()