""" Action handlers — one function per distinct behaviour the bot performs. Handlers receive the incoming update (the platform's `message` envelope) and carry out exactly the side effects of the live system: fetching a price and sending the reply. """ import re from urllib.parse import urlparse from .. import config, messages from ..services.telegram_api import TelegramAPI from ..services import ( price_service, group_links, referral_state, explorer_service, explorer_state, usage_log, ) tg = TelegramAPI() # Legacy Telegram Markdown treats these as formatting delimiters; a lone, # unpaired one (e.g. the "_" in "@FinanceGPTlab_bot") makes Telegram reject # the whole message with "can't parse entities". Escaping them in the # signature keeps it literal regardless of parse_mode. _MARKDOWN_SPECIAL_RE = re.compile(r"([_*`\[])") def _escape_markdown(text: str) -> str: return _MARKDOWN_SPECIAL_RE.sub(r"\\\1", text) # --- small helpers ------------------------------------------------------------- def _send(chat_id, text, reply_markup=None, reply_to_message_id=None, parse_mode=None, retry=True): """Every outgoing message goes through here so the signature is never missed. If this is a reply and Telegram rejects it (e.g. "message to be replied not found" because the original was deleted or is otherwise inaccessible), fall back to sending the same message as a normal, non-reply message instead of failing the whole update. """ signature = ( _escape_markdown(config.BOT_SIGNATURE) if parse_mode == "Markdown" else config.BOT_SIGNATURE ) full_text = f"{text}\n{signature}" try: return tg.send_message( chat_id, full_text, reply_markup=reply_markup, reply_to_message_id=reply_to_message_id, parse_mode=parse_mode, retry=retry, ) except Exception: if reply_to_message_id is None: raise return tg.send_message( chat_id, full_text, reply_markup=reply_markup, parse_mode=parse_mode, retry=retry, ) def _get(d, *path): """Safe nested lookup; returns None if any segment is missing.""" cur = d for key in path: if not isinstance(cur, dict): return None cur = cur.get(key) return cur # --- usage report log ------------------------------------------------------------ # # Every button press / command / recognized free-text action gets one row in # data/usage_log.db (chat id, first/last name, action, optional detail) — see # services/usage_log.py. Query it directly with the sqlite3 CLI; no UI. def _log(update: dict, action: str, detail: str = None) -> None: frm = _get(update, "message", "from") or _get(update, "callback_query", "from") or {} usage_log.log_action(frm.get("id"), frm.get("first_name"), frm.get("last_name"), action, detail) # --- membership gate ----------------------------------------------------------- def _activate(user_id) -> None: _send(user_id, messages.MEMBERSHIP_CONFIRMED_TEXT, messages.MAIN_MENU_KEYBOARD) def _require_join(user_id) -> None: _send( user_id, messages.MEMBERSHIP_REQUIRED_TEXT, messages.MEMBERSHIP_REQUIRED_KEYBOARD, ) # Statuses that mean the user is currently a member of the channel. _MEMBER_STATUSES = {"creator", "administrator", "member", "restricted"} def _is_channel_member(user_id) -> bool: try: member = tg.get_chat_member(config.GEEVEX_CHANNEL_ID, user_id) except Exception: return False status = _get(member, "result", "status") return status in _MEMBER_STATUSES def check_membership_for_start(update: dict) -> None: """ Membership check triggered by /start: look up the sender in the channel. On success activate the bot, otherwise prompt to join. """ _log(update, "start") user_id = _get(update, "message", "from", "id") if _is_channel_member(user_id): _activate(user_id) else: _require_join(user_id) def check_membership_for_callback(update: dict) -> None: """ Membership check triggered by the "عضو شدم ✔️" button. The user id is taken from the Telegram callback payload (`callback_query.from.id`). """ _log(update, "membership_check_button") user_id = _get(update, "callback_query", "from", "id") if _is_channel_member(user_id): _activate(user_id) else: _require_join(user_id) # --- main menu --------------------------------------------------------------------- def handle_price_menu_button(update: dict) -> None: """"قیمت لحظه ای ارز های دیجیتال" was pressed: ask for a coin name/symbol. No special "awaiting" state is needed — every other private-chat text is already routed to handle_price_lookup, so the user's next message is picked up by that same flow regardless. """ _log(update, "price_menu_button") user_id = _get(update, "message", "from", "id") _send(user_id, messages.ASK_COIN_TEXT) def handle_earn_button(update: dict) -> None: """"کسب درامد" was pressed: show the add-to-group instructions.""" _log(update, "earn_button") user_id = _get(update, "message", "from", "id") _send( user_id, messages.ADD_TO_GROUP_PROMPT_TEXT, reply_markup=messages.add_to_group_keyboard(), ) # --- blockchain explorer ----------------------------------------------------------- def handle_explorer_menu_button(update: dict) -> None: """"بلاکچین اکسپلور" was pressed: ask which network the tx hash is on.""" _log(update, "explorer_menu_button") user_id = _get(update, "message", "from", "id") _send( user_id, messages.EXPLORER_ASK_NETWORK_TEXT, reply_markup=messages.explorer_network_keyboard(), ) def handle_explorer_network_selected(update: dict) -> None: """A network button (callback data "explorer_network:") was pressed.""" data = _get(update, "callback_query", "data") or "" network = data.split(":", 1)[1] if ":" in data else "" if network not in explorer_service.NETWORKS: return _log(update, "explorer_network_selected", network) user_id = _get(update, "callback_query", "from", "id") explorer_state.mark_awaiting_hash(user_id, network) _send(user_id, messages.explorer_ask_hash_text(network)) def handle_explorer_hash_reply(update: dict, chat_id, text) -> bool: """ If `chat_id` is currently waiting on a tx hash (after picking a network), consume this text as that hash, look up the transaction, and reply with the result. Returns False when nothing was pending, so the caller can fall through normally. """ network = explorer_state.consume_awaiting_hash(chat_id) if network is None: return False tx_hash = (text or "").strip() try: result = explorer_service.fetch_transaction(network, tx_hash) except explorer_service.TransactionNotFound: _log(update, "explorer_hash_lookup", f"{network}:{tx_hash}:not_found") _send(chat_id, messages.EXPLORER_NOT_FOUND_TEXT) return True except Exception: _log(update, "explorer_hash_lookup", f"{network}:{tx_hash}:error") _send(chat_id, messages.EXPLORER_ERROR_TEXT) return True _log(update, "explorer_hash_lookup", f"{network}:{tx_hash}:found") results = result if isinstance(result, list) else [result] body = "\n\n".join(messages.explorer_result_text(r) for r in results) _send(chat_id, body, parse_mode="Markdown") return True # --- group referral link ---------------------------------------------------------- # Statuses that mean the bot is now a member of the chat. _JOINED_STATUSES = {"member", "administrator"} # Statuses that mean the bot was not previously in the chat. _NOT_PREVIOUSLY_IN_CHAT_STATUSES = {"left", "kicked"} def handle_bot_added_to_group(update: dict) -> None: """ Fired on a `my_chat_member` update: reacts only to the bot being newly added to a group/supergroup. Asks whoever added it (if they're an admin) for their referral link; tells them if they're not an admin. """ payload = update.get("my_chat_member") or {} chat = payload.get("chat") or {} if chat.get("type") not in ("group", "supergroup"): return old_status = _get(payload, "old_chat_member", "status") new_status = _get(payload, "new_chat_member", "status") if old_status not in _NOT_PREVIOUSLY_IN_CHAT_STATUSES or new_status not in _JOINED_STATUSES: return # not a fresh join (e.g. admin rights changed, or bot left) group_chat_id = chat.get("id") group_title = chat.get("title") or "" adder_id = _get(payload, "from", "id") if adder_id is None: return usage_log.log_action( adder_id, _get(payload, "from", "first_name"), _get(payload, "from", "last_name"), "add_to_group", group_title, ) try: member = tg.get_chat_member(group_chat_id, adder_id) except Exception: return is_admin = _get(member, "result", "status") in ("creator", "administrator") if not is_admin: try: _send(adder_id, messages.NOT_GROUP_ADMIN_TEXT, retry=False) except Exception: _send(group_chat_id, messages.NOT_GROUP_ADMIN_TEXT) return ask_text = messages.ask_referral_link_text(group_title) try: _send(adder_id, ask_text, retry=False) reply_chat_id = adder_id except Exception: _send(group_chat_id, messages.bot_added_to_group_group_message()) _send(group_chat_id, ask_text) reply_chat_id = group_chat_id referral_state.mark_awaiting_link(reply_chat_id, adder_id, group_chat_id, group_title) def _is_valid_button_url(link: str) -> bool: """ Reject anything Telegram would refuse as an inline-button `url` (e.g. a stray space breaking the host, or a scheme-only string with no host). """ if " " in link or "\t" in link or "\n" in link: return False parsed = urlparse(link) return parsed.scheme in ("http", "https") and bool(parsed.netloc) def handle_referral_link_reply(update: dict, chat_id, sender_id, text) -> bool: """ If `chat_id` is currently waiting on a referral link from `sender_id`, consume this text as that reply and return True (handled either way — valid link stored, or an "invalid link" nudge sent). Returns False when there was nothing pending, so the caller can fall through normally. """ pending = referral_state.get_pending(chat_id, sender_id) if pending is None: return False link = (text or "").strip() if not _is_valid_button_url(link): _send(chat_id, messages.REFERRAL_LINK_INVALID_TEXT) return True group_links.set_link(pending["group_chat_id"], link) referral_state.clear(chat_id) _log(update, "referral_link_saved", link) _send(chat_id, messages.REFERRAL_LINK_SAVED_TEXT) return True # --- price lookup ---------------------------------------------------------------- def handle_price_lookup(update: dict) -> None: """Free-form coin-name text in a private chat: show the 3-exchange price.""" text = _get(update, "message", "text") chat_id = _get(update, "message", "from", "id") symbol = price_service.extract_symbol(text) if symbol is None: return # not a recognizable coin -> stay silent amount = price_service.extract_amount(text) if amount is not None: body = price_service.build_amount_message(amount, symbol) _log(update, "price_lookup", f"{symbol} amount={amount}") else: body = price_service.build_price_message(symbol) _log(update, "price_lookup", symbol) if body is None: return # every source failed or doesn't list this coin keyboard = messages.buy_coin_keyboard(price_service.display_name(symbol)) _send(chat_id, body, reply_markup=keyboard, parse_mode="Markdown") def handle_group_price_lookup(update: dict) -> None: """Free-form coin-name text inside a group/supergroup: reply with the price.""" text = _get(update, "message", "text") chat_id = _get(update, "message", "chat", "id") message_id = _get(update, "message", "message_id") symbol = price_service.extract_symbol(text) if symbol is None: return amount = price_service.extract_amount(text) if amount is not None: body = price_service.build_amount_message(amount, symbol) _log(update, "group_price_lookup", f"{symbol} amount={amount}") else: body = price_service.build_price_message(symbol) _log(update, "group_price_lookup", symbol) if body is None: return buy_url = group_links.get_link(chat_id) keyboard = messages.buy_coin_keyboard(price_service.display_name(symbol), buy_url) _send( chat_id, body, reply_markup=keyboard, reply_to_message_id=message_id, parse_mode="Markdown", )