Spaces:
Sleeping
Sleeping
File size: 2,395 Bytes
429a26d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 | """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()
|