{html.escape(v)}")
return "\n".join(lines)
def _plain_record(idx: int, row: dict) -> str:
"""Plain-text version for copy button."""
lines = [f"#{idx}"]
for key, _emoji, label in FIELD_LABELS:
v = _tv(row.get(key), TRUNC.get(key, 60))
if v:
lines.append(f"{label}: {v}")
return "\n".join(lines)
def render_page(results, page, total_pages, phone, elapsed, source_note=""):
start = page * PAGE_SIZE
chunk = results[start:start + PAGE_SIZE]
head = (
f"📱 Phone: {html.escape(phone)}\n"
f"📄 Page {page+1}/{total_pages} · "
f"{len(results)} result(s) · ⏱️ {elapsed:.1f}s"
)
if source_note:
head += f"\n⚠️ {source_note}"
body = "\n\n".join(format_result(start + i + 1, r) for i, r in enumerate(chunk))
return f"{head}\n\n{body}"
def page_keyboard(search_id, page, total_pages, has_export=True):
nav = []
if page > 0:
nav.append({"text": "◀ Prev", "callback_data": f"p:{search_id}:{page-1}"})
nav.append({"text": f"{page+1} / {total_pages}", "callback_data": "noop"})
if page < total_pages - 1:
nav.append({"text": "Next ▶", "callback_data": f"p:{search_id}:{page+1}"})
rows = [nav]
extras = []
extras.append({"text": "📋 Copy page", "callback_data": f"c:{search_id}:{page}"})
if has_export:
extras.append({"text": "📥 Export", "callback_data": f"e:{search_id}"})
rows.append(extras)
return {"inline_keyboard": rows}
# ─────────────────────────────────────────────
# ACTIVITY TRACKING
# ─────────────────────────────────────────────
ACTIVITY_LOG = deque(maxlen=10000)
ACTIVE_USERS = {}
UNIQUE_USERS = {}
PHONE_COUNTS = Counter()
PEAK_CONCURRENT = 0
_activity_lock = threading.Lock()
_admin_lock = threading.Lock()
_admin_sessions = {}
_admin_states = {}
# ── User management ──
BANNED_USERS: set = set()
BAN_REASONS: dict = {}
USER_HISTORY: dict = {} # user_id -> deque of {phone, ts, count, elapsed, source}
# ── System flags ──
MAINTENANCE_MODE = False
MAINTENANCE_REASON = ""
_state_lock = threading.Lock()
def _short_user(user_id: int, user: dict | None = None) -> str:
info = UNIQUE_USERS.get(user_id) or {}
username = (user or {}).get("username") or info.get("username")
first = (user or {}).get("first_name") or info.get("first_name")
if username: return f"@{username} ({user_id})"
if first: return f"{first} ({user_id})"
return f"({user_id})"
def _log_activity(user: dict, action: str, input_val: str, reply_type: str,
summary: str = "", elapsed_ms: int = 0, source: str = ""):
if not user or "id" not in user:
return
now = time.time()
user_id = user["id"]
with _activity_lock:
info = UNIQUE_USERS.setdefault(user_id, {
"username": None, "first_name": None, "last_name": None,
"count": 0, "first_seen": now, "last_seen": now,
})
if user.get("username"): info["username"] = user["username"]
if user.get("first_name"): info["first_name"] = user["first_name"]
if user.get("last_name"): info["last_name"] = user["last_name"]
info["count"] += 1
info["last_seen"] = now
ACTIVE_USERS[user_id] = now
if action == "search" and isinstance(input_val, str) and input_val.isdigit() and len(input_val) == 10:
PHONE_COUNTS[input_val] += 1
ACTIVITY_LOG.append({
"ts": now, "user_id": user_id,
"username": user.get("username"), "first_name": user.get("first_name"),
"action": action, "input": (input_val or "")[:80],
"reply_type": reply_type, "summary": (summary or "")[:150],
"elapsed_ms": int(elapsed_ms), "source": source,
})
def _get_live_users():
cutoff = time.time() - LIVE_WINDOW
with _activity_lock:
live = [(uid, ts) for uid, ts in ACTIVE_USERS.items() if ts >= cutoff]
live.sort(key=lambda x: -x[1])
return live
def _get_stats():
now = time.time()
today_start = now - (now % 86400)
with _activity_lock:
today_users = set(); today_searches = 0; total_searches = 0
for e in ACTIVITY_LOG:
if e["action"] == "search":
total_searches += 1
if e["ts"] >= today_start:
today_searches += 1; today_users.add(e["user_id"])
return {
"unique_users": len(UNIQUE_USERS),
"live_users": sum(1 for ts in ACTIVE_USERS.values() if ts >= now - LIVE_WINDOW),
"today_users": len(today_users),
"today_searches": today_searches,
"total_searches": total_searches,
"peak_concurrent": PEAK_CONCURRENT,
"total_log_entries": len(ACTIVITY_LOG),
"banned_count": len(BANNED_USERS),
"maintenance": MAINTENANCE_MODE,
}
def _add_to_history(user_id: int, phone: str, count: int, elapsed: float, source: str):
if user_id is None: return
with _activity_lock:
h = USER_HISTORY.setdefault(user_id, deque(maxlen=HISTORY_LEN))
h.appendleft({"phone": phone, "ts": time.time(), "count": count,
"elapsed": elapsed, "source": source})
# ─────────────────────────────────────────────
# BAN / MAINTENANCE HELPERS
# ─────────────────────────────────────────────
def _is_banned(user_id: int) -> bool:
with _state_lock:
return user_id in BANNED_USERS
def _ban_user(user_id: int, reason: str = ""):
with _state_lock:
BANNED_USERS.add(user_id)
BAN_REASONS[user_id] = reason
def _unban_user(user_id: int) -> bool:
with _state_lock:
if user_id in BANNED_USERS:
BANNED_USERS.discard(user_id)
BAN_REASONS.pop(user_id, None)
return True
return False
def _get_maintenance():
with _state_lock:
return MAINTENANCE_MODE, MAINTENANCE_REASON
def _set_maintenance(on: bool, reason: str = ""):
global MAINTENANCE_MODE, MAINTENANCE_REASON
with _state_lock:
MAINTENANCE_MODE = on
MAINTENANCE_REASON = reason if on else ""
# ─────────────────────────────────────────────
# ADMIN AUTH & STATE
# ─────────────────────────────────────────────
def _is_admin_session(user_id: int) -> bool:
if user_id is None: return False
with _admin_lock:
exp = _admin_sessions.get(user_id)
if exp is None: return False
if time.time() > exp:
_admin_sessions.pop(user_id, None); return False
return True
def _grant_admin(user_id: int):
with _admin_lock:
_admin_sessions[user_id] = time.time() + ADMIN_SESSION_TTL
def _revoke_admin(user_id: int):
with _admin_lock:
_admin_sessions.pop(user_id, None)
def _admin_state_get(user_id: int):
with _admin_lock:
entry = _admin_states.get(user_id)
if entry is None: return None
state, set_at = entry
if time.time() - set_at > ADMIN_PROMPT_TTL:
_admin_states.pop(user_id, None); return None
return state
def _admin_state_set(user_id: int, state: str):
with _admin_lock:
_admin_states[user_id] = (state, time.time())
def _admin_state_clear(user_id: int):
with _admin_lock:
_admin_states.pop(user_id, None)
# ─────────────────────────────────────────────
# ADMIN RENDERING
# ─────────────────────────────────────────────
def _fmt_ts(ts: float) -> str:
return time.strftime("%H:%M:%S", time.localtime(ts))
def admin_menu_kb():
m_on, _ = _get_maintenance()
maint_label = "🟢 Maintenance: OFF" if not m_on else "🔴 Maintenance: ON"
return {
"inline_keyboard": [
[{"text": "🔴 Live Users", "callback_data": "adm:live"},
{"text": "📈 Stats", "callback_data": "adm:stats"}],
[{"text": "📜 Activity Log", "callback_data": "adm:log:0"},
{"text": "👥 All Users", "callback_data": "adm:users:0"}],
[{"text": "🔍 Top Searches", "callback_data": "adm:top"},
{"text": "🚫 Banned", "callback_data": "adm:banned"}],
[{"text": "🚫 Ban User", "callback_data": "adm:ban"},
{"text": "✅ Unban User", "callback_data": "adm:unban"}],
[{"text": "📢 Broadcast", "callback_data": "adm:broadcast"},
{"text": "📥 Export Log", "callback_data": "adm:export"}],
[{"text": "🔎 Log by Phone", "callback_data": "adm:logphone"},
{"text": maint_label, "callback_data": "adm:maint"}],
[{"text": "🔄 Refresh", "callback_data": "adm:menu"}],
]
}
def render_admin_menu() -> str:
m_on, reason = _get_maintenance()
m_line = f"🔴 Maintenance ON — {html.escape(reason)}" if m_on else "🟢 Operational"
return (
f"🛡️ Admin Panel — v{VERSION}\n"
"━━━━━━━━━━━━━━━━━━\n"
f"{m_line}\n"
f"👥 Users: {len(UNIQUE_USERS)} · 🚫 Banned: {len(BANNED_USERS)}\n\n"
"Tap a button or use commands:\n"
"• /live /stats /log /users /top\n"
"• /ban <id> [reason] · /unban <id>\n"
"• /broadcast <msg>\n"
"• /export · /logphone <num>\n"
"• /maintenance on|off [reason]\n"
"• /logout"
)
def render_live() -> str:
live = _get_live_users()
if not live:
return f"🔴 Live Users\n\nNo one active in the last {LIVE_WINDOW // 60} minutes."
lines = [f"🔴 Live Users (last {LIVE_WINDOW // 60} min): {len(live)}\n"]
for uid, ts in live[:25]:
lines.append(f"• {_short_user(uid)} — {_fmt_ts(ts)}")
return "\n".join(lines)
def render_stats() -> str:
s = _get_stats()
uptime = (time.time() - STATS["started"]) / 60
return (
"📈 Bot Statistics\n"
"━━━━━━━━━━━━━━━━━━\n"
f"⏱️ Uptime: {uptime:.1f} min\n"
f"👥 Unique users: {s['unique_users']}\n"
f"🔴 Live now: {s['live_users']}\n"
f"📅 Active today: {s['today_users']}\n"
f"🔍 Searches today: {s['today_searches']}\n"
f"🔍 Total searches: {s['total_searches']}\n"
f"📊 Peak concurrent: {s['peak_concurrent']}\n"
f"🚫 Banned users: {s['banned_count']}\n"
f"💾 Cached queries: {len(SEARCH_CACHE)}\n"
f"📜 Log entries: {s['total_log_entries']}\n"
f"⏰ Timeouts: {STATS['timeouts']}\n"
f"❌ Errors: {STATS['errors']}"
)
def render_log(entries, page=0, page_size=8):
total = len(entries)
total_pages = max(1, (total + page_size - 1) // page_size)
page = max(0, min(page, total_pages - 1))
chunk = entries[page * page_size:(page + 1) * page_size]
if not chunk:
return "📜 Activity Log\n\nNo activity yet.", None
lines = [f"📜 Activity Log — page {page + 1}/{total_pages} ({total} entries)\n"]
for e in chunk:
uname = e.get("username"); first = e.get("first_name")
name_str = f"@{uname}" if uname else (first or "(no name)")
uid = e["user_id"]
reply = e.get("summary") or e.get("reply_type") or "-"
elapsed = e.get("elapsed_ms", 0); source = e.get("source") or ""
block = f"{_fmt_ts(e['ts'])} · {html.escape(name_str)} · {uid}\n"
block += f" 📥 {html.escape(e['input'] or '-')}\n"
reply_line = f" 📤 {html.escape(reply)}"
if elapsed: reply_line += f" · {elapsed}ms"
if source: reply_line += f" · {source}"
block += reply_line
lines.append(block)
text = "\n\n".join(lines)
nav = []
if page > 0:
nav.append({"text": "◀ Newer", "callback_data": f"adm:log:{page - 1}"})
nav.append({"text": f"{page + 1}/{total_pages}", "callback_data": "adm:noop"})
if page < total_pages - 1:
nav.append({"text": "Older ▶", "callback_data": f"adm:log:{page + 1}"})
kb = {"inline_keyboard": [nav]} if len(nav) > 1 else None
return text, kb
def render_users(page=0, page_size=8):
users = sorted(UNIQUE_USERS.items(), key=lambda kv: -kv[1]["last_seen"])
total = len(users)
total_pages = max(1, (total + page_size - 1) // page_size)
page = max(0, min(page, total_pages - 1))
chunk = users[page * page_size:(page + 1) * page_size]
if not chunk:
return "👥 All Users\n\nNo users yet.", None
lines = [f"👥 All Users — page {page + 1}/{total_pages} ({total} total)\n"]
for uid, info in chunk:
username = info.get("username"); first = info.get("first_name") or ""
name_str = f"@{username}" if username else (first or "(no name)")
ban_mark = " 🚫" if _is_banned(uid) else ""
lines.append(
f"{html.escape(name_str)}{ban_mark} · {uid}\n"
f" Requests: {info['count']} · Last: {_fmt_ts(info['last_seen'])}"
)
text = "\n\n".join(lines)
nav = []
if page > 0:
nav.append({"text": "◀ Prev", "callback_data": f"adm:users:{page - 1}"})
nav.append({"text": f"{page + 1}/{total_pages}", "callback_data": "adm:noop"})
if page < total_pages - 1:
nav.append({"text": "Next ▶", "callback_data": f"adm:users:{page + 1}"})
kb = {"inline_keyboard": [nav]} if len(nav) > 1 else None
return text, kb
def render_top() -> str:
if not PHONE_COUNTS:
return "🔍 Top Searched Numbers\n\nNo searches yet."
lines = ["🔍 Top Searched Numbers\n"]
for i, (num, cnt) in enumerate(PHONE_COUNTS.most_common(10), 1):
lines.append(f"{i}. {num} — {cnt} time(s)")
return "\n".join(lines)
def render_banned() -> str:
with _state_lock:
if not BANNED_USERS:
return "🚫 Banned Users\n\nNo banned users."
lines = [f"🚫 Banned Users ({len(BANNED_USERS)})\n"]
for uid in list(BANNED_USERS)[:30]:
reason = BAN_REASONS.get(uid) or "—"
lines.append(f"• {_short_user(uid)} — {html.escape(reason)}")
return "\n".join(lines)
# ─────────────────────────────────────────────
# TELEGRAM HELPERS
# ─────────────────────────────────────────────
async def tg(method: str, payload: dict, timeout: float = 30) -> dict:
try:
async with httpx.AsyncClient(timeout=timeout) as client:
r = await client.post(f"{TELEGRAM_API}/{method}", json=payload)
return r.json()
except Exception as e:
log.error(f"❌ tg({method}) error: {e!r}")
return {"ok": False, "error": str(e)}
async def send(chat_id, text, **kw):
kw.setdefault("parse_mode", "HTML")
kw.setdefault("disable_web_page_preview", True)
kw = {k: v for k, v in kw.items() if v is not None}
res = await tg("sendMessage", {"chat_id": chat_id, "text": text, **kw})
if res.get("ok"):
return res["result"]["message_id"]
log.warning(f"⚠️ sendMessage failed: {res}")
return None
async def edit(chat_id, message_id, text, **kw):
kw.setdefault("parse_mode", "HTML")
kw.setdefault("disable_web_page_preview", True)
kw = {k: v for k, v in kw.items() if v is not None}
res = await tg("editMessageText", {"chat_id": chat_id, "message_id": message_id, "text": text, **kw})
if not res.get("ok") and "message is not modified" not in str(res):
log.warning(f"⚠️ editMessageText failed: {res}")
async def send_document(chat_id: int, filename: str, content: bytes, caption: str = ""):
"""Send a document via multipart/form-data."""
try:
files = {"document": (filename, content, "application/octet-stream")}
data = {"chat_id": str(chat_id)}
if caption:
data["caption"] = caption
async with httpx.AsyncClient(timeout=60) as client:
r = await client.post(f"{TELEGRAM_API}/sendDocument", data=data, files=files)
res = r.json()
if not res.get("ok"):
log.warning(f"⚠️ sendDocument failed: {res}")
return res
except Exception as e:
log.error(f"❌ sendDocument error: {e!r}")
return {"ok": False}
# ─────────────────────────────────────────────
# INPUT NORMALIZATION — "Did you mean?"
# ─────────────────────────────────────────────
def normalize_phone(text: str) -> tuple[Optional[str], Optional[str]]:
"""
Returns (10-digit phone, hint_message) or (None, error).
Accepts: 9876543210, +91 9876543210, 09876543210, 919876543210,
98765-43210, (98765) 43210, etc.
"""
raw = text.strip()
digits = "".join(ch for ch in raw if ch.isdigit())
if not digits:
return None, "not a valid number"
# Strip leading 0 (local format)
if len(digits) == 11 and digits.startswith("0"):
digits = digits[1:]
# Strip 91 country code
if len(digits) == 12 and digits.startswith("91"):
digits = digits[2:]
elif len(digits) == 11 and digits.startswith("91"):
digits = digits[1:]
if len(digits) == 10:
if PHONE_REGEX.match(digits):
return digits, None
return None, "not a valid 10-digit number"
return None, f"expected 10 digits, got {len(digits)}"
# ─────────────────────────────────────────────
# STATE
# ─────────────────────────────────────────────
SEARCH_CACHE = {}
USER_LAST = {}
STATS = {"queries": 0, "timeouts": 0, "errors": 0, "started": time.time()}
def cache_gc():
global PEAK_CONCURRENT
while True:
time.sleep(60)
now = time.time()
dead = [k for k, v in SEARCH_CACHE.items() if now - v["ts"] > CACHE_TTL]
for k in dead:
SEARCH_CACHE.pop(k, None)
with _activity_lock:
cutoff = now - LIVE_WINDOW
for uid in list(ACTIVE_USERS.keys()):
if ACTIVE_USERS[uid] < cutoff:
del ACTIVE_USERS[uid]
if len(ACTIVE_USERS) > PEAK_CONCURRENT:
PEAK_CONCURRENT = len(ACTIVE_USERS)
# ─────────────────────────────────────────────
# BATCH SEARCH
# ─────────────────────────────────────────────
async def _do_single_search(phone: str) -> tuple[list, float, str]:
"""Returns (results, elapsed, source). Uses DuckDB if warmed, else API fallback."""
loop = asyncio.get_running_loop()
t0 = time.time()
source = "duckdb"
if not _warmed.is_set():
results = await search_apis(phone)
source = "api_fallback"
else:
results = await asyncio.wait_for(
loop.run_in_executor(None, run_search_safe, phone, FETCH_LIMIT),
timeout=QUERY_TIMEOUT + 10,
)
elapsed = time.time() - t0
return dedup_records(results), elapsed, source
# ─────────────────────────────────────────────
# HANDLERS
# ─────────────────────────────────────────────
WELCOME_TEMPLATE = (
"👋 Phone Lookup Bot v{v}\n\n"
"Send me any 10-digit mobile number and I'll return every "
"record we have on file.\n\n"
"📌 Examples\n"
"• 9876543210\n"
"• +91 98765 43210\n"
"• 9876543210,9876543211 (batch)\n\n"
"Commands:\n"
"• /start — this message\n"
"• /info — bot info & credits\n"
"• /status — bot health\n"
"• /history — your last 10 searches\n"
"• /myid — your Telegram ID"
)
async def handle_message(msg: dict):
try:
chat_id = msg["chat"]["id"]
user = msg.get("from") or {}
user_id = user.get("id")
text = (msg.get("text") or "").strip()
log.info(f"💬 Message from {user_id}: {text[:50]!r}")
if not text or user_id is None:
return
# ── BAN CHECK ──
if _is_banned(user_id):
with _state_lock:
reason = BAN_REASONS.get(user_id) or "no reason given"
await send(chat_id, f"🚫 You are banned.\n\nReason: {html.escape(reason)}")
_log_activity(user, "banned_hit", text[:40], "blocked", reason)
return
# ── ADMIN TOKEN INPUT ──
state = _admin_state_get(user_id)
if state and _is_admin_session(user_id) is False:
if state == "await_admin_token":
_admin_state_clear(user_id)
if ADMIN_TOKEN and text == ADMIN_TOKEN:
_grant_admin(user_id)
_log_activity(user, "admin_login", "***", "granted", "session started")
await send(chat_id, "✅ Admin access granted",
reply_markup=admin_menu_kb())
await send(chat_id, render_admin_menu())
else:
_log_activity(user, "admin_login", "***", "denied", "invalid token")
await send(chat_id, "❌ Invalid token.\nTry /admin again.")
return
# ── ADMIN INPUT CAPTURE (after login) ──
if state and _is_admin_session(user_id):
_admin_state_clear(user_id)
if state == "await_ban_id":
parts = text.split(None, 1)
target = parts[0].lstrip("@")
reason = parts[1] if len(parts) > 1 else ""
try:
tid = int(target)
except ValueError:
await send(chat_id, "❗ Invalid user ID.")
return
_ban_user(tid, reason)
_log_activity(user, "admin", f"/ban {tid}", "banned", reason or "no reason")
await send(chat_id, f"🚫 Banned {tid} — {html.escape(reason or 'no reason')}")
return
if state == "await_unban_id":
try:
tid = int(text.strip())
except ValueError:
await send(chat_id, "❗ Invalid user ID.")
return
ok = _unban_user(tid)
_log_activity(user, "admin", f"/unban {tid}", "unbanned")
await send(chat_id, "✅ Unbanned." if ok else "ℹ️ User was not banned.")
return
if state == "await_broadcast":
asyncio.create_task(_run_broadcast(chat_id, text))
_log_activity(user, "admin", "/broadcast", "sending", f"{len(text)} chars")
return
if state == "await_logphone":
phone, hint = normalize_phone(text)
if not phone:
await send(chat_id, f"❗ {hint}")
return
entries = [e for e in ACTIVITY_LOG
if e["action"] == "search" and e["input"] == phone]
entries.reverse()
if not entries:
await send(chat_id, f"🔎 No searches logged for {phone}.")
return
lines = [f"🔎 Searches for {phone} ({len(entries)} total)\n"]
for e in entries[:20]:
uname = e.get("username"); first = e.get("first_name")
nm = f"@{uname}" if uname else (first or "(anon)")
lines.append(f"• {_fmt_ts(e['ts'])} · {html.escape(nm)} · {e['user_id']}")
await send(chat_id, "\n".join(lines))
_log_activity(user, "admin", f"logphone {phone}", "text")
return
if state == "await_maint_reason":
_set_maintenance(True, text[:100])
_log_activity(user, "admin", "maintenance on", "state", text[:100])
await send(chat_id, f"🔴 Maintenance ON\nReason: {html.escape(text[:100])}")
return
# ── MAINTENANCE GATE (public blocked, admin allowed) ──
m_on, m_reason = _get_maintenance()
if m_on and not _is_admin_session(user_id):
if text in ("/start", "/help", "/status", "/info", "/myid"):
pass # allow basic commands
else:
await send(chat_id, f"🛠️ Bot under maintenance\n\n{html.escape(m_reason)}")
_log_activity(user, "maintenance_block", text[:40], "blocked")
return
# ── /admin ──
if text == "/admin":
if not ADMIN_TOKEN:
await send(chat_id, "⚠️ Admin panel is not configured.")
return
if _is_admin_session(user_id):
_log_activity(user, "admin", "/admin", "menu")
await send(chat_id, render_admin_menu(), reply_markup=admin_menu_kb())
else:
_admin_state_set(user_id, "await_admin_token")
await send(chat_id, "🔐 Admin access required\n\n"
"Send the admin token to continue. "
"It expires in 3 minutes.")
return
# ── /logout ──
if text == "/logout":
if _is_admin_session(user_id):
_revoke_admin(user_id)
_log_activity(user, "admin_logout", "/logout", "revoked")
await send(chat_id, "✅ Admin session ended.")
else:
await send(chat_id, "You are not logged in as admin.")
return
# ── ADMIN COMMANDS (session required) ──
if text.startswith("/") and _is_admin_session(user_id):
cmd = text.split()[0].lower()
args = text.split()[1:]
if cmd == "/live":
_log_activity(user, "admin", "/live", "text")
await send(chat_id, render_live()); return
if cmd == "/stats":
_log_activity(user, "admin", "/stats", "text")
await send(chat_id, render_stats()); return
if cmd == "/users":
_log_activity(user, "admin", "/users", "text")
t, kb = render_users(0); await send(chat_id, t, reply_markup=kb); return
if cmd == "/top":
_log_activity(user, "admin", "/top", "text")
await send(chat_id, render_top()); return
if cmd == "/log":
entries = list(ACTIVITY_LOG); entries.reverse()
label = ""
if args:
try:
tid = int(args[0])
entries = [e for e in entries if e["user_id"] == tid]
label = f" (user {tid})"
except ValueError:
target = args[0].lstrip("@").lower()
entries = [e for e in entries if (e.get("username") or "").lower() == target]
label = f" (@{target})"
_log_activity(user, "admin", text, "log_view")
t, kb = render_log(entries, 0)
if label: t = t.replace("Activity Log", f"Activity Log{label}", 1)
await send(chat_id, t, reply_markup=kb); return
if cmd == "/banned":
_log_activity(user, "admin", "/banned", "text")
await send(chat_id, render_banned()); return
if cmd == "/ban":
if not args:
_admin_state_set(user_id, "await_ban_id")
await send(chat_id, "Send user ID (and optional reason):")
return
tid_s = args[0].lstrip("@")
reason = " ".join(args[1:]) if len(args) > 1 else ""
try: tid = int(tid_s)
except ValueError:
await send(chat_id, "❗ Invalid user ID."); return
_ban_user(tid, reason)
_log_activity(user, "admin", f"/ban {tid}", "banned", reason or "no reason")
await send(chat_id, f"🚫 Banned {tid} — {html.escape(reason or 'no reason')}")
return
if cmd == "/unban":
if not args:
_admin_state_set(user_id, "await_unban_id")
await send(chat_id, "Send user ID to unban:")
return
try: tid = int(args[0])
except ValueError:
await send(chat_id, "❗ Invalid user ID."); return
ok = _unban_user(tid)
_log_activity(user, "admin", f"/unban {tid}", "unbanned")
await send(chat_id, "✅ Unbanned." if ok else "ℹ️ User was not banned.")
return
if cmd == "/broadcast":
if not args:
_admin_state_set(user_id, "await_broadcast")
await send(chat_id, "Send the message to broadcast to all users:")
return
await send(chat_id, "ℹ️ Use /broadcast with no args, then send the message.")
return
if cmd == "/export":
_log_activity(user, "admin", "/export", "file")
await _send_log_export(chat_id); return
if cmd == "/logphone":
if not args:
_admin_state_set(user_id, "await_logphone")
await send(chat_id, "Send a 10-digit number to search the log:")
return
phone, hint = normalize_phone(args[0])
if not phone:
await send(chat_id, f"❗ {hint}"); return
entries = [e for e in ACTIVITY_LOG
if e["action"] == "search" and e["input"] == phone]
entries.reverse()
if not entries:
await send(chat_id, f"🔎 No searches for {phone}."); return
lines = [f"🔎 Searches for {phone} ({len(entries)})\n"]
for e in entries[:20]:
uname = e.get("username"); first = e.get("first_name")
nm = f"@{uname}" if uname else (first or "(anon)")
lines.append(f"• {_fmt_ts(e['ts'])} · {html.escape(nm)} · {e['user_id']}")
await send(chat_id, "\n".join(lines)); return
if cmd == "/maintenance":
if not args:
m_on, m_reason = _get_maintenance()
st = "ON" if m_on else "OFF"
await send(chat_id, f"🛠️ Maintenance: {st}"
+ (f"\nReason: {html.escape(m_reason)}" if m_on else ""))
return
mode = args[0].lower()
if mode == "on":
reason = " ".join(args[1:]) if len(args) > 1 else ""
if reason:
_set_maintenance(True, reason)
_log_activity(user, "admin", "maintenance on", "state", reason)
await send(chat_id, f"🔴 Maintenance ON — {html.escape(reason)}")
else:
_admin_state_set(user_id, "await_maint_reason")
await send(chat_id, "Send a short reason for maintenance:")
return
elif mode == "off":
_set_maintenance(False)
_log_activity(user, "admin", "maintenance off", "state")
await send(chat_id, "🟢 Maintenance OFF"); return
await send(chat_id, "Usage: /maintenance on|off [reason]"); return
# ── /status ──
if text == "/status":
uptime = (time.time() - STATS["started"]) / 60
m_on, _ = _get_maintenance()
_log_activity(user, "status", "/status", "text")
await send(chat_id, (
f"📊 Bot Status — v{VERSION}\n\n"
f"⏱️ Uptime: {uptime:.1f} min\n"
f"🔌 DuckDB: {_con_ready.is_set()}\n"
f"🔥 Cache warmed: {_warmed.is_set()}\n"
f"📂 Files: {len(PARQUET_URLS)} (HiTeck only)\n"
f"🛠️ Maintenance: {m_on}\n"
f"🔍 Queries: {STATS['queries']}\n"
f"⏰ Timeouts: {STATS['timeouts']}\n"
f"❌ Errors: {STATS['errors']}"
)); return
# ── /info ──
if text == "/info":
uptime = (time.time() - STATS["started"]) / 60
s = _get_stats()
_log_activity(user, "info", "/info", "text")
await send(chat_id, (
f"ℹ️ About This Bot\n"
"━━━━━━━━━━━━━━━━━━\n"
f"🤖 Version: {VERSION}\n"
f"📊 Sources: HiTeck (7 files) + 2 APIs\n"
f"👥 Unique users: {s['unique_users']}\n"
f"🔍 Searches: {s['total_searches']}\n"
f"⏱️ Avg latency (today): "
f"{_avg_latency_today():.1f}s\n"
f"🕒 Uptime: {uptime:.0f} min\n"
"━━━━━━━━━━━━━━━━━━\n"
"Built with ❤️ for fast, private lookups."
)); return
# ── /history ──
if text == "/history":
with _activity_lock:
h = list(USER_HISTORY.get(user_id, []))
_log_activity(user, "history", "/history", "text")
if not h:
await send(chat_id, "📭 No history yet\n\nSearch a number to start.")
return
rows = []
for i, item in enumerate(h, 1):
rows.append([{
"text": f"{i}. {item['phone']} ({item['count']} res)",
"callback_data": f"h:{item['phone']}"
}])
await send(chat_id,
"🕘 Your last searches\n\nTap to search again:",
reply_markup={"inline_keyboard": rows})
return
# ── /myid ──
if text == "/myid":
_log_activity(user, "myid", "/myid", "text")
await send(chat_id, f"🆔 Your Telegram ID\n{user_id}")
return
# ── /start, /help ──
if text in ("/start", "/help"):
_log_activity(user, "start", text, "text")
await send(chat_id, WELCOME_TEMPLATE.format(v=VERSION))
if not _warmed.is_set() and not _warm_thread_started:
threading.Thread(target=warm_phone_cache, daemon=True).start()
return
# ── BATCH SEARCH (comma-separated) ──
if "," in text:
parts = [p.strip() for p in text.split(",") if p.strip()]
if len(parts) < 2:
await send(chat_id, "❌ Batch format\n\nExample: 9876543210,9876543211")
return
if len(parts) > BATCH_MAX:
await send(chat_id, f"❌ Max {BATCH_MAX} numbers per batch.")
return
phones = []
for p in parts:
ph, hint = normalize_phone(p)
if not ph:
await send(chat_id, f"❌ Invalid number: {html.escape(p)} — {hint}")
return
phones.append(ph)
await _handle_batch(chat_id, user, phones)
return
# ── SINGLE SEARCH ──
phone, hint = normalize_phone(text)
if not phone:
_log_activity(user, "invalid", text, "error", hint)
await send(chat_id,
"❌ Invalid input\n\n"
"Send a 10-digit number.\n"
"Also accepts: +91 98765 43210, "
"09876543210, 9876543210,9876543211")
return
# Show hint if user typed a non-10-digit but we still parsed it
if text.strip() != phone and text.strip().isdigit() is False:
await send(chat_id, f"ℹ️ Interpreted as {phone}")
# Cooldown
now = time.time()
if now - USER_LAST.get(user_id, 0) < USER_COOLDOWN:
wait = USER_COOLDOWN - (now - USER_LAST.get(user_id, 0))
_log_activity(user, "cooldown", phone, "text", f"wait {wait:.0f}s")
await send(chat_id, f"⏳ Wait {wait:.0f}s before next search.")
return
USER_LAST[user_id] = now
if not _warmed.is_set() and not _warm_thread_started:
threading.Thread(target=warm_phone_cache, daemon=True).start()
placeholder = await send(
chat_id,
f"🔍 Searching…\n\n📱 {html.escape(phone)}\n\n"
f"First query may take up to {QUERY_TIMEOUT}s. Repeats are fast."
)
if placeholder is None:
return
t0 = time.time()
try:
results, elapsed, source_used = await _do_single_search(phone)
except asyncio.TimeoutError:
STATS["timeouts"] += 1
_log_activity(user, "search", phone, "timeout", f"after {QUERY_TIMEOUT}s")
await edit(chat_id, placeholder, f"⏱️ Timed out after {QUERY_TIMEOUT}s. Try again.")
return
except TimeoutError:
STATS["timeouts"] += 1
_log_activity(user, "search", phone, "timeout", f"after {QUERY_TIMEOUT}s")
await edit(chat_id, placeholder, f"⏱️ Timed out after {QUERY_TIMEOUT}s. Try again.")
return
except Exception as e:
STATS["errors"] += 1
log.error(f"❌ search error: {e!r}\n{traceback.format_exc()}")
_log_activity(user, "search", phone, "error", str(e)[:100])
await edit(chat_id, placeholder, f"❌ Error: {html.escape(str(e)[:200])}")
return
STATS["queries"] += 1
log.info(f"✅ Search returned {len(results)} rows in {elapsed:.1f}s")
_add_to_history(user_id, phone, len(results), elapsed, source_used)
if not results:
_log_activity(user, "search", phone, "no_results",
"no records found", int(elapsed * 1000), source_used)
await edit(chat_id, placeholder,
f"❌ No record found\n\n"
f"📱 {html.escape(phone)}\n"
f"⏱️ {elapsed:.1f}s")
return
search_id = uuid.uuid4().hex[:10]
total_pages = (len(results) + PAGE_SIZE - 1) // PAGE_SIZE
source_note = "Using API fallback (DuckDB warming)" if source_used == "api_fallback" else ""
SEARCH_CACHE[search_id] = {
"ts": time.time(), "results": results, "query": phone,
"elapsed": elapsed, "source_note": source_note,
}
_log_activity(user, "search", phone, "results",
f"{len(results)} result(s)", int(elapsed * 1000), source_used)
await edit(
chat_id, placeholder,
render_page(results, 0, total_pages, phone, elapsed, source_note),
reply_markup=page_keyboard(search_id, 0, total_pages),
)
except Exception as e:
log.error(f"❌ handle_message fatal: {e!r}\n{traceback.format_exc()}")
async def _handle_batch(chat_id, user, phones):
await send(chat_id, f"🔍 Batch search — {len(phones)} number(s)…")
t0 = time.time()
tasks = [asyncio.create_task(_safe_batch_one(p)) for p in phones]
results = await asyncio.gather(*tasks, return_exceptions=True)
total_elapsed = time.time() - t0
lines = [f"📊 Batch results ({len(phones)} numbers) — {total_elapsed:.1f}s\n"]
for phone, res in zip(phones, results):
if isinstance(res, Exception):
lines.append(f"❌ {phone} — error: {html.escape(str(res)[:60])}")
else:
n, ok = res
if ok:
lines.append(f"✅ {phone} — {n} record(s)")
else:
lines.append(f"❌ {phone} — no results")
kb_rows = []
for phone in phones:
kb_rows.append([{"text": f"View {phone}", "callback_data": f"h:{phone}"}])
_log_activity(user, "batch_search", ",".join(phones), "text",
f"{len(phones)} nums in {total_elapsed:.1f}s")
await send(chat_id, "\n".join(lines),
reply_markup={"inline_keyboard": kb_rows})
async def _safe_batch_one(phone: str) -> tuple[int, bool]:
try:
results, _, _ = await asyncio.wait_for(_do_single_search(phone), timeout=BATCH_TIMEOUT)
return len(results), len(results) > 0
except Exception:
return 0, False
async def _send_log_export(chat_id: int):
"""Generate CSV + JSON, send as documents."""
entries = list(ACTIVITY_LOG)
if not entries:
await send(chat_id, "📭 No activity to export.")
return
# CSV
buf = io.StringIO()
w = csv.writer(buf)
w.writerow(["timestamp", "user_id", "username", "first_name",
"action", "input", "reply_type", "summary", "elapsed_ms", "source"])
for e in entries:
w.writerow([
time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(e["ts"])),
e["user_id"], e.get("username") or "", e.get("first_name") or "",
e["action"], e.get("input", ""), e.get("reply_type", ""),
e.get("summary", ""), e.get("elapsed_ms", 0), e.get("source", ""),
])
csv_bytes = buf.getvalue().encode("utf-8")
ts = time.strftime("%Y%m%d_%H%M%S")
await send_document(chat_id, f"activity_log_{ts}.csv", csv_bytes,
caption=f"📥 Activity log — {len(entries)} entries")
async def _run_broadcast(admin_chat_id: int, message: str):
"""Background broadcast with rate limit. Sends a summary to admin."""
with _activity_lock:
recipients = list(UNIQUE_USERS.keys())
if not recipients:
await send(admin_chat_id, "📭 No users to broadcast to.")
return
await send(admin_chat_id, f"📢 Broadcasting to {len(recipients)} users…")
sent = 0; failed = 0
for uid in recipients:
if _is_banned(uid):
continue
try:
res = await tg("sendMessage", {
"chat_id": uid,
"text": f"📢 Announcement\n\n{message}",
"parse_mode": "HTML",
"disable_web_page_preview": True,
}, timeout=15)
if res.get("ok"):
sent += 1
else:
failed += 1
except Exception:
failed += 1
await asyncio.sleep(BROADCAST_DELAY)
await send(admin_chat_id,
f"✅ Broadcast complete\n\n"
f"Sent: {sent}\n"
f"Failed: {failed}")
_log_activity({"id": 0}, "admin", "broadcast", "done",
f"sent={sent} failed={failed}")
def _avg_latency_today() -> float:
now = time.time()
today_start = now - (now % 86400)
with _activity_lock:
vals = [e["elapsed_ms"] for e in ACTIVITY_LOG
if e["action"] == "search" and e["ts"] >= today_start and e["elapsed_ms"] > 0]
if not vals:
return 0.0
return (sum(vals) / len(vals)) / 1000.0
async def handle_callback(cb: dict):
try:
data = cb.get("data", "")
cb_id = cb["id"]
msg = cb.get("message") or {}
chat_id = msg.get("chat", {}).get("id")
message_id = msg.get("message_id")
user = cb.get("from") or {}
user_id = user.get("id")
if not chat_id or not message_id:
await tg("answerCallbackQuery", {"callback_query_id": cb_id})
return
if _is_banned(user_id):
await tg("answerCallbackQuery", {
"callback_query_id": cb_id, "text": "🚫 You are banned.",
"show_alert": True,
}); return
# ── Admin callbacks ──
if data.startswith("adm:"):
if not _is_admin_session(user_id):
await tg("answerCallbackQuery", {
"callback_query_id": cb_id,
"text": "Admin session expired — send /admin again.",
"show_alert": True,
}); return
parts = data.split(":")
action = parts[1] if len(parts) > 1 else "menu"
if action == "menu":
await edit(chat_id, message_id, render_admin_menu(), reply_markup=admin_menu_kb())
elif action == "live":
await edit(chat_id, message_id, render_live(), reply_markup=admin_menu_kb())
elif action == "stats":
await edit(chat_id, message_id, render_stats(), reply_markup=admin_menu_kb())
elif action == "top":
await edit(chat_id, message_id, render_top(), reply_markup=admin_menu_kb())
elif action == "banned":
await edit(chat_id, message_id, render_banned(), reply_markup=admin_menu_kb())
elif action == "log":
page = int(parts[2]) if len(parts) > 2 else 0
entries = list(ACTIVITY_LOG); entries.reverse()
t, kb = render_log(entries, page)
await edit(chat_id, message_id, t, reply_markup=kb)
elif action == "users":
page = int(parts[2]) if len(parts) > 2 else 0
t, kb = render_users(page)
await edit(chat_id, message_id, t, reply_markup=kb)
elif action == "ban":
_admin_state_set(user_id, "await_ban_id")
await edit(chat_id, message_id, "🚫 Send user ID (and optional reason):")
elif action == "unban":
_admin_state_set(user_id, "await_unban_id")
await edit(chat_id, message_id, "✅ Send user ID to unban:")
elif action == "broadcast":
_admin_state_set(user_id, "await_broadcast")
await edit(chat_id, message_id, "📢 Send the message to broadcast:")
elif action == "export":
await edit(chat_id, message_id, "📥 Generating export…",
reply_markup=admin_menu_kb())
await _send_log_export(chat_id)
elif action == "logphone":
_admin_state_set(user_id, "await_logphone")
await edit(chat_id, message_id, "🔎 Send a 10-digit number:")
elif action == "maint":
m_on, _ = _get_maintenance()
if m_on:
_set_maintenance(False)
await edit(chat_id, message_id, "🟢 Maintenance OFF",
reply_markup=admin_menu_kb())
else:
_admin_state_set(user_id, "await_maint_reason")
await edit(chat_id, message_id, "Send a short maintenance reason:")
elif action == "noop":
pass
await tg("answerCallbackQuery", {"callback_query_id": cb_id})
_log_activity(user, "admin", data, "menu")
return
# ── Search history re-run ──
if data.startswith("h:"):
phone = data.split(":", 1)[1]
await tg("answerCallbackQuery", {
"callback_query_id": cb_id,
"text": f"Searching {phone}…",
})
# Reuse the single-search flow by faking a message
fake_msg = {"chat": {"id": chat_id}, "from": user, "text": phone}
await handle_message(fake_msg)
return
# ── Copy page ──
if data.startswith("c:"):
try:
_, sid, page_s = data.split(":")
page = int(page_s)
except Exception:
await tg("answerCallbackQuery", {"callback_query_id": cb_id})
return
entry = SEARCH_CACHE.get(sid)
if not entry:
await tg("answerCallbackQuery", {
"callback_query_id": cb_id,
"text": "Session expired.", "show_alert": True,
}); return
results = entry["results"]
total_pages = (len(results) + PAGE_SIZE - 1) // PAGE_SIZE
page = max(0, min(page, total_pages - 1))
start = page * PAGE_SIZE
chunk = results[start:start + PAGE_SIZE]
text = "\n\n".join(_plain_record(start + i + 1, r)
for i, r in enumerate(chunk))
await send(chat_id, f"📋 Copy text for page {page + 1}\n\n"
f"{html.escape(text[:3500])}")
await tg("answerCallbackQuery", {"callback_query_id": cb_id,
"text": "Sent as a copyable message"})
_log_activity(user, "copy", entry["query"], "text")
return
# ── Export ──
if data.startswith("e:"):
sid = data.split(":", 1)[1]
entry = SEARCH_CACHE.get(sid)
if not entry:
await tg("answerCallbackQuery", {
"callback_query_id": cb_id,
"text": "Session expired.", "show_alert": True,
}); return
phone = entry["query"]
results = entry["results"]
payload = json.dumps(results, indent=2, ensure_ascii=False).encode("utf-8")
await send_document(chat_id, f"results_{phone}.json", payload,
caption=f"📥 {len(results)} record(s) for {phone}")
await tg("answerCallbackQuery", {"callback_query_id": cb_id,
"text": "Sent as file"})
_log_activity(user, "export", phone, "file", f"{len(results)} records")
return
# ── Pagination ──
if data == "noop" or not data.startswith("p:"):
await tg("answerCallbackQuery", {"callback_query_id": cb_id})
return
try:
_, search_id, page_str = data.split(":")
page = int(page_str)
except Exception:
await tg("answerCallbackQuery", {"callback_query_id": cb_id})
return
entry = SEARCH_CACHE.get(search_id)
if not entry:
await tg("answerCallbackQuery", {
"callback_query_id": cb_id,
"text": "Session expired — search again.", "show_alert": True,
}); return
results = entry["results"]
total_pages = (len(results) + PAGE_SIZE - 1) // PAGE_SIZE
page = max(0, min(page, total_pages - 1))
await edit(
chat_id, message_id,
render_page(results, page, total_pages, entry["query"],
entry["elapsed"], entry.get("source_note", "")),
reply_markup=page_keyboard(search_id, page, total_pages),
)
await tg("answerCallbackQuery", {"callback_query_id": cb_id})
_log_activity(user, "paginate", entry["query"], "page",
f"page {page + 1}/{total_pages}")
except Exception as e:
log.error(f"❌ handle_callback fatal: {e!r}\n{traceback.format_exc()}")
# ─────────────────────────────────────────────
# WEBHOOK & HEALTH
# ─────────────────────────────────────────────
async def set_webhook():
if not SPACE_URL: return
url = f"{SPACE_URL}/webhook"
payload = {
"url": url, "secret_token": WEBHOOK_SECRET,
"drop_pending_updates": True,
"allowed_updates": ["message", "callback_query"],
}
for attempt in range(1, 4):
try:
async with httpx.AsyncClient(timeout=30) as client:
r = await client.post(f"{TELEGRAM_API}/setWebhook", json=payload)
data = r.json()
log.info(f"🔗 setWebhook → {data}")
if data.get("ok"): return
except Exception as e:
log.warning(f"⚠️ setWebhook attempt {attempt}: {e}")
await asyncio.sleep(3)
async def _health_endpoint(request: Request):
s = _get_stats()
return JSONResponse({
"status": "ok",
"version": VERSION,
"duckdb_ready": _con_ready.is_set(),
"warmed": _warmed.is_set(),
"files": len(PARQUET_URLS),
"cached_searches": len(SEARCH_CACHE),
"live_users": s["live_users"],
"unique_users": s["unique_users"],
"banned_users": s["banned_count"],
"maintenance": s["maintenance"],
"stats": STATS,
})
def _esc_html(s):
return html.escape(str(s)) if s is not None else ""
async def _admin_dashboard(request: Request):
token = request.query_params.get("token", "")
if not WEBHOOK_SECRET or token != WEBHOOK_SECRET:
return HTMLResponse("{e['user_id']}{_esc_html(e['input'])}Auto-refresh 10s. Uptime: {uptime:.1f} min
| Time | User | User ID | Action | Input | Reply | Latency | Source |
|---|---|---|---|---|---|---|---|
| No activity yet | |||||||