初次提交

simulator.py

@@ -0,0 +1,436 @@
+import tkinter as tk
+from tkinter import ttk, messagebox, scrolledtext
+import requests
+import threading
+import json
+
+class BYWSongRequestSimulator:
+    def __init__(self, root):
+        self.root = root
+        self.root.title("BYWVoiceHub 综合测试模拟器 (Full Features)")
+        self.root.geometry("950x700")
+        
+        self.base_url = "https://bywmusic.bzw-hheb.top/api"
+        self.session = requests.Session()
+        self.session.headers.update({
+            "User-Agent": "BYWVoiceHub-Simulator/2.0"
+        })
+        self.current_user = None
+        
+        self.create_widgets()
+        self.log("[*] 欢迎使用 BYWVoiceHub 全功能测试模拟器！")
+        self.log(f"[*] 目标域名: {self.base_url}")
+        
+    def log(self, message):
+        self.log_text.config(state="normal")
+        self.log_text.insert(tk.END, message + "\n")
+        self.log_text.see(tk.END)
+        self.log_text.config(state="disabled")
+        
+    def run_async(self, target, *args):
+        threading.Thread(target=target, args=args, daemon=True).start()
+
+    def create_widgets(self):
+        style = ttk.Style()
+        style.theme_use('clam')
+        
+        # Notebook for Tabs
+        self.notebook = ttk.Notebook(self.root)
+        self.notebook.pack(fill="both", expand=True, padx=10, pady=5)
+        
+        self.tab_auth = ttk.Frame(self.notebook)
+        self.tab_public = ttk.Frame(self.notebook)
+        self.tab_search = ttk.Frame(self.notebook)
+        self.tab_my = ttk.Frame(self.notebook)
+        
+        self.notebook.add(self.tab_auth, text=" 认证管理 (Auth) ")
+        self.notebook.add(self.tab_public, text=" 广场大厅 (Public) ")
+        self.notebook.add(self.tab_search, text=" 搜歌与投稿 (Search & Request) ")
+        self.notebook.add(self.tab_my, text=" 我的 (My Profile) ")
+        
+        self.setup_auth_tab()
+        self.setup_public_tab()
+        self.setup_search_tab()
+        self.setup_my_tab()
+        
+        # Log area at the bottom
+        log_frame = ttk.LabelFrame(self.root, text=" 运行日志 (Logs) ")
+        log_frame.pack(fill="x", padx=10, pady=5, side="bottom")
+        
+        self.log_text = scrolledtext.ScrolledText(log_frame, height=12, wrap="word", state="disabled", bg="#f4f4f4")
+        self.log_text.pack(fill="both", expand=True, padx=5, pady=5)
+
+    def setup_auth_tab(self):
+        # 登录
+        login_frame = ttk.LabelFrame(self.tab_auth, text=" 登录 ")
+        login_frame.pack(fill="x", padx=10, pady=5)
+        ttk.Label(login_frame, text="账号:").grid(row=0, column=0, padx=5, pady=5)
+        self.ent_login_user = ttk.Entry(login_frame, width=15)
+        self.ent_login_user.grid(row=0, column=1, padx=5, pady=5)
+        self.ent_login_user.insert(0, "admin")
+        
+        ttk.Label(login_frame, text="密码:").grid(row=0, column=2, padx=5, pady=5)
+        self.ent_login_pass = ttk.Entry(login_frame, show="*", width=15)
+        self.ent_login_pass.grid(row=0, column=3, padx=5, pady=5)
+        self.ent_login_pass.insert(0, "123456")
+        
+        ttk.Button(login_frame, text="登录", command=self.do_login).grid(row=0, column=4, padx=5, pady=5)
+        ttk.Button(login_frame, text="验证登录状态", command=self.do_verify).grid(row=0, column=5, padx=5, pady=5)
+
+        # 注册
+        reg_frame = ttk.LabelFrame(self.tab_auth, text=" 注册 ")
+        reg_frame.pack(fill="x", padx=10, pady=5)
+        ttk.Label(reg_frame, text="新账号:").grid(row=0, column=0, padx=5, pady=5)
+        self.ent_reg_user = ttk.Entry(reg_frame, width=15)
+        self.ent_reg_user.grid(row=0, column=1, padx=5, pady=5)
+        ttk.Label(reg_frame, text="新密码:").grid(row=0, column=2, padx=5, pady=5)
+        self.ent_reg_pass = ttk.Entry(reg_frame, show="*", width=15)
+        self.ent_reg_pass.grid(row=0, column=3, padx=5, pady=5)
+        ttk.Button(reg_frame, text="注册", command=self.do_register).grid(row=0, column=4, padx=5, pady=5)
+        
+        # 账号管理
+        profile_frame = ttk.LabelFrame(self.tab_auth, text=" 账号管理 (修改资料) ")
+        profile_frame.pack(fill="x", padx=10, pady=5)
+        ttk.Label(profile_frame, text="昵称:").grid(row=0, column=0, padx=5, pady=5)
+        self.ent_prof_name = ttk.Entry(profile_frame, width=15)
+        self.ent_prof_name.grid(row=0, column=1, padx=5, pady=5)
+        ttk.Label(profile_frame, text="年级:").grid(row=0, column=2, padx=5, pady=5)
+        self.ent_prof_grade = ttk.Entry(profile_frame, width=10)
+        self.ent_prof_grade.grid(row=0, column=3, padx=5, pady=5)
+        ttk.Label(profile_frame, text="班级:").grid(row=0, column=4, padx=5, pady=5)
+        self.ent_prof_class = ttk.Entry(profile_frame, width=10)
+        self.ent_prof_class.grid(row=0, column=5, padx=5, pady=5)
+        ttk.Button(profile_frame, text="更新资料", command=self.do_update_profile).grid(row=0, column=6, padx=5, pady=5)
+
+    def setup_public_tab(self):
+        btn_frame = ttk.Frame(self.tab_public)
+        btn_frame.pack(fill="x", padx=10, pady=5)
+        ttk.Button(btn_frame, text="获取播出排期 (Schedules)", command=self.get_schedules).pack(side="left", padx=5)
+        ttk.Button(btn_frame, text="获取公共歌曲列表 (Public Songs)", command=self.get_public_songs).pack(side="left", padx=5)
+        
+        self.tree_public = ttk.Treeview(self.tab_public, columns=("ID", "Title", "Artist", "Info"), show="headings", height=15)
+        self.tree_public.heading("ID", text="ID")
+        self.tree_public.heading("Title", text="歌曲名")
+        self.tree_public.heading("Artist", text="歌手")
+        self.tree_public.heading("Info", text="其他信息(排期/状态/投稿人)")
+        self.tree_public.column("ID", width=50)
+        self.tree_public.column("Title", width=200)
+        self.tree_public.column("Artist", width=150)
+        self.tree_public.column("Info", width=400)
+        
+        scrollbar = ttk.Scrollbar(self.tab_public, orient=tk.VERTICAL, command=self.tree_public.yview)
+        self.tree_public.configure(yscroll=scrollbar.set)
+        
+        self.tree_public.pack(side="left", fill="both", expand=True, padx=10, pady=5)
+        scrollbar.pack(side="right", fill="y", pady=5)
+
+    def setup_search_tab(self):
+        # 搜索与链接解析
+        search_frame = ttk.LabelFrame(self.tab_search, text=" 搜歌与解析 ")
+        search_frame.pack(fill="x", padx=10, pady=5)
+        
+        ttk.Label(search_frame, text="关键词/链接:").grid(row=0, column=0, padx=5, pady=5)
+        self.ent_search = ttk.Entry(search_frame, width=30)
+        self.ent_search.grid(row=0, column=1, padx=5, pady=5)
+        self.ent_search.insert(0, "周杰伦")
+        
+        self.cb_platform = ttk.Combobox(search_frame, values=["网易云", "酷狗", "哔哩哔哩", "酷狗链接解析"], width=12, state="readonly")
+        self.cb_platform.current(0)
+        self.cb_platform.grid(row=0, column=2, padx=5, pady=5)
+        
+        ttk.Button(search_frame, text="搜索 / 解析", command=self.do_search).grid(row=0, column=3, padx=5, pady=5)
+        
+        # 结果列表
+        self.tree_search = ttk.Treeview(self.tab_search, columns=("Title", "Artist", "ID", "Platform", "Ext"), show="headings", height=8)
+        self.tree_search.heading("Title", text="歌曲名")
+        self.tree_search.heading("Artist", text="歌手")
+        self.tree_search.heading("ID", text="歌曲ID")
+        self.tree_search.heading("Platform", text="平台")
+        self.tree_search.heading("Ext", text="附加信息")
+        self.tree_search.column("ID", width=150)
+        self.tree_search.column("Ext", width=100)
+        
+        scrollbar = ttk.Scrollbar(self.tab_search, orient=tk.VERTICAL, command=self.tree_search.yview)
+        self.tree_search.configure(yscroll=scrollbar.set)
+        
+        self.tree_search.pack(fill="both", expand=True, padx=10, pady=5)
+        
+        # 投稿
+        req_frame = ttk.LabelFrame(self.tab_search, text=" 投稿 (点歌) ")
+        req_frame.pack(fill="x", padx=10, pady=5)
+        ttk.Label(req_frame, text="留言:").grid(row=0, column=0, padx=5, pady=5)
+        self.ent_note = ttk.Entry(req_frame, width=40)
+        self.ent_note.grid(row=0, column=1, padx=5, pady=5)
+        self.ent_note.insert(0, "来自模拟器的全功能测试投稿")
+        ttk.Button(req_frame, text="提交选中歌曲", command=self.do_request).grid(row=0, column=2, padx=5, pady=5)
+
+    def setup_my_tab(self):
+        btn_frame = ttk.Frame(self.tab_my)
+        btn_frame.pack(fill="x", padx=10, pady=5)
+        ttk.Button(btn_frame, text="获取我的投稿歌曲页 (My Requests)", command=self.get_my_requests).pack(side="left", padx=5)
+        ttk.Button(btn_frame, text="获取我的消息页 (Notifications)", command=self.get_my_notifications).pack(side="left", padx=5)
+        
+        self.tree_my = ttk.Treeview(self.tab_my, columns=("Title", "Content", "Date", "Status"), show="headings", height=15)
+        self.tree_my.heading("Title", text="标题/类型")
+        self.tree_my.heading("Content", text="内容/歌曲")
+        self.tree_my.heading("Date", text="日期")
+        self.tree_my.heading("Status", text="状态")
+        self.tree_my.column("Title", width=150)
+        self.tree_my.column("Content", width=350)
+        self.tree_my.column("Date", width=150)
+        self.tree_my.column("Status", width=100)
+        
+        scrollbar = ttk.Scrollbar(self.tab_my, orient=tk.VERTICAL, command=self.tree_my.yview)
+        self.tree_my.configure(yscroll=scrollbar.set)
+        
+        self.tree_my.pack(fill="both", expand=True, padx=10, pady=5)
+
+    # ================= 认证相关 =================
+    def do_login(self):
+        def _run():
+            u, p = self.ent_login_user.get().strip(), self.ent_login_pass.get().strip()
+            self.log(f"\n[*] 正在登录: {u}")
+            try:
+                res = self.session.post(f"{self.base_url}/auth/login", json={"username": u, "password": p})
+                if res.status_code == 200:
+                    data = res.json()
+                    if data.get("success"):
+                        self.current_user = data.get('user', {})
+                        self.log(f"[+] 登录成功！欢迎: {self.current_user.get('name', u)}")
+                        # Fill profile
+                        self.ent_prof_name.delete(0, tk.END)
+                        self.ent_prof_name.insert(0, self.current_user.get('name', ''))
+                        self.ent_prof_grade.delete(0, tk.END)
+                        self.ent_prof_grade.insert(0, self.current_user.get('grade', '') or '')
+                        self.ent_prof_class.delete(0, tk.END)
+                        self.ent_prof_class.insert(0, self.current_user.get('class', '') or '')
+                    else:
+                        self.log(f"[-] 登录失败 (可能需要2FA): {data}")
+                else:
+                    self.log(f"[-] 登录失败: {res.text}")
+            except Exception as e:
+                self.log(f"[!] 错误: {e}")
+        self.run_async(_run)
+
+    def do_register(self):
+        def _run():
+            u, p = self.ent_reg_user.get().strip(), self.ent_reg_pass.get().strip()
+            if not u or not p:
+                self.log("[!] 账号和密码不能为空！")
+                return
+            self.log(f"\n[*] 正在注册: {u}")
+            try:
+                res = self.session.post(f"{self.base_url}/auth/register", json={"username": u, "password": p})
+                if res.status_code == 200:
+                    self.log(f"[+] 注册成功！并且已自动登录。")
+                else:
+                    self.log(f"[-] 注册失败: {res.text}")
+            except Exception as e:
+                self.log(f"[!] 错误: {e}")
+        self.run_async(_run)
+
+    def do_verify(self):
+        def _run():
+            self.log("\n[*] 验证登录状态...")
+            try:
+                res = self.session.get(f"{self.base_url}/auth/verify")
+                if res.status_code == 200:
+                    data = res.json()
+                    self.log(f"[+] 登录有效。当前用户: {data.get('user', {}).get('username')}")
+                else:
+                    self.log(f"[-] 未登录或令牌失效: {res.text}")
+            except Exception as e:
+                self.log(f"[!] 错误: {e}")
+        self.run_async(_run)
+
+    def do_update_profile(self):
+        def _run():
+            payload = {
+                "name": self.ent_prof_name.get(),
+                "grade": self.ent_prof_grade.get(),
+                "class": self.ent_prof_class.get()
+            }
+            self.log("\n[*] 正在更新账号资料...")
+            try:
+                res = self.session.put(f"{self.base_url}/user/profile", json=payload)
+                if res.status_code == 200:
+                    self.log("[+] 资料更新成功！")
+                else:
+                    self.log(f"[-] 更新失败: {res.text}")
+            except Exception as e:
+                self.log(f"[!] 错误: {e}")
+        self.run_async(_run)
+
+    # ================= 广场相关 =================
+    def get_schedules(self):
+        def _run():
+            self.log("\n[*] 获取播出排期 (Open API)...")
+            try:
+                res = self.session.get(f"{self.base_url}/open/schedules")
+                if res.status_code == 200:
+                    data = res.json()
+                    schedules = data.get("data", {}).get("schedules", [])
+                    self.log(f"[+] 成功获取 {len(schedules)} 条排期。")
+                    for item in self.tree_public.get_children():
+                        self.tree_public.delete(item)
+                    for s in schedules:
+                        song = s.get("song", {})
+                        info = f"日期:{s.get('playDate')} 序号:{s.get('sequence')} 播放:{s.get('played')}"
+                        self.tree_public.insert("", "end", values=(s.get('id'), song.get('title'), song.get('artist'), info))
+                else:
+                    self.log(f"[-] 获取排期失败: {res.text}")
+            except Exception as e:
+                self.log(f"[!] 错误: {e}")
+        self.run_async(_run)
+
+    def get_public_songs(self):
+        def _run():
+            self.log("\n[*] 获取公共歌曲列表 (Open API)...")
+            try:
+                res = self.session.get(f"{self.base_url}/open/songs", params={"limit": 50})
+                if res.status_code == 200:
+                    data = res.json()
+                    songs = data.get("data", {}).get("songs", [])
+                    self.log(f"[+] 成功获取 {len(songs)} 首公共歌曲。")
+                    for item in self.tree_public.get_children():
+                        self.tree_public.delete(item)
+                    for s in songs:
+                        req = s.get("requester", {})
+                        req_name = req.get("name", "未知") if req else "未知"
+                        info = f"排期:{s.get('scheduled')} 播放:{s.get('played')} 投稿人:{req_name}"
+                        self.tree_public.insert("", "end", values=(s.get('id'), s.get('title'), s.get('artist'), info))
+                else:
+                    self.log(f"[-] 获取歌曲列表失败: {res.text}")
+            except Exception as e:
+                self.log(f"[!] 错误: {e}")
+        self.run_async(_run)
+
+    # ================= 搜歌与点歌 =================
+    def do_search(self):
+        def _run():
+            kw = self.ent_search.get().strip()
+            plat = self.cb_platform.get()
+            if not kw:
+                self.log("[!] 请输入搜索内容！")
+                return
+                
+            self.log(f"\n[*] 正在搜索: '{kw}' [{plat}]...")
+            for item in self.tree_search.get_children():
+                self.tree_search.delete(item)
+                
+            try:
+                if plat == "网易云":
+                    res = self.session.get(f"{self.base_url}/native-api/search/wy", params={"str": kw, "limit": 15})
+                    if res.status_code == 200:
+                        songs = res.json().get("list", [])
+                        for s in songs:
+                            self.tree_search.insert("", "end", values=(s.get("name"), s.get("singer"), s.get("songmid"), "netease", ""))
+                        self.log(f"[+] 网易云找到 {len(songs)} 首歌")
+                elif plat == "酷狗":
+                    res = self.session.get(f"{self.base_url}/music/kg/search", params={"keyword": kw})
+                    if res.status_code == 200:
+                        songs = res.json().get("data", {}).get("info", [])
+                        for s in songs:
+                            self.tree_search.insert("", "end", values=(s.get("songname"), s.get("singername"), s.get("hash"), "kugou", ""))
+                        self.log(f"[+] 酷狗找到 {len(songs)} 首歌")
+                elif plat == "哔哩哔哩":
+                    res = self.session.get(f"{self.base_url}/bilibili/search", params={"keyword": kw})
+                    if res.status_code == 200:
+                        songs = res.json()
+                        for s in songs:
+                            self.tree_search.insert("", "end", values=(s.get("title"), s.get("author"), s.get("bvid"), "bilibili", f"CID:{s.get('cid','')}"))
+                        self.log(f"[+] B站找到 {len(songs)} 个视频")
+                elif plat == "酷狗链接解析":
+                    res = self.session.get(f"{self.base_url}/music/kg/parse-link", params={"url": kw})
+                    if res.status_code == 200:
+                        s = res.json()
+                        self.tree_search.insert("", "end", values=(s.get("title"), s.get("artist"), s.get("musicId"), "kugou", "解析成功"))
+                        self.log(f"[+] 链接解析成功！")
+                    else:
+                        self.log(f"[-] 链接解析失败: {res.text}")
+            except Exception as e:
+                self.log(f"[!] 错误: {e}")
+        self.run_async(_run)
+
+    def do_request(self):
+        selected = self.tree_search.selection()
+        if not selected:
+            messagebox.showwarning("提示", "请先在上方搜索并选中一首歌曲！")
+            return
+            
+        item = self.tree_search.item(selected[0])
+        title, artist, music_id, platform, ext = item['values']
+        note = self.ent_note.get().strip()
+        
+        def _run():
+            self.log(f"\n[*] 尝试提交点歌: 《{title}》- {artist} ...")
+            payload = {
+                "title": title,
+                "artist": artist,
+                "musicPlatform": platform,
+                "musicId": str(music_id),
+                "submissionNote": note,
+                "submissionNotePublic": True
+            }
+            # 如果是 B 站，尝试把 CID 放进去
+            if platform == 'bilibili' and "CID:" in ext:
+                cid = ext.replace("CID:", "").strip()
+                if cid: payload["bilibiliCid"] = cid
+                
+            try:
+                res = self.session.post(f"{self.base_url}/songs/request", json=payload)
+                if res.status_code == 200:
+                    self.log(f"[+] 点歌成功！歌曲已加入列表。")
+                else:
+                    try:
+                        err_msg = res.json().get('message', res.text)
+                    except:
+                        err_msg = res.text
+                    self.log(f"[-] 点歌失败 (HTTP {res.status_code}): {err_msg}")
+            except Exception as e:
+                self.log(f"[!] 错误: {e}")
+        self.run_async(_run)
+
+    # ================= 我的 =================
+    def get_my_requests(self):
+        def _run():
+            self.log("\n[*] 获取我的投稿歌曲页 (My Requests)...")
+            try:
+                res = self.session.get(f"{self.base_url}/songs/index", params={"scope": "mine"})
+                if res.status_code == 200:
+                    songs = res.json().get("data", {}).get("songs", [])
+                    self.log(f"[+] 成功获取，共 {len(songs)} 首。")
+                    for item in self.tree_my.get_children():
+                        self.tree_my.delete(item)
+                    for s in songs:
+                        status = "已播" if s.get("played") else ("排期" if s.get("scheduled") else "等待")
+                        self.tree_my.insert("", "end", values=("我的点歌", f"《{s.get('title')}》- {s.get('artist')}", s.get("requestedAt", ""), status))
+                else:
+                    self.log(f"[-] 获取失败: {res.text}")
+            except Exception as e:
+                self.log(f"[!] 错误: {e}")
+        self.run_async(_run)
+
+    def get_my_notifications(self):
+        def _run():
+            self.log("\n[*] 获取我的消息页 (Notifications)...")
+            try:
+                res = self.session.get(f"{self.base_url}/notifications/index")
+                if res.status_code == 200:
+                    notifs = res.json().get("notifications", [])
+                    self.log(f"[+] 成功获取，共 {len(notifs)} 条通知。")
+                    for item in self.tree_my.get_children():
+                        self.tree_my.delete(item)
+                    for n in notifs:
+                        status = "已读" if n.get("read") else "未读"
+                        self.tree_my.insert("", "end", values=(n.get("type", "通知"), n.get("message", ""), n.get("createdAt", "")[:10], status))
+                else:
+                    self.log(f"[-] 获取失败: {res.text}")
+            except Exception as e:
+                self.log(f"[!] 错误: {e}")
+        self.run_async(_run)
+
+if __name__ == "__main__":
+    root = tk.Tk()
+    app = BYWSongRequestSimulator(root)
+    root.mainloop()