#!/usr/bin/env python3
"""SPG Multi-Manager Support Bot for Telegram / Android Termux.
Features:
- Optional channel links; users never have to join to request support
- One open ticket per user; sticky assignment per ticket
- Least-open-ticket balancing among managers who are on duty and in their shift
- Configurable 1–24-hour shift windows (supports overnight shifts)
- System-admin/head-manager fallback and urgent-keyword escalation
- Durable SQLite outbox with automatic retry; replies mapped to assigned ticket; reply to copied message
- SQLite transcript and staff-action audit logs
This is a support-ticket/audit system, not covert monitoring. It records messages sent through
this bot only. Audio is relayed as audio; voice transcription is not enabled in this version.
"""
from __future__ import annotations
import asyncio
import html
import logging
import time
import os
import re
import sqlite3
from datetime import datetime, timedelta, time as dtime
from pathlib import Path
from typing import Any, Optional
from zoneinfo import ZoneInfo
from telegram import (
BotCommand,
BotCommandScopeChat,
BotCommandScopeDefault,
InlineKeyboardButton,
InlineKeyboardMarkup,
Update,
)
from telegram.constants import ChatType
from telegram.error import TelegramError
from telegram.ext import (
Application,
CallbackQueryHandler,
CommandHandler,
ContextTypes,
MessageHandler,
filters,
)
import config
BASE_DIR = Path(__file__).resolve().parent
DB_PATH = Path(config.DATABASE_PATH)
if not DB_PATH.is_absolute():
DB_PATH = BASE_DIR / DB_PATH
IMAGE_PATH = Path(config.WELCOME_IMAGE_PATH)
if not IMAGE_PATH.is_absolute():
IMAGE_PATH = BASE_DIR / IMAGE_PATH
TZ = ZoneInfo(config.TIMEZONE)
logging.basicConfig(
format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
level=logging.INFO,
)
logger = logging.getLogger("spg_support_bot")
_ticket_create_lock = asyncio.Lock()
# HTTPX debug/info request URLs can contain the bot token; keep them out of logs.
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("httpcore").setLevel(logging.WARNING)
logging.getLogger("telegram.request").setLevel(logging.WARNING)
logging.getLogger("telegram.ext.ExtBot").setLevel(logging.WARNING)
CATEGORIES = {
"finance": "Finance / Deposit",
"withdrawal": "Withdrawal",
"payment": "Payment Pending",
"scam": "Scam / Fraud Report",
"account": "Account / Login Problem",
"other": "Other Issue",
}
def now_dt() -> datetime:
return datetime.now(TZ)
def now_text() -> str:
return now_dt().isoformat(timespec="seconds")
class ClosingConnection(sqlite3.Connection):
"""Connection context manager that commits/rolls back and always closes."""
def __exit__(self, exc_type, exc_value, traceback):
try:
return super().__exit__(exc_type, exc_value, traceback)
finally:
self.close()
def db_connect() -> sqlite3.Connection:
conn = sqlite3.connect(DB_PATH, timeout=30, factory=ClosingConnection)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys = ON")
conn.execute("PRAGMA busy_timeout = 30000")
return conn
def db_init() -> None:
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
with db_connect() as conn:
conn.executescript(
"""
CREATE TABLE IF NOT EXISTS managers (
telegram_id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
shift_start TEXT NOT NULL DEFAULT '00:00',
shift_hours INTEGER NOT NULL DEFAULT 12 CHECK (shift_hours BETWEEN 1 AND 24),
enabled INTEGER NOT NULL DEFAULT 1,
on_duty INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS staff_roles (
telegram_id INTEGER PRIMARY KEY,
role TEXT NOT NULL CHECK (role IN ('owner','head')),
enabled INTEGER NOT NULL DEFAULT 1,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS tickets (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
username TEXT,
full_name TEXT NOT NULL,
category TEXT NOT NULL,
assigned_to INTEGER,
assigned_role TEXT NOT NULL DEFAULT 'unassigned',
status TEXT NOT NULL DEFAULT 'open',
priority TEXT NOT NULL DEFAULT 'normal',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
closed_at TEXT,
escalated_from INTEGER,
last_escalation_reason TEXT
);
CREATE INDEX IF NOT EXISTS idx_tickets_user_status ON tickets(user_id, status);
CREATE INDEX IF NOT EXISTS idx_tickets_assignee_status ON tickets(assigned_to, status);
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ticket_id INTEGER NOT NULL,
sender_id INTEGER NOT NULL,
sender_role TEXT NOT NULL,
message_type TEXT NOT NULL,
text_content TEXT,
file_id TEXT,
telegram_message_id INTEGER,
created_at TEXT NOT NULL,
FOREIGN KEY(ticket_id) REFERENCES tickets(id)
);
CREATE INDEX IF NOT EXISTS idx_messages_ticket ON messages(ticket_id, id);
CREATE TABLE IF NOT EXISTS manager_message_map (
manager_id INTEGER NOT NULL,
copied_message_id INTEGER NOT NULL,
ticket_id INTEGER NOT NULL,
created_at TEXT NOT NULL,
PRIMARY KEY(manager_id, copied_message_id),
FOREIGN KEY(ticket_id) REFERENCES tickets(id)
);
CREATE TABLE IF NOT EXISTS settings (
setting_key TEXT PRIMARY KEY,
setting_value TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS audit_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
actor_id INTEGER NOT NULL,
action TEXT NOT NULL,
details TEXT,
ticket_id INTEGER,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS delivery_outbox (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ticket_id INTEGER NOT NULL,
kind TEXT NOT NULL CHECK (kind IN ('notice','user_message')),
source_chat_id INTEGER,
source_message_id INTEGER,
dedupe_key TEXT NOT NULL UNIQUE,
status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN ('pending','processing','delivered')),
attempts INTEGER NOT NULL DEFAULT 0,
next_attempt_at REAL NOT NULL DEFAULT 0,
locked_at REAL,
last_error TEXT,
created_at TEXT NOT NULL,
delivered_at TEXT,
FOREIGN KEY(ticket_id) REFERENCES tickets(id)
);
CREATE INDEX IF NOT EXISTS idx_outbox_ready ON delivery_outbox(status, next_attempt_at, id);
CREATE INDEX IF NOT EXISTS idx_outbox_ticket ON delivery_outbox(ticket_id, id, status);
CREATE TABLE IF NOT EXISTS manager_ticket_notices (
ticket_id INTEGER NOT NULL,
manager_id INTEGER NOT NULL,
message_id INTEGER NOT NULL,
created_at TEXT NOT NULL,
PRIMARY KEY(ticket_id, manager_id),
FOREIGN KEY(ticket_id) REFERENCES tickets(id)
);
CREATE TABLE IF NOT EXISTS staff_lead_map (
lead_id INTEGER NOT NULL,
copied_message_id INTEGER NOT NULL,
staff_id INTEGER NOT NULL,
created_at TEXT NOT NULL,
PRIMARY KEY(lead_id, copied_message_id)
);
CREATE TABLE IF NOT EXISTS staff_delivery_outbox (
id INTEGER PRIMARY KEY AUTOINCREMENT,
staff_id INTEGER NOT NULL,
source_chat_id INTEGER NOT NULL,
source_message_id INTEGER NOT NULL,
target_id INTEGER NOT NULL,
dedupe_key TEXT NOT NULL UNIQUE,
status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN ('pending','processing','delivered')),
attempts INTEGER NOT NULL DEFAULT 0,
next_attempt_at REAL NOT NULL DEFAULT 0,
locked_at REAL,
last_error TEXT,
created_at TEXT NOT NULL,
delivered_at TEXT
);
CREATE INDEX IF NOT EXISTS idx_staff_outbox_ready ON staff_delivery_outbox(status, next_attempt_at, id);
CREATE TABLE IF NOT EXISTS webhook_inbox (
update_id INTEGER PRIMARY KEY,
payload TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN ('pending','processing','done')),
attempts INTEGER NOT NULL DEFAULT 0,
next_attempt_at REAL NOT NULL DEFAULT 0,
locked_at REAL,
received_at TEXT NOT NULL,
processed_at TEXT,
last_error TEXT
);
CREATE INDEX IF NOT EXISTS idx_webhook_inbox_ready ON webhook_inbox(status, next_attempt_at, update_id);
"""
)
# Migrate old v1 DBs whose shift duration was restricted to 8 or 12 hours.
manager_sql_row = conn.execute("SELECT sql FROM sqlite_master WHERE type='table' AND name='managers'").fetchone()
manager_sql = (manager_sql_row["sql"] or "") if manager_sql_row else ""
if "IN (8, 12)" in manager_sql or "IN(8,12)" in manager_sql.replace(" ", ""):
conn.execute("PRAGMA foreign_keys = OFF")
conn.execute("BEGIN IMMEDIATE")
conn.execute("CREATE TABLE managers_v2 (telegram_id INTEGER PRIMARY KEY, name TEXT NOT NULL, shift_start TEXT NOT NULL DEFAULT '00:00', shift_hours INTEGER NOT NULL DEFAULT 12 CHECK (shift_hours BETWEEN 1 AND 24), enabled INTEGER NOT NULL DEFAULT 1, on_duty INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL)")
conn.execute("INSERT INTO managers_v2 (telegram_id,name,shift_start,shift_hours,enabled,on_duty,created_at) SELECT telegram_id,name,shift_start,CASE WHEN shift_hours BETWEEN 1 AND 24 THEN shift_hours ELSE 12 END,enabled,on_duty,created_at FROM managers")
conn.execute("DROP TABLE managers")
conn.execute("ALTER TABLE managers_v2 RENAME TO managers")
conn.commit()
conn.execute("PRAGMA foreign_keys = ON")
conn.execute("PRAGMA journal_mode = WAL")
# Keep SQLite useful under concurrent handlers and ensure a Telegram update is not
# recorded twice when long polling retries the same update after a network hiccup.
conn.execute("DELETE FROM messages WHERE telegram_message_id IS NOT NULL AND id NOT IN (SELECT MIN(id) FROM messages WHERE telegram_message_id IS NOT NULL GROUP BY ticket_id, sender_id, telegram_message_id)")
conn.execute("CREATE UNIQUE INDEX IF NOT EXISTS uq_message_update ON messages(ticket_id, sender_id, telegram_message_id) WHERE telegram_message_id IS NOT NULL")
for uid in getattr(config, "OWNER_IDS", []):
if isinstance(uid, int) and uid > 0:
conn.execute(
"INSERT OR IGNORE INTO staff_roles (telegram_id, role, enabled, created_at) VALUES (?, 'owner', 1, ?)",
(uid, now_text()),
)
for uid in getattr(config, "HEAD_MANAGER_IDS", []):
if isinstance(uid, int) and uid > 0:
conn.execute(
"INSERT OR IGNORE INTO staff_roles (telegram_id, role, enabled, created_at) VALUES (?, 'head', 1, ?)",
(uid, now_text()),
)
# Privileged staff must not also appear in the normal manager assignment pool.
privileged_ids = [x for x in (list(getattr(config, "OWNER_IDS", [])) + list(getattr(config, "HEAD_MANAGER_IDS", []))) if isinstance(x, int) and x > 0]
if privileged_ids:
placeholders = ",".join("?" for _ in privileged_ids)
conn.execute(f"UPDATE managers SET enabled=0, on_duty=0 WHERE telegram_id IN ({placeholders})", privileged_ids)
# Configured system-admin IDs are authoritative. Dynamic Head Managers remain intact.
configured_owners = [x for x in getattr(config, "OWNER_IDS", []) if isinstance(x, int) and x > 0]
if configured_owners:
placeholders = ",".join("?" for _ in configured_owners)
conn.execute(f"DELETE FROM staff_roles WHERE role='owner' AND telegram_id NOT IN ({placeholders})", configured_owners)
conn.execute("INSERT OR IGNORE INTO settings (setting_key, setting_value) VALUES ('last_manager_id', '0')")
def audit(actor_id: int, action: str, details: str = "", ticket_id: Optional[int] = None) -> None:
with db_connect() as conn:
conn.execute(
"INSERT INTO audit_logs (actor_id, action, details, ticket_id, created_at) VALUES (?, ?, ?, ?, ?)",
(actor_id, action, details[:1000], ticket_id, now_text()),
)
def get_setting(key: str) -> Optional[str]:
with db_connect() as conn:
row = conn.execute("SELECT setting_value FROM settings WHERE setting_key=?", (key,)).fetchone()
return str(row["setting_value"]) if row else None
def set_setting(key: str, value: str) -> None:
with db_connect() as conn:
conn.execute(
"INSERT INTO settings(setting_key, setting_value) VALUES(?, ?) "
"ON CONFLICT(setting_key) DO UPDATE SET setting_value=excluded.setting_value",
(key, value),
)
def enqueue_delivery(ticket_id: int, kind: str, source_chat_id: Optional[int] = None,
source_message_id: Optional[int] = None) -> None:
"""Durably save a notification/message before any Telegram delivery attempt."""
if kind == "notice":
dedupe = f"notice:{ticket_id}"
else:
dedupe = f"message:{source_chat_id}:{source_message_id}"
with db_connect() as conn:
conn.execute(
"INSERT OR IGNORE INTO delivery_outbox "
"(ticket_id, kind, source_chat_id, source_message_id, dedupe_key, status, next_attempt_at, created_at) "
"VALUES (?, ?, ?, ?, ?, 'pending', 0, ?)",
(ticket_id, kind, source_chat_id, source_message_id, dedupe, now_text()),
)
def staff_role(uid: int) -> Optional[str]:
with db_connect() as conn:
row = conn.execute("SELECT role FROM staff_roles WHERE telegram_id=? AND enabled=1", (uid,)).fetchone()
if row:
return row["role"]
row = conn.execute("SELECT enabled FROM managers WHERE telegram_id=?", (uid,)).fetchone()
if row and row["enabled"]:
return "manager"
return None
def is_owner(uid: int) -> bool:
return staff_role(uid) == "owner"
def is_head(uid: int) -> bool:
return staff_role(uid) == "head"
def is_manager(uid: int) -> bool:
return staff_role(uid) == "manager"
def is_staff(uid: int) -> bool:
return staff_role(uid) is not None
def staff_name(uid: Optional[int]) -> str:
if uid is None:
return "Unassigned"
role = staff_role(uid)
if role == "owner":
return "System Admin"
if role == "head":
return "Head Manager"
with db_connect() as conn:
row = conn.execute("SELECT name FROM managers WHERE telegram_id=?", (uid,)).fetchone()
return row["name"] if row else f"Staff {uid}"
def public_assignee_label(ticket: sqlite3.Row) -> str:
"""Avoid exposing internal staff hierarchy in customer-facing messages."""
if ticket["assigned_role"] == "manager" and ticket["assigned_to"]:
return "Dedicated Support Manager"
if ticket["assigned_to"]:
return "Senior Support Team"
return "Support Queue"
def owner_ids() -> list[int]:
with db_connect() as conn:
rows = conn.execute("SELECT telegram_id FROM staff_roles WHERE role='owner' AND enabled=1 ORDER BY telegram_id").fetchall()
return [int(r["telegram_id"]) for r in rows]
def head_ids() -> list[int]:
with db_connect() as conn:
rows = conn.execute("SELECT telegram_id FROM staff_roles WHERE role='head' AND enabled=1 ORDER BY telegram_id").fetchall()
return [int(r["telegram_id"]) for r in rows]
def escalation_candidates(exclude: Optional[int] = None) -> list[int]:
ordered = head_ids() + owner_ids()
result: list[int] = []
for uid in ordered:
if uid != exclude and uid not in result:
result.append(uid)
return result
def public_commands() -> list[BotCommand]:
return [BotCommand("start", "Open SPG support"), BotCommand("help", "Help and support options")]
def staff_commands(uid: int) -> list[BotCommand]:
role = staff_role(uid)
if role == "owner":
pairs = [
("start", "Open support"), ("help", "System help"), ("panel", "Control panel"),
("addmanager", "Add manager"), ("removemanager", "Disable manager"), ("managers", "Manager roster"),
("shift", "Set shift hours"), ("sethead", "Set head manager"), ("assign", "Assign ticket"),
("takeover", "Take over ticket"), ("tickets", "All open tickets"), ("transcript", "Ticket transcript"),
("stats", "Support statistics"), ("queue", "Delivery queue"), ("close", "Close ticket"),
("escalate", "Escalate ticket"), ("reply", "Reply to ticket"), ("audit", "Staff action audit"),
("channels", "Optional channel links"), ("setwelcomevideo", "Set welcome video"),
("clearwelcomevideo", "Clear welcome video"), ("contactlead", "Contact senior support"), ("cancel", "Cancel current mode")
]
elif role == "head":
pairs = [("start", "Open support"), ("help", "Permitted commands"), ("panel", "Control panel"),
("mytickets", "My tickets"), ("tickets", "All open tickets"), ("managers", "Manager roster"),
("stats", "Support statistics"), ("queue", "Delivery queue"), ("takeover", "Take over ticket"),
("assign", "Assign ticket"), ("transcript", "Ticket transcript"), ("audit", "Staff action audit"),
("close", "Close ticket"), ("escalate", "Escalate ticket"), ("reply", "Reply to ticket"),
("contactlead", "Contact senior support"), ("cancel", "Cancel current mode")]
elif role == "manager":
pairs = [("start", "Open support"), ("help", "Permitted commands"), ("panel", "Manager panel"),
("mytickets", "My tickets"), ("duty", "Set duty on or off"), ("escalate", "Escalate ticket"),
("close", "Close assigned ticket"), ("reply", "Reply to ticket"), ("contactlead", "Contact senior support"), ("cancel", "Cancel current mode")]
else:
return public_commands()
return [BotCommand(cmd, desc) for cmd, desc in pairs]
_COMMAND_MENU_SYNCED: dict[int, str] = {}
def _has_staff_record(uid: int) -> bool:
"""Detect a previously privileged user so stale per-chat commands can be reset."""
with db_connect() as conn:
row = conn.execute(
"SELECT 1 FROM managers WHERE telegram_id=? UNION SELECT 1 FROM staff_roles WHERE telegram_id=? LIMIT 1",
(uid, uid),
).fetchone()
return row is not None
async def sync_command_menu(bot, uid: int) -> None:
# The global bot menu already contains only /start and /help for normal customers.
# Avoid one extra Bot API round-trip for every public /start; set a private scope only
# for active staff or a previously registered staff account needing its menu cleared.
role = staff_role(uid)
if role is None and not _has_staff_record(uid):
return
effective_role = role or "public"
if _COMMAND_MENU_SYNCED.get(uid) == effective_role:
return
try:
commands = staff_commands(uid) if role else public_commands()
await bot.set_my_commands(commands, scope=BotCommandScopeChat(chat_id=uid))
_COMMAND_MENU_SYNCED[uid] = effective_role
except TelegramError as exc:
logger.warning("Could not update private command menu for user %s: %s", uid, exc)
def parse_start_time(value: str) -> Optional[dtime]:
try:
parsed = datetime.strptime(value, "%H:%M").time()
return parsed
except (ValueError, TypeError):
return None
def shift_is_active(manager: sqlite3.Row, at: Optional[datetime] = None) -> bool:
at = at or now_dt()
start_clock = parse_start_time(manager["shift_start"])
if start_clock is None:
return False
start_today = datetime.combine(at.date(), start_clock, tzinfo=TZ)
duration = timedelta(hours=int(manager["shift_hours"]))
for start in (start_today, start_today - timedelta(days=1)):
if start <= at < start + duration:
return True
return False
def manager_available(uid: int) -> bool:
with db_connect() as conn:
row = conn.execute("SELECT * FROM managers WHERE telegram_id=? AND enabled=1 AND on_duty=1", (uid,)).fetchone()
return bool(row and shift_is_active(row))
def available_managers() -> list[sqlite3.Row]:
with db_connect() as conn:
rows = conn.execute("SELECT * FROM managers WHERE enabled=1 AND on_duty=1 ORDER BY telegram_id").fetchall()
return [r for r in rows if shift_is_active(r)]
def active_ticket_for_user(user_id: int) -> Optional[sqlite3.Row]:
with db_connect() as conn:
return conn.execute(
"SELECT * FROM tickets WHERE user_id=? AND status='open' ORDER BY id DESC LIMIT 1", (user_id,)
).fetchone()
def get_ticket(ticket_id: int) -> Optional[sqlite3.Row]:
with db_connect() as conn:
return conn.execute("SELECT * FROM tickets WHERE id=?", (ticket_id,)).fetchone()
def open_ticket_count(assignee: int) -> int:
with db_connect() as conn:
row = conn.execute(
"SELECT COUNT(*) AS n FROM tickets WHERE assigned_to=? AND status='open'", (assignee,)
).fetchone()
return int(row["n"])
def choose_manager() -> Optional[int]:
pool = available_managers()
if not pool:
return None
counts = {int(m["telegram_id"]): open_ticket_count(int(m["telegram_id"])) for m in pool}
least = min(counts.values())
candidates = sorted([uid for uid, count in counts.items() if count == least])
with db_connect() as conn:
row = conn.execute("SELECT setting_value FROM settings WHERE setting_key='last_manager_id'").fetchone()
last = int(row["setting_value"]) if row else 0
choice = next((uid for uid in candidates if uid > last), candidates[0])
conn.execute("UPDATE settings SET setting_value=? WHERE setting_key='last_manager_id'", (str(choice),))
return choice
def add_message(ticket_id: int, sender_id: int, sender_role: str, message_type: str, text_content: str,
file_id: Optional[str], telegram_message_id: Optional[int]) -> None:
with db_connect() as conn:
conn.execute(
"INSERT OR IGNORE INTO messages (ticket_id, sender_id, sender_role, message_type, text_content, file_id, telegram_message_id, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
(ticket_id, sender_id, sender_role, message_type, (text_content or "")[:4000], file_id, telegram_message_id, now_text()),
)
conn.execute("UPDATE tickets SET updated_at=? WHERE id=?", (now_text(), ticket_id))
def message_file(message) -> tuple[str, Optional[str]]:
if message.photo:
return "photo", message.photo[-1].file_id
for attr, kind in [
("document", "document"), ("video", "video"), ("animation", "animation"),
("voice", "voice"), ("audio", "audio"), ("video_note", "video_note"),
("sticker", "sticker"),
]:
obj = getattr(message, attr, None)
if obj:
return kind, getattr(obj, "file_id", None)
if message.location:
return "location", None
if message.contact:
return "contact", None
if message.poll:
return "poll", None
if message.text:
return "text", None
return "other", None
def classify_message(text: str) -> tuple[str, bool]:
low = (text or "").lower()
if any(word.lower() in low for word in config.EMERGENCY_KEYWORDS):
# Urgency is detected from message content; there is deliberately no emergency button.
return "other", True
if any(word.lower() in low for word in config.WITHDRAWAL_KEYWORDS):
return "withdrawal", False
if any(word.lower() in low for word in config.DEPOSIT_KEYWORDS):
return "finance", False
if any(word.lower() in low for word in config.ACCOUNT_KEYWORDS):
return "account", False
return "other", False
def membership_chat_id_configured(channel: dict[str, Any]) -> bool:
return channel.get("chat_id") is not None and str(channel.get("chat_id")).strip() != ""
async def check_membership(bot, user_id: int) -> tuple[bool, str]:
"""Channel membership is optional. This function remains for compatibility with callbacks."""
return True, "Channel joining is optional; support is open to everyone."
def join_keyboard() -> InlineKeyboardMarkup:
"""Optional channel shortcuts; never a gate to support."""
return home_keyboard()
def home_keyboard() -> InlineKeyboardMarkup:
rows = [
[InlineKeyboardButton("💬 Open Support", callback_data="open_support"), InlineKeyboardButton("🎫 My Ticket", callback_data="my_ticket")],
[InlineKeyboardButton("💰 Deposit / Finance", callback_data="category:finance"), InlineKeyboardButton("🏦 Withdrawal", callback_data="category:withdrawal")],
[InlineKeyboardButton("⏳ Payment Pending", callback_data="category:payment"), InlineKeyboardButton("👤 Account / Login", callback_data="category:account")],
[InlineKeyboardButton("🛡️ Scam / Fraud", callback_data="category:scam"), InlineKeyboardButton("💬 Other Issue", callback_data="category:other")],
]
channels = getattr(config, "REQUIRED_CHANNELS", [])
if channels:
rows.append([InlineKeyboardButton("📢 Prediction", url=channels[0]["url"])])
if len(channels) > 1:
rows[-1].append(InlineKeyboardButton("📰 Updates", url=channels[1]["url"]))
if len(channels) > 2:
rows[-1].append(InlineKeyboardButton("🎁 Gift Codes", url=channels[2]["url"]))
return InlineKeyboardMarkup(rows)
def category_keyboard() -> InlineKeyboardMarkup:
return InlineKeyboardMarkup([
[InlineKeyboardButton("💰 Deposit / Finance", callback_data="category:finance"), InlineKeyboardButton("🏦 Withdrawal", callback_data="category:withdrawal")],
[InlineKeyboardButton("⏳ Payment Pending", callback_data="category:payment"), InlineKeyboardButton("🛡️ Scam / Fraud", callback_data="category:scam")],
[InlineKeyboardButton("👤 Account / Login", callback_data="category:account"), InlineKeyboardButton("💬 Other Issue", callback_data="category:other")],
[InlineKeyboardButton("⬅️ Main Menu", callback_data="home")],
])
async def send_welcome_banner(bot, chat_id: int, reply_markup: InlineKeyboardMarkup) -> bool:
"""Send banner first, reusing Telegram file_id after the initial upload."""
cached = get_setting("welcome_photo_file_id")
short_caption = "SEIKH PRIME GAME | SPG SUPPORT CENTER\nChoose an option below / नीचे विकल्प चुनें"
try:
if cached:
try:
await bot.send_photo(chat_id=chat_id, photo=cached, caption=short_caption, parse_mode="HTML",
reply_markup=reply_markup, connect_timeout=5, read_timeout=8, write_timeout=10, pool_timeout=5)
return True
except TelegramError as exc:
logger.warning("Cached welcome banner failed; refreshing file id: %s", exc)
set_setting("welcome_photo_file_id", "")
if IMAGE_PATH.exists():
with IMAGE_PATH.open("rb") as photo:
sent = await bot.send_photo(chat_id=chat_id, photo=photo, caption=short_caption, parse_mode="HTML",
reply_markup=reply_markup, connect_timeout=5, read_timeout=8, write_timeout=10, pool_timeout=5)
if sent.photo:
set_setting("welcome_photo_file_id", sent.photo[-1].file_id)
return True
except TelegramError as exc:
logger.warning("Welcome banner failed for chat %s; falling back to text: %s", chat_id, exc)
return False
async def send_home(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
user = update.effective_user
chat = update.effective_chat
if not user or not chat:
return
if chat.type != ChatType.PRIVATE:
await update.effective_message.reply_text("Please open this bot in a private chat to use support.")
return
keyboard = home_keyboard()
banner_sent = await send_welcome_banner(context.bot, chat.id, keyboard)
if not banner_sent:
# Fail open: users still receive support choices if image delivery fails.
await context.bot.send_message(chat_id=chat.id, text="SEIKH PRIME GAME | SPG SUPPORT CENTER",
parse_mode="HTML", reply_markup=keyboard, connect_timeout=5, read_timeout=8,
write_timeout=10, pool_timeout=5)
await context.bot.send_message(chat_id=chat.id, text=config.WELCOME_TEXT, parse_mode="HTML",
connect_timeout=5, read_timeout=10, write_timeout=10, pool_timeout=5)
saved_video = get_setting("welcome_video_file_id")
if saved_video:
try:
await context.bot.send_video(chat_id=chat.id, video=saved_video, connect_timeout=5, read_timeout=10, write_timeout=10, pool_timeout=5)
except TelegramError as exc:
logger.warning("Configured welcome video could not be sent: %s", exc)
async def cmd_start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
user = update.effective_user
if user:
await sync_command_menu(context.bot, user.id)
# Avoid duplicate slow welcome deliveries when users tap /start repeatedly.
last_start = float(context.user_data.get("last_start_sent", 0) or 0)
if time.time() - last_start < 5:
return
context.user_data["last_start_sent"] = time.time()
audit(user.id, "start", "User opened support bot")
await send_home(update, context)
async def cmd_id(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
user = update.effective_user
if user:
await update.effective_message.reply_text(f"Your Telegram user ID: {user.id}\nName: {user.full_name}")
def staff_panel_keyboard(uid: int) -> InlineKeyboardMarkup:
if is_owner(uid) or is_head(uid):
rows = [
[InlineKeyboardButton("🎫 My Tickets", callback_data="panel:mytickets"), InlineKeyboardButton("📋 All Tickets", callback_data="panel:alltickets")],
[InlineKeyboardButton("📊 Stats", callback_data="panel:stats"), InlineKeyboardButton("📬 Queue", callback_data="panel:queue")],
[InlineKeyboardButton("👥 Managers", callback_data="panel:managers"), InlineKeyboardButton("📨 Senior Support", callback_data="panel:contactlead")],
[InlineKeyboardButton("❔ Help", callback_data="panel:help")],
]
elif is_manager(uid):
rows = [
[InlineKeyboardButton("🎫 My Tickets", callback_data="panel:mytickets")],
[InlineKeyboardButton("🟢 Duty ON", callback_data="panel:duty_on"), InlineKeyboardButton("🔴 Duty OFF", callback_data="panel:duty_off")],
[InlineKeyboardButton("📨 Senior Support", callback_data="panel:contactlead")],
[InlineKeyboardButton("❔ Help", callback_data="panel:help")],
]
else:
rows = [[InlineKeyboardButton("🎧 Open Support / सहायता", callback_data="open_support")]]
return InlineKeyboardMarkup(rows)
async def cmd_panel(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
uid = update.effective_user.id
if not is_staff(uid):
await update.effective_message.reply_text("Customer support ke liye /start dabayein.")
return
title = "SPG SUPPORT CONTROL PANEL" if (is_owner(uid) or is_head(uid)) else "SPG MANAGER PANEL"
await update.effective_message.reply_text(title, reply_markup=staff_panel_keyboard(uid))
async def cmd_help(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
uid = update.effective_user.id
await sync_command_menu(context.bot, uid)
if is_owner(uid):
text = (
"🛠️ SYSTEM ADMIN COMMANDS\n/panel\n/addmanager ID Name\n/removemanager ID\n/managers\n"
"/shift ID HH:MM HOURS (1–24)\n/sethead ID\n/assign TICKET_ID STAFF_ID\n/takeover TICKET_ID\n"
"/tickets\n/transcript TICKET_ID\n/stats\n/queue\n/close TICKET_ID\n/escalate TICKET_ID reason\n"
"/reply TICKET_ID message\n/audit\n/channels\n/setwelcomevideo (reply to a video)\n/clearwelcomevideo\n"
"/contactlead\n/cancel\n\nStaff can reply directly to a copied ticket message. Use /contactlead to message senior support privately."
)
elif is_head(uid):
text = (
"🧭 HEAD MANAGER COMMANDS\n/panel\n/mytickets\n/tickets\n/managers\n/stats\n/queue\n"
"/takeover TICKET_ID\n/assign TICKET_ID STAFF_ID\n/transcript TICKET_ID\n/close TICKET_ID\n"
"/escalate TICKET_ID reason\n/reply TICKET_ID message\n/audit\n/contactlead\n/cancel\n\n"
"Reply to a copied ticket message to send a reply. Use /contactlead for private staff-to-senior communication."
)
elif is_manager(uid):
text = (
"👨💼 MANAGER COMMANDS\n/panel\n/mytickets\n/duty on\n/duty off\n"
"/escalate TICKET_ID reason\n/close TICKET_ID\n/reply TICKET_ID message\n/contactlead\n/cancel\n\n"
"Reply to a copied user message to respond. Use /contactlead to privately contact the senior team."
)
else:
text = (
"Welcome to SPG Support. Use /start to open the support menu, choose a category, or send your issue directly.\n\n"
"SPG channels are optional. You never need to join a channel to request support.\n"
"सहायता के लिए /start दबाएँ, कोई विषय चुनें या अपनी समस्या सीधे भेजें। चैनल से जुड़ना ज़रूरी नहीं है।"
)
await update.effective_message.reply_text(text)
async def staff_guard(update: Update, owner_only: bool = False) -> bool:
uid = update.effective_user.id
if owner_only and not is_owner(uid):
await update.effective_message.reply_text("⛔ This command is not enabled for your staff role.")
return False
if not owner_only and not is_staff(uid):
await update.effective_message.reply_text("⛔ Ye command authorized staff ke liye hai.")
return False
return True
async def cmd_addmanager(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if not await staff_guard(update, owner_only=True):
return
if len(context.args) < 2 or not context.args[0].isdigit():
await update.effective_message.reply_text("Usage: /addmanager TELEGRAM_ID Manager_Name")
return
mid = int(context.args[0])
name = " ".join(context.args[1:]).strip()
if mid <= 0 or mid in owner_ids() or mid in head_ids():
await update.effective_message.reply_text("Invalid ID or this user already has a privileged support role.")
return
with db_connect() as conn:
conn.execute(
"INSERT INTO managers (telegram_id, name, shift_start, shift_hours, enabled, on_duty, created_at) VALUES (?, ?, '00:00', 12, 1, 0, ?) "
"ON CONFLICT(telegram_id) DO UPDATE SET name=excluded.name, enabled=1",
(mid, name, now_text()),
)
audit(update.effective_user.id, "add_manager", f"Added manager {mid} ({name})")
await update.effective_message.reply_text(
f"✅ Manager {name} ({mid}) add ho gaya. Default shift 00:00–12:00, 12 hours hai.\n"
"Manager ko bot open karke /start karna hoga; system admin shift set kare aur manager /duty on use kare."
)
try:
await context.bot.send_message(mid, "You have been added as an SPG Support Manager. Open this bot and send /start, then see /help. Set availability with /duty on when your configured shift is active.")
except TelegramError:
await update.effective_message.reply_text("Note: Manager ne bot ko /start nahi kiya hoga. Unse pehle bot open karke /start karwao.")
async def cmd_removemanager(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if not await staff_guard(update, owner_only=True):
return
if len(context.args) != 1 or not context.args[0].isdigit():
await update.effective_message.reply_text("Usage: /removemanager TELEGRAM_ID")
return
mid = int(context.args[0])
with db_connect() as conn:
conn.execute("UPDATE managers SET enabled=0, on_duty=0 WHERE telegram_id=?", (mid,))
tickets = conn.execute("SELECT id FROM tickets WHERE assigned_to=? AND assigned_role='manager' AND status='open'", (mid,)).fetchall()
audit(update.effective_user.id, "remove_manager", f"Disabled manager {mid}; open tickets={len(tickets)}")
await update.effective_message.reply_text(f"✅ Manager {mid} disabled. Open tickets: {len(tickets)}. Ab un tickets ka agla user message senior support fallback par jayega; doosre managers ko silently forward nahi hoga.")
async def cmd_managers(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if not (is_owner(update.effective_user.id) or is_head(update.effective_user.id)):
await update.effective_message.reply_text("Manager roster is not enabled for your staff role.")
return
with db_connect() as conn:
rows = conn.execute("SELECT * FROM managers ORDER BY name COLLATE NOCASE").fetchall()
if not rows:
await update.effective_message.reply_text("Abhi koi manager add nahi hai. Authorized setup command: /addmanager ID Name")
return
lines = ["👨💼 MANAGER ROSTER"]
for m in rows:
uid = int(m["telegram_id"])
duty = "ON DUTY" if manager_available(uid) else ("Marked ON, shift inactive" if m["on_duty"] else "OFF DUTY")
count = open_ticket_count(uid)
status = "enabled" if m["enabled"] else "disabled"
lines.append(f"• {m['name']} | ID {uid} | shift {m['shift_start']} / {m['shift_hours']}h | {duty} | open tickets {count} | {status}")
await update.effective_message.reply_text("\n".join(lines))
async def cmd_shift(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if not await staff_guard(update, owner_only=True):
return
if len(context.args) != 3 or not context.args[0].isdigit() or parse_start_time(context.args[1]) is None or not context.args[2].isdigit() or not 1 <= int(context.args[2]) <= 24:
await update.effective_message.reply_text("Usage: /shift TELEGRAM_ID HH:MM HOURS (1–24)\nExample: /shift 123456789 14:00 10")
return
mid, start, hours = int(context.args[0]), context.args[1], int(context.args[2])
with db_connect() as conn:
cur = conn.execute("UPDATE managers SET shift_start=?, shift_hours=? WHERE telegram_id=? AND enabled=1", (start, hours, mid))
if cur.rowcount == 0:
await update.effective_message.reply_text("Enabled manager not found. Pehle /addmanager use karo.")
return
audit(update.effective_user.id, "set_shift", f"Manager {mid} shift={start}/{hours}h")
await update.effective_message.reply_text(f"✅ Manager {mid} ki shift {start} se {hours} hours set ho gayi ({config.TIMEZONE}). Manager ko /duty on mark karna hoga.")
async def cmd_duty(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
uid = update.effective_user.id
if not is_manager(uid):
await update.effective_message.reply_text("Ye command registered managers ke liye hai.")
return
if len(context.args) != 1 or context.args[0].lower() not in ("on", "off"):
await update.effective_message.reply_text("Usage: /duty on ya /duty off")
return
state = 1 if context.args[0].lower() == "on" else 0
with db_connect() as conn:
conn.execute("UPDATE managers SET on_duty=? WHERE telegram_id=?", (state, uid))
audit(uid, "duty_change", "on" if state else "off")
if state and not manager_available(uid):
with db_connect() as conn:
row = conn.execute("SELECT * FROM managers WHERE telegram_id=?", (uid,)).fetchone()
await update.effective_message.reply_text(f"🕒 Aap ON mark ho gaye, lekin aapki scheduled shift abhi active nahi hai ({row['shift_start']}, {row['shift_hours']}h, {config.TIMEZONE}). Is waqt naye tickets aapko assign nahi honge.")
else:
await update.effective_message.reply_text("✅ Aap ab " + ("ON DUTY" if state else "OFF DUTY") + " hain.")
async def cmd_sethead(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if not await staff_guard(update, owner_only=True):
return
if len(context.args) != 1 or not context.args[0].isdigit():
await update.effective_message.reply_text("Usage: /sethead TELEGRAM_ID")
return
hid = int(context.args[0])
if hid in owner_ids():
await update.effective_message.reply_text("Ye ID pehle se privileged staff hai.")
return
with db_connect() as conn:
conn.execute("INSERT INTO staff_roles (telegram_id, role, enabled, created_at) VALUES (?, 'head', 1, ?) ON CONFLICT(telegram_id) DO UPDATE SET role='head', enabled=1", (hid, now_text()))
# A Head Manager must not simultaneously remain in the regular manager routing pool.
conn.execute("UPDATE managers SET enabled=0, on_duty=0 WHERE telegram_id=?", (hid,))
audit(update.effective_user.id, "set_head", f"Head manager {hid}")
await update.effective_message.reply_text(f"✅ {hid} Head Manager bana diya. Unhe is bot par /start bhejna hoga.")
try:
await context.bot.send_message(hid, "You have been added as an SPG Head Manager. Send /start to this bot, then /help.")
except TelegramError:
await update.effective_message.reply_text("Note: Head Manager ko pehle bot open karke /start karna hoga.")
async def cmd_mytickets(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
uid = update.effective_user.id
if not is_staff(uid):
await update.effective_message.reply_text("Ye command staff ke liye hai.")
return
with db_connect() as conn:
rows = conn.execute("SELECT * FROM tickets WHERE assigned_to=? AND status='open' ORDER BY updated_at DESC LIMIT 30", (uid,)).fetchall()
if not rows:
await update.effective_message.reply_text("Aapke paas abhi koi open ticket nahi hai.")
return
lines = ["🎫 AAPKE OPEN TICKETS"]
for t in rows:
lines.append(f"#{t['id']} | {t['category']} | {t['priority']} | user {t['user_id']} | updated {t['updated_at']}")
await update.effective_message.reply_text("\n".join(lines))
async def cmd_tickets(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if not (is_owner(update.effective_user.id) or is_head(update.effective_user.id)):
await update.effective_message.reply_text("All-ticket list is not enabled for your staff role. Use /mytickets for your assigned tickets.")
return
with db_connect() as conn:
rows = conn.execute("SELECT * FROM tickets WHERE status='open' ORDER BY priority DESC, updated_at DESC LIMIT 50").fetchall()
if not rows:
await update.effective_message.reply_text("Koi open ticket nahi hai.")
return
lines = ["🎫 OPEN TICKETS (max 50)"]
for t in rows:
lines.append(f"#{t['id']} | {t['category']} | {t['priority']} | user {t['user_id']} | assigned {staff_name(t['assigned_to'])} ({t['assigned_to']}) | {t['status']}")
await update.effective_message.reply_text("\n".join(lines))
async def cmd_stats(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if not (is_owner(update.effective_user.id) or is_head(update.effective_user.id)):
await update.effective_message.reply_text("Support statistics is not enabled for your staff role.")
return
with db_connect() as conn:
total_users = conn.execute("SELECT COUNT(DISTINCT user_id) n FROM tickets").fetchone()["n"]
open_t = conn.execute("SELECT COUNT(*) n FROM tickets WHERE status='open'").fetchone()["n"]
closed_t = conn.execute("SELECT COUNT(*) n FROM tickets WHERE status='closed'").fetchone()["n"]
escalated = conn.execute("SELECT COUNT(*) n FROM tickets WHERE priority='emergency' AND status='open'").fetchone()["n"]
enabled_m = conn.execute("SELECT COUNT(*) n FROM managers WHERE enabled=1").fetchone()["n"]
on_duty = len(available_managers())
await update.effective_message.reply_text(
"📊 SPG SUPPORT STATS\n"
f"Unique users: {total_users}\nOpen tickets: {open_t}\nClosed tickets: {closed_t}\n"
f"Open emergency tickets: {escalated}\nEnabled managers: {enabled_m}\nManagers on duty/in shift: {on_duty}"
)
async def cmd_close(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
uid = update.effective_user.id
if not is_staff(uid):
await update.effective_message.reply_text("Ye command staff ke liye hai.")
return
if len(context.args) != 1 or not context.args[0].isdigit():
await update.effective_message.reply_text("Usage: /close TICKET_ID")
return
tid = int(context.args[0])
ticket = get_ticket(tid)
if not ticket or ticket["status"] != "open":
await update.effective_message.reply_text("Open ticket nahi mila.")
return
if not (is_owner(uid) or is_head(uid) or int(ticket["assigned_to"] or 0) == uid):
await update.effective_message.reply_text("⛔ Sirf assigned staff ya senior support ticket close kar sakta hai.")
return
with db_connect() as conn:
conn.execute("UPDATE tickets SET status='closed', closed_at=?, updated_at=? WHERE id=?", (now_text(), now_text(), tid))
audit(uid, "close_ticket", "Ticket closed", tid)
try:
await context.bot.send_message(int(ticket["user_id"]), f"✅ Aapka SPG support ticket #{tid} resolved/closed mark kiya gaya hai. Nayi problem ke liye /start se naya ticket khol sakte hain.")
except TelegramError:
pass
await update.effective_message.reply_text(f"✅ Ticket #{tid} close kar diya.")
async def cmd_takeover(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
uid = update.effective_user.id
if not (is_owner(uid) or is_head(uid)):
await update.effective_message.reply_text("Ticket takeover is not enabled for your staff role.")
return
if len(context.args) != 1 or not context.args[0].isdigit():
await update.effective_message.reply_text("Usage: /takeover TICKET_ID")
return
tid = int(context.args[0])
ticket = get_ticket(tid)
if not ticket or ticket["status"] != "open":
await update.effective_message.reply_text("Open ticket nahi mila.")
return
with db_connect() as conn:
conn.execute("UPDATE tickets SET assigned_to=?, assigned_role=?, escalated_from=?, updated_at=? WHERE id=?", (uid, staff_role(uid), ticket["assigned_to"], now_text(), tid))
audit(uid, "takeover", f"Ticket taken over by {uid}", tid)
await update.effective_message.reply_text(f"✅ Ticket #{tid} ab aapke paas assigned hai. User ke agle messages seedha aapko milenge.")
try:
await context.bot.send_message(int(ticket["user_id"]), f"Aapka ticket #{tid} ab senior support team handle karegi.")
except TelegramError:
pass
async def cmd_assign(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
uid = update.effective_user.id
if not (is_owner(uid) or is_head(uid)):
await update.effective_message.reply_text("Manual assignment is not enabled for your staff role.")
return
if len(context.args) != 2 or not all(x.isdigit() for x in context.args):
await update.effective_message.reply_text("Usage: /assign TICKET_ID STAFF_TELEGRAM_ID")
return
tid, target = int(context.args[0]), int(context.args[1])
ticket = get_ticket(tid)
role = staff_role(target)
if not ticket or ticket["status"] != "open" or role not in ("manager", "owner", "head"):
await update.effective_message.reply_text("Ticket ya target staff invalid hai.")
return
if role == "manager" and not manager_available(target):
await update.effective_message.reply_text("Ye manager abhi available/on-shift nahi hai. Duty aur schedule check karo.")
return
with db_connect() as conn:
conn.execute("UPDATE tickets SET assigned_to=?, assigned_role=?, escalated_from=?, updated_at=? WHERE id=?", (target, role, ticket["assigned_to"], now_text(), tid))
audit(uid, "manual_assign", f"Ticket assigned to {target} role={role}", tid)
await update.effective_message.reply_text(f"✅ Ticket #{tid} {staff_name(target)} ({target}) ko assign ho gaya.")
try:
await context.bot.send_message(target, f"🔁 Ticket #{tid} aapko senior support ne assign kiya. Category: {ticket['category']}\nUser ID: {ticket['user_id']}\nReply to a copied user message to respond, or use /reply TICKET_ID message.")
except TelegramError:
await update.effective_message.reply_text("Target ne bot par /start nahi kiya ya DM unavailable hai.")
try:
await context.bot.send_message(int(ticket["user_id"]), f"Aapke support ticket #{tid} ko authorized staff ke paas transfer kiya gaya hai.")
except TelegramError:
pass
async def cmd_escalate(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
uid = update.effective_user.id
if not is_staff(uid):
await update.effective_message.reply_text("Ye command staff ke liye hai.")
return
if len(context.args) < 1 or not context.args[0].isdigit():
await update.effective_message.reply_text("Usage: /escalate TICKET_ID optional_reason")
return
tid = int(context.args[0])
reason = " ".join(context.args[1:]).strip() or "Manager requested escalation"
ticket = get_ticket(tid)
if not ticket or ticket["status"] != "open":
await update.effective_message.reply_text("Open ticket nahi mila.")
return
if not (is_owner(uid) or is_head(uid) or int(ticket["assigned_to"] or 0) == uid):
await update.effective_message.reply_text("⛔ Aap sirf apne assigned ticket ko escalate kar sakte hain.")
return
targets = escalation_candidates(exclude=uid)
if not targets:
await update.effective_message.reply_text("Senior support recipients configure nahi hain. config.py mein system admin/head IDs check karo.")
return
target = targets[0]
with db_connect() as conn:
conn.execute("UPDATE tickets SET assigned_to=?, assigned_role=?, priority='high', escalated_from=?, last_escalation_reason=?, updated_at=? WHERE id=?", (target, staff_role(target) or "owner", uid, reason, now_text(), tid))
audit(uid, "escalate_ticket", f"To {target}: {reason}", tid)
alert = f"🚨 ESCALATED TICKET #{tid}\nFrom staff ID: {uid}\nReason: {reason}\nUser ID: {ticket['user_id']}\nCategory: {ticket['category']}\nPriority: HIGH\nUse /takeover {tid} if you need to take it over."
try:
await context.bot.send_message(target, alert)
await context.bot.send_message(int(ticket["user_id"]), f"Aapka ticket #{tid} senior support team ko escalate kar diya gaya hai. Aage ke messages assigned senior ko jayenge.")
except TelegramError:
pass
await update.effective_message.reply_text(f"✅ Ticket #{tid} {staff_name(target)} ({target}) ko escalate kar diya gaya.")
async def cmd_reply(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
uid = update.effective_user.id
if not is_staff(uid):
await update.effective_message.reply_text("Ye command staff ke liye hai.")
return
if len(context.args) < 2 or not context.args[0].isdigit():
await update.effective_message.reply_text("Usage: /reply TICKET_ID your_message_here")
return
tid = int(context.args[0])
text = " ".join(context.args[1:]).strip()
ticket = get_ticket(tid)
if not ticket or ticket["status"] != "open":
await update.effective_message.reply_text("Open ticket nahi mila.")
return
if int(ticket["assigned_to"] or 0) != uid:
await update.effective_message.reply_text("⛔ Ye ticket aapko assign nahi hai. Senior support se /takeover ya /assign karwayein; reply cross-route nahi hoga.")
return
try:
await context.bot.send_message(int(ticket["user_id"]), f"💬 SPG Support | Ticket #{tid}\n\n{html.escape(text)}", parse_mode="HTML")
except TelegramError:
await update.effective_message.reply_text("User ko message deliver nahi hua. Ho sakta hai user ne bot block kiya ho.")
return
add_message(tid, uid, staff_role(uid) or "staff", "text", text, None, update.effective_message.message_id)
audit(uid, "reply_to_ticket", "Text reply sent", tid)
await update.effective_message.reply_text(f"✅ Reply ticket #{tid} ke user ko bhej diya.")
async def cmd_transcript(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
uid = update.effective_user.id
if not (is_owner(uid) or is_head(uid)):
await update.effective_message.reply_text("Transcript is not enabled for your staff role.")
return
if len(context.args) != 1 or not context.args[0].isdigit():
await update.effective_message.reply_text("Usage: /transcript TICKET_ID")
return
tid = int(context.args[0])
ticket = get_ticket(tid)
if not ticket:
await update.effective_message.reply_text("Ticket nahi mila.")
return
with db_connect() as conn:
rows = conn.execute("SELECT * FROM messages WHERE ticket_id=? ORDER BY id DESC LIMIT 30", (tid,)).fetchall()
rows = list(reversed(rows))
head = f"TRANSCRIPT TICKET #{tid}\nUser ID: {ticket['user_id']}\nCategory: {ticket['category']}\nStatus: {ticket['status']}\nAssigned: {staff_name(ticket['assigned_to'])}\n\n"
parts = [head]
for m in rows:
content = (m["text_content"] or "").strip()
if m["file_id"]:
content += f" [attachment: {m['message_type']} / file_id stored]"
if not content:
content = f"[{m['message_type']}]"
parts.append(f"[{m['created_at']}] {m['sender_role']} {m['sender_id']}: {content[:700]}\n")
output = "".join(parts)
# Telegram text-message limit safety: split the transcript into chunks.
for i in range(0, len(output), 3800):
await update.effective_message.reply_text(output[i:i+3800])
async def cmd_audit(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
uid = update.effective_user.id
if not (is_owner(uid) or is_head(uid)):
await update.effective_message.reply_text("Audit log is not enabled for your staff role.")
return
with db_connect() as conn:
rows = conn.execute("SELECT * FROM audit_logs ORDER BY id DESC LIMIT 20").fetchall()
if not rows:
await update.effective_message.reply_text("Abhi audit entries nahi hain.")
return
lines = ["🧾 LAST 20 STAFF / BOT AUDIT EVENTS"]
for r in rows:
lines.append(f"{r['created_at']} | actor {r['actor_id']} | {r['action']} | ticket {r['ticket_id']} | {r['details'] or ''}")
text = "\n".join(lines)
for i in range(0, len(text), 3800):
await update.effective_message.reply_text(text[i:i+3800])
async def cmd_channels(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if not await staff_guard(update, owner_only=True):
return
lines = ["OPTIONAL SPG CHANNEL LINKS (JOIN NOT REQUIRED)"]
for c in getattr(config, "REQUIRED_CHANNELS", []):
configured = "SET" if membership_chat_id_configured(c) else "MISSING CHAT_ID"
lines.append(f"• {c['name']}: {c['url']}")
lines.append("Users can contact support without joining any channel.")
await update.effective_message.reply_text("\n".join(lines))
async def handle_channel_post_id(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Privileged helper: after bot admin access, a new channel post reveals chat.id."""
post = update.channel_post
if not post or not post.chat or post.chat.type != ChatType.CHANNEL:
return
message = (
"🔎 SPG channel ID detected\n"
f"Title: {post.chat.title or 'Channel'}\n"
f"Chat ID: {post.chat.id}\n\n"
"Copy this numeric ID into the matching REQUIRED_CHANNELS entry in config.py, save, then restart the bot."
)
for oid in owner_ids():
try:
await context.bot.send_message(oid, message)
except TelegramError:
logger.warning("Could not DM configured admin %s with detected channel ID; make sure the configured admin sent /start", oid)
async def cmd_contactlead(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
uid = update.effective_user.id
if not is_staff(uid):
await update.effective_message.reply_text("This option is available to support staff only.")
return
if not escalation_candidates(exclude=uid):
await update.effective_message.reply_text("Senior support recipients are not configured yet. Ask the system administrator to configure the senior team.")
return
context.user_data["lead_contact_mode"] = True
await update.effective_message.reply_text(
"🔒 Private Senior Support Desk\nAb apna message, photo, video, voice note ya file bhejo. Ye senior team ko privately forward hoga.\n"
"Ticket reply mode mein wapas jaane ke liye /cancel bhejo."
)
async def cmd_cancel(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
context.user_data.pop("lead_contact_mode", None)
await update.effective_message.reply_text("Current mode band kar diya gaya. Normal ticket replies wapas active hain.")
async def cmd_setwelcomevideo(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
uid = update.effective_user.id
if not is_owner(uid):
await update.effective_message.reply_text("This command is not enabled for your staff role.")
return
source = update.effective_message.reply_to_message
if not source:
await update.effective_message.reply_text("Pehle bot ko video bhejo, phir us video par Reply karke /setwelcomevideo bhejo.")
return
file_id = source.video.file_id if source.video else (source.document.file_id if source.document and (source.document.mime_type or '').startswith('video/') else None)
if not file_id:
await update.effective_message.reply_text("Reply ki gayi message mein video/video file nahi mili.")
return
set_setting("welcome_video_file_id", file_id)
audit(uid, "set_welcome_video", "Welcome video file_id saved")
await update.effective_message.reply_text("✅ Welcome video save ho gaya. Ab naye /start par banner aur welcome message ke baad bheja jayega.")
async def cmd_clearwelcomevideo(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if not is_owner(update.effective_user.id):
await update.effective_message.reply_text("This command is not enabled for your staff role.")
return
set_setting("welcome_video_file_id", "")
audit(update.effective_user.id, "clear_welcome_video", "Welcome video cleared")
await update.effective_message.reply_text("✅ Welcome video disabled.")
async def cmd_queue(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
uid = update.effective_user.id
if not (is_owner(uid) or is_head(uid)):
await update.effective_message.reply_text("Queue status is not enabled for your staff role.")
return
with db_connect() as conn:
pending = conn.execute("SELECT COUNT(*) AS n FROM delivery_outbox WHERE status='pending'").fetchone()["n"]
processing = conn.execute("SELECT COUNT(*) AS n FROM delivery_outbox WHERE status='processing'").fetchone()["n"]
delivered = conn.execute("SELECT COUNT(*) AS n FROM delivery_outbox WHERE status='delivered'").fetchone()["n"]
oldest = conn.execute("SELECT created_at FROM delivery_outbox WHERE status='pending' ORDER BY id LIMIT 1").fetchone()
await update.effective_message.reply_text(
f"📬 SPG DELIVERY QUEUE\nPending: {pending}\nProcessing: {processing}\nDelivered (all-time): {delivered}\n"
f"Oldest pending: {oldest['created_at'] if oldest else 'None'}\n\n"
"Queued messages are saved on this device and retry automatically when recipients/connectivity become available."
)
async def create_ticket(user, category_key: str, emergency: bool, context: ContextTypes.DEFAULT_TYPE) -> Optional[sqlite3.Row]:
"""Create exactly one open ticket per user in this bot process; commit before delivery."""
async with _ticket_create_lock:
existing = active_ticket_for_user(user.id)
if existing:
return existing
category = CATEGORIES.get(category_key, CATEGORIES["other"])
assigned = None
role = "unassigned"
priority = "emergency" if emergency else "normal"
if priority == "emergency":
targets = escalation_candidates()
if targets:
assigned, role = targets[0], staff_role(targets[0]) or "owner"
else:
assigned = choose_manager()
if assigned is not None:
role = "manager"
else:
targets = escalation_candidates()
if targets:
assigned, role, priority = targets[0], staff_role(targets[0]) or "owner", "high"
with db_connect() as conn:
cur = conn.execute(
"INSERT INTO tickets (user_id, username, full_name, category, assigned_to, assigned_role, status, priority, created_at, updated_at) "
"VALUES (?, ?, ?, ?, ?, ?, 'open', ?, ?, ?)",
(user.id, user.username, user.full_name, category, assigned, role, priority, now_text(), now_text()),
)
tid = int(cur.lastrowid)
audit(user.id, "create_ticket", f"Category={category}; assigned={assigned}; priority={priority}", tid)
enqueue_delivery(tid, "notice")
return get_ticket(tid)
async def transfer_to_fallback(ticket_id: int, context: ContextTypes.DEFAULT_TYPE, exclude: Optional[int], reason: str,
emergency_alert: bool = False) -> Optional[int]:
ticket = get_ticket(ticket_id)
if not ticket:
return None
candidates = escalation_candidates(exclude=exclude)
alert_text = (
f"{'🚨 EMERGENCY' if emergency_alert else '⚠️ SUPPORT ESCALATION'} — TICKET #{ticket_id}\n"
f"Reason: {reason}\nUser ID: {ticket['user_id']}\nCategory: {ticket['category']}\n"
f"Previous assignee: {ticket['assigned_to']}\nPriority: {ticket['priority']}"
)
for target in candidates:
try:
await context.bot.send_message(target, alert_text)
with db_connect() as conn:
conn.execute(
"UPDATE tickets SET assigned_to=?, assigned_role=?, priority=?, escalated_from=?, last_escalation_reason=?, updated_at=? WHERE id=?",
(target, staff_role(target) or "owner", "emergency" if emergency_alert else "high", ticket["assigned_to"], reason, now_text(), ticket_id),
)
audit(0, "auto_escalate", f"ticket transferred to {target}: {reason}", ticket_id)
return target
except TelegramError as exc:
logger.warning("Escalation recipient %s cannot be reached: %s", target, exc)
return None
def claim_outbox_job() -> Optional[sqlite3.Row]:
"""Claim one due job atomically; earlier jobs for the same ticket must finish first."""
now = time.time()
with db_connect() as conn:
conn.execute("BEGIN IMMEDIATE")
row = conn.execute(
"SELECT d.* FROM delivery_outbox d "
"WHERE d.status='pending' AND d.next_attempt_at<=? "
"AND NOT EXISTS (SELECT 1 FROM delivery_outbox p WHERE p.ticket_id=d.ticket_id "
"AND p.id None:
with db_connect() as conn:
if priority:
conn.execute("UPDATE tickets SET assigned_to=?, assigned_role=?, priority=?, updated_at=? WHERE id=?",
(uid, role, priority, now_text(), ticket_id))
else:
conn.execute("UPDATE tickets SET assigned_to=?, assigned_role=?, updated_at=? WHERE id=?",
(uid, role, now_text(), ticket_id))
def delivery_candidates(ticket: sqlite3.Row) -> list[int]:
"""Emergency goes only to senior support. Normal tickets stay assigned; unavailable staff escalate."""
assigned = int(ticket["assigned_to"] or 0)
role = ticket["assigned_role"]
priority = ticket["priority"]
fallbacks = escalation_candidates()
if priority == "emergency":
candidates = ([assigned] if assigned in fallbacks else []) + fallbacks
elif role == "manager" and assigned and manager_available(assigned):
candidates = [assigned] + fallbacks
elif role == "manager" and assigned and not manager_available(assigned):
# An off-duty manager must not receive fresh messages; escalate only to senior support.
candidates = fallbacks
elif role in ("owner", "head") and assigned:
candidates = [assigned] + [x for x in fallbacks if x != assigned]
else:
available = [int(m["telegram_id"]) for m in available_managers()]
chosen = choose_manager() if not assigned or role == "unassigned" else None
candidates = ([assigned] if assigned else []) + ([chosen] if chosen else []) + available + fallbacks
unique = []
for uid in candidates:
if uid and uid not in unique:
unique.append(uid)
return unique
async def send_ticket_header(bot, ticket: sqlite3.Row, recipient: int) -> bool:
with db_connect() as conn:
notice = conn.execute("SELECT message_id FROM manager_ticket_notices WHERE ticket_id=? AND manager_id=?", (ticket["id"], recipient)).fetchone()
if notice:
return True
header = (
f"🎫 SPG SUPPORT TICKET #{ticket['id']}\n"
f"User: {ticket['full_name']} (@{ticket['username'] or 'no username'})\n"
f"User ID: {ticket['user_id']}\nCategory: {ticket['category']}\nPriority: {ticket['priority'].upper()}\n\n"
"Reply to a copied user message to respond. Commands: /reply TICKET_ID message | /close TICKET_ID | /escalate TICKET_ID reason"
)
sent = await bot.send_message(
chat_id=recipient, text=header, connect_timeout=6, read_timeout=12,
write_timeout=12, pool_timeout=8,
)
with db_connect() as conn:
conn.execute("INSERT OR IGNORE INTO manager_ticket_notices(ticket_id, manager_id, message_id, created_at) VALUES(?, ?, ?, ?)",
(ticket["id"], recipient, sent.message_id, now_text()))
return True
async def process_outbox_job(bot, job: sqlite3.Row) -> None:
ticket = get_ticket(int(job["ticket_id"]))
if not ticket or ticket["status"] != "open":
with db_connect() as conn:
conn.execute("UPDATE delivery_outbox SET status='delivered', delivered_at=?, last_error='Ticket closed or missing; no delivery needed' WHERE id=?",
(now_text(), job["id"]))
return
candidates = delivery_candidates(ticket)
if not candidates:
raise RuntimeError("No configured senior recipient and no on-duty manager")
last_exc: Optional[BaseException] = None
for recipient in candidates:
role = staff_role(recipient) or "owner"
# Do not send emergency tickets to regular managers.
if (ticket["priority"] == "emergency") and role not in ("owner", "head"):
continue
try:
await send_ticket_header(bot, ticket, recipient)
set_assignee(int(ticket["id"]), recipient, role, "emergency" if ticket["priority"] == "emergency" else None)
if job["kind"] == "user_message":
source_chat = int(job["source_chat_id"])
source_message = int(job["source_message_id"])
copied = await bot.copy_message(
chat_id=recipient, from_chat_id=source_chat, message_id=source_message,
connect_timeout=6, read_timeout=15, write_timeout=15, pool_timeout=8,
)
with db_connect() as conn:
conn.execute("INSERT OR REPLACE INTO manager_message_map(manager_id, copied_message_id, ticket_id, created_at) VALUES(?, ?, ?, ?)",
(recipient, copied.message_id, ticket["id"], now_text()))
with db_connect() as conn:
conn.execute("UPDATE delivery_outbox SET status='delivered', delivered_at=?, locked_at=NULL, last_error=NULL WHERE id=?",
(now_text(), job["id"]))
conn.execute("UPDATE tickets SET updated_at=? WHERE id=?", (now_text(), ticket["id"]))
return
except TelegramError as exc:
last_exc = exc
logger.warning("Outbox job %s delivery to %s failed: %s", job["id"], recipient, exc)
# Try senior support backup for regular tickets, and other senior support for emergencies.
if role in ("owner", "head"):
continue
if last_exc:
raise last_exc
raise RuntimeError("No valid delivery recipient was available for this ticket")
async def outbox_worker(application: Application, worker_no: int) -> None:
logger.info("Delivery worker %s started", worker_no)
prefer_staff = (worker_no % 2 == 0)
while True:
# Alternate queues so a continuous ticket stream cannot starve staff-to-senior messages.
job = claim_staff_delivery_job() if prefer_staff else claim_outbox_job()
prefer_staff = not prefer_staff
if job:
is_staff_job = "target_id" in job.keys()
try:
if is_staff_job:
await process_staff_delivery_job(application.bot, job)
else:
await process_outbox_job(application.bot, job)
except Exception as exc:
attempts = int(job["attempts"]) + 1
delay = min(300, (5 if is_staff_job else 3) * (2 ** min(attempts, 7)))
retry_after = getattr(exc, "retry_after", None)
if retry_after:
try:
delay = max(delay, int(retry_after) + 1)
except Exception:
pass
table = "staff_delivery_outbox" if is_staff_job else "delivery_outbox"
with db_connect() as conn:
conn.execute(f"UPDATE {table} SET status='pending', attempts=?, next_attempt_at=?, locked_at=NULL, last_error=? WHERE id=?",
(attempts, time.time() + delay, str(exc)[:700], job["id"]))
logger.warning("%s job %s retained for retry #%s in %ss: %s", table, job["id"], attempts, delay, exc)
await asyncio.sleep(0.1)
continue
await asyncio.sleep(0.35)
async def post_init(application: Application) -> None:
# Default command menu is public-safe; each staff member gets a private role-specific scope on /start.
try:
await application.bot.set_my_commands(public_commands(), scope=BotCommandScopeDefault())
except TelegramError as exc:
logger.warning("Could not set default command menu: %s", exc)
# Recover jobs interrupted by a forced Termux close/reboot; jobs remain in SQLite.
with db_connect() as conn:
conn.execute("UPDATE delivery_outbox SET status='pending', locked_at=NULL WHERE status='processing'")
for worker_no in range(1, 4):
application.create_task(outbox_worker(application, worker_no), name=f"spg-outbox-{worker_no}")
logger.info("Persistent outbox workers started")
async def handle_user_message(message, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Store every user message first; a durable outbox delivers it in the background."""
user = message.from_user
if not user:
return
body = message.text or message.caption or ""
msg_type, file_id = message_file(message)
category_key, emergency = classify_message(body)
ticket = active_ticket_for_user(user.id)
is_new = ticket is None
if not ticket:
ticket = await create_ticket(user, category_key, emergency, context)
if not ticket:
await message.reply_text("Ticket create nahi ho saka. Please kuch der baad phir try karein.")
return
tid = int(ticket["id"])
# An explicit emergency always routes to senior support and never to a regular manager.
if emergency:
with db_connect() as conn:
conn.execute("UPDATE tickets SET assigned_to=NULL, assigned_role='unassigned', priority='emergency', last_escalation_reason=?, updated_at=? WHERE id=?",
("Emergency keyword detected; queued for senior support", now_text(), tid))
else:
ticket = get_ticket(tid)
if ticket and ticket["assigned_role"] == "manager" and ticket["assigned_to"] and not manager_available(int(ticket["assigned_to"])):
fallbacks = escalation_candidates()
with db_connect() as conn:
if fallbacks:
uid = fallbacks[0]
conn.execute("UPDATE tickets SET assigned_to=?, assigned_role=?, priority='high', last_escalation_reason='Assigned manager is off duty; routed to senior support', updated_at=? WHERE id=?",
(uid, staff_role(uid) or "owner", now_text(), tid))
else:
conn.execute("UPDATE tickets SET assigned_to=NULL, assigned_role='unassigned', priority='high', last_escalation_reason='No staff currently configured; keep pending', updated_at=? WHERE id=?",
(now_text(), tid))
ticket = get_ticket(tid)
if ticket and ticket["assigned_to"] is None:
manager = choose_manager()
fallback = escalation_candidates()
recipient = manager if manager is not None else (fallback[0] if fallback else None)
if recipient:
with db_connect() as conn:
conn.execute("UPDATE tickets SET assigned_to=?, assigned_role=?, updated_at=? WHERE id=?",
(recipient, staff_role(recipient) or ("manager" if recipient == manager else "owner"), now_text(), tid))
# Persist transcript and delivery job before acknowledging the message.
add_message(tid, user.id, "user", msg_type, body, file_id, message.message_id)
enqueue_delivery(tid, "user_message", int(message.chat_id), int(message.message_id))
audit(user.id, "message_queued", f"type={msg_type}; emergency={emergency}", tid)
if is_new:
updated = get_ticket(tid)
await message.reply_text(
f"✅ SPG support ticket #{tid} create ho gaya aur aapka message save ho gaya.\n"
f"Category: {updated['category']}\n"
"Aapka message support queue mein save ho gaya hai. Jawab isi Telegram chat mein milega.\n\n"
"सुरक्षा: OTP, पासवर्ड, PIN या बैंक जानकारी साझा न करें। कोई भी staff member आपसे निजी UPI/बैंक खाते में भुगतान नहीं मांगेगा।"
)
elif emergency:
# Emergency confirmation only once for this detected escalation; the message itself is queued.
await message.reply_text(f"🚨 Emergency ticket #{tid} ko high priority par support queue mein bhej diya gaya hai.")
async def queue_staff_contact(message, context: ContextTypes.DEFAULT_TYPE) -> None:
uid = message.from_user.id
targets = [target for target in escalation_candidates(exclude=uid) if target != uid]
if not targets:
await message.reply_text("Senior support recipients available nahi hain. Message is baar forward nahi hua; administrator ko notify karein.")
return
# Persist the requested delivery before telling the staff member it was accepted.
with db_connect() as conn:
for target in targets:
dedupe = f"staff-contact:{uid}:{message.chat_id}:{message.message_id}:{target}"
conn.execute(
"INSERT OR IGNORE INTO staff_delivery_outbox (staff_id, source_chat_id, source_message_id, target_id, dedupe_key, status, next_attempt_at, created_at) VALUES (?, ?, ?, ?, ?, 'pending', 0, ?)",
(uid, int(message.chat_id), int(message.message_id), target, dedupe, now_text()),
)
audit(uid, "contact_senior", f"Queued staff message for {len(targets)} senior recipient(s)")
await message.reply_text("✅ Message senior support queue mein save ho gaya hai. Aap aur text/media bhej sakte hain. Mode band karne ke liye /cancel bhejein.")
def claim_staff_delivery_job() -> Optional[sqlite3.Row]:
now = time.time()
with db_connect() as conn:
conn.execute("BEGIN IMMEDIATE")
conn.execute("UPDATE staff_delivery_outbox SET status='pending', locked_at=NULL WHERE status='processing' AND locked_at None:
target = int(job["target_id"])
staff_id = int(job["staff_id"])
await bot.send_message(target, f"📨 PRIVATE STAFF MESSAGE\nFrom: {staff_name(staff_id)}\nStaff ID: {staff_id}\nReply to the copied message below to respond privately.")
copied = await bot.copy_message(chat_id=target, from_chat_id=int(job["source_chat_id"]), message_id=int(job["source_message_id"]))
with db_connect() as conn:
conn.execute("INSERT OR REPLACE INTO staff_lead_map (lead_id, copied_message_id, staff_id, created_at) VALUES (?, ?, ?, ?)", (target, copied.message_id, staff_id, now_text()))
conn.execute("UPDATE staff_delivery_outbox SET status='delivered', delivered_at=?, locked_at=NULL, last_error=NULL WHERE id=?", (now_text(), job["id"]))
audit(staff_id, "contact_senior_delivered", f"Delivered to senior recipient {target}")
async def handle_staff_reply(message, context: ContextTypes.DEFAULT_TYPE) -> None:
uid = message.from_user.id
if message.reply_to_message:
with db_connect() as conn:
lead_mapping = conn.execute("SELECT staff_id FROM staff_lead_map WHERE lead_id=? AND copied_message_id=?", (uid, message.reply_to_message.message_id)).fetchone()
if lead_mapping:
target_staff = int(lead_mapping["staff_id"])
try:
await context.bot.copy_message(chat_id=target_staff, from_chat_id=message.chat_id, message_id=message.message_id)
await message.reply_text("✅ Reply staff member ko bhej diya gaya.")
except TelegramError as exc:
await message.reply_text("Reply deliver nahi hua; queue/retry ya dobara text bhejne ki koshish karein.")
logger.warning("Senior-to-staff reply failed: %s", exc)
return
with db_connect() as conn:
mapping = conn.execute("SELECT ticket_id FROM manager_message_map WHERE manager_id=? AND copied_message_id=?", (uid, message.reply_to_message.message_id)).fetchone()
if mapping:
ticket = get_ticket(int(mapping["ticket_id"]))
if not ticket or ticket["status"] != "open":
await message.reply_text("Ye ticket ab open nahi hai.")
return
if int(ticket["assigned_to"] or 0) != uid:
await message.reply_text("⛔ Ye ticket ab aapko assigned nahi hai. Reply send nahi kiya gaya; senior support se takeover karwayein.")
return
try:
copied = await context.bot.copy_message(chat_id=int(ticket["user_id"]), from_chat_id=message.chat_id, message_id=message.message_id)
except TelegramError:
text = message.text or message.caption
if text:
await context.bot.send_message(int(ticket["user_id"]), f"SPG Support | Ticket #{ticket['id']}\n\n{text}")
else:
await message.reply_text("User tak media deliver nahi hua. Text reply try karein.")
return
kind, file_id = message_file(message)
add_message(int(ticket["id"]), uid, staff_role(uid) or "staff", kind, message.text or message.caption or "", file_id, message.message_id)
audit(uid, "ticket_reply", f"Reply via reply-to message_id={message.message_id}", int(ticket["id"]))
await message.reply_text(f"✅ Aapka reply ticket #{ticket['id']} ke user ko bhej diya gaya.")
return
# Contact mode lets staff send text/media privately to the senior team from this same bot.
if context.user_data.get("lead_contact_mode"):
await queue_staff_contact(message, context)
return
await message.reply_text(
"Ticket reply ke liye copied user message par Reply karein ya /reply TICKET_ID message use karein.\n"
"Senior support se private baat ke liye /contactlead bhejein. Apne assigned tickets ke liye /mytickets use karein."
)
async def handle_non_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
message = update.effective_message
user = update.effective_user
chat = update.effective_chat
if not message or not user or not chat:
return
if chat.type != ChatType.PRIVATE:
return
if is_staff(user.id):
await handle_staff_reply(message, context)
else:
await handle_user_message(message, context)
async def handle_callback(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
query = update.callback_query
if not query or not query.from_user:
return
await query.answer()
uid = query.from_user.id
data = query.data or ""
if data.startswith("panel:"):
action = data.split(":", 1)[1]
if not is_staff(uid):
await query.message.reply_text("Customer support ke liye /start dabayein.")
return
if action == "mytickets":
await cmd_mytickets_from_callback(query, context)
elif action == "duty_on" or action == "duty_off":
if not is_manager(uid):
await query.message.reply_text("Duty controls registered managers ke liye hain.")
return
state = 1 if action == "duty_on" else 0
with db_connect() as conn:
conn.execute("UPDATE managers SET on_duty=? WHERE telegram_id=?", (state, uid))
audit(uid, "duty_change", "on" if state else "off")
await query.message.reply_text("✅ Duty ON mark ho gaya. Agar scheduled shift active hai to aap new ticket assignments ke liye available hain." if state else "✅ Duty OFF mark ho gaya. Naye tickets aapko assign nahi honge.")
elif action == "alltickets" and (is_owner(uid) or is_head(uid)):
with db_connect() as conn:
rows = conn.execute("SELECT * FROM tickets WHERE status='open' ORDER BY updated_at DESC LIMIT 30").fetchall()
lines = ["📋 OPEN TICKETS (max 30)"] + [f"#{t['id']} | {t['category']} | {t['priority']} | user {t['user_id']} | {staff_name(t['assigned_to'])}" for t in rows]
await query.message.reply_text("\n".join(lines) if rows else "Koi open ticket nahi hai.")
elif action == "stats" and (is_owner(uid) or is_head(uid)):
with db_connect() as conn:
ot = conn.execute("SELECT COUNT(*) n FROM tickets WHERE status='open'").fetchone()["n"]
ct = conn.execute("SELECT COUNT(*) n FROM tickets WHERE status='closed'").fetchone()["n"]
pending = conn.execute("SELECT COUNT(*) n FROM delivery_outbox WHERE status='pending'").fetchone()["n"]
await query.message.reply_text(f"📊 Support stats\nOpen tickets: {ot}\nClosed tickets: {ct}\nPending delivery jobs: {pending}\nManagers on duty/in shift: {len(available_managers())}")
elif action == "queue" and (is_owner(uid) or is_head(uid)):
with db_connect() as conn:
pending = conn.execute("SELECT COUNT(*) n FROM delivery_outbox WHERE status='pending'").fetchone()["n"]
processing = conn.execute("SELECT COUNT(*) n FROM delivery_outbox WHERE status='processing'").fetchone()["n"]
await query.message.reply_text(f"📬 Queue status\nPending: {pending}\nProcessing: {processing}\nUse /queue for more details.")
elif action == "managers" and (is_owner(uid) or is_head(uid)):
with db_connect() as conn:
rows = conn.execute("SELECT * FROM managers WHERE enabled=1 ORDER BY name COLLATE NOCASE").fetchall()
lines = ["👥 MANAGERS"] + [f"{m['name']} | {m['telegram_id']} | shift {m['shift_start']} ({m['shift_hours']}h) | {'ON DUTY' if manager_available(int(m['telegram_id'])) else 'OFF DUTY'}" for m in rows]
await query.message.reply_text("\n".join(lines) if rows else "Abhi manager add nahi hai. /addmanager ID Name")
elif action == "contactlead":
context.user_data["lead_contact_mode"] = True
await query.message.reply_text("🔒 Senior Support Desk: ab apna text/media bhejo. Mode band karne ke liye /cancel.")
elif action == "help":
await query.message.reply_text("Use /help to see only the commands available for your staff role.")
else:
await query.message.reply_text("Is panel action ke liye permission nahi hai.")
return
if data == "home":
await query.edit_message_reply_markup(reply_markup=home_keyboard())
return
if data == "check_join":
await query.message.reply_text("📢 SPG channels optional hain. Support use karne ke liye join karna zaroori nahi hai.", reply_markup=home_keyboard())
return
if data == "my_ticket":
if not is_staff(uid):
ticket = active_ticket_for_user(uid)
if not ticket:
await query.message.reply_text("Aapka koi open ticket nahi hai. Support category choose karein:", reply_markup=category_keyboard())
else:
await query.message.reply_text(
f"🎫 Ticket #{ticket['id']}\nCategory: {ticket['category']}\nStatus: {ticket['status']}\n"
f"Handling: {public_assignee_label(ticket)}\nPriority: {ticket['priority']}"
)
else:
await cmd_mytickets_from_callback(query, context)
return
if data == "open_support":
ticket = active_ticket_for_user(uid) if not is_staff(uid) else None
if ticket:
await query.message.reply_text(f"Aapka ticket #{ticket['id']} already open hai. Messages isi assigned manager ko jayenge.")
else:
await query.message.reply_text("SPG Support category choose karein:", reply_markup=category_keyboard())
return
if data.startswith("category:"):
key = data.split(":", 1)[1]
if key not in CATEGORIES:
await query.message.reply_text("Unknown category. /start dabayein.")
return
if is_staff(uid):
await query.message.reply_text("Staff accounts customer ticket create nahi karte. /help use karein.")
return
emergency = key == "emergency"
ticket = await create_ticket(query.from_user, key, emergency, context)
if not ticket:
await query.message.reply_text("Ticket create nahi ho saka; please /start dobara try karein.")
return
if ticket["assigned_to"] is None:
await query.message.reply_text(f"Ticket #{ticket['id']} save ho gaya hai aur support queue mein pending hai. Aapko dobara request bhejne ki zaroorat nahi.")
else:
await query.message.reply_text(
f"✅ Aapka support ticket #{ticket['id']} create ho gaya.\nCategory: {ticket['category']}\n"
f"Handling: {public_assignee_label(ticket)}\n\nAb aap text, voice/audio, video, photo, document ya doosri supported file bhej sakte hain."
)
return
async def cmd_mytickets_from_callback(query, context) -> None:
uid = query.from_user.id
with db_connect() as conn:
rows = conn.execute("SELECT * FROM tickets WHERE assigned_to=? AND status='open' ORDER BY updated_at DESC LIMIT 20", (uid,)).fetchall()
if not rows:
await query.message.reply_text("Aapke paas abhi koi open ticket nahi hai.")
else:
text = "\n".join(["🎫 AAPKE OPEN TICKETS"] + [f"#{r['id']} | {r['category']} | User {r['user_id']} | {r['priority']}" for r in rows])
await query.message.reply_text(text)
async def error_handler(update: object, context: ContextTypes.DEFAULT_TYPE) -> None:
logger.error("Unhandled error while processing update: %s", context.error, exc_info=context.error)
try:
uid = update.effective_user.id if isinstance(update, Update) and update.effective_user else 0
audit(uid, "runtime_error", str(context.error)[:500])
except Exception:
pass
def validate_config() -> None:
token = config.BOT_TOKEN.strip()
if not token or token == "PASTE_SUPPORT_BOT_TOKEN_HERE":
raise SystemExit("Please open config.py and set BOT_TOKEN first.")
if not any(isinstance(x, int) and x > 0 for x in getattr(config, "OWNER_IDS", [])):
raise SystemExit("Please set OWNER_IDS in config.py to your numeric Telegram user ID.")
def build_application() -> Application:
"""Create the PTB application and handlers. Shared by polling and cPanel webhook modes."""
app = (Application.builder().token(config.BOT_TOKEN.strip()).concurrent_updates(16)
.connect_timeout(8).read_timeout(12).write_timeout(12).pool_timeout(8)
.post_init(post_init).build())
app.add_handler(CommandHandler("start", cmd_start))
app.add_handler(CommandHandler("help", cmd_help))
app.add_handler(CommandHandler("id", cmd_id))
app.add_handler(CommandHandler("panel", cmd_panel))
app.add_handler(CommandHandler("addmanager", cmd_addmanager))
app.add_handler(CommandHandler("removemanager", cmd_removemanager))
app.add_handler(CommandHandler("managers", cmd_managers))
app.add_handler(CommandHandler("shift", cmd_shift))
app.add_handler(CommandHandler("duty", cmd_duty))
app.add_handler(CommandHandler("sethead", cmd_sethead))
app.add_handler(CommandHandler("mytickets", cmd_mytickets))
app.add_handler(CommandHandler("tickets", cmd_tickets))
app.add_handler(CommandHandler("stats", cmd_stats))
app.add_handler(CommandHandler("close", cmd_close))
app.add_handler(CommandHandler("takeover", cmd_takeover))
app.add_handler(CommandHandler("assign", cmd_assign))
app.add_handler(CommandHandler("escalate", cmd_escalate))
app.add_handler(CommandHandler("reply", cmd_reply))
app.add_handler(CommandHandler("transcript", cmd_transcript))
app.add_handler(CommandHandler("audit", cmd_audit))
app.add_handler(CommandHandler("channels", cmd_channels))
app.add_handler(CommandHandler("queue", cmd_queue))
app.add_handler(CommandHandler("contactlead", cmd_contactlead))
app.add_handler(CommandHandler("cancel", cmd_cancel))
app.add_handler(CommandHandler("setwelcomevideo", cmd_setwelcomevideo))
app.add_handler(CommandHandler("clearwelcomevideo", cmd_clearwelcomevideo))
app.add_handler(CallbackQueryHandler(handle_callback))
app.add_handler(MessageHandler(filters.ChatType.CHANNEL, handle_channel_post_id))
app.add_handler(MessageHandler(filters.ALL & ~filters.COMMAND, handle_non_command))
app.add_error_handler(error_handler)
return app
def main() -> None:
validate_config()
db_init()
app = build_application()
logger.info("SPG Support Bot starting. Database=%s", DB_PATH)
app.run_polling(allowed_updates=Update.ALL_TYPES, drop_pending_updates=False)
if __name__ == "__main__":
main()