"""Register a Telegram chat_id for a farmer by updating the instance DB. This is a small admin helper which writes the chat id directly into `instance/farm_management.db` to avoid needing to authenticate via the web admin UI. Usage: python scripts/register_telegram_chat.py --farmer-id 1 --chat-id 123456789 It will print what it changed and exit with a non-zero code on error. """ import argparse import sqlite3 import os import sys def register_chat(farmer_id: int, chat_id: str, db_path: str): if not os.path.exists(db_path): print(f"Database not found: {db_path}") return 2 con = sqlite3.connect(db_path) cur = con.cursor() # Check farmer exists cur.execute("SELECT id, name, telegram_chat_id FROM farmers WHERE id = ?", (farmer_id,)) row = cur.fetchone() if not row: print(f"Farmer id {farmer_id} not found in {db_path}") con.close() return 3 old = row[2] cur.execute("UPDATE farmers SET telegram_chat_id = ? WHERE id = ?", (str(chat_id), farmer_id)) con.commit() con.close() print(f"Updated farmer {farmer_id} telegram_chat_id: {old} -> {chat_id}") return 0 def main(): p = argparse.ArgumentParser() p.add_argument('--farmer-id', '-f', type=int, required=False, help='Farmer id to update') p.add_argument('--chat-id', '-c', required=True, help='Telegram chat id to set') p.add_argument('--all', '-a', action='store_true', help='Apply chat id to all farmers') p.add_argument('--db', default=os.path.join('instance', 'farm_management.db'), help='Path to instance DB') args = p.parse_args() if args.all: # Update all farmers if not os.path.exists(args.db): print(f"Database not found: {args.db}") sys.exit(2) con = sqlite3.connect(args.db) cur = con.cursor() cur.execute("UPDATE farmers SET telegram_chat_id = ?", (str(args.chat_id),)) con.commit() con.close() print(f"Updated telegram_chat_id for ALL farmers -> {args.chat_id}") rc = 0 else: if not args.farmer_id: print('Either --farmer-id or --all must be provided') sys.exit(1) rc = register_chat(args.farmer_id, args.chat_id, args.db) sys.exit(rc) if __name__ == '__main__': main()