#!/usr/bin/env python3 # -*- coding: utf-8 -*- import telebot import os import sqlite3 import threading import requests import json import re import time import base64 import hashlib import hmac import random import struct import asyncio import aiohttp import ssl from datetime import datetime, timedelta from telebot import types import urllib3 import sys import ssl ssl._create_default_https_context = ssl._create_unverified_context urllib3.disable_warnings() API_TOKEN = '8869295385:AAEaQWgheEJkNxWxp9MUOmTXTPqxnDt7ytY' ADMIN_ID = 8208453819 CHANNEL_USERNAME = "@JiangQ888" CHANNEL_LINK = "https://t.me/JiangQ888" BOT_USERNAME = "@MRXHP3Bot" DB_FILE = "bot_data.db" DIR = os.path.dirname(os.path.abspath(__file__)) bot = telebot.TeleBot(API_TOKEN) db_lock = threading.Lock() user_states = {} p3_sessions = {} # ==================== SM4 ==================== SM4_SBOX = bytes([ 0xd6,0x90,0xe9,0xfe,0xcc,0xe1,0x3d,0xb7,0x16,0xb6,0x14,0xc2,0x28,0xfb,0x2c,0x05, 0x2b,0x67,0x9a,0x76,0x2a,0xbe,0x04,0xc3,0xaa,0x44,0x13,0x26,0x49,0x86,0x06,0x99, 0x9c,0x42,0x50,0xf4,0x91,0xef,0x98,0x7a,0x33,0x54,0x0b,0x43,0xed,0xcf,0xac,0x62, 0xe4,0xb3,0x1c,0xa9,0xc9,0x08,0xe8,0x95,0x80,0xdf,0x94,0xfa,0x75,0x8f,0x3f,0xa6, 0x47,0x07,0xa7,0xfc,0xf3,0x73,0x17,0xba,0x83,0x59,0x3c,0x19,0xe6,0x85,0x4f,0xa8, 0x68,0x6b,0x81,0xb2,0x71,0x64,0xda,0x8b,0xf8,0xeb,0x0f,0x4b,0x70,0x56,0x9d,0x35, 0x1e,0x24,0x0e,0x5e,0x63,0x58,0xd1,0xa2,0x25,0x22,0x7c,0x3b,0x01,0x21,0x78,0x87, 0xd4,0x00,0x46,0x57,0x9f,0xd3,0x27,0x52,0x4c,0x36,0x02,0xe7,0xa0,0xc4,0xc8,0x9e, 0xea,0xbf,0x8a,0xd2,0x40,0xc7,0x38,0xb5,0xa3,0xf7,0xf2,0xce,0xf9,0x61,0x15,0xa1, 0xe0,0xae,0x5d,0xa4,0x9b,0x34,0x1a,0x55,0xad,0x93,0x32,0x30,0xf5,0x8c,0xb1,0xe3, 0x1d,0xf6,0xe2,0x2e,0x82,0x66,0xca,0x60,0xc0,0x29,0x23,0xab,0x0d,0x53,0x4e,0x6f, 0xd5,0xdb,0x37,0x45,0xde,0xfd,0x8e,0x2f,0x03,0xff,0x6a,0x72,0x6d,0x6c,0x5b,0x51, 0x8d,0x1b,0xaf,0x92,0xbb,0xdd,0xbc,0x7f,0x11,0xd9,0x5c,0x41,0x1f,0x10,0x5a,0xd8, 0x0a,0xc1,0x31,0x88,0xa5,0xcd,0x7b,0xbd,0x2d,0x74,0xd0,0x12,0xb8,0xe5,0xb4,0xb0, 0x89,0x69,0x97,0x4a,0x0c,0x96,0x77,0x7e,0x65,0xb9,0xf1,0x09,0xc5,0x6e,0xc6,0x84, 0x18,0xf0,0x7d,0xec,0x3a,0xdc,0x4d,0x20,0x79,0xee,0x5f,0x3e,0xd7,0xcb,0x39,0x48, ]) SM4_FK = (0xA3B1BAC6, 0x56AA3350, 0x677D9197, 0xB27022DC) SM4_CK = ( 0x00070E15, 0x1C232A31, 0x383F464D, 0x545B6269, 0x70777E85, 0x8C939AA1, 0xA8AFB6BD, 0xC4CBD2D9, 0xE0E7EEF5, 0xFC030A11, 0x181F262D, 0x343B4249, 0x50575E65, 0x6C737A81, 0x888F969D, 0xA4ABB2B9, 0xC0C7CED5, 0xDCE3EAF1, 0xF8FF060D, 0x141B2229, 0x30373E45, 0x4C535A61, 0x686F767D, 0x848B9299, 0xA0A7AEB5, 0xBCC3CAD1, 0xD8DFE6ED, 0xF4FB0209, 0x10171E25, 0x2C333A41, 0x484F565D, 0x646B7279, ) def _sm4_rotl(x: int, n: int) -> int: return ((x << n) | (x >> (32 - n))) & 0xFFFFFFFF def _sm4_tau(a: int) -> int: return ( (SM4_SBOX[(a >> 24) & 0xFF] << 24) | (SM4_SBOX[(a >> 16) & 0xFF] << 16) | (SM4_SBOX[(a >> 8) & 0xFF] << 8) | (SM4_SBOX[a & 0xFF]) ) def _sm4_l(b: int) -> int: return b ^ _sm4_rotl(b, 2) ^ _sm4_rotl(b, 10) ^ _sm4_rotl(b, 18) ^ _sm4_rotl(b, 24) def _sm4_lp(b: int) -> int: return b ^ _sm4_rotl(b, 13) ^ _sm4_rotl(b, 23) def _sm4_t(x: int) -> int: return _sm4_l(_sm4_tau(x)) def _sm4_tp(x: int) -> int: return _sm4_lp(_sm4_tau(x)) def _sm4_key_schedule(key: bytes) -> list: mk = struct.unpack(">4I", key) k = [0] * 36 k[0] = mk[0] ^ SM4_FK[0] k[1] = mk[1] ^ SM4_FK[1] k[2] = mk[2] ^ SM4_FK[2] k[3] = mk[3] ^ SM4_FK[3] rk = [0] * 32 for i in range(32): k[i + 4] = k[i] ^ _sm4_tp(k[i + 1] ^ k[i + 2] ^ k[i + 3] ^ SM4_CK[i]) rk[i] = k[i + 4] return rk def sm4_encrypt_block(block: bytes, key: bytes) -> bytes: x = list(struct.unpack(">4I", block)) rk = _sm4_key_schedule(key) for i in range(32): x.append(x[i] ^ _sm4_t(x[i + 1] ^ x[i + 2] ^ x[i + 3] ^ rk[i])) return struct.pack(">4I", x[35], x[34], x[33], x[32]) def sm4_decrypt_block(block: bytes, key: bytes) -> bytes: x = list(struct.unpack(">4I", block)) rk = _sm4_key_schedule(key) for i in range(32): x.append(x[i] ^ _sm4_t(x[i + 1] ^ x[i + 2] ^ x[i + 3] ^ rk[31 - i])) return struct.pack(">4I", x[35], x[34], x[33], x[32]) _BLOCK = 16 def _xor(a: bytes, b: bytes) -> bytes: return bytes(x ^ y for x, y in zip(a, b)) def pkcs7_pad(data: bytes, block_size: int = 16) -> bytes: n = block_size - (len(data) % block_size) return data + bytes([n] * n) def pkcs7_unpad(data: bytes) -> bytes: n = data[-1] if n < 1 or n > 16: raise ValueError(f"PKCS7 填充错误: 最后一字节={n}") return data[:-n] def sm4_cbc_encrypt(plain: bytes, key: bytes, iv: bytes) -> bytes: padded = pkcs7_pad(plain) out, prev = bytearray(), iv for i in range(0, len(padded), _BLOCK): blk = _xor(padded[i:i + _BLOCK], prev) prev = sm4_encrypt_block(blk, key) out.extend(prev) return bytes(out) def sm4_cbc_decrypt(ciphertext: bytes, key: bytes, iv: bytes) -> bytes: out, prev = bytearray(), iv for i in range(0, len(ciphertext), _BLOCK): blk = ciphertext[i:i + _BLOCK] pt = _xor(sm4_decrypt_block(blk, key), prev) out.extend(pt) prev = blk return pkcs7_unpad(bytes(out)) KEY = bytes.fromhex("97b7ce068e8d424edbbf81d19f9e8340") def encode_uri_component(s: str) -> str: from urllib.parse import quote return quote(s, safe="!~*'()") def decode_uri_component(s: str) -> str: from urllib.parse import unquote return unquote(s) def obfuscate_iv(iv: bytes) -> bytes: r = bytearray(32) for i in range(16): rb = random.randint(0, 255) r[i * 2] = ((iv[i] >> 4) | (rb & 0xF0)) & 0xFF r[i * 2 + 1] = ((iv[i] & 0x0F) | ((rb << 4) & 0xF0)) & 0xFF return bytes(r) def recover_iv(obf: bytes) -> bytes: iv = bytearray(16) for i in range(16): iv[i] = ((obf[i * 2] & 0x0F) << 4) | (obf[i * 2 + 1] & 0x0F) return bytes(iv) def derive_hmac_key(iv: bytes, rounds: int) -> bytes: padded = iv + bytes([0x10] * 16) enc = sm4_encrypt_block(padded[:16], KEY) + sm4_encrypt_block(padded[16:], KEY) d = enc for _ in range(rounds): d = hashlib.sha256(d).digest() return _xor(d[:16], d[16:]) def encrypt(data, rounds=5) -> str: json_str = data if isinstance(data, str) else json.dumps(data, ensure_ascii=False) plain_bytes = encode_uri_component(json_str).encode("latin-1") iv = os.urandom(16) ciphertext = sm4_cbc_encrypt(plain_bytes, KEY, iv) hmac_key = derive_hmac_key(iv, rounds) mac = hmac.digest(hmac_key, ciphertext, hashlib.sha256) return ( obfuscate_iv(iv).hex() + base64.b64encode(ciphertext).decode() + "-" + base64.b64encode(mac).decode() ) def decrypt(cipher_str: str) -> dict: iv_hex = cipher_str[:64] obf_iv = bytes.fromhex(iv_hex) if len(obf_iv) != 32: raise ValueError(f"IV 长度错误: {len(obf_iv)} (期望 32)") iv = recover_iv(obf_iv) parts = cipher_str[64:].split("-") if len(parts) != 2: raise ValueError("密文格式错误: 缺少 HMAC 分隔符") ct_bytes = base64.b64decode(parts[0]) mac_bytes = base64.b64decode(parts[1]) ok = False for rounds in (5, 6): hk = derive_hmac_key(iv, rounds) if hmac.compare_digest(hmac.digest(hk, ct_bytes, hashlib.sha256), mac_bytes): ok = True break if not ok: raise ValueError("HMAC 验证失败") pt = sm4_cbc_decrypt(ct_bytes, KEY, iv) try: return json.loads(pt.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError): pass try: return json.loads(decode_uri_component(pt.decode("latin-1"))) except (UnicodeDecodeError, json.JSONDecodeError): pass raise ValueError("解密后无法解析为 JSON") TEMPLATE = { "DeptSeq": 47, "model": "microsoft", "ClientBaseVersion": "3.16.2", "MobileBrand": "microsoft", "Langage": "zh_CN", "SystemType": "Windows Unknown x64", "SystemVersion": "Windows Unknown x64", "DevicePixelRatio": 1, "ClientPlatform": "windows", "PhoneHeight": None, "PhoneWidth": None, "IPADD": "", "LONGITUDE": None, "LATITUDE": None, "geohash6": "wwdzec", "UnionId": "oCGpF5x8dKQm7vL2nBZ-aH9KsQw", "MinAppOpenId": "", "LocationName": "", "GpsCountry": "", "GpsProvince": "", "GpsCity": "", "District": "", "MacAddress": "", "Coordinate": "", } URL = "https://wxapp.js96008.com/wxminapp/MobAuthtication.do" HEADERS = { "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36 " "MicroMessenger/7.0.20.1781(0x6700143B) NetType/WIFI " "MiniProgramEnv/Windows WindowsWechat/WMPF " "WindowsWechat(0x63090a13) UnifiedPCWindowsWechat(0xf2541b18) XWEB/17071" ), "Content-Type": "application/json; charset=UTF-8", "xweb_xhr": "1", "Sec-Fetch-Site": "cross-site", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Dest": "empty", "Referer": "https://servicewechat.com/wxb1d504f9ec37735c/267/page-frame.html", "Accept-Language": "zh-CN,zh;q=0.9", } def query_sync(name: str, phone: str, idno: str) -> dict: data = {**TEMPLATE, "IdNo": idno, "CifName": name, "MobilePhone": phone} cipher = encrypt(data) payload = { "ChannelId": "W", "UnionId": "oCGpF523dYUCh6ba7nBZ-jHG8KuM", "Data": cipher, } try: resp = requests.post(URL, data=json.dumps(payload), headers=HEADERS, timeout=30, verify=False) resp.raise_for_status() raw = resp.json() if raw.get("DecodeData"): return decrypt(raw["DecodeData"]) return raw except Exception as e: return {"ReturnCode": "ERROR", "ReturnMsg": str(e)} # ==================== 数据库 ==================== def init_db(): with db_lock: conn = sqlite3.connect(DB_FILE, check_same_thread=False) c = conn.cursor() c.execute('''CREATE TABLE IF NOT EXISTS users ( user_id INTEGER PRIMARY KEY, username TEXT, auth_days INTEGER DEFAULT 0, auth_expire TEXT, is_banned INTEGER DEFAULT 0, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP )''') c.execute('''CREATE TABLE IF NOT EXISTS auth_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER, days INTEGER, expire_time TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP )''') # 给管理员授权永久 c.execute("INSERT OR IGNORE INTO users (user_id, username, auth_days, auth_expire) VALUES (?, ?, ?, ?)", (ADMIN_ID, "admin", -1, "永久")) conn.commit() conn.close() print("数据库初始化完成") init_db() class DBUtils: @staticmethod def get_user(user_id): with db_lock: conn = sqlite3.connect(DB_FILE, check_same_thread=False) c = conn.cursor() c.execute("SELECT * FROM users WHERE user_id = ?", (user_id,)) row = c.fetchone() conn.close() if row: return { 'user_id': row[0], 'username': row[1], 'auth_days': row[2], 'auth_expire': row[3], 'is_banned': row[4] } return None @staticmethod def create_user(user_id, username): with db_lock: conn = sqlite3.connect(DB_FILE, check_same_thread=False) c = conn.cursor() c.execute("INSERT OR IGNORE INTO users (user_id, username) VALUES (?, ?)", (user_id, username)) conn.commit() conn.close() @staticmethod def is_authorized(user_id): user = DBUtils.get_user(user_id) if not user: return False if user['is_banned']: return False if user['auth_days'] == -1: return True if not user['auth_expire']: return False try: expire = datetime.strptime(user['auth_expire'], "%Y-%m-%d %H:%M:%S") return expire > datetime.now() except: return False @staticmethod def get_auth_remain(user_id): user = DBUtils.get_user(user_id) if not user: return 0 if user['auth_days'] == -1: return -1 if not user['auth_expire']: return 0 try: expire = datetime.strptime(user['auth_expire'], "%Y-%m-%d %H:%M:%S") remain = (expire - datetime.now()).days return max(0, remain) except: return 0 @staticmethod def set_auth(user_id, days): with db_lock: conn = sqlite3.connect(DB_FILE, check_same_thread=False) c = conn.cursor() if days == -1: c.execute("UPDATE users SET auth_days = -1, auth_expire = '永久' WHERE user_id = ?", (user_id,)) else: expire = (datetime.now() + timedelta(days=days)).strftime("%Y-%m-%d %H:%M:%S") c.execute("UPDATE users SET auth_days = ?, auth_expire = ? WHERE user_id = ?", (days, expire, user_id)) c.execute("INSERT INTO auth_logs (user_id, days, expire_time) VALUES (?, ?, ?)", (user_id, days, expire if days != -1 else '永久')) conn.commit() conn.close() @staticmethod def ban_user(user_id): with db_lock: conn = sqlite3.connect(DB_FILE, check_same_thread=False) c = conn.cursor() c.execute("UPDATE users SET is_banned = 1 WHERE user_id = ?", (user_id,)) conn.commit() conn.close() @staticmethod def unban_user(user_id): with db_lock: conn = sqlite3.connect(DB_FILE, check_same_thread=False) c = conn.cursor() c.execute("UPDATE users SET is_banned = 0 WHERE user_id = ?", (user_id,)) conn.commit() conn.close() @staticmethod def get_all_users(): with db_lock: conn = sqlite3.connect(DB_FILE, check_same_thread=False) c = conn.cursor() c.execute("SELECT user_id, username, auth_days, auth_expire, is_banned FROM users") rows = c.fetchall() conn.close() return rows # ==================== 辅助函数 ==================== def check_channel_membership(user_id): try: member = bot.get_chat_member(CHANNEL_USERNAME, user_id) return member.status in ['member', 'administrator', 'creator'] except Exception: return False def send_channel_required_message(chat_id): keyboard = telebot.types.InlineKeyboardMarkup(row_width=1) keyboard.add( telebot.types.InlineKeyboardButton("📢 加入频道", url=CHANNEL_LINK), telebot.types.InlineKeyboardButton("✅ 我已加入", callback_data="check_joined") ) bot.send_message(chat_id, f"⚠️ 请先加入频道才能使用机器人：\n\n{CHANNEL_LINK}\n\n加入后请点击「✅ 我已加入」按钮", reply_markup=keyboard) def extract_ids_from_text(content): pattern = r'[1-9]\d{5}(?:19|20)\d{2}(?:0[1-9]|1[0-2])(?:0[1-9]|[12]\d|3[01])\d{3}[\dXx]' id_cards = re.findall(pattern, content, re.IGNORECASE) return list(set([id_card.upper() for id_card in id_cards])) def create_progress_bar(current, total, length=20): if total == 0: return "[░░░░░░░░░░░░░░░░░░░░] 0%" percent = current / total filled = int(percent * length) bar = "█" * filled + "░" * (length - filled) return f"[{bar}] {percent*100:.1f}% ({current}/{total})" # ==================== 命令处理器 ==================== @bot.message_handler(commands=['start']) def send_welcome(message): user_id = message.from_user.id username = message.from_user.username or "未知用户" DBUtils.create_user(user_id, username) if not check_channel_membership(user_id): send_channel_required_message(message.chat.id) return if DBUtils.is_authorized(user_id): remain = DBUtils.get_auth_remain(user_id) remain_text = "永久" if remain == -1 else f"{remain}天" text = f"✅ 欢迎使用P3核验机器人！\n\n您的授权状态：已授权（剩余 {remain_text}）\n\n可用命令：\n/p3 姓名 手机号 - 批量核验\n/sys 姓名 手机号 身份证 - 单次三要素核验\n/id_extract - 上传TXT提取身份证\n/me - 查看授权信息" else: text = "❌ 您未授权，请联系管理员 @JiangQ888 购买授权" bot.reply_to(message, text) @bot.message_handler(commands=['me']) def handle_me(message): user_id = message.from_user.id user = DBUtils.get_user(user_id) if not user: bot.reply_to(message, "请先发送 /start 初始化") return if DBUtils.is_authorized(user_id): remain = DBUtils.get_auth_remain(user_id) remain_text = "永久" if remain == -1 else f"{remain}天" text = f"✅ 授权状态：已授权（剩余 {remain_text}）\n用户ID：{user_id}\n用户名：{user['username']}" else: text = f"❌ 未授权\n用户ID：{user_id}\n用户名：{user['username']}\n请联系管理员 @JiangQ888 购买授权" bot.reply_to(message, text) @bot.message_handler(commands=['p3']) def handle_p3(message): user_id = message.from_user.id if not check_channel_membership(user_id): send_channel_required_message(message.chat.id) return if not DBUtils.is_authorized(user_id): bot.reply_to(message, "❌ 您未授权，请联系管理员 @JiangQ888 购买授权") return args = message.text.split() if len(args) != 3: bot.reply_to(message, "❌ 格式错误！\n正确格式：/p3 姓名 手机号\n示例：/p3 张三 13800138000") return name = args[1].strip() phone = args[2].strip() if not re.match(r'^1[3-9]\d{9}$', phone): bot.reply_to(message, "❌ 手机号格式错误！请输入11位正确手机号") return p3_sessions[user_id] = {'name': name, 'phone': phone, 'ids': [], 'step': 'waiting_ids'} bot.reply_to(message, f"✅ 已接收：姓名 {name}，手机号 {phone}\n\n请发送身份证号列表（每行一个，用换行分隔），或上传TXT文件\n\n💡 也可直接粘贴身份证号，每行一个，支持18位身份证") @bot.message_handler(commands=['sys']) def handle_sys(message): user_id = message.from_user.id if not check_channel_membership(user_id): send_channel_required_message(message.chat.id) return if not DBUtils.is_authorized(user_id): bot.reply_to(message, "❌ 您未授权，请联系管理员 @JiangQ888 购买授权") return args = message.text.split() if len(args) != 4: bot.reply_to(message, "❌ 格式错误！\n支持以下格式：\n/sys 姓名 手机号 身份证\n/sys 姓名 身份证 手机号\n示例：/sys 张三 13800138000 110101199001011234") return name = args[1].strip() param2 = args[2].strip() param3 = args[3].strip().upper() phone = None idno = None if re.match(r'^1[3-9]\d{9}$', param2): phone = param2 idno = param3 elif re.match(r'^1[3-9]\d{9}$', param3): phone = param3 idno = param2 else: for p in [param2, param3]: if re.match(r'^1[3-9]\d{9}$', p): phone = p elif re.match(r'^[1-9]\d{5}(?:19|20)\d{2}(?:0[1-9]|1[0-2])(?:0[1-9]|[12]\d|3[01])\d{3}[\dXx]$', p, re.I): idno = p.upper() if not phone or not idno: bot.reply_to(message, "❌ 无法识别手机号或身份证！请检查格式\n支持：/sys 姓名 手机号 身份证 或 /sys 姓名 身份证 手机号") return if len(idno) != 18: bot.reply_to(message, "❌ 身份证必须为18位！") return msg = bot.reply_to(message, f"🔍 正在核验三要素...\n姓名：{name}\n手机号：{phone}\n身份证：{idno}") try: result = query_sync(name, phone, idno) ret_code = result.get("ReturnCode", "") ret_msg = result.get("ReturnMsg", "") if ret_code == "000000": bot.edit_message_text( f"✅ 三要素核验一致！\n姓名：{name}\n手机号：{phone}\n身份证：{idno}", user_id, msg.message_id ) else: bot.edit_message_text( f"❌ 三要素核验失败\n姓名：{name}\n手机号：{phone}\n身份证：{idno}\n\n错误：{ret_msg}", user_id, msg.message_id ) except Exception as e: bot.edit_message_text(f"❌ 核验异常：{str(e)}", user_id, msg.message_id) @bot.message_handler(commands=['id_extract']) def handle_id_extract(message): user_id = message.from_user.id if not check_channel_membership(user_id): send_channel_required_message(message.chat.id) return if not DBUtils.is_authorized(user_id): bot.reply_to(message, "❌ 您未授权，请联系管理员 @JiangQ888 购买授权") return user_states[user_id] = {'waiting_id_file': True} bot.reply_to(message, "📁 请上传包含身份证号的TXT文件\n\n我会自动提取所有18位身份证号，并生成纯净的身份证列表文件") # ==================== 管理员命令 ==================== @bot.message_handler(commands=['admin']) def handle_admin(message): user_id = message.from_user.id if user_id != ADMIN_ID: bot.reply_to(message, "❌ 无权限操作！") return all_users = DBUtils.get_all_users() total = len(all_users) authorized = sum(1 for u in all_users if DBUtils.is_authorized(u[0])) text = f""" 👑 管理员面板 ━━━━━━━━━━━━━━━━━━━━ 📊 系统统计 • 用户总数：{total} • 已授权用户：{authorized} ━━━━━━━━━━━━━━━━━━━━ 📋 管理指令： /auth 用户ID 天数 - 授权用户（-1为永久） /ban 用户ID - 封禁用户 /unban 用户ID - 解封用户 /list - 查看用户列表 """ bot.reply_to(message, text) @bot.message_handler(commands=['auth']) def handle_auth(message): user_id = message.from_user.id if user_id != ADMIN_ID: return args = message.text.split() if len(args) != 3: bot.reply_to(message, "❌ 格式错误！\n/auth 用户ID 天数\n示例：/auth 123456 30 (-1为永久)") return try: target_id = int(args[1]) days = int(args[2]) DBUtils.set_auth(target_id, days) remain = "永久" if days == -1 else f"{days}天" bot.reply_to(message, f"✅ 用户 {target_id} 已授权 {remain}") except: bot.reply_to(message, "❌ 参数错误！") @bot.message_handler(commands=['ban']) def handle_ban(message): user_id = message.from_user.id if user_id != ADMIN_ID: return args = message.text.split() if len(args) != 2: bot.reply_to(message, "❌ 格式错误！\n/ban 用户ID") return try: target_id = int(args[1]) DBUtils.ban_user(target_id) bot.reply_to(message, f"✅ 用户 {target_id} 已封禁") except: bot.reply_to(message, "❌ 参数错误！") @bot.message_handler(commands=['unban']) def handle_unban(message): user_id = message.from_user.id if user_id != ADMIN_ID: return args = message.text.split() if len(args) != 2: bot.reply_to(message, "❌ 格式错误！\n/unban 用户ID") return try: target_id = int(args[1]) DBUtils.unban_user(target_id) bot.reply_to(message, f"✅ 用户 {target_id} 已解封") except: bot.reply_to(message, "❌ 参数错误！") @bot.message_handler(commands=['list']) def handle_list(message): user_id = message.from_user.id if user_id != ADMIN_ID: return users = DBUtils.get_all_users() if not users: bot.reply_to(message, "暂无用户") return text = "📋 用户列表\n━━━━━━━━━━━━━━━━━━━━\n" for uid, username, auth_days, auth_expire, banned in users: status = "❌封禁" if banned else "✅正常" if auth_days == -1: auth = "永久" elif auth_expire: try: expire = datetime.strptime(auth_expire, "%Y-%m-%d %H:%M:%S") days = (expire - datetime.now()).days auth = f"{days}天" except: auth = "过期" else: auth = "未授权" text += f"{status} {uid} | {username} | {auth}\n" bot.reply_to(message, text) # ==================== P3批量核验逻辑 ==================== def process_p3_batch(user_id, name, phone, id_list): msg = bot.send_message(user_id, "🔍 开始核验...\n" + create_progress_bar(0, len(id_list))) for i, idno in enumerate(id_list, 1): try: result = query_sync(name, phone, idno) ret_code = result.get("ReturnCode", "") ret_msg = result.get("ReturnMsg", "") if ret_code == "000000": bot.edit_message_text( f"✅ 核验成功！找到匹配身份证！\n━━━━━━━━━━━━━━━━━━━━\n姓名：{name}\n手机号：{phone}\n身份证：{idno}\n━━━━━━━━━━━━━━━━━━━━\n已核验 {i}/{len(id_list)} 个", user_id, msg.message_id ) return elif "手机号和证件号认证一致" in ret_msg: bot.edit_message_text( f"✅ 机主二要素一致！\n姓名：{name}\n手机号：{phone}\n身份证：{idno}", user_id, msg.message_id ) return progress = create_progress_bar(i, len(id_list)) bot.edit_message_text( f"🔍 正在核验...\n{progress}\n当前：{idno}\n状态：{ret_msg[:30] if ret_msg else '核验中'}", user_id, msg.message_id ) except Exception as e: progress = create_progress_bar(i, len(id_list)) bot.edit_message_text( f"🔍 正在核验...\n{progress}\n当前：{idno}\n异常：{str(e)[:30]}", user_id, msg.message_id ) time.sleep(0.3) bot.edit_message_text( f"❌ 核验完成，未找到匹配身份证\n姓名：{name}\n手机号：{phone}\n已核验 {len(id_list)} 个身份证", user_id, msg.message_id ) # ==================== 消息处理器 ==================== @bot.message_handler(content_types=['text']) def handle_text(message): user_id = message.from_user.id if user_id in p3_sessions and p3_sessions[user_id].get('step') == 'waiting_ids': session = p3_sessions[user_id] text = message.text.strip() if text.lower() == '取消': del p3_sessions[user_id] bot.reply_to(message, "❌ 操作已取消") return id_list = extract_ids_from_text(text) if not id_list: bot.reply_to(message, "❌ 未找到有效的18位身份证号，请重新输入") return if len(id_list) > 100: bot.reply_to(message, f"⚠️ 身份证数量过多（{len(id_list)}个），限制最多100个") id_list = id_list[:100] session['ids'] = id_list session['step'] = 'processing' bot.reply_to(message, f"✅ 已识别 {len(id_list)} 个身份证，开始核验...") threading.Thread(target=process_p3_batch, args=(user_id, session['name'], session['phone'], id_list), daemon=True).start() return if message.text and message.text.startswith('/'): return @bot.message_handler(content_types=['document']) def handle_document(message): user_id = message.from_user.id if not message.document.file_name.lower().endswith('.txt'): bot.reply_to(message, "❌ 请上传TXT格式的文件！") return if user_id in user_states and user_states.get(user_id, {}).get('waiting_id_file'): try: file_info = bot.get_file(message.document.file_id) downloaded_file = bot.download_file(file_info.file_path) content = downloaded_file.decode('utf-8', errors='ignore') id_list = extract_ids_from_text(content) if not id_list: bot.reply_to(message, "❌ 文件中未找到有效的18位身份证号") del user_states[user_id] return timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"id_cards_{timestamp}.txt" with open(filename, 'w', encoding='utf-8') as f: for idno in id_list: f.write(f"{idno}\n") with open(filename, 'rb') as f: bot.send_document(user_id, f, caption=f"✅ 提取完成！共 {len(id_list)} 个身份证") os.remove(filename) del user_states[user_id] except Exception as e: bot.reply_to(message, f"❌ 处理失败：{str(e)}") if user_id in user_states: del user_states[user_id] return if user_id in p3_sessions and p3_sessions[user_id].get('step') == 'waiting_ids': try: file_info = bot.get_file(message.document.file_id) downloaded_file = bot.download_file(file_info.file_path) content = downloaded_file.decode('utf-8', errors='ignore') id_list = extract_ids_from_text(content) if not id_list: bot.reply_to(message, "❌ 文件中未找到有效的18位身份证号") return if len(id_list) > 100: bot.reply_to(message, f"⚠️ 身份证数量过多（{len(id_list)}个），限制最多100个") id_list = id_list[:100] session = p3_sessions[user_id] session['ids'] = id_list session['step'] = 'processing' bot.reply_to(message, f"✅ 已识别 {len(id_list)} 个身份证，开始核验...") threading.Thread(target=process_p3_batch, args=(user_id, session['name'], session['phone'], id_list), daemon=True).start() except Exception as e: bot.reply_to(message, f"❌ 处理失败：{str(e)}") @bot.message_handler(content_types=['photo']) def handle_photo(message): pass # ==================== 回调处理器 ==================== @bot.callback_query_handler(func=lambda call: call.data == "check_joined") def check_joined_callback(call): user_id = call.from_user.id if check_channel_membership(user_id): bot.answer_callback_query(call.id, "✅ 验证成功！") send_welcome(call.message) else: bot.answer_callback_query(call.id, "❌ 请先加入频道！", show_alert=True) # ==================== 启动 ==================== if __name__ == '__main__': print("🤖 P3核验机器人启动中...") print(f"📢 必加频道: {CHANNEL_LINK}") print(f"👑 管理员ID: {ADMIN_ID}") print(f"🤖 Bot用户名: {BOT_USERNAME}") try: bot.infinity_polling(none_stop=True, timeout=60) except Exception as e: print(f"❌ 异常：{e}")