diff --git a/script_library/index.json b/script_library/index.json index 6a9d179..d136863 100644 --- a/script_library/index.json +++ b/script_library/index.json @@ -1,7 +1,7 @@ { "format_version": 1, - "data_version": 81, - "updated": "2026-04-05T09:58:07Z", + "data_version": 82, + "updated": "2026-04-05T16:20:03.492Z", "announcement": null, "categories": [ { @@ -1328,6 +1328,29 @@ "updated": null, "status": "active", "lines": 99 + }, + { + "id": "subinfo_web_mnlyw592", + "name": "subinfo_web", + "name_en": "subinfo_web", + "desc": "订阅链接查询", + "desc_en": "订阅链接查询", + "category": "ui", + "file": "scripts/ui/subinfo_web_mnlyw592.py", + "thumbnail": null, + "version": 1, + "file_type": "py", + "author": "啥也不是", + "author_en": "啥也不是", + "tags": [ + "community" + ], + "requires": [], + "min_app_version": "1.5.0", + "added": "2026-04-05", + "updated": null, + "status": "active", + "lines": 568 } ] } diff --git a/script_library/scripts/ui/subinfo_web_mnlyw592.py b/script_library/scripts/ui/subinfo_web_mnlyw592.py new file mode 100644 index 0000000..cfa5ca4 --- /dev/null +++ b/script_library/scripts/ui/subinfo_web_mnlyw592.py @@ -0,0 +1,567 @@ +import json +import re +from datetime import datetime +from pathlib import Path +from typing import Dict, List, Optional +from urllib.parse import urlparse +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from socket import error as SocketError + +import requests + + +USER_AGENT = "ClashforWindows/0.19.21" +SCHEME_PATTERN = re.compile(r"https?://", re.IGNORECASE) +URL_CHAR_PATTERN = re.compile(r"[A-Za-z0-9\-._~:/?#\[\]@!$&'()*+,;=%]") +TRAILING_CHARS = ",.;:!?)]}>'\"，。；：！？】）〉》、" +LEADING_CHARS = "<([{\"'【（〈《" +EXPORT_DIR = Path.cwd() + + +def format_size(num_bytes: int) -> str: + negative = num_bytes < 0 + size = float(abs(num_bytes)) + units = ["B", "KB", "MB", "GB", "TB", "PB"] + for unit in units: + if size < 1024 or unit == units[-1]: + text = f"{size:.2f}{unit}" + return f"-{text}" if negative else text + size /= 1024 + return "0.00B" + + +def parse_subscription_userinfo(header_value: str) -> Dict[str, int]: + result: Dict[str, int] = {} + matches = re.findall(r"([A-Za-z]+)=(\d+)", header_value) + for key, value in matches: + result[key.lower()] = int(value) + return result + + +def format_remaining_seconds(seconds: int) -> str: + if seconds < 0: + seconds = abs(seconds) + prefix = "已过期" + else: + prefix = "距离到期还有" + + days = seconds // 86400 + hours = (seconds % 86400) // 3600 + minutes = (seconds % 3600) // 60 + secs = seconds % 60 + + if days > 0: + return f"{prefix}{days}天{hours}小时{minutes}分{secs}秒" + if hours > 0: + return f"{prefix}{hours}小时{minutes}分{secs}秒" + if minutes > 0: + return f"{prefix}{minutes}分{secs}秒" + return f"{prefix}{secs}秒" + + +def clean_candidate_url(url: str) -> str: + url = url.strip().strip(LEADING_CHARS).rstrip(TRAILING_CHARS) + while url and url[-1] in TRAILING_CHARS: + url = url[:-1] + return url + + +def looks_like_http_url(url: str) -> bool: + try: + parsed = urlparse(url) + except Exception: + return False + return parsed.scheme in {"http", "https"} and bool(parsed.netloc) + + +def extract_urls(text: str) -> List[str]: + result: List[str] = [] + seen = set() + starts = [m.start() for m in SCHEME_PATTERN.finditer(text)] + + for i, start in enumerate(starts): + next_start = starts[i + 1] if i + 1 < len(starts) else len(text) + end = start + + while end < len(text): + if end > start and end < next_start and text[end:end + 7].lower() == "http://": + break + if end > start and end < next_start and text[end:end + 8].lower() == "https://": + break + if end >= next_start: + break + if not URL_CHAR_PATTERN.match(text[end]): + break + end += 1 + + raw = text[start:end] + url = clean_candidate_url(raw) + if not url or not looks_like_http_url(url): + continue + if url not in seen: + seen.add(url) + result.append(url) + + stripped = clean_candidate_url(text.strip()) + if not result and looks_like_http_url(stripped): + result.append(stripped) + + return result + + +def export_positive_urls(urls: List[str], base_dir: Optional[Path] = None) -> Optional[Path]: + if not urls: + return None + + output_dir = Path(base_dir) if base_dir else EXPORT_DIR + output_dir.mkdir(parents=True, exist_ok=True) + file_path = output_dir / f"positive_sub_urls_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt" + file_path.write_text("\n".join(urls) + "\n", encoding="utf-8") + return file_path + + +def query_sub_info(url: str, timeout: int = 12) -> Dict[str, object]: + headers = {"User-Agent": USER_AGENT} + response = requests.get(url, headers=headers, timeout=timeout) + response.raise_for_status() + + sub_header: Optional[str] = response.headers.get("Subscription-Userinfo") + if not sub_header: + raise ValueError("未获取到 Subscription-Userinfo，可能该订阅不提供流量或到期信息。") + + info = parse_subscription_userinfo(sub_header) + upload = info.get("upload", 0) + download = info.get("download", 0) + total = info.get("total", 0) + expire = info.get("expire") + + used = upload + download + remain = total - used + + result: Dict[str, object] = { + "订阅链接": url, + "总流量": format_size(total) if total else "未知", + "剩余流量": format_size(remain) if total else "未知", + "已上传": format_size(upload), + "已下载": format_size(download), + "到期时间": "未知", + "到期说明": "可能是无限期订阅，或服务端未返回 expire", + "剩余流量字节": remain, + "查询成功": True, + } + + if expire: + expire_dt = datetime.fromtimestamp(expire).astimezone() + now_ts = int(datetime.now().astimezone().timestamp()) + result["到期时间"] = expire_dt.strftime("%Y-%m-%d %H:%M:%S %Z") + result["到期说明"] = format_remaining_seconds(expire - now_ts) + + return result + + +def render_success_block(data: Dict[str, object], index: int, total_count: int) -> str: + return ( + f"[{index}/{total_count}] 查询成功\n" + f"订阅链接: {data['订阅链接']}\n" + f"总流量: {data['总流量']}\n" + f"剩余流量: {data['剩余流量']}\n" + f"已上传: {data['已上传']}\n" + f"已下载: {data['已下载']}\n" + f"到期时间: {data['到期时间']}\n" + f"说明: {data['到期说明']}\n" + ) + + +def render_error_block(url: str, error_message: str, index: int, total_count: int) -> str: + return ( + f"[{index}/{total_count}] 查询失败\n" + f"订阅链接: {url}\n" + f"错误信息: {error_message}\n" + ) + + +def run_batch_query(raw_text: str) -> Dict[str, object]: + raw_text = raw_text.strip() + if not raw_text: + return {"ok": False, "message": "请先输入或粘贴内容。"} + + urls = extract_urls(raw_text) + if not urls: + return {"ok": False, "message": "未识别到可查询的 http/https 链接。"} + + results: List[str] = [] + success_count = 0 + positive_urls: List[str] = [] + + for index, url in enumerate(urls, start=1): + try: + data = query_sub_info(url) + results.append(render_success_block(data, index, len(urls))) + success_count += 1 + remain_bytes = int(data.get("剩余流量字节", 0)) + if remain_bytes > 0 and str(data["订阅链接"]) not in positive_urls: + positive_urls.append(str(data["订阅链接"])) + except requests.HTTPError as exc: + status_code = exc.response.status_code if exc.response is not None else "未知" + results.append(render_error_block(url, f"HTTP 请求失败，状态码: {status_code}", index, len(urls))) + except requests.RequestException as exc: + results.append(render_error_block(url, f"网络请求失败: {exc}", index, len(urls))) + except Exception as exc: + results.append(render_error_block(url, f"查询失败: {exc}", index, len(urls))) + + export_text = ("\n".join(positive_urls) + "\n") if positive_urls else "" + export_path = export_positive_urls(positive_urls) + summary_lines = [ + f"共识别 {len(urls)} 个链接，成功 {success_count} 个，失败 {len(urls) - success_count} 个。", + f"剩余流量大于 0 的可用链接：{len(positive_urls)} 个。", + ] + if export_path: + summary_lines.append(f"已同时保存到脚本目录: {export_path.name}") + else: + summary_lines.append("没有可导出的可用链接。") + + summary = "\n".join(summary_lines) + f"\n{'=' * 72}\n\n" + final_text = summary + (f"\n{'-' * 72}\n\n".join(results) if results else "没有可显示的结果。\n") + return { + "ok": True, + "urls": urls, + "success_count": success_count, + "total_count": len(urls), + "positive_count": len(positive_urls), + "export_path": str(export_path) if export_path else "", + "export_name": export_path.name if export_path else "", + "export_text": export_text, + "positive_urls": positive_urls, + "text": final_text, + } + + +HTML_PAGE = """ + + + + +订阅查询工具 + + + +
+
+

