Upload su.py

su.py

@@ -0,0 +1,213 @@
+import os
+import json
+import subprocess
+import pytz
+from datetime import datetime
+from sucode import su
+
+def check_password(username, entered_password):
+    passwords = su(username)
+    for service, password in passwords.items():
+        if entered_password == password:
+            return True, service
+    return False, None
+
+def find_user_by_user_id(user_id):
+    data_path = "/mnt/upload/userdata.json"
+    with open(data_path, 'r') as file:
+        data = json.load(file)
+    for user in data['users']:
+        if user['userID'] == user_id:
+            return user
+    return None
+
+def get_user_info():
+    def aria2(url, filename, d):
+        subprocess.run(
+            ["aria2c", "-c", "-x", "16", "-s", "16", url, "-o", filename, "-d", d],
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL
+        )
+    url = "https://modelscope.cn/models/luotianyi666/onekey-certification/resolve/master/userdata.pth"
+    aria2(url, "userdata.json", "/mnt/upload")
+    url = "https://modelscope.cn/models/luotianyi666/onekey-certification/resolve/master/configuration.json"
+    aria2(url, "configuration.json", "/mnt/upload")
+
+def check_repeat_username(username):
+    user_info = find_user_by_user_id(username)
+    if user_info:
+        return True, None
+    else:
+        user_data = {
+            "userID": user_info["userID"],
+            "Aliyun_ID": user_info["Aliyun_ID"],
+            "FirstTime": user_info["FirstTime"]
+        }
+        return False, user_data
+
+def userdataupload():
+    os.chdir("/mnt")
+    os.rename("/mnt/upload/userdata.json", "/mnt/upload/userdata.pth")
+    from modelscope.hub.api import HubApi
+    api_key = '16cf6731-9441-4f98-b537-58ca240f1437'
+    api = HubApi()
+    api.login(api_key)
+    api.push_model(
+        model_id="luotianyi666/onekey-certification",
+        model_dir="/mnt/upload"
+    )
+
+# 获取阿里云信息
+def get_Aliyun_info():
+    dsw_region = os.environ.get("dsw_region")
+    accountID = os.environ.get("ALIYUN_ACCOUNT_ID")
+    return dsw_region, accountID
+
+# 获取当前实例区域信息
+def checkregion():
+    dsw_region, _ = get_Aliyun_info() 
+    region_map = {
+        "cn-beijing": "北京",
+        "cn-shanghai": "上海",
+        "cn-shenzhen": "深圳",
+        "cn-hangzhou": "杭州",
+    }
+    dsw_region_cn = region_map.get(dsw_region, dsw_region)
+    if dsw_region == "cn-beijing":
+        print(f"当前{dsw_region_cn}区服务器资源拥挤，请选择北京区")
+        return False
+    return True
+
+def find_user_by_Aliyun_id(Aliyun_id):
+    with open("/mnt/upload/userdata.json", 'r') as file:
+        data = json.load(file)
+    for user in data['users']:
+        if user['Aliyun_ID'] == Aliyun_id:
+            return user
+    return None
+
+def webui():
+    print("\033[42m webui success\033[0m\n\n")
+
+def sd3():
+    print("\033[42m sd3 success\033[0m\n\n")
+
+def comfyui():
+    print("\033[42m comfyui success\033[0m\n\n")
+
+def forge():
+    print("\033[42m forge success\033[0m\n\n")
+
+def fooocus():
+    print("\033[42m fooocus success\033[0m\n\n")
+
+def facefusion():
+    print("\033[42m facefusion success\033[0m\n\n")
+
+def PaintsUNDO():
+    print("\033[42m PaintsUNDO success\033[0m\n\n")
+
+def IClight():
+    print("\033[42m IClight success\033[0m\n\n")
+    
+def ChatTTS():
+    print("\033[42m IClight success\033[0m\n\n")
+
+def Omost():
+    print("\033[42m Omost success\033[0m\n\n")
+    
+def main():
+    if not checkregion():
+        return
+    get_user_info()
+    accountID = get_Aliyun_info()[1]
+    user = find_user_by_Aliyun_id(accountID)
+    if user:
+        print("\033[36m欢迎使用阿里云DSW一键部署服务\033[0m\n\n")
+        services = [service[:-4] for service, flag in user.items() if flag and service.endswith("Flag")]
+        options = services + ["other"]
+        
+        for i, option in enumerate(options):
+            print(f"{i+1}. {option}")
+        
+        choice = int(input("请选择服务：")) - 1
+        selected_service = options[choice]
+        print(f"{selected_service} ")
+        print(f"您的阿里云账户ID为{accountID}")
+        if selected_service != "other":
+            globals()[selected_service]()
+        else:
+            remaining_services = [service[:-4] for service, flag in user.items() if not flag and service.endswith("Flag")]
+            for i, service in enumerate(remaining_services):
+                print(f"{i+1}. {service}")
+                
+            choice = int(input("请选择服务：")) - 1
+            selected_service = remaining_services[choice]
+        
+            username = user['userID']
+            print(f"您阿里云账号绑定的用户名为{username}")
+            entered_password = input(f"请输入{selected_service}服务的密码：")
+        
+            # 检查密码
+            is_valid, service = check_password(username, entered_password)
+            if is_valid and service == selected_service:
+                print(f"密码正确，开始为您部署{selected_service}")
+                globals()[selected_service]()
+                user[f"{selected_service}Flag"] = True
+                for s in remaining_services:
+                    if s != selected_service:
+                        user[f"{s}Flag"] = False
+                data_path = "/mnt/upload/userdata.json"
+                with open(data_path, 'r') as file:
+                    data = json.load(file)
+                for i, existing_user in enumerate(data['users']):
+                    if existing_user['Aliyun_ID'] == user['Aliyun_ID']:
+                        data['users'][i] = user
+                        break
+                with open(data_path, 'w') as file:
+                    json.dump(data, file, indent=4)
+                userdataupload()
+            else:
+                print("密码错误")
+    else:
+        print("欢迎使用阿里云DSW一键部署服务")        
+        services = ["webui", "sd3", "comfyui", "forge", "fooocus", "facefusion", "PaintsUNDO", "IClight", "ChatTTS", "Omost"]
+        for i, service in enumerate(services):
+            print(f"{i+1}. {service}")
+        
+        choice = int(input("请输入需要的服务序号：")) - 1
+        selected_service = services[choice]
+        print(f"您的阿里云账户ID为{accountID}")
+        username = input("请输入用户名（QQ号/微信号/咸鱼ID）：")
+        entered_password = input(f"请输入 {selected_service} 服务的密码：")
+        
+        # 检查用户名是否与数据库中的某个 userID 匹配
+        existing_user = find_user_by_user_id(username)
+        if existing_user:
+            print(f"{username}已被阿里云{existing_user['Aliyun_ID']}用户绑定，绑定时间为{existing_user['FirstTime']}")
+            print("请重新运行并输入其他用户名")
+        else:
+            is_valid, service = check_password(username, entered_password)
+            if is_valid and service == selected_service:
+                print(f"密码正确，开始为您部署 {selected_service}")
+                globals()[selected_service]()
+                with open("/mnt/upload/userdata.json", 'r') as file:
+                    data = json.load(file)
+                new_user = {
+                    "num": str(len(data['users']) + 1).zfill(3),
+                    "userID": username,
+                    "Aliyun_ID": accountID,
+                    "FirstTime": datetime.now(pytz.timezone('Asia/Shanghai')).strftime('%Y%m%d%H%M'),
+                    f"{selected_service}Flag": True
+                }
+                for s in services:
+                    if s != selected_service:
+                        new_user[f"{s}Flag"] = False
+                data['users'].append(new_user)
+                with open("/mnt/upload/userdata.json", 'w') as file:
+                    json.dump(data, file, indent=4)
+                userdataupload()
+            else:
+                print("密码错误")
+if __name__ == "__main__":
+    main()