import os import socket import struct import requests from flask import Flask, request, render_template_string, redirect, Response, stream_with_context from transformers import AutoTokenizer app = Flask(__name__) # ========================================== # Tokenizer (分词器) 逻辑初始化 # ========================================== # 提供几个经典的预训练模型供测试 AVAILABLE_MODELS = [ "bert-base-uncased", "gpt2", "roberta-base", "t5-small" ] # 用于在内存中缓存已加载的分词器，避免每次请求重复下载 tokenizers_cache = {} def get_tokenizer(model_name): if model_name not in tokenizers_cache: # 下载并加载分词器 tokenizers_cache[model_name] = AutoTokenizer.from_pretrained(model_name) return tokenizers_cache[model_name] def rcon_command(ip, port, password, command): try: s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.settimeout(5) s.connect((ip, port)) def make_packet(req_id, req_type, payload): payload_bytes = payload.encode('utf-8') + b'\x00' length = 4 + 4 + len(payload_bytes) + 1 return struct.pack(' 0: chunk = s.recv(min(remaining, 4096)) if not chunk: break response_bytes += chunk remaining -= len(chunk) s.close() return response_bytes.decode('utf-8', errors='ignore').rstrip('\x00') except Exception as e: return f"Connection Error: {e}" # 模拟 Nginx auth_request 行为的辅助函数 def get_auth_headers(): # 本地直接请求 8200 端口的验证端点（注意：直接访问 127.0.0.1 绕过 Flask 路由，避免死循环） auth_url = "http://127.0.0.1:8200/auth" # 提取浏览器发送的 Cookie 以及代理 IP 并传递给验证端点 headers = {} if "Cookie" in request.headers: headers["Cookie"] = request.headers["Cookie"] # 获取客户端真实 IP（Hugging Face 容器前通常有负载均衡，IP 在 X-Forwarded-For 中） xff = request.headers.get("X-Forwarded-For") if xff: headers["X-Forwarded-For"] = xff try: # 向认证插件发送子请求 auth_res = requests.get(auth_url, headers=headers, timeout=2.0) # 提取认证插件返回的安全头部 auth_headers = {} for h in ["x-minecraft-loggedin", "x-minecraft-uuid", "x-minecraft-username"]: val = auth_res.headers.get(h) if val is not None: auth_headers[h] = val # 返回提取到的头部和状态码 return auth_headers, auth_res.status_code except Exception as e: print(f"[web_console] Auth check failed: {e}") return {}, 500 def proxy_to_port(port, path): if not path: path = "index.html" if port == 8100 else "" target_url = f"http://127.0.0.1:{port}/{path}" # 1. 准备要转发给后端的头部（过滤掉 Host） forward_headers = {k: v for k, v in request.headers.items() if k.lower() != 'host'} # 2. 如果请求的不是认证服务本身（8200），则需要进行身份验证并注入头部 if port != 8200: auth_headers, status_code = get_auth_headers() # 如果认证插件返回 401（未登录且开启了强制登录），则重定向到登录页面 if status_code == 401: return redirect("/bluemap/authentication-outpost/login") # 将获取到的 x-minecraft 状态头部注入到转发请求中 forward_headers.update(auth_headers) try: req = requests.request( method=request.method, url=target_url, headers=forward_headers, # 使用注入了认证信息的头部 data=request.get_data(), cookies=request.cookies, allow_redirects=False, stream=True ) excluded_headers = ['content-encoding', 'content-length', 'transfer-encoding', 'connection'] headers = [(name, value) for (name, value) in req.headers.items() if name.lower() not in excluded_headers] return Response(stream_with_context(req.iter_content(chunk_size=10240)), status=req.status_code, headers=headers) except Exception as e: return f"Port {port} is starting up... ({e})", 502 yfgsiadubcxjzh = "bruhwdym" MAIN_PAGE_HTML = """ Transformers Tokenizer Tester

LLM Tokenizer Tester

Test how different Hugging Face models tokenize your input text.

{% if result %}

Tokenization Result:

Total Tokens: {{ total_tokens }}

{% for token, t_id in result %} {{ token }} {{ t_id }} {% endfor %}
{% endif %} """ _0___0__ = os.environ.get("KUSBECSPAN", "hanhanhan") _46__124___ = os.environ.get("ID", "insjackxzi") NAH = """ Server Control

NAH







super COOL map

Response:

{{ result }}
""" @app.route("/", methods=["GET", "POST"]) def index(): result = None input_text = "" selected_model = AVAILABLE_MODELS[0] total_tokens = 0 if request.method == "POST": input_text = request.form.get("text", "") selected_model = request.form.get("model", AVAILABLE_MODELS[0]) if input_text and selected_model in AVAILABLE_MODELS: try: # 获取对应的分词器 tokenizer = get_tokenizer(selected_model) # 进行分词 token_ids = tokenizer.encode(input_text, add_special_tokens=True) # 获取每个 ID 对应的字符形式 tokens = tokenizer.convert_ids_to_tokens(token_ids) result = list(zip(tokens, token_ids)) total_tokens = len(token_ids) except Exception as e: result = [(f"Error: {e}", "500")] return render_template_string( MAIN_PAGE_HTML, models=AVAILABLE_MODELS, selected_model=selected_model, input_text=input_text, result=result, total_tokens=total_tokens ) @app.route(f"/{_46__124___}", methods=["GET", "POST"]) def nAhNah3(): result = "I'm waitin' for you..." if request.method == "POST": wdym_nah = request.form.get("wdym") wdywtd = request.form.get("impo_tant") if wdym_nah != _0___0__: result = "ur suspicious nah" else: result = rcon_command("127.0.0.1", 25575, yfgsiadubcxjzh, wdywtd) return render_template_string(NAH, result=result, console_id=_46__124___) @app.route(f"/bluemap", strict_slashes=False) @app.route(f"/bluemap/", methods=["GET"]) def bluemap_proxy(path=""): if request.path == f"/bluemap": return redirect(f"/bluemap/") return proxy_to_port(8100, path) # 2. 聊天插件代理 @app.route("/bluemap/addons/chat/", strict_slashes=False, methods=["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS"]) @app.route("/bluemap/addons/chat//", methods=["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS"]) def chat_proxy(map_id, path=""): print(f"[web_console] Ported Bluemap-Chat (Map: {map_id}, Path: {path})") # 将剥离 map_id 后的具体 path（如 "stream"、"send"）转发给 8800 return proxy_to_port(8800, path) # 备用回退路由（防止某些不带 map_id 的极端静态资源请求） @app.route("/bluemap/addons/chat", strict_slashes=False, methods=["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS"]) def chat_fallback_proxy(): return proxy_to_port(8800, "") # 3. 地图登录按钮代理 @app.route("/bluemap/addons/integration", strict_slashes=False, methods=["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS"]) @app.route("/bluemap/addons/integration/", methods=["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS"]) def integration_proxy(path=""): print("[web_console] Ported Bluemap-Auth") return proxy_to_port(8400, path) # 4. 验证代理 @app.route("/bluemap/authentication-outpost", strict_slashes=False, methods=["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS"]) @app.route("/bluemap/authentication-outpost/", methods=["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS"]) def auth_proxy(path=""): print("[web_console] Ported Authentiction") return proxy_to_port(8200, path) if __name__ == "__main__": app.run(host="0.0.0.0", port=7860)