457 lines
19 KiB
Python
457 lines
19 KiB
Python
import os, json, time, base64, asyncio, re, urllib.request, logging, threading
|
||
from http.server import HTTPServer, BaseHTTPRequestHandler
|
||
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
|
||
from telegram.ext import Application, CommandHandler, CallbackQueryHandler, MessageHandler, filters, ContextTypes
|
||
|
||
logging.basicConfig(level=logging.INFO, format='%(asctime)s %(message)s')
|
||
log = logging.getLogger(__name__)
|
||
|
||
TOKEN = os.environ.get('BOT_TOKEN', '')
|
||
ADMIN_ID = int(os.environ.get('ADMIN_ID', '0'))
|
||
DATA_FILE = os.environ.get('DATA_FILE', '/opt/sub-bot/data.json')
|
||
PROTOS = ['ss://', 'vmess://', 'vless://', 'trojan://', 'hysteria2://', 'hy2://', 'tuic://']
|
||
AUTO_DEL = 60
|
||
SUB_SECRET = os.environ.get('SUB_SECRET', 'change_me_to_random_string')
|
||
SUB_HOST = os.environ.get('SUB_HOST', '0.0.0.0:18888')
|
||
WAITING_ADD = set() # user_ids waiting to add sub
|
||
|
||
def load_data():
|
||
if os.path.exists(DATA_FILE):
|
||
with open(DATA_FILE) as f: return json.load(f)
|
||
return {'subs': [], 'groups': []}
|
||
|
||
def save_data(data):
|
||
os.makedirs(os.path.dirname(DATA_FILE), exist_ok=True)
|
||
with open(DATA_FILE, 'w') as f: json.dump(data, f, ensure_ascii=False, indent=2)
|
||
|
||
def detect_type(link):
|
||
for p in PROTOS:
|
||
if link.startswith(p): return p.replace('://', '')
|
||
return 'unknown'
|
||
|
||
def parse_surge_ss(text):
|
||
"""Parse Surge format: Name = ss, server, port, encrypt-method=x, password=y"""
|
||
m = re.match(r'^(.+?)\s*=\s*ss\s*,\s*([^,]+)\s*,\s*(\d+)\s*,\s*(.+)$', text.strip())
|
||
if not m: return None, None
|
||
name = m.group(1).strip()
|
||
server = m.group(2).strip()
|
||
port = m.group(3).strip()
|
||
params = dict(re.findall(r'([\w-]+)\s*=\s*([^,]+)', m.group(4)))
|
||
method = params.get('encrypt-method', '').strip()
|
||
pwd = params.get('password', '').strip()
|
||
if not method or not pwd: return None, None
|
||
# build ss:// link
|
||
raw = f'{method}:{pwd}@{server}:{port}'
|
||
encoded = base64.b64encode(raw.encode()).decode().rstrip('=')
|
||
ss_link = f'ss://{encoded}#{urllib.request.quote(name)}'
|
||
return ss_link, name
|
||
|
||
def extract_name(link):
|
||
if '#' in link: return urllib.request.unquote(link.split('#')[-1])
|
||
return link[:30]
|
||
|
||
async def auto_del(msg, delay=AUTO_DEL):
|
||
await asyncio.sleep(delay)
|
||
try: await msg.delete()
|
||
except: pass
|
||
|
||
async def send_del(target, text, delay=AUTO_DEL, **kw):
|
||
msg = await target.reply_text(text, **kw)
|
||
asyncio.ensure_future(auto_del(msg, delay))
|
||
return msg
|
||
|
||
async def is_member(update, context):
|
||
uid = update.effective_user.id
|
||
if uid == ADMIN_ID: return True
|
||
data = load_data()
|
||
groups = data.get('groups', [])
|
||
if not groups: return True
|
||
for gid in groups:
|
||
try:
|
||
m = await context.bot.get_chat_member(gid, uid)
|
||
if m.status in ('member', 'administrator', 'creator'): return True
|
||
except: pass
|
||
return False
|
||
|
||
def add_sub(link, user):
|
||
proto = detect_type(link)
|
||
if proto == 'unknown': return None
|
||
data = load_data()
|
||
for s in data['subs']:
|
||
if s['link'] == link: return 'dup'
|
||
name = extract_name(link)
|
||
data['subs'].append({'link': link, 'type': proto, 'name': name,
|
||
'added_by': user.id, 'added_name': user.first_name,
|
||
'added_at': int(time.time()), 'alive': True})
|
||
save_data(data)
|
||
return f'✅ 已添加 [{proto}] {name}'
|
||
|
||
# --- /help: main menu with buttons ---
|
||
async def cmd_help(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
if not await is_member(update, context):
|
||
return await update.message.reply_text('⛔ 仅限群成员使用')
|
||
buttons = [
|
||
[InlineKeyboardButton("📋 订阅列表", callback_data='menu_list'),
|
||
InlineKeyboardButton("📥 获取订阅", callback_data='menu_get')],
|
||
[InlineKeyboardButton("➕ 添加订阅", callback_data='menu_add'),
|
||
InlineKeyboardButton("🗑 删除订阅", callback_data='menu_del')],
|
||
[InlineKeyboardButton("🔍 检测存活", callback_data='menu_check')],
|
||
]
|
||
if update.effective_user.id == ADMIN_ID:
|
||
buttons.append([InlineKeyboardButton("⚙️ 绑定当前群", callback_data='menu_setgroup')])
|
||
await send_del(update.message,
|
||
'🚀 *订阅管理 Bot*\n\n'
|
||
'直接发订阅链接自动入库\n'
|
||
'或点击下方按钮操作:',
|
||
parse_mode='Markdown', reply_markup=InlineKeyboardMarkup(buttons))
|
||
|
||
# --- menu callbacks ---
|
||
async def cb_menu(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
q = update.callback_query
|
||
await q.answer()
|
||
action = q.data
|
||
data = load_data()
|
||
|
||
if action == 'menu_list':
|
||
if not data['subs']:
|
||
return await send_del(q.message, '📭 暂无订阅')
|
||
lines = [f"`{i+1}` {'🟢' if s.get('alive',True) else '🔴'} [{s['type']}] {s['name']}"
|
||
for i, s in enumerate(data['subs'])]
|
||
await send_del(q.message, '📋 *订阅列表*\n\n' + '\n'.join(lines), parse_mode='Markdown')
|
||
|
||
elif action == 'menu_get':
|
||
alive = [s for s in data['subs'] if s.get('alive', True)]
|
||
if not alive:
|
||
return await send_del(q.message, '📭 暂无可用订阅')
|
||
types = sorted(set(s['type'] for s in alive))
|
||
buttons = [[InlineKeyboardButton(f"📦 全部 ({len(alive)})", callback_data='get_all_raw')]]
|
||
for t in types:
|
||
cnt = sum(1 for s in alive if s['type'] == t)
|
||
buttons.append([
|
||
InlineKeyboardButton(f"🔗 {t} ({cnt})", callback_data=f'get_{t}_raw'),
|
||
InlineKeyboardButton(f"📄 {t} Base64", callback_data=f'get_{t}_b64')])
|
||
buttons.append([InlineKeyboardButton("⚡ Clash Meta 订阅", callback_data='get_all_clash')])
|
||
await send_del(q.message, '选择格式:', reply_markup=InlineKeyboardMarkup(buttons))
|
||
|
||
elif action == 'menu_del':
|
||
if not data['subs']:
|
||
return await send_del(q.message, '📭 暂无订阅')
|
||
uid = q.from_user.id
|
||
buttons = []
|
||
for i, s in enumerate(data['subs']):
|
||
if uid != ADMIN_ID and uid != s.get('added_by'): continue
|
||
st = '🟢' if s.get('alive', True) else '🔴'
|
||
buttons.append([InlineKeyboardButton(f"{st} [{s['type']}] {s['name']}", callback_data=f'del_{i}')])
|
||
if not buttons:
|
||
return await send_del(q.message, '没有可删除的订阅')
|
||
buttons.append([InlineKeyboardButton("🗑 删除所有不可用", callback_data='del_alldown')])
|
||
await send_del(q.message, '选择要删除的:', reply_markup=InlineKeyboardMarkup(buttons))
|
||
|
||
elif action == 'menu_add':
|
||
WAITING_ADD.add(q.from_user.id)
|
||
await send_del(q.message, '📎 请发送订阅链接（支持多条，空格或换行分隔）\n支持: ss/vmess/vless/trojan/hy2/tuic')
|
||
|
||
elif action == 'menu_check':
|
||
msg = await send_del(q.message, '🔍 检测中...', delay=120)
|
||
results = []
|
||
for s in data['subs']:
|
||
ok = await check_node(s['link'], s['type'])
|
||
s['alive'] = ok
|
||
results.append(f"{'🟢' if ok else '🔴'} [{s['type']}] {s['name']}")
|
||
save_data(data)
|
||
try: await msg.delete()
|
||
except: pass
|
||
await send_del(q.message, '📊 *检测结果*\n\n' + '\n'.join(results), parse_mode='Markdown')
|
||
|
||
elif action == 'menu_setgroup':
|
||
if q.from_user.id != ADMIN_ID:
|
||
return await send_del(q.message, '⛔ 仅管理员可用')
|
||
cid = q.message.chat.id
|
||
if cid not in data.get('groups', []):
|
||
data.setdefault('groups', []).append(cid)
|
||
save_data(data)
|
||
await send_del(q.message, f'✅ 已绑定群 `{cid}`', parse_mode='Markdown')
|
||
|
||
# --- get/pick/del callbacks ---
|
||
async def cb_getsub(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
q = update.callback_query
|
||
await q.answer()
|
||
data = load_data()
|
||
alive = [s for s in data['subs'] if s.get('alive', True)]
|
||
action = q.data
|
||
|
||
if action.startswith('pick_'):
|
||
idx = int(action.split('_')[1])
|
||
if idx < len(alive):
|
||
buttons = [
|
||
[InlineKeyboardButton("🔗 原始链接", callback_data=f'fmt_{idx}_raw')],
|
||
[InlineKeyboardButton("⚡ Clash Meta 订阅", callback_data=f'fmt_{idx}_clash')],
|
||
]
|
||
await send_del(q.message, f'📍 {alive[idx]["name"]}\n选择输出格式:',
|
||
reply_markup=InlineKeyboardMarkup(buttons))
|
||
return
|
||
|
||
if action.startswith('fmt_'):
|
||
parts = action.split('_')
|
||
idx, fmt = int(parts[1]), parts[2]
|
||
if idx < len(alive):
|
||
s = alive[idx]
|
||
if fmt == 'raw':
|
||
await send_del(q.message, s["link"])
|
||
else:
|
||
# single node clash meta - generate inline
|
||
cm = gen_clash_meta([s])
|
||
await send_del(q.message, f'```yaml\n{cm}\n```', parse_mode='Markdown')
|
||
return
|
||
|
||
if action.startswith('send_'):
|
||
parts = action.split('_')
|
||
proto, fmt = parts[1], parts[2]
|
||
subs = alive if proto == 'all' else [s for s in alive if s['type'] == proto]
|
||
# ask format first
|
||
if fmt == 'raw':
|
||
buttons = [
|
||
[InlineKeyboardButton("🔗 原始链接", callback_data=f'out_{proto}_raw')],
|
||
[InlineKeyboardButton("⚡ Clash Meta", callback_data=f'out_{proto}_clash')],
|
||
[InlineKeyboardButton("📄 Base64", callback_data=f'out_{proto}_b64')],
|
||
]
|
||
return await send_del(q.message, f'📦 全部 ({len(subs)}) — 选择格式:',
|
||
reply_markup=InlineKeyboardMarkup(buttons))
|
||
|
||
if action.startswith('out_'):
|
||
parts = action.split('_')
|
||
proto, fmt = parts[1], parts[2]
|
||
subs = alive if proto == 'all' else [s for s in alive if s['type'] == proto]
|
||
if fmt == 'clash':
|
||
url = f'http://{SUB_HOST}/{SUB_SECRET}/download?target=ClashMeta'
|
||
if proto != 'all': url += f'&type={proto}'
|
||
return await send_del(q.message, f'📎 Clash Meta 订阅链接:\n{url}')
|
||
links = '\n'.join(s['link'] for s in subs)
|
||
if fmt == 'b64':
|
||
return await send_del(q.message, f'```\n{base64.b64encode(links.encode()).decode()}\n```', parse_mode='Markdown')
|
||
if len(links) > 4000:
|
||
for s in subs:
|
||
await send_del(q.message, s['link'])
|
||
else:
|
||
await send_del(q.message, links)
|
||
return
|
||
|
||
if action == 'get_all_clash':
|
||
url = f'http://{SUB_HOST}/{SUB_SECRET}/download?target=ClashMeta'
|
||
await send_del(q.message, f'📎 Clash Meta 订阅链接:\n{url}')
|
||
return
|
||
|
||
parts = action.split('_')
|
||
proto, fmt = parts[1], parts[2]
|
||
subs = alive if proto == 'all' else [s for s in alive if s['type'] == proto]
|
||
if not subs:
|
||
return await send_del(q.message, '📭 无匹配')
|
||
|
||
if fmt == 'raw' and len(subs) > 1:
|
||
buttons = []
|
||
for i, s in enumerate(subs):
|
||
gi = alive.index(s)
|
||
buttons.append([InlineKeyboardButton(
|
||
f"{'🟢' if s.get('alive',True) else '🔴'} {s['name']}", callback_data=f'pick_{gi}')])
|
||
buttons.append([InlineKeyboardButton(f"📦 全部 ({len(subs)})", callback_data=f'send_{proto}_raw')])
|
||
return await send_del(q.message, '选择节点:', reply_markup=InlineKeyboardMarkup(buttons))
|
||
|
||
if fmt == 'raw' and len(subs) == 1:
|
||
buttons = [
|
||
[InlineKeyboardButton("🔗 原始链接", callback_data=f'fmt_{alive.index(subs[0])}_raw')],
|
||
[InlineKeyboardButton("⚡ Clash Meta 订阅", callback_data=f'fmt_{alive.index(subs[0])}_clash')],
|
||
]
|
||
return await send_del(q.message, f'📍 {subs[0]["name"]}\n选择输出格式:',
|
||
reply_markup=InlineKeyboardMarkup(buttons))
|
||
|
||
links = '\n'.join(s['link'] for s in subs)
|
||
if fmt == 'b64':
|
||
await send_del(q.message, base64.b64encode(links.encode()).decode())
|
||
else:
|
||
await send_del(q.message, links)
|
||
|
||
async def cb_delsub(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
q = update.callback_query
|
||
await q.answer()
|
||
data = load_data()
|
||
action = q.data
|
||
uid = q.from_user.id
|
||
if action == 'del_alldown':
|
||
dead = [s for s in data['subs'] if not s.get('alive', True)]
|
||
if uid != ADMIN_ID:
|
||
dead = [s for s in dead if s.get('added_by') == uid]
|
||
if not dead:
|
||
return await send_del(q.message, '没有不可用节点')
|
||
names = [s['name'] for s in dead]
|
||
for s in dead: data['subs'].remove(s)
|
||
save_data(data)
|
||
return await send_del(q.message, f"🗑 已删除 {len(names)} 个:\n" + '\n'.join(f'• {n}' for n in names))
|
||
idx = int(action.split('_')[1])
|
||
if idx >= len(data['subs']):
|
||
return await send_del(q.message, '❌ 已失效')
|
||
s = data['subs'][idx]
|
||
if uid != ADMIN_ID and uid != s.get('added_by'):
|
||
return await send_del(q.message, '⛔ 只能删除自己上传的')
|
||
removed = data['subs'].pop(idx)
|
||
save_data(data)
|
||
await send_del(q.message, f"🗑 已删除: {removed['name']}")
|
||
|
||
# --- auto detect links in messages ---
|
||
async def auto_detect(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
if not update.message or not update.message.text: return
|
||
text = update.message.text.strip()
|
||
uid = update.effective_user.id
|
||
# check standard protocol links
|
||
found = [w for w in text.split() if any(w.startswith(p) for p in PROTOS)]
|
||
# check Surge format (line-based)
|
||
surge_links = []
|
||
if not found:
|
||
for line in text.split('\n'):
|
||
ss_link, name = parse_surge_ss(line)
|
||
if ss_link: surge_links.append(ss_link)
|
||
if not found and not surge_links:
|
||
if uid in WAITING_ADD:
|
||
WAITING_ADD.discard(uid)
|
||
await send_del(update.message, '❌ 未识别到订阅链接')
|
||
return
|
||
if not await is_member(update, context): return
|
||
WAITING_ADD.discard(uid)
|
||
results = []
|
||
for link in found + surge_links:
|
||
r = add_sub(link, update.effective_user)
|
||
if r and r != 'dup': results.append(r)
|
||
elif r == 'dup': results.append(f'⚠️ 已存在: {extract_name(link)}')
|
||
if results:
|
||
await send_del(update.message, '\n'.join(results))
|
||
|
||
# --- node health check ---
|
||
async def check_node(link, proto):
|
||
try:
|
||
server, port = parse_sp(link, proto)
|
||
if not server: return False
|
||
_, w = await asyncio.wait_for(asyncio.open_connection(server, int(port)), timeout=5)
|
||
w.close(); await w.wait_closed()
|
||
return True
|
||
except: return False
|
||
|
||
def parse_sp(link, proto):
|
||
try:
|
||
if proto == 'ss':
|
||
raw = link[5:]
|
||
if '#' in raw: raw = raw.split('#')[0]
|
||
decoded = base64.b64decode(raw + '==').decode()
|
||
_, rest = decoded.split(':', 1)
|
||
_, sp = rest.rsplit('@', 1)
|
||
return sp.rsplit(':', 1)
|
||
m = re.search(r'@([^:/?#]+):(\d+)', link)
|
||
if m: return m.group(1), m.group(2)
|
||
except: pass
|
||
return None, None
|
||
|
||
def parse_to_clash_proxy(s):
|
||
try:
|
||
if s['type'] == 'ss':
|
||
raw = s['link'][5:]
|
||
name = ''
|
||
if '#' in raw: raw, name = raw.rsplit('#', 1); name = urllib.request.unquote(name)
|
||
decoded = base64.b64decode(raw + '==').decode()
|
||
method, rest = decoded.split(':', 1)
|
||
pwd, sp = rest.rsplit('@', 1)
|
||
server, port = sp.rsplit(':', 1)
|
||
return {'name': name or server, 'type': 'ss', 'server': server, 'port': int(port),
|
||
'cipher': method, 'password': pwd}
|
||
except: pass
|
||
return None
|
||
|
||
def gen_clash_meta(subs):
|
||
lines = ['proxies:']
|
||
for s in subs:
|
||
p = parse_to_clash_proxy(s)
|
||
if p: lines.append(f' - {json.dumps(p, ensure_ascii=False)}')
|
||
return '\n'.join(lines)
|
||
|
||
# --- HTTP subscription endpoint ---
|
||
class SubHandler(BaseHTTPRequestHandler):
|
||
def do_GET(self):
|
||
path = self.path.split('?')[0]
|
||
if path != f'/{SUB_SECRET}/download':
|
||
self.send_response(404)
|
||
self.end_headers()
|
||
return
|
||
data = load_data()
|
||
alive = [s for s in data['subs'] if s.get('alive', True)]
|
||
qs = {}
|
||
if '?' in self.path:
|
||
qs = dict(x.split('=',1) for x in self.path.split('?',1)[1].split('&') if '=' in x)
|
||
ftype = qs.get('type', '')
|
||
if ftype: alive = [s for s in alive if s['type'] == ftype]
|
||
target = qs.get('target', 'ClashMeta')
|
||
body = gen_clash_meta(alive) if target == 'ClashMeta' else '\n'.join(s['link'] for s in alive)
|
||
self.send_response(200)
|
||
self.send_header('Content-Type', 'text/plain; charset=utf-8')
|
||
self.end_headers()
|
||
self.wfile.write(body.encode())
|
||
def log_message(self, *a): pass
|
||
|
||
def start_http():
|
||
srv = HTTPServer(('0.0.0.0', 18888), SubHandler)
|
||
log.info('Sub HTTP on :18888')
|
||
srv.serve_forever()
|
||
|
||
# --- legacy commands still work ---
|
||
async def cmd_setgroup(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
if update.effective_user.id != ADMIN_ID:
|
||
return await update.message.reply_text('⛔ 仅管理员可用')
|
||
cid = update.effective_chat.id
|
||
data = load_data()
|
||
if cid not in data.get('groups', []):
|
||
data.setdefault('groups', []).append(cid)
|
||
save_data(data)
|
||
await send_del(update.message, f'✅ 已绑定群 `{cid}`', parse_mode='Markdown')
|
||
|
||
async def auto_cleanup(bot_token):
|
||
"""Every 6h: check nodes, auto-delete dead ones, notify group"""
|
||
import httpx
|
||
await asyncio.sleep(60)
|
||
while True:
|
||
data = load_data()
|
||
if data['subs']:
|
||
dead = []
|
||
for s in list(data['subs']):
|
||
ok = await check_node(s['link'], s['type'])
|
||
if not ok:
|
||
dead.append(s)
|
||
data['subs'].remove(s)
|
||
if dead:
|
||
save_data(data)
|
||
names = '\n'.join(f"• [{s['type']}] {s['name']}" for s in dead)
|
||
msg = f'🗑 自动清理 — 已删除 {len(dead)} 个不可用节点:\n\n{names}'
|
||
for gid in data.get('groups', []):
|
||
try:
|
||
async with httpx.AsyncClient() as c:
|
||
await c.post(f'https://api.telegram.org/bot{bot_token}/sendMessage',
|
||
json={'chat_id': gid, 'text': msg})
|
||
except: pass
|
||
log.info(f'Auto cleanup: removed {len(dead)} dead nodes')
|
||
await asyncio.sleep(21600)
|
||
|
||
def run_cleanup():
|
||
asyncio.run(auto_cleanup(TOKEN))
|
||
|
||
def main():
|
||
app = Application.builder().token(TOKEN).build()
|
||
app.add_handler(CommandHandler('start', cmd_help))
|
||
app.add_handler(CommandHandler('vps', cmd_help))
|
||
app.add_handler(CommandHandler('setgroup', cmd_setgroup))
|
||
app.add_handler(CallbackQueryHandler(cb_menu, pattern='^menu_'))
|
||
app.add_handler(CallbackQueryHandler(cb_delsub, pattern='^del_'))
|
||
app.add_handler(CallbackQueryHandler(cb_getsub, pattern='^(get_|pick_|send_|fmt_|out_)'))
|
||
app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, auto_detect))
|
||
log.info('Sub Bot starting polling...')
|
||
threading.Thread(target=start_http, daemon=True).start()
|
||
threading.Thread(target=run_cleanup, daemon=True).start()
|
||
app.run_polling(drop_pending_updates=True)
|
||
|
||
if __name__ == '__main__':
|
||
main()
|