订阅链接查询

支持：单链接查询、混杂文本自动提取、多链接批量查询、页面内直接导出可用链接。

+
+ +
+ +
+ + + + +
+
就绪
+
+
自动识别 http / https
+
页面内导出 + 一键复制
+
+
+ 小提示：查询完成后，可用链接会直接显示在页面下方，可一键复制。 +
+
+ +
+ 
结果会显示在这里。
+
+ +
+
可用订阅链接（剩余流量大于 0）
+ +
+ + +
+
还没有导出内容
+
+
+ + + +""" + + +class WebHandler(BaseHTTPRequestHandler): + def _set_common_headers(self, content_type: str, content_length: int, status: int = 200, extra_headers: Optional[Dict[str, str]] = None) -> None: + self.send_response(status) + self.send_header("Content-Type", content_type) + self.send_header("Content-Length", str(content_length)) + self.send_header("Cache-Control", "no-store") + self.send_header("Access-Control-Allow-Origin", "*") + self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") + self.send_header("Access-Control-Allow-Headers", "Content-Type") + if extra_headers: + for key, value in extra_headers.items(): + self.send_header(key, value) + self.end_headers() + + def _safe_write(self, data: bytes) -> None: + try: + self.wfile.write(data) + except (BrokenPipeError, ConnectionResetError, SocketError): + return + + def _send_json(self, payload: Dict[str, object], status: int = 200) -> None: + data = json.dumps(payload, ensure_ascii=False).encode("utf-8") + self._set_common_headers("application/json; charset=utf-8", len(data), status) + self._safe_write(data) + + def do_OPTIONS(self) -> None: + self._set_common_headers("text/plain; charset=utf-8", 0, 204) + + def do_GET(self) -> None: + if self.path == "/": + body = HTML_PAGE.encode("utf-8") + self._set_common_headers("text/html; charset=utf-8", len(body), 200) + self._safe_write(body) + return + + if self.path.startswith("/download/"): + filename = self.path.split("/download/", 1)[1] + if not filename: + self._send_json({"ok": False, "message": "文件名无效"}, status=400) + return + file_path = EXPORT_DIR / Path(filename).name + if not file_path.exists() or not file_path.is_file(): + self._send_json({"ok": False, "message": "文件不存在"}, status=404) + return + data = file_path.read_bytes() + headers = {"Content-Disposition": f'attachment; filename="{file_path.name}"'} + self._set_common_headers("text/plain; charset=utf-8", len(data), 200, headers) + self._safe_write(data) + return + + self._send_json({"ok": False, "message": "Not Found"}, status=404) + + def do_POST(self) -> None: + if self.path != "/query": + self._send_json({"ok": False, "message": "Not Found"}, status=404) + return + + try: + length = int(self.headers.get("Content-Length", "0")) + raw = self.rfile.read(length).decode("utf-8") + data = json.loads(raw) if raw else {} + text = str(data.get("text", "")) + result = run_batch_query(text) + self._send_json(result) + except Exception as exc: + self._send_json({"ok": False, "message": f"服务端处理失败: {exc}"}, status=500) + + def log_message(self, format: str, *args) -> None: + return + + +def try_open_browser(url: str) -> None: + try: + import webbrowser + webbrowser.open(url) + except Exception: + pass + + +def run_web_mode() -> None: + host = "127.0.0.1" + server = None + port = 8765 + for candidate in [8765, 8766, 8767, 8080, 8000]: + try: + server = ThreadingHTTPServer((host, candidate), WebHandler) + port = candidate + break + except OSError: + continue + + if server is None: + raise RuntimeError("无法启动本地网页服务，请检查端口占用情况。") + + url = f"http://{host}:{port}/" + print("启动成功：HTML 查询界面已就绪") + print(f"请在浏览器或应用内网页预览打开：{url}") + print("按停止运行即可结束服务。") + try_open_browser(url) + server.serve_forever() + + +if __name__ == "__main__": + run_web_mode()