134 lines
4.0 KiB
Python
134 lines
4.0 KiB
Python
"""Local dev CLI for the CM bot. Mirrors Telegram /1, /2, /3 plus
|
|
operational commands. No-arg invocation drops into a stdlib TUI menu.
|
|
"""
|
|
|
|
import argparse
|
|
import sys
|
|
|
|
from .cm_bot_hal import CM_BOT_HAL
|
|
|
|
|
|
def cmd_interactive(_args):
|
|
raise NotImplementedError("cmd_interactive is implemented in a later task")
|
|
|
|
|
|
def _print_user(user: dict) -> None:
|
|
print(f"Username: {user['username']}")
|
|
print(f"Password: {user['password']}")
|
|
print(f"Link: {user['link']}")
|
|
|
|
|
|
def cmd_register(_args):
|
|
bot = CM_BOT_HAL()
|
|
_print_user(bot.get_user_api())
|
|
|
|
|
|
def cmd_set_pin(args):
|
|
bot = CM_BOT_HAL()
|
|
if not bot.is_whatsapp_url(args.link):
|
|
print(f"ERROR: not a WhatsApp URL: {args.link}", file=sys.stderr)
|
|
sys.exit(2)
|
|
t_username, f_username = bot.get_whatsapp_link_username(args.link)
|
|
success = bot.set_security_pin_api(args.link)
|
|
if not success:
|
|
print("ERROR: set_security_pin_api returned a falsy result", file=sys.stderr)
|
|
sys.exit(1)
|
|
print(f"OK: f_username={f_username} t_username={t_username}")
|
|
|
|
|
|
def cmd_insert_user(args):
|
|
bot = CM_BOT_HAL()
|
|
f_password = bot.get_user_pass_from_acc(args.f_username)
|
|
if not f_password:
|
|
print(f"ERROR: no password for {args.f_username}", file=sys.stderr)
|
|
sys.exit(2)
|
|
success = bot.insert_user_to_table_user({
|
|
"f_username": args.f_username,
|
|
"f_password": f_password,
|
|
"t_username": args.t_username,
|
|
"t_password": bot.security_pin,
|
|
})
|
|
if not success:
|
|
print("ERROR: insert failed", file=sys.stderr)
|
|
sys.exit(1)
|
|
print(f"OK: inserted {args.f_username} → {args.t_username}")
|
|
|
|
|
|
def cmd_credit(args):
|
|
bot = CM_BOT_HAL()
|
|
print(f"Credit: {bot.get_user_credit(args.username, args.password)}")
|
|
|
|
|
|
def cmd_transfer(args):
|
|
bot = CM_BOT_HAL()
|
|
print(bot.transfer_credit_api(
|
|
args.f_username, args.f_password,
|
|
args.t_username, args.t_password,
|
|
))
|
|
|
|
|
|
def cmd_monitor_once(args):
|
|
bot = CM_BOT_HAL()
|
|
available = bot.get_all_available_acc()
|
|
print(f"Available accounts: {len(available)} (target: {args.target})")
|
|
if len(available) >= args.target:
|
|
print("Already at target; nothing to do.")
|
|
return
|
|
for _ in range(len(available), args.target):
|
|
try:
|
|
user = bot.create_new_acc()
|
|
print(f"Created: {user['username']}")
|
|
except Exception as exc:
|
|
print(f"ERROR creating account: {exc}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
p = argparse.ArgumentParser(
|
|
prog="bot_cli",
|
|
description="CM Bot dev CLI (mirrors Telegram triggers).",
|
|
)
|
|
sub = p.add_subparsers(dest="command")
|
|
|
|
sp = sub.add_parser("register", aliases=["get-acc"], help="Get next available account (Telegram /1).")
|
|
sp.set_defaults(func=cmd_register)
|
|
|
|
sp = sub.add_parser("set-pin", help="Set security PIN from a WhatsApp link (Telegram /2).")
|
|
sp.add_argument("link")
|
|
sp.set_defaults(func=cmd_set_pin)
|
|
|
|
sp = sub.add_parser("insert-user", help="Insert into user table (Telegram /3).")
|
|
sp.add_argument("f_username")
|
|
sp.add_argument("t_username")
|
|
sp.set_defaults(func=cmd_insert_user)
|
|
|
|
sp = sub.add_parser("credit", help="Read account credit balance.")
|
|
sp.add_argument("username")
|
|
sp.add_argument("password")
|
|
sp.set_defaults(func=cmd_credit)
|
|
|
|
sp = sub.add_parser("transfer", help="One-shot credit transfer.")
|
|
sp.add_argument("f_username")
|
|
sp.add_argument("f_password")
|
|
sp.add_argument("t_username")
|
|
sp.add_argument("t_password")
|
|
sp.set_defaults(func=cmd_transfer)
|
|
|
|
sp = sub.add_parser("monitor-once", aliases=["monitor"], help="One iteration of the auto-create monitor.")
|
|
sp.add_argument("--target", type=int, default=20)
|
|
sp.set_defaults(func=cmd_monitor_once)
|
|
|
|
return p
|
|
|
|
|
|
def main(argv=None) -> int:
|
|
parser = build_parser()
|
|
args = parser.parse_args(argv)
|
|
if args.command is None:
|
|
return cmd_interactive(args) or 0
|
|
return args.func(args) or 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|