From f9293251c10cd08c07b6da2df2f270e4f716a447 Mon Sep 17 00:00:00 2001 From: Richard <63699118+Herr-Richard@users.noreply.github.com> Date: Thu, 23 Apr 2026 15:12:23 +0800 Subject: [PATCH] Add Xiaomi FDS Client GUI functionality Implement Xiaomi FDS Client GUI for resource access and download. --- Xiaomi_FDS_Client_GUI.py | 387 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 387 insertions(+) create mode 100644 Xiaomi_FDS_Client_GUI.py diff --git a/Xiaomi_FDS_Client_GUI.py b/Xiaomi_FDS_Client_GUI.py new file mode 100644 index 0000000..73c5c7b --- /dev/null +++ b/Xiaomi_FDS_Client_GUI.py @@ -0,0 +1,387 @@ +#!/usr/bin python3 +# 功能用途：实现对XIAOMI FDS中的对象文件进行资源访问和下载 +# Developer ：zhupengfei7 +# Date : 2026-04-17 15:48:03 +# 使用pyinstaller打包Windows平台exe包命令:(.venv) PS D:\Project\Python\Project\XIAOMI\Xiaomi_FDS_Client_GUI> pyinstaller -F -w -i NONE "D:\Project\Python\Project\XIAOMI\Xiaomi_FDS_Client_GUI\Xiaomi_FDS_Client_GUI.py" + +import requests +import hmac +import hashlib +import base64 +import json +import os +import threading +from datetime import datetime, UTC +from urllib.parse import quote +import tkinter as tk +from tkinter import ttk, scrolledtext, messagebox, filedialog + +# 配置文件 +CONFIG_PATH = "ApplicationConfig.json" + + +# 账号配置管理，管理FDS鉴权信息的工具类 +class AccountManager: + def __init__(self): + self.accounts = [] + self.load() + + # 从配置文件加载FDS鉴权信息，初始化，创建空账号列表+配置文件 + def load(self): + if os.path.exists(CONFIG_PATH): + with open(CONFIG_PATH, "r", encoding="utf-8") as f: + self.accounts = json.load(f) + + # 保存FDS鉴权信息到配置文件 + def save(self): + with open(CONFIG_PATH, "w", encoding="utf-8") as f: + json.dump(self.accounts, f, ensure_ascii=False, indent=2) + + # 添加或更新FDS鉴权信息到配置文件 + def add_or_update(self, name, ak, sk, bucket, endpoint): + for item in self.accounts: + if item["name"] == name: + item.update({"ak": ak, "sk": sk, "bucket": bucket, "endpoint": endpoint}) + self.save() + return + self.accounts.append({ + "name": name, + "ak": ak, + "sk": sk, + "bucket": bucket, + "endpoint": endpoint + }) + self.save() + + # 删除指定名称的FDS鉴权信息 + def delete(self, name): + self.accounts = [x for x in self.accounts if x["name"] != name] + self.save() + + +# FDS签名生成函数 +def sign(sk, msg): + return base64.b64encode(hmac.new(sk.encode(), msg.encode(), hashlib.sha1).digest()).decode() + + +# 生成FDS请求鉴权头，核心鉴权功能，必须引用说明文档的方式 +def auth_header(ak, sk, method, bucket, key=""): + date = datetime.now(UTC).strftime("%a, %d %b %Y %H:%M:%S GMT") + canonical = f"/{bucket}" + (f"/{quote(key)}" if key else "") + sign_str = f"{method}\n\n\n{date}\n{canonical}" + return { + "Date": date, + "Authorization": f"Galaxy-V2 {ak}:{sign(sk, sign_str)}" + } + + +# 分页获取FDS文件列表 +def fetch_page(ak, sk, endpoint, bucket, marker=""): + params = {"prefix": "", "max-keys": 1000} + if marker: + params["marker"] = marker + + pengfei_res = requests.get( + f"https://{endpoint}/{bucket}", + headers=auth_header(ak, sk, "GET", bucket), + params=params, + timeout=15 + ) + if pengfei_res.status_code != 200: + return [], False, "" + + data = pengfei_res.json() + objects = data.get("objects", []) + truncated = data.get("truncated", False) or data.get("is_truncated", False) + next_marker = data.get("nextMarker", "") or data.get("next_marker", "") + return objects, truncated, next_marker + + +# 软件信息 +def show_about_info(): + info = """Version ：vlatest_v2604-1701 +Developer ：zhupengfei7 +Copyright: Xiaomi + +Just for info： +1.galaxy-fds-sdk-python(https://github.com/XiaoMi/galaxy-fds-sdk-python) +2.fds-node(https://git.n.xiaomi.com/miui_security_fe/shared/fds-node) + +Dev Docs : +1. Xiaomi Galaxy FDS Book(http://docs.api.xiaomi.net/fds) +""" + messagebox.showinfo("关于软件：Xiaomi Galaxy File Storage Service Client （Download only）", info) + + +# FDS客户端主界面 +class FDSClient(tk.Tk): + def __init__(self): + super().__init__() + self.title("Xiaomi Galaxy File Storage Service Client（For EOL use only）") + self.geometry("1000x720") + # 程序启动时最大化窗口 + self.state('zoomed') + self.configure(bg="#F5F7FA") + + self.pengfei_acc = AccountManager() + self.pengfei_all_files_cache = [] + self.pengfei_is_loading = False + self.setup_style() + self.build_menu() + self.build_ui() + + # 初始化UI + def setup_style(self): + style = ttk.Style() + style.theme_use("clam") + style.configure(".", bg="#F5F7FA", fg="#333") + style.configure("Card.TFrame", bg="white", relief="flat") + style.configure("Primary.TButton", background="#409EFF", foreground="white") + style.map("Primary.TButton", background=[("active", "#338BEF")]) + style.configure("Treeview", bg="white", rowheight=24) + style.configure("Treeview.Heading", background="#409EFF", foreground="white") + + # 顶部菜单 + def build_menu(self): + menubar = tk.Menu(self) + help_menu = tk.Menu(menubar, tearoff=0) + help_menu.add_command(label="关于软件", command=show_about_info) + menubar.add_cascade(label="帮助", menu=help_menu) + self.config(menu=menubar) + + # 主界面所有UI组件 + def build_ui(self): + # FDS鉴权信息配置 + card1 = ttk.Frame(self, style="Card.TFrame", padding=12) + card1.pack(fill="x", padx=16, pady=8) + + ttk.Label(card1, text="配置名称").grid(row=0, column=0, sticky="w") + self.pengfei_zh_CN_name = tk.StringVar() + ttk.Entry(card1, textvariable=self.pengfei_zh_CN_name, width=16).grid(row=0, column=1, padx=6) + + self.pengfei_zh_CN_list = tk.StringVar() + self.pengfei_combo = ttk.Combobox(card1, textvariable=self.pengfei_zh_CN_list, state="readonly", width=16) + self.pengfei_combo.grid(row=0, column=2, padx=6) + self.refresh_combo() + + ttk.Button(card1, text="保存FDS鉴权信息配置", command=self.save_account, style="Primary.TButton").grid(row=0, column=3, padx=4) + ttk.Button(card1, text="删除FDS鉴权信息配置", command=self.del_account).grid(row=0, column=4, padx=4) + ttk.Button(card1, text="读取FDS鉴权信息配置", command=self.load_account).grid(row=0, column=5, padx=4) + + # 鉴权信息 + card2 = ttk.Frame(self, style="Card.TFrame", padding=12) + card2.pack(fill="x", padx=16, pady=4) + + ttk.Label(card2, text="AK").grid(row=0, column=0, sticky="w") + self.pengfei_zh_CN_ak = tk.StringVar() + ttk.Entry(card2, textvariable=self.pengfei_zh_CN_ak, width=26).grid(row=0, column=1, padx=6) + + ttk.Label(card2, text="SK").grid(row=0, column=2, sticky="w") + self.pengfei_zh_CN_sk = tk.StringVar() + ttk.Entry(card2, textvariable=self.pengfei_zh_CN_sk, width=26, show="*").grid(row=0, column=3, padx=6) + + ttk.Label(card2, text="Bucket").grid(row=1, column=0, sticky="w", pady=6) + self.pengfei_zh_CN_bucket = tk.StringVar() + ttk.Entry(card2, textvariable=self.pengfei_zh_CN_bucket, width=26).grid(row=1, column=1, padx=6) + + ttk.Label(card2, text="Endpoint").grid(row=1, column=2, sticky="w", pady=6) + self.pengfei_zh_CN_end = tk.StringVar() + ttk.Entry(card2, textvariable=self.pengfei_zh_CN_end, width=26).grid(row=1, column=3, padx=6) + + ttk.Button(card2, text="进行身份鉴权并读取FDS对象存储文件列表", command=self.start_load_files, style="Primary.TButton").grid(row=0, column=4, rowspan=2, padx=10, ipady=4) + + # 搜索 + search = ttk.Frame(self, padding=12) + search.pack(fill="x", padx=16, pady=4) + ttk.Label(search, text="模糊搜索").pack(side="left") + self.pengfei_zh_CN_search = tk.StringVar() + ttk.Entry(search, textvariable=self.pengfei_zh_CN_search, width=60).pack(side="left", padx=8, fill="x", expand=True) + ttk.Button(search, text="搜索", command=self.do_fuzzy_search, style="Primary.TButton").pack(side="left") + + # 文件列表展示 + card3 = ttk.Frame(self, style="Card.TFrame", padding=8) + card3.pack(fill="both", expand=True, padx=16, pady=6) + + tree_frame = ttk.Frame(card3) + tree_frame.pack(fill="both", expand=True) + + cols = ("name",) + self.pengfei_tree = ttk.Treeview(tree_frame, columns=cols, show="headings") + self.pengfei_tree.heading("name", text="文件名称") + self.pengfei_tree.column("name", width=950) + + pengfei_vsb = ttk.Scrollbar(tree_frame, orient="vertical", command=self.pengfei_tree.yview) + self.pengfei_tree.configure(yscrollcommand=pengfei_vsb.set) + + self.pengfei_tree.pack(side="left", fill="both", expand=True) + pengfei_vsb.pack(side="right", fill="y") + + # 下载 + btn_bar = ttk.Frame(self) + btn_bar.pack(fill="x", padx=16, pady=4) + ttk.Button(btn_bar, text="下载选中文件", command=self.do_download, style="Primary.TButton").pack(side="right") + + # 日志 + self.pengfei_log_text = scrolledtext.ScrolledText(self, height=4, bg="white", relief="flat") + self.pengfei_log_text.pack(fill="x", padx=16, pady=8) + + # 日志输出 + def log(self, msg): + self.pengfei_log_text.insert("end", msg + "\n") + self.pengfei_log_text.see("end") + + # 刷新FDS鉴权信息配置下拉框 + def refresh_combo(self): + self.pengfei_combo["values"] = [x["name"] for x in self.pengfei_acc.accounts] + + # 保存当前FDS鉴权信息 + def save_account(self): + name = self.pengfei_zh_CN_name.get().strip() + if not name: + messagebox.showwarning("提示", "请输入配置名称") + return + self.pengfei_acc.add_or_update( + name, self.pengfei_zh_CN_ak.get(), self.pengfei_zh_CN_sk.get(), + self.pengfei_zh_CN_bucket.get(), self.pengfei_zh_CN_end.get() + ) + self.refresh_combo() + self.log(f"✅ 已保存：{name}") + + # 加载选中的FDS鉴权信息 + def load_account(self): + name = self.pengfei_zh_CN_list.get() + for item in self.pengfei_acc.accounts: + if item["name"] == name: + self.pengfei_zh_CN_name.set(name) + self.pengfei_zh_CN_ak.set(item["ak"]) + self.pengfei_zh_CN_sk.set(item["sk"]) + self.pengfei_zh_CN_bucket.set(item["bucket"]) + self.pengfei_zh_CN_end.set(item["endpoint"]) + self.log(f"✅ 已加载：{name}") + return + + # 删除选中的FDS鉴权信息 + def del_account(self): + name = self.pengfei_zh_CN_list.get() + if not name: + return + if messagebox.askyesno("确认", f"删除配置：{name}？"): + self.pengfei_acc.delete(name) + self.refresh_combo() + self.log("🗑 已删除") + + # 后台线程加载文件列表 + def start_load_files(self): + if self.pengfei_is_loading: + messagebox.showinfo("提示", "正在加载中，请稍候...") + return + + self.pengfei_is_loading = True + for i in self.pengfei_tree.get_children(): + self.pengfei_tree.delete(i) + self.pengfei_all_files_cache = [] + + threading.Thread(target=self.load_files_background, daemon=True).start() + + # 后台线程分页拉取文件列表 + def load_files_background(self): + ak = self.pengfei_zh_CN_ak.get() + sk = self.pengfei_zh_CN_sk.get() + endpoint = self.pengfei_zh_CN_end.get() + bucket = self.pengfei_zh_CN_bucket.get() + + marker = "" + + while True: + try: + objects, truncated, next_marker = fetch_page(ak, sk, endpoint, bucket, marker) + if not objects: + break + + page_files = [obj["name"] for obj in objects] + self.pengfei_all_files_cache.extend(page_files) + + # 主线程更新UI，避免界面卡顿 + self.after(0, lambda files=page_files: self.append_to_tree(files)) + + self.log(f"⏳ 已加载 {len(self.pengfei_all_files_cache)} 条文件...") + + if not truncated or not next_marker: + break + marker = next_marker + + except Exception as e: + self.log(f"❌ 加载异常：{e}") + break + + self.pengfei_is_loading = False + self.log(f"✅ 全部加载完成，共 {len(self.pengfei_all_files_cache)} 条文件") + + # 将文件列表追加到UI展示 + def append_to_tree(self, files): + for f in files: + self.pengfei_tree.insert("", "end", values=(f,)) + + # 模糊搜索文件 + def do_fuzzy_search(self): + keyword = self.pengfei_zh_CN_search.get().strip().lower() + + if not self.pengfei_all_files_cache: + self.log("⚠️ 未检测到文件缓存，正在自动加载...") + self.start_load_files() + return + + for i in self.pengfei_tree.get_children(): + self.pengfei_tree.delete(i) + + if not keyword: + filtered = self.pengfei_all_files_cache + else: + filtered = [f for f in self.pengfei_all_files_cache if keyword in f.lower()] + + for f in filtered: + self.pengfei_tree.insert("", "end", values=(f,)) + self.log(f"🔍 模糊搜索结果：{len(filtered)} 个文件") + + # 下载选中的文件 + def do_download(self): + pengfei_sel = self.pengfei_tree.selection() + if not pengfei_sel: + messagebox.showwarning("提示", "请选择文件") + return + path = filedialog.askdirectory() + if not path: + return + key = self.pengfei_tree.item(pengfei_sel[0])["values"][0] + ok = download_file( + self.pengfei_zh_CN_ak.get(), self.pengfei_zh_CN_sk.get(), + self.pengfei_zh_CN_end.get(), self.pengfei_zh_CN_bucket.get(), + key, path + ) + if ok: + self.log(f"📥 下载完成：{key}") + else: + self.log(f"❌ 下载失败：{key}") + + +# 下载单个FDS中的文件 +def download_file(ak, sk, endpoint, bucket, key, save_path): + try: + pengfei_res = requests.get( + f"https://{endpoint}/{bucket}/{quote(key)}", + headers=auth_header(ak, sk, "GET", bucket, key), + stream=True, + timeout=30 + ) + filename = key.split("/")[-1] + with open(os.path.join(save_path, filename), "wb") as f: + for chunk in pengfei_res.iter_content(1024 * 1024): + f.write(chunk) + return True + except: + return False + + +if __name__ == "__main__": + app = FDSClient() + app.mainloop()