上传文件至「/」

login.py

@@ -0,0 +1,704 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+import requests
+import time
+import urllib.parse
+import sys
+import io
+import uuid
+import subprocess
+import concurrent.futures
+import re
+import threading
+import json
+import os
+from io import BytesIO
+
+if sys.platform == 'win32':
+    try:
+        sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', line_buffering=True)
+    except Exception:
+        pass
+
+if getattr(sys, 'frozen', False):
+    BASE_DIR = os.path.dirname(sys.executable)
+else:
+    BASE_DIR = os.path.dirname(os.path.abspath(__file__))
+CONFIG_FILE = os.path.join(BASE_DIR, "config.json")
+STATE_FILE = os.path.join(BASE_DIR, ".eportal_state.json")
+
+# 模块级默认值，main() 会从 config.json 覆盖
+PORTAL_HOST = "http://192.168.200.100"
+PORTAL_BASE_URL = PORTAL_HOST + "/eportal"
+USERNAME = ""
+PASSWORD = ""
+MAX_ATTEMPTS = 60
+SLEEP_SECONDS = 3
+PING_HOST = "www.dufe.edu.cn"
+PUBLIC_KEY_FALLBACK = "10001&94dd2a8675fb779e6b9f7103698634cd400f27a154afa67af6166a43fc26417222a79506d34cacc7641946abda1785b7acf9910ad6a0978c91ec84d40b71d2891379af19ffb333e7517e390bd26ac312fe940c340466b4a5d4af1d65c3b5944078f96a1a51a5a53e4bc302818b7c9f63c4a1b07bd7d874cef1c3d4b2f5eb7871"
+
+
+DEFAULT_CONFIG = {
+    "username": "",
+    "password": "",
+    "portal_host": "http://192.168.200.100",
+    "max_attempts": 60,
+    "sleep_seconds": 3,
+    "ping_host": "www.dufe.edu.cn",
+    "public_key_fallback": "10001&94dd2a8675fb779e6b9f7103698634cd400f27a154afa67af6166a43fc26417222a79506d34cacc7641946abda1785b7acf9910ad6a0978c91ec84d40b71d2891379af19ffb333e7517e390bd26ac312fe940c340466b4a5d4af1d65c3b5944078f96a1a51a5a53e4bc302818b7c9f63c4a1b07bd7d874cef1c3d4b2f5eb7871",
+}
+
+
+def load_config():
+    def _write_default():
+        try:
+            with open(CONFIG_FILE, "w", encoding="utf-8") as f:
+                json.dump(DEFAULT_CONFIG, f, ensure_ascii=False, indent=4)
+            _print(f"已创建默认配置文件: {CONFIG_FILE}")
+        except Exception:
+            pass
+
+    if not os.path.exists(CONFIG_FILE):
+        _write_default()
+        return dict(DEFAULT_CONFIG)
+
+    # 检测空文件或内容为空
+    try:
+        if os.path.getsize(CONFIG_FILE) == 0:
+            _print("config.json 为空，重新生成默认配置", "ERROR")
+            _write_default()
+            return dict(DEFAULT_CONFIG)
+    except Exception:
+        pass
+
+    cfg = dict(DEFAULT_CONFIG)
+    try:
+        with open(CONFIG_FILE, "r", encoding="utf-8") as f:
+            data = json.load(f)
+        if not isinstance(data, dict):
+            _print("config.json 格式错误，使用默认配置", "ERROR")
+            return dict(DEFAULT_CONFIG)
+        cfg.update(data)
+    except json.JSONDecodeError as e:
+        _print(f"config.json 解析失败: {e}，使用默认配置", "ERROR")
+    except Exception as e:
+        _print(f"读取 config.json 失败: {e}", "ERROR")
+    return cfg
+
+
+def _print(msg, level="INFO"):
+    print(f"[{level}] {msg}", flush=True)
+
+
+_skip_wait = False  # Ctrl+C 退出时跳过等待
+
+
+def _wait_exit():
+    global _skip_wait
+    if _skip_wait:
+        return
+    try:
+        input("按任意键退出...")
+    except (EOFError, KeyboardInterrupt):
+        pass
+
+
+def save_state(mac, user_index):
+    try:
+        with open(STATE_FILE, "w") as f:
+            json.dump({"mac": mac, "userIndex": user_index}, f)
+    except Exception:
+        pass
+
+
+def load_state():
+    try:
+        if os.path.exists(STATE_FILE):
+            with open(STATE_FILE, "r") as f:
+                return json.load(f)
+    except Exception:
+        pass
+    return None
+
+
+def detect_portal():
+    try:
+        requests.get(PORTAL_HOST + "/", timeout=(2, 2), allow_redirects=False)
+        return None, None
+    except Exception:
+        pass
+
+    for url in [
+        "http://www.baidu.com",
+        "http://detectportal.firefox.com/canonical.html",
+        "http://www.msftconnecttest.com/redirect",
+    ]:
+        try:
+            resp = requests.get(url, timeout=(2, 2), allow_redirects=False)
+            if resp.status_code in (301, 302, 303, 307, 308):
+                loc = resp.headers.get("Location", "")
+                m = re.match(r'(https?://[^/]+(?:/eportal)?)', loc)
+                if m:
+                    host = re.match(r'https?://[^/]+', m.group(1)).group(0)
+                    base = m.group(1)
+                    return host, base
+        except Exception:
+            pass
+    return None, None
+
+
+_page_info_cache = None
+
+
+def _fetch_page_info():
+    global _page_info_cache
+    if _page_info_cache:
+        return _page_info_cache
+
+    info = {"public_key": None, "query_string": "", "portal_mac": None}
+
+    def parse_page(resp, extract_qs=True):
+        html = resp.text
+        if "?" in resp.url:
+            info["query_string"] = resp.url.split("?", 1)[1]
+        m = re.search(r'[?&]mac=([^&]+)', resp.url)
+        if m:
+            info["portal_mac"] = urllib.parse.unquote(m.group(1)).replace(':', '').replace('-', '').lower()
+        # 从 HTML 提取 publicKey
+        for pid in ['publicKey', 'publicKeyModulus', 'publicKeyExponent']:
+            m = re.search(rf'(?:id|name)=["\']{pid}["\'][^>]*value=["\']([^"\']+)', html)
+            if m:
+                if pid == 'publicKey' and '&' in m.group(1):
+                    info["public_key"] = m.group(1)
+                    break
+        if not info["public_key"]:
+            exp_match = re.search(r'(?:id|name)=["\']publicKeyExponent["\'][^>]*value=["\']([^"\']+)', html)
+            mod_match = re.search(r'(?:id|name)=["\']publicKeyModulus["\'][^>]*value=["\']([^"\']+)', html)
+            if exp_match and mod_match:
+                info["public_key"] = exp_match.group(1) + "&" + mod_match.group(1)
+        if not extract_qs:
+            return
+        # 从 HTML 提取 queryString（JS 重定向: location.href='...?wlanuserip=...'）
+        if not info["query_string"]:
+            m = re.search(r"location\.href\s*=\s*['\"]([^'\"]*\?[^'\"]+)", html)
+            if m:
+                qs_url = m.group(1)
+                if "?" in qs_url:
+                    info["query_string"] = qs_url.split("?", 1)[1]
+        if not info["query_string"]:
+            params = []
+            for key in ['wlanuserip', 'wlanacname', 'wlanacip', 'wlanusermac', 'wlanuserfirsturl',
+                        'ssid', 'nasip', 'snmpagentip', 'mac', 't', 'url']:
+                m = re.search(rf'(?:id|name)=["\']{key}["\'][^>]*value=["\']([^"\']+)', html)
+                if m:
+                    params.append(f"{key}={m.group(1)}")
+            if params:
+                info["query_string"] = "&".join(params)
+
+    # 方法1: 跟随 portal 根路径重定向链，提取 JS 重定向中的 queryString
+    try:
+        resp = requests.get(PORTAL_HOST + "/", timeout=10, allow_redirects=True)
+        parse_page(resp)
+    except Exception:
+        pass
+
+    # 方法2: 访问 eportal 页面获取 publicKey（同时尝试提取 queryString）
+    try:
+        resp = requests.get(PORTAL_BASE_URL + "/", timeout=10, allow_redirects=True)
+        parse_page(resp, extract_qs=not bool(info["query_string"]))
+    except Exception:
+        pass
+
+    # 方法3: 对 HTTP 外部 URL 发起请求触发 DNS 劫持，从返回的 HTML 提取 queryString
+    if not info["query_string"]:
+        for test_url in ["http://www.baidu.com", "http://detectportal.firefox.com/canonical.html"]:
+            try:
+                resp = requests.get(test_url, timeout=5, allow_redirects=False)
+                if resp.status_code in (301, 302, 303, 307, 308):
+                    loc = resp.headers.get("Location", "")
+                    if "?" in loc:
+                        info["query_string"] = loc.split("?", 1)[1]
+                    m = re.search(r'[?&]mac=([^&]+)', loc)
+                    if m:
+                        info["portal_mac"] = urllib.parse.unquote(m.group(1)).replace(':', '').replace('-', '').lower()
+                    break
+                elif resp.status_code == 200:
+                    parse_page(resp)
+                    if info["query_string"]:
+                        break
+            except Exception:
+                pass
+
+    _page_info_cache = info
+    return info
+
+
+def get_public_key():
+    info = _fetch_page_info()
+    if info["public_key"]:
+        return info["public_key"]
+    return PUBLIC_KEY_FALLBACK
+
+
+def get_query_string():
+    return _fetch_page_info()["query_string"]
+
+
+def get_portal_mac():
+    return _fetch_page_info()["portal_mac"]
+
+
+def js_rsa_encrypt(data_str, modulus_hex, exponent_hex):
+    # 模拟 JS RSAUtils.encryptedString：原始RSA，零填充，每2个charCode打包为1个16bit digit
+    modulus = int(modulus_hex, 16)
+    exponent = int(exponent_hex, 16)
+
+    modulus_bytes = (modulus.bit_length() + 7) // 8
+    js_digits = (modulus_bytes + 1) // 2
+    chunk_size = 2 * (js_digits - 1)
+
+    a = [ord(ch) for ch in data_str]
+    while len(a) % chunk_size != 0:
+        a.append(0)
+
+    parts = []
+    for i in range(0, len(a), chunk_size):
+        val = 0
+        j = 0
+        for k in range(i, i + chunk_size, 2):
+            val += (a[k] + (a[k + 1] << 8)) << (16 * j)
+            j += 1
+        parts.append(hex(pow(val, exponent, modulus))[2:])
+
+    return " ".join(parts)
+
+
+def double_encode(s):
+    return urllib.parse.quote(urllib.parse.quote(s, safe=''), safe='')
+
+
+def _ocr_single(img_bytes):
+    try:
+        from PIL import Image
+        import pytesseract
+        img = Image.open(BytesIO(img_bytes))
+        gray = img.convert("L")
+        binary = gray.point(lambda x: 0 if x < 200 else 255)
+        return pytesseract.image_to_string(
+            binary,
+            config="--psm 7 -c tessedit_char_whitelist=0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ"
+        ).strip()
+    except (ImportError, Exception):
+        return ""
+
+
+def _fetch_captcha():
+    import random
+    url = f"{PORTAL_BASE_URL}/validcode?rnd={random.random()}"
+    resp = requests.get(url, timeout=10)
+    return resp.content, url
+
+
+def recognize_captcha(first_url):
+    try:
+        if not first_url.startswith("http"):
+            first_url = urllib.parse.urljoin(PORTAL_BASE_URL, first_url)
+        r = requests.get(first_url, timeout=10)
+        text = _ocr_single(r.content)
+        if text:
+            _print(f"验证码 OCR 识别: {text}")
+            return text
+    except Exception as e:
+        _print(f"验证码下载失败: {e}", "ERROR")
+
+    for i in range(2, 4):
+        _print(f"OCR未识别，刷新验证码重试({i}/3)...")
+        try:
+            img_bytes, _ = _fetch_captcha()
+            text = _ocr_single(img_bytes)
+            if text:
+                _print(f"OCR第{i}次: {text}")
+                return text
+        except Exception as e:
+            _print(f"验证码刷新失败: {e}", "ERROR")
+
+    _print("OCR 3次均失败，切换为手动输入...", "ERROR")
+    try:
+        from PIL import Image
+        img_bytes, _ = _fetch_captcha()
+        Image.open(BytesIO(img_bytes)).show()
+    except Exception:
+        pass
+    return input("      请手动输入验证码: ").strip()
+
+
+def get_all_macs():
+    macs = []
+    mac_raw = set()
+
+    def add_mac(m):
+        if not m:
+            return
+        m = m.replace(':', '').replace('-', '').replace('.', '').lower().strip()
+        if len(m) == 12 and all(c in '0123456789abcdef' for c in m):
+            if m not in mac_raw and m != '000000000000':
+                mac_raw.add(m)
+                macs.append(m)
+
+    # ipconfig /all（Windows 最可靠）
+    try:
+        output = subprocess.run(
+            ['ipconfig', '/all'],
+            capture_output=True, text=True, timeout=5
+        ).stdout
+        for line in output.split('\n'):
+            if '物理地址' in line or 'Physical' in line:
+                parts = line.split(':')
+                if len(parts) >= 2:
+                    add_mac(parts[-1])
+    except Exception:
+        pass
+
+    # getmac 库
+    try:
+        from getmac import get_mac_address
+        m = get_mac_address()
+        if m:
+            add_mac(m)
+    except Exception:
+        pass
+
+    # uuid（不可靠，放最后）
+    try:
+        m = format(uuid.getnode(), '012x')
+        if m != '000000000000':
+            add_mac(m)
+    except Exception:
+        pass
+
+    # PowerShell 备用
+    if not macs:
+        try:
+            output = subprocess.run(
+                ['powershell', '-Command',
+                 'Get-NetAdapter | Select-Object -ExpandProperty MacAddress'],
+                capture_output=True, text=True, timeout=5
+            ).stdout
+            for line in output.strip().split('\n'):
+                add_mac(line)
+        except Exception:
+            pass
+
+    return macs
+
+
+def is_authenticated():
+    test_urls = [
+        ("https://detectportal.firefox.com/canonical.html", "success"),
+        ("https://www.msftconnecttest.com/connecttest.txt", "Microsoft"),
+        ("https://www.baidu.com", None),
+    ]
+    passed = 0
+    for url, keyword in test_urls:
+        try:
+            r = requests.get(url, allow_redirects=False, timeout=3)
+            if r.status_code == 200:
+                if keyword:
+                    if keyword in r.text:
+                        passed += 1
+                else:
+                    passed += 1
+        except Exception:
+            continue
+    return passed >= 2
+
+
+def do_login(mac, query_string=""):
+    exponent_hex, modulus_hex = get_public_key().split('&')
+    plain = f"{PASSWORD}>{mac}"
+    encrypted = js_rsa_encrypt(plain[::-1], modulus_hex, exponent_hex)
+
+    def build_body(validcode=""):
+        return "&".join([
+            f"userId={double_encode(USERNAME)}",
+            f"password={double_encode(encrypted)}",
+            f"service={double_encode('')}",
+            f"queryString={double_encode(query_string)}",
+            "operatorPwd=",
+            "operatorUserId=",
+            f"validcode={double_encode(validcode)}",
+            f"passwordEncrypt={double_encode('true')}",
+        ])
+
+    def parse_result(result):
+        ok = result.get("result") == "success"
+        ui = result.get("userIndex", "")
+        ki = result.get("keepaliveInterval", 0)
+        return ok, result.get("message", ""), ui, ki
+
+    try:
+        resp = requests.post(
+            f"{PORTAL_BASE_URL}/InterFace.do?method=login",
+            data=build_body(),
+            headers={"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8"},
+            timeout=10,
+        )
+        resp.encoding = 'utf-8'
+        result = resp.json()
+
+        if result.get("result") == "fail" and result.get("validCodeUrl"):
+            captcha_url = result["validCodeUrl"]
+            if captcha_url and captcha_url.strip():
+                _print(f"需要验证码 ({captcha_url})")
+                code = recognize_captcha(captcha_url)
+                if code:
+                    _print(f"输入验证码: {code}，重试...")
+                    resp2 = requests.post(
+                        f"{PORTAL_BASE_URL}/InterFace.do?method=login",
+                        data=build_body(code),
+                        headers={"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8"},
+                        timeout=10,
+                    )
+                    resp2.encoding = 'utf-8'
+                    result = resp2.json()
+
+        return parse_result(result)
+    except Exception as e:
+        return False, str(e), "", 0
+
+
+def ping_test(host=None, count=3):
+    if host is None:
+        host = PING_HOST
+    try:
+        cmd = ["ping", "-n", str(count), host] if sys.platform == "win32" else ["ping", "-c", str(count), host]
+        output = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
+        if output.returncode == 0:
+            # 提取平均延迟
+            m = re.search(r'平均\s*=\s*(\d+)\s*ms|avg.*?(\d+\.?\d*).*?ms', output.stdout)
+            if m:
+                avg = m.group(1) or m.group(2)
+                _print(f"Ping {host} 平均延迟: {avg}ms")
+            else:
+                _print(f"Ping {host} 成功（延迟未解析）")
+        else:
+            _print(f"Ping {host} 失败: {output.stderr.strip()}", "ERROR")
+    except Exception as e:
+        _print(f"Ping {host} 异常: {e}", "ERROR")
+
+
+def register_mac(mac, user_index):
+    try:
+        resp = requests.post(
+            f"{PORTAL_BASE_URL}/InterFace.do?method=registerMac",
+            data=f"mac={mac}&userIndex={user_index}",
+            headers={"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8"},
+            timeout=10,
+        )
+        resp.encoding = 'utf-8'
+        result = resp.json()
+        if result.get("result") == "success":
+            _print(f"MAC快速认证已注册: {mac}")
+        else:
+            _print(f"MAC注册: {result.get('message', result)}", "ERROR")
+    except Exception as e:
+        _print(f"MAC注册异常: {e}", "ERROR")
+
+
+def start_keepalive(user_index, interval_seconds):
+    interval = max(60, min(600, int(interval_seconds) if interval_seconds else 180))
+
+    def _loop():
+        count = 0
+        while True:
+            time.sleep(interval)
+            count += 1
+            try:
+                resp = requests.post(
+                    f"{PORTAL_BASE_URL}/InterFace.do?method=keepalive",
+                    data=f"userIndex={user_index}",
+                    headers={"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8"},
+                    timeout=10,
+                )
+                resp.encoding = 'utf-8'
+                result = resp.json()
+                _print(f"保活 #{count} → {result.get('result', '?')}")
+            except Exception as e:
+                _print(f"保活 #{count} 异常: {e}", "ERROR")
+
+    t = threading.Thread(target=_loop, daemon=True)
+    t.start()
+    _print(f"保活线程已启动（间隔 {interval}s）")
+
+
+def main():
+    global PORTAL_HOST, PORTAL_BASE_URL, _page_info_cache
+    global USERNAME, PASSWORD, MAX_ATTEMPTS, SLEEP_SECONDS
+    global PING_HOST, PUBLIC_KEY_FALLBACK
+
+    cfg = load_config()
+    if not cfg["username"] or not cfg["password"]:
+        _print("请编辑 config.json 填入账号密码后重新运行", "ERROR")
+        return False
+
+    PORTAL_HOST = cfg["portal_host"]
+    PORTAL_BASE_URL = PORTAL_HOST + "/eportal"
+    USERNAME = cfg["username"]
+    PASSWORD = cfg["password"]
+    MAX_ATTEMPTS = int(cfg["max_attempts"])
+    SLEEP_SECONDS = int(cfg["sleep_seconds"])
+    PING_HOST = cfg["ping_host"]
+    PUBLIC_KEY_FALLBACK = cfg["public_key_fallback"]
+
+    _print(f"校园网认证登录")
+    _print(f"目标: {PORTAL_HOST}    用户: {USERNAME}")
+
+    print("[INFO] 自动检测 Portal 地址...", end=" ", flush=True)
+    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
+        future = executor.submit(detect_portal)
+        try:
+            host, base = future.result(timeout=5)
+        except concurrent.futures.TimeoutError:
+            host, base = None, None
+    if host and base:
+        PORTAL_HOST = host
+        PORTAL_BASE_URL = base
+        _page_info_cache = None
+        print(f"→ {base}")
+    else:
+        print("→ 使用默认")
+
+    if is_authenticated():
+        _print("已认证，无需登录！")
+        return True
+
+    saved = load_state()
+    if saved and saved.get("userIndex"):
+        try:
+            resp = requests.post(
+                f"{PORTAL_BASE_URL}/InterFace.do?method=getOnlineUserInfo",
+                data=f"userIndex={saved['userIndex']}",
+                headers={"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8"},
+                timeout=10,
+            )
+            resp.encoding = 'utf-8'
+            info = resp.json()
+            if info.get("result") == "success":
+                _print(f"恢复上次会话 userIndex={saved['userIndex']}")
+                ping_test()
+                start_keepalive(saved["userIndex"], info.get("keepaliveInterval", 180))
+                _print("按 Ctrl+C 退出...")
+                try:
+                    while True:
+                        time.sleep(60)
+                except KeyboardInterrupt:
+                    _print("退出保活")
+                return True
+            else:
+                _print("上次会话已失效，重新登录...")
+        except Exception:
+            pass
+
+    all_macs = []
+    if saved and saved.get("mac"):
+        all_macs.append(saved["mac"])
+        _print(f"[上次MAC] {saved['mac']}")
+
+    portal_mac = get_portal_mac()
+    if portal_mac:
+        all_macs.append(portal_mac)
+        _print(f"[Portal MAC] {portal_mac}")
+
+    local_macs = get_all_macs()
+    for m in local_macs:
+        if m not in all_macs:
+            all_macs.append(m)
+
+    _print(f"[扫描] 发现 {len(local_macs)} 个本地网卡MAC:")
+    for i, m in enumerate(local_macs):
+        print(f"  {i+1}. {m}")
+
+    qs = get_query_string()
+    if qs:
+        _print(f"[queryString] {qs[:80]}...")
+
+    fallbacks = ["111111111111", "000000000000", "ffffffffffff",
+                 "111111111", "123456789abc", "abcdef123456",
+                 "000000000001", "deadbeef0000"]
+    for m in fallbacks:
+        if m not in all_macs:
+            all_macs.append(m)
+
+    _print(f"共 {len(all_macs)} 种MAC方案，最多尝试 {MAX_ATTEMPTS} 次\n")
+
+    last_msg = ""
+    mac_index = 0
+    delay = SLEEP_SECONDS
+    consecutive_fails = 0
+
+    for attempt in range(1, MAX_ATTEMPTS + 1):
+        mac = all_macs[mac_index % len(all_macs)]
+        mac_index += 1
+
+        if attempt % (len(all_macs) + 1) == 0:
+            fresh = get_all_macs()
+            for m in fresh:
+                if m not in all_macs:
+                    all_macs.append(m)
+                    _print(f"[新增MAC] {m}")
+
+        print(f"[{attempt:02d}/{MAX_ATTEMPTS}] MAC={mac:14s}", end=" ")
+
+        ok, msg, user_idx, keep_interval = do_login(mac, query_string=qs)
+        print(f"→ {msg}")
+
+        if ok:
+            _print(f"登录成功！MAC={mac}")
+            ping_test()
+            save_state(mac, user_idx)
+            if user_idx:
+                _print(f"userIndex: {user_idx}")
+                register_mac(mac, user_idx)
+                start_keepalive(user_idx, keep_interval)
+            _print("按 Ctrl+C 退出保活...")
+            try:
+                while True:
+                    time.sleep(60)
+            except KeyboardInterrupt:
+                global _skip_wait
+                _skip_wait = True
+                _print("退出保活")
+            return True
+
+        consecutive_fails += 1
+        if msg != last_msg:
+            delay = SLEEP_SECONDS
+            consecutive_fails = 0
+        else:
+            delay = min(SLEEP_SECONDS * (1.5 ** (consecutive_fails // 3)), 60)
+
+        last_msg = msg
+
+        if attempt % 5 == 0 and is_authenticated():
+            _print("检测到已认证重定向！")
+            return True
+
+        time.sleep(delay)
+
+    if is_authenticated():
+        _print("最终检查：已认证")
+        return True
+
+    _print(f"{MAX_ATTEMPTS} 次尝试失败。最后: {last_msg}", "ERROR")
+    return False
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except KeyboardInterrupt:
+        pass
+    except Exception as e:
+        _print(f"异常退出: {e}", "ERROR")
+    _wait_exit()