#!/usr/bin/env python3 # Minimal Telegram input bridge for Health_1990_input_bot # - Reads token from ~/.credentials/health_input_bot_telegram_token # - Polls getUpdates, checks for 'food' messages, creates event JSONL lines, # appends atomically to inbox/events.jsonl, keeps offset, and ACKs once. import os import sys import json import time import uuid import logging import datetime import urllib.request import urllib.parse import ssl import traceback HOME = os.path.expanduser('~') TOKEN_PATH = os.path.join(HOME, '.credentials', 'health_input_bot_telegram_token') BASE_DIR = os.path.join(HOME, 'AI-Coach', 'health-watcher-prod') INBOX_DIR = os.path.join(BASE_DIR, 'inbox') EVENTS_PATH = os.path.join(INBOX_DIR, 'events.jsonl') OFFSET_PATH = os.path.join(INBOX_DIR, 'telegram_offset.txt') LOG_DIR = os.path.join(BASE_DIR, 'logs') LOG_PATH = os.path.join(LOG_DIR, 'health_input_bridge.log') # ACK text (Hebrew) ACK_TEXT = "קיבלתי, מעבד את הדיווח..." # Ensure directories os.makedirs(INBOX_DIR, exist_ok=True) os.makedirs(LOG_DIR, exist_ok=True) # Logging setup logger = logging.getLogger('health_input_bridge') logger.setLevel(logging.INFO) fh = logging.FileHandler(LOG_PATH, encoding='utf-8') fh.setFormatter(logging.Formatter('%(asctime)s %(levelname)s %(message)s')) logger.addHandler(fh) # Helper: read token def read_token(path): try: with open(path, 'r', encoding='utf-8') as f: token = f.read().strip() if not token: raise ValueError('token file empty') return token except Exception as e: logger.exception('Failed to read token from %s: %s', path, e) raise # Simple heuristic for likely-food in Hebrew QUESTION_WORDS = {'מה','איך','מתי','כמה','האם','מי','למה','איזה'} FOOD_KEYWORDS = ['אכל', 'אכלתי', 'אכלנו', 'אכלת', 'אוכל', 'אוכלתי', 'שתה', 'שתיתי', 'שתיתי', 'שתינו', 'קפה', 'ביצה', 'ביצים', 'יוגורט', 'לחם', 'חלב', 'גרם', 'מ"ל', 'מל', 'חטיף', 'פרוסה', 'פרוסות', 'בננה', 'תפוח', 'ירקות', 'סלט', 'עוף', 'בשר', 'דגים', 'קוסקוס'] def is_likely_food(text): if not text or not text.strip(): return False t = text.strip() # Quick reject: question mark or starts with question word first = t.split()[0].strip(' ?!.,') if first in QUESTION_WORDS: return False if '?' in t or t.endswith('?'): return False low = t.lower() # if contains any food keyword, consider it food for kw in FOOD_KEYWORDS: if kw in low: return True # contains numbers + unit (e.g., "150 גרם") if any(u in low for u in ('גרם', 'מ"ל', 'מל', 'ק"ג', 'קג')) and any(ch.isdigit() for ch in low): return True return False # Atomic append to jsonl using os.open O_APPEND def append_event_line(path, line): import os, errno b = (line.rstrip('\n') + '\n').encode('utf-8') flags = os.O_WRONLY | os.O_APPEND | os.O_CREAT mode = 0o600 fd = os.open(path, flags, mode) try: os.write(fd, b) finally: os.close(fd) # Read/Write offset def read_offset(path): try: with open(path, 'r', encoding='utf-8') as f: return int(f.read().strip()) except Exception: return None def write_offset(path, offset): tmp = path + '.tmp' with open(tmp, 'w', encoding='utf-8') as f: f.write(str(int(offset))) os.replace(tmp, path) # HTTP helpers ctx = ssl.create_default_context() def telegram_api(token, method, params=None): base = f'https://api.telegram.org/bot{token}/{method}' data = None if params is not None: data = urllib.parse.urlencode(params).encode('utf-8') req = urllib.request.Request(base, data=data) with urllib.request.urlopen(req, context=ctx, timeout=35) as resp: return json.load(resp) import threading # Background poller to watch outbox/results.jsonl and create pending approval class PendingWatcher(threading.Thread): def __init__(self, event_id, user_id, timeout=300, poll_interval=1): super().__init__(daemon=True) self.event_id = event_id self.user_id = user_id self.timeout = timeout self.poll_interval = poll_interval self.base = os.path.join(HOME, 'AI-Coach', 'health-watcher-prod') self.outbox_path = os.path.join(self.base, 'outbox', 'results.jsonl') self.pending_path = os.path.join(self.base, 'outbox', 'pending_approval.jsonl') def run(self): logger.info('PendingWatcher started for %s user=%s', self.event_id, self.user_id) start = time.time() seen = False while time.time() - start <= self.timeout: try: if os.path.exists(self.outbox_path): with open(self.outbox_path, 'r', encoding='utf-8') as f: for line in f: if not line.strip(): continue try: obj = json.loads(line) except Exception: continue if str(obj.get('event_id')) == str(self.event_id): # found matching outbox result if obj.get('dashboard_updated'): pending = { 'event_id': self.event_id, 'user_id': str(self.user_id), 'timestamp': datetime.datetime.now(datetime.timezone.utc).astimezone().isoformat(), 'status': 'pending', 'result': obj } try: append_event_line(self.pending_path, json.dumps(pending, ensure_ascii=False)) logger.info('Wrote pending approval for %s', self.event_id) except Exception: logger.exception('Failed to write pending approval for %s', self.event_id) return else: # found result but not dashboard_updated yet; keep waiting seen = True time.sleep(self.poll_interval) except Exception: logger.exception('PendingWatcher loop error') time.sleep(1) logger.warning('PendingWatcher timeout for %s (seen=%s)', self.event_id, seen) # Main loop def main(): try: token = read_token(TOKEN_PATH) except Exception: print('Failed to read token; see log for details', file=sys.stderr) sys.exit(1) # getMe for friendliness (don't log token) try: me = telegram_api(token, 'getMe') bot_username = me.get('result', {}).get('username') logger.info('Bot getMe: %s', bot_username) except Exception: logger.exception('getMe failed') bot_username = None offset = read_offset(OFFSET_PATH) if offset is None: offset = 0 logger.info('Starting poll loop with offset=%s', offset) print('Health input bridge started. Logging to', LOG_PATH) # remembered watchers: event_id -> thread watchers = {} try: while True: try: params = {'timeout': 30, 'allowed_updates': json.dumps(['message'])} if offset: params['offset'] = offset res = telegram_api(token, 'getUpdates', params) updates = res.get('result', []) if not updates: continue last_update_id = None for upd in updates: last_update_id = upd.get('update_id') # Only process standard messages with text msg = upd.get('message') or {} chat = msg.get('chat') or {} text = msg.get('text') chat_id = chat.get('id') logger.info('recv update_id=%s chat_id=%s text_len=%s', last_update_id, chat_id, len(text) if text else 0) # For this controlled test we optionally log raw text logger.info('raw_text: %s', text) if text and is_likely_food(text): # Build event ev = { 'source_bot': 'Health Input Bot', 'event_type': 'food_log', 'event_id': 'hib-' + uuid.uuid4().hex, 'timestamp': datetime.datetime.now(datetime.timezone.utc).astimezone().isoformat(), 'raw_text': text, 'status': 'received_by_input_bot' } # Append atomically try: append_event_line(EVENTS_PATH, json.dumps(ev, ensure_ascii=False)) logger.info('Appended event_id=%s raw_text=%s', ev['event_id'], text) except Exception: logger.exception('Failed to append event') # Start pending watcher for this event try: w = PendingWatcher(ev['event_id'], chat_id) w.start() watchers[ev['event_id']] = w except Exception: logger.exception('Failed to start PendingWatcher for %s', ev.get('event_id')) # Send immediate ACK try: telegram_api(token, 'sendMessage', {'chat_id': chat_id, 'text': ACK_TEXT}) logger.info('Sent ACK to chat_id=%s', chat_id) except Exception: logger.exception('Failed to send ACK') else: logger.info('Not classified as food (update_id=%s)', last_update_id) # advance offset if last_update_id is not None: offset = int(last_update_id) + 1 try: write_offset(OFFSET_PATH, offset) logger.info('Wrote offset=%s', offset) except Exception: logger.exception('Failed to write offset') except Exception: logger.exception('Error during poll loop') time.sleep(2) except KeyboardInterrupt: logger.info('Interrupted by user; exiting') print('\nStopped by user') # Approve helper: send final Telegram and write delivered.log def approve_and_send(event_id): # find pending entry pending_path = os.path.join(HOME, 'AI-Coach', 'health-watcher-prod', 'outbox', 'pending_approval.jsonl') delivered_path = os.path.join(HOME, 'AI-Coach', 'health-watcher-prod', 'outbox', 'delivered.log') outbox_path = os.path.join(HOME, 'AI-Coach', 'health-watcher-prod', 'outbox', 'results.jsonl') # check delivered log for idempotency try: if os.path.exists(delivered_path): with open(delivered_path, 'r', encoding='utf-8') as f: if event_id in f.read(): print('Already delivered:', event_id) return True except Exception: logger.exception('Failed to read delivered.log') pending = None if os.path.exists(pending_path): with open(pending_path, 'r', encoding='utf-8') as f: for line in f: if not line.strip(): continue try: p = json.loads(line) except Exception: continue if p.get('event_id') == event_id: pending = p break if not pending: print('No pending approval found for', event_id) return False # find full outbox result result = None if os.path.exists(outbox_path): with open(outbox_path, 'r', encoding='utf-8') as f: for line in f: if not line.strip(): continue try: r = json.loads(line) except Exception: continue if r.get('event_id') == event_id: result = r break if not result: print('No outbox result found for', event_id) return False # send final Telegram using user_message_he try: token = read_token(TOKEN_PATH) msg = result.get('user_message_he') or '' chat_id = int(pending.get('user_id')) if not msg: print('No user_message_he to send for', event_id) return False res = telegram_api(token, 'sendMessage', {'chat_id': chat_id, 'text': msg}) ok = res.get('ok') if ok: # append to delivered.log atomically append_event_line(delivered_path, event_id) print('Delivered:', event_id) return True else: print('Telegram send failed:', res) return False except Exception: logger.exception('Failed during approve_and_send') return False if __name__ == '__main__': # CLI: allow approve mode if len(sys.argv) >= 3 and sys.argv[1] == '--approve': eid = sys.argv[2] ok = approve_and_send(eid) sys.exit(0 if ok else 2) main()