perf(core): optimize bulk user URI generation in wrapper_uri.py

core/cli.py

@@ -160,7 +160,7 @@ def bulk_user_add(traffic_gb: float, expiration_days: int, count: int, prefix: s
 @click.option('--new-expiration-days', '-ne', required=False, help='Expiration days for the new user', type=int)
 @click.option('--renew-password', '-rp', is_flag=True, help='Renew password for the user')
 @click.option('--renew-creation-date', '-rc', is_flag=True, help='Renew creation date for the user')
-@click.option('--blocked/--unblocked', 'blocked', default=None, help='Block or unblock the user.')
+@click.option('--blocked/--unblocked', 'blocked', '-b', default=None, help='Block or unblock the user.')
 @click.option('--unlimited-ip/--limited-ip', 'unlimited_ip', default=None, help='Set user to be exempt from or subject to IP limits.')
 def edit_user(username: str, new_username: str, new_traffic_limit: int, new_expiration_days: int, renew_password: bool, renew_creation_date: bool, blocked: bool | None, unlimited_ip: bool | None):
     try:

core/scripts/hysteria2/wrapper_uri.py

@@ -1,65 +1,152 @@
-import subprocess
+#!/usr/bin/env python3
-import concurrent.futures
+
-import re
+import os
-import json
 import sys
+import json
+import argparse
+from functools import lru_cache
+from typing import Dict, List, Any
 
 from init_paths import *
 from paths import *
 
-DEFAULT_ARGS = ["-a", "-n", "-s"]
+@lru_cache(maxsize=None)
-
+def load_json_file(file_path: str) -> Any:
-def run_show_uri(username):
+    if not os.path.exists(file_path):
+        return None
     try:
-        cmd = ["python3", CLI_PATH, "show-user-uri", "-u", username] + DEFAULT_ARGS
+        with open(file_path, 'r') as f:
-        result = subprocess.run(cmd, capture_output=True, text=True, check=True)
+            content = f.read()
-        output = result.stdout
+            return json.loads(content) if content else None
-        if "Invalid username" in output:
+    except (json.JSONDecodeError, IOError):
-            return {"username": username, "error": "User not found"}
+        return None
-        return parse_output(username, output)
-    except subprocess.CalledProcessError as e:
-        return {"username": username, "error": e.stderr.strip()}
 
-def parse_output(username, output):
+@lru_cache(maxsize=None)
-    ipv4 = None
+def load_env_file(env_file: str) -> Dict[str, str]:
-    ipv6 = None
+    env_vars = {}
-    normal_sub = None
+    if os.path.exists(env_file):
-    nodes = []
+        with open(env_file, 'r') as f:
+            for line in f:
+                line = line.strip()
+                if line and not line.startswith('#') and '=' in line:
+                    key, value = line.split('=', 1)
+                    env_vars[key] = value.strip()
+    return env_vars
 
-    ipv4_match = re.search(r"IPv4:\s*(hy2://[^\s]+)", output)
+def generate_uri(username: str, auth_password: str, ip: str, port: str, 
-    ipv6_match = re.search(r"IPv6:\s*(hy2://[^\s]+)", output)
+                 obfs_password: str, sha256: str, sni: str, ip_version: int, 
-    normal_sub_match = re.search(r"Normal-SUB Sublink:\s*(https?://[^\s]+)", output)
+                 insecure: bool, fragment_tag: str) -> str:
-
+    ip_part = f"[{ip}]" if ip_version == 6 and ':' in ip else ip
-    if ipv4_match:
+    uri_base = f"hy2://{username}:{auth_password}@{ip_part}:{port}"
-        ipv4 = ipv4_match.group(1)
+    
-    if ipv6_match:
+    params = {
-        ipv6 = ipv6_match.group(1)
+        "insecure": "1" if insecure else "0",
-    if normal_sub_match:
+        "sni": sni
-        normal_sub = normal_sub_match.group(1)
-
-    node_matches = re.findall(r"Node: (.+?) \(IPv[46]\):\s*(hy2://[^\s]+)", output)
-    for name, uri in node_matches:
-        nodes.append({"name": name.strip(), "uri": uri})
-
-
-    return {
-        "username": username,
-        "ipv4": ipv4,
-        "ipv6": ipv6,
-        "nodes": nodes,
-        "normal_sub": normal_sub
     }
+    if obfs_password:
+        params["obfs"] = "salamander"
+        params["obfs-password"] = obfs_password
+    if sha256:
+        params["pinSHA256"] = sha256
+    
+    query_string = "&".join([f"{k}={v}" for k, v in params.items()])
+    return f"{uri_base}?{query_string}#{fragment_tag}"
 
-def batch_show_uri(usernames, max_workers=20):
+def process_users(target_usernames: List[str]) -> List[Dict[str, Any]]:
-    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
+    config = load_json_file(CONFIG_FILE)
-        results = list(executor.map(run_show_uri, usernames))
+    all_users = load_json_file(USERS_FILE)
-    return results
+    nodes = load_json_file(NODES_JSON_PATH) or []
-
+    
-if __name__ == "__main__":
+    if not config or not all_users:
-    if len(sys.argv) < 2:
+        print("Error: Could not load hysteria2 configuration or user files.", file=sys.stderr)
-        print("Usage: python3 show_uri_json.py user1 user2 ...")
         sys.exit(1)
 
-    usernames = sys.argv[1:]
+    port = config.get("listen", "").split(":")[-1]
-    output_list = batch_show_uri(usernames)
+    tls_config = config.get("tls", {})
-    print(json.dumps(output_list, indent=2))
+    sha256 = tls_config.get("pinSHA256", "")
+    insecure = tls_config.get("insecure", True)
+    obfs_password = config.get("obfs", {}).get("salamander", {}).get("password", "")
+    
+    hy2_env = load_env_file(CONFIG_ENV)
+    ip4 = hy2_env.get('IP4')
+    ip6 = hy2_env.get('IP6')
+    sni = hy2_env.get('SNI', '')
+
+    ns_env = load_env_file(NORMALSUB_ENV)
+    ns_domain = ns_env.get('HYSTERIA_DOMAIN')
+    ns_port = ns_env.get('HYSTERIA_PORT')
+    ns_subpath = ns_env.get('SUBPATH')
+
+    results = []
+    
+    for username in target_usernames:
+        user_data = all_users.get(username)
+        if not user_data or "password" not in user_data:
+            results.append({"username": username, "error": "User not found or password not set"})
+            continue
+
+        auth_password = user_data["password"]
+        user_output = {
+            "username": username,
+            "ipv4": None,
+            "ipv6": None,
+            "nodes": [],
+            "normal_sub": None
+        }
+
+        if ip4 and ip4 != "None":
+            user_output["ipv4"] = generate_uri(
+                username, auth_password, ip4, port, obfs_password, sha256, sni, 4, insecure, f"{username}-IPv4"
+            )
+        
+        if ip6 and ip6 != "None":
+            user_output["ipv6"] = generate_uri(
+                username, auth_password, ip6, port, obfs_password, sha256, sni, 6, insecure, f"{username}-IPv6"
+            )
+
+        for node in nodes:
+            node_name, node_ip = node.get("name"), node.get("ip")
+            if not (node_name and node_ip):
+                continue
+            ip_v = 6 if ':' in node_ip else 4
+            tag = f"{username}-{node_name}"
+            uri = generate_uri(
+                username, auth_password, node_ip, port, obfs_password, sha256, sni, ip_v, insecure, tag
+            )
+            user_output["nodes"].append({"name": node_name, "uri": uri})
+        
+        if ns_domain and ns_port and ns_subpath:
+            user_output["normal_sub"] = f"https://{ns_domain}:{ns_port}/{ns_subpath}/sub/normal/{auth_password}#{username}"
+
+        results.append(user_output)
+        
+    return results
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Efficiently generate Hysteria2 URIs for multiple users.",
+        formatter_class=argparse.RawTextHelpFormatter
+    )
+    parser.add_argument('usernames', nargs='*', help="A list of usernames to process.")
+    parser.add_argument('--all', action='store_true', help="Process all users from users.json.")
+    
+    args = parser.parse_args()
+    
+    target_usernames = args.usernames
+    if args.all:
+        all_users = load_json_file(USERS_FILE)
+        if all_users:
+            target_usernames = list(all_users.keys())
+        else:
+            print("Error: Could not load users.json to process all users.", file=sys.stderr)
+            sys.exit(1)
+            
+    if not target_usernames:
+        parser.print_help()
+        sys.exit(1)
+
+    output_list = process_users(target_usernames)
+    print(json.dumps(output_list, indent=2))
+
+if __name__ == "__main__":
+    main()