"""Check Telegram chats for replies and update all aima_far_v* log CSVs.""" import asyncio import csv import os import sqlite3 import sys from datetime import datetime, timezone from pathlib import Path if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8", errors="replace") BASE_DIR = Path(__file__).parent.parent DEFAULT_ENV_FILES = [ BASE_DIR / ".env", Path(r"C:\python_scripts\top_1_telegram_signals\.env"), ] SESSION = BASE_DIR / "data/processed/telegram/aima_support_session" LOG_GLOB = "aima_far_v*_telegram_log.csv" DB_PATH = BASE_DIR / "data/processed/aima_conversion_shadow.sqlite" def load_env(): # type: () -> None for path in DEFAULT_ENV_FILES: if not path.exists(): continue for raw in path.read_text(encoding="utf-8", errors="ignore").splitlines(): line = raw.strip() if not line or line.startswith("#") or "=" not in line: continue k, v = line.split("=", 1) os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'")) def parse_dt(s): # type: (str) -> object if not s: return None try: dt = datetime.fromisoformat(s) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt except Exception: return None async def check_all_replies(): # type: () -> None load_env() api_id = os.environ.get("TG_API_ID") api_hash = os.environ.get("TG_API_HASH") if not api_id or not api_hash: raise SystemExit("TG_API_ID/TG_API_HASH missing from .env") from telethon import TelegramClient # noqa: PLC0415 client = TelegramClient(str(SESSION), int(api_id), api_hash) await client.connect() if not await client.is_user_authorized(): raise SystemExit("Session not authorized — run aima_telegram_login.py first") me = await client.get_me() print("[auth] {} (@{})".format( getattr(me, "first_name", ""), getattr(me, "username", "?"))) proc_dir = BASE_DIR / "data/processed" log_files = sorted(proc_dir.glob(LOG_GLOB)) if not log_files: print("[warn] no log CSVs found for pattern {}".format(LOG_GLOB)) await client.disconnect() return total_checked = 0 total_new_replies = 0 for log_csv in log_files: with log_csv.open(encoding="utf-8-sig") as f: reader = csv.DictReader(f) fieldnames = list(reader.fieldnames or []) rows = [dict(r) for r in reader] if not rows: continue changed = False for row in rows: if row.get("gate1_message_sent", "").lower() != "yes": continue if row.get("replied", "").lower() == "yes": continue # already confirmed, skip recipient_id = row.get("recipient_id", "").strip() if not recipient_id: continue sent_dt = parse_dt(row.get("sent_at_utc", "")) total_checked += 1 try: entity = await client.get_entity(int(recipient_id)) reply_found = False async for msg in client.iter_messages(entity, limit=25): if msg.out: continue # skip our own messages msg_dt = msg.date if msg_dt and msg_dt.tzinfo is None: msg_dt = msg_dt.replace(tzinfo=timezone.utc) if sent_dt and msg_dt and msg_dt < sent_dt: break # message is older than our send — no reply # incoming message after our send → reply reply_found = True row["replied"] = "yes" row["chat_started"] = "yes" row["reply_text"] = (msg.text or "")[:300] row["replied_at_utc"] = msg_dt.isoformat() if msg_dt else "" changed = True total_new_replies += 1 print("[reply] {} pilot={}: {}".format( log_csv.stem, row.get("pilot_id"), (msg.text or "")[:80])) break if not reply_found and not row.get("replied"): row["replied"] = "no" changed = True except Exception as exc: print("[warn] pilot={} id={}: {}".format( row.get("pilot_id"), recipient_id, exc)) if changed: with log_csv.open("w", encoding="utf-8-sig", newline="") as f: writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore") writer.writeheader() writer.writerows(rows) print("[saved] {}".format(log_csv.name)) else: print("[skip] {} — nothing changed".format(log_csv.name)) await client.disconnect() print("[done] checked={} new_replies={}".format(total_checked, total_new_replies)) _sync_to_sqlite(proc_dir) def _sync_to_sqlite(log_dir): # type: (Path) -> None """Upsert outreach+reply data from all FAR log CSVs into SQLite aima_outreach_events.""" if not DB_PATH.exists(): print("[warn] SQLite DB not found, skipping sync") return conn = sqlite3.connect(str(DB_PATH)) cur = conn.cursor() cur.execute(""" CREATE TABLE IF NOT EXISTS aima_outreach_events ( pilot_id INTEGER PRIMARY KEY, batch_name TEXT, lead_id TEXT, dataset TEXT, hypothesis TEXT, outreach_status TEXT, reply_text TEXT, sent_at_utc TEXT, replied_at_utc TEXT, updated_at_utc TEXT ) """) synced = 0 for csv_path in sorted(log_dir.glob("*_telegram_log.csv")): if "test" in csv_path.stem: continue with csv_path.open(encoding="utf-8-sig") as f: for row in csv.DictReader(f): pid = row.get("pilot_id", "").strip() if not pid: continue sent = row.get("gate1_message_sent", "").lower() == "yes" if not sent: continue replied = row.get("replied", "").lower() == "yes" status = "replied" if replied else "sent" try: cur.execute(""" INSERT INTO aima_outreach_events (pilot_id, batch_name, lead_id, dataset, hypothesis, outreach_status, reply_text, sent_at_utc, replied_at_utc, updated_at_utc) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(pilot_id) DO UPDATE SET outreach_status = excluded.outreach_status, reply_text = CASE WHEN excluded.reply_text != '' THEN excluded.reply_text ELSE aima_outreach_events.reply_text END, replied_at_utc = CASE WHEN excluded.replied_at_utc != '' THEN excluded.replied_at_utc ELSE aima_outreach_events.replied_at_utc END, updated_at_utc = excluded.updated_at_utc """, ( int(pid), csv_path.stem.replace("_telegram_log", ""), row.get("lead_id", ""), row.get("dataset", ""), row.get("hypothesis", "").strip(), status, (row.get("reply_text") or "").strip()[:300], row.get("sent_at_utc", ""), row.get("replied_at_utc", ""), datetime.now(timezone.utc).isoformat(), )) synced += 1 except Exception as exc: print("[warn] sqlite upsert pilot={}: {}".format(pid, exc)) conn.commit() conn.close() print("[sqlite] synced {} outreach rows → aima_outreach_events".format(synced)) def main(): # type: () -> None loop = asyncio.get_event_loop() loop.run_until_complete(check_all_replies()) if __name__ == "__main__": main()