Add Xiaomi FDS Client GUI functionality

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 <zhupengfei7@xiaomi.com>
+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()