([^<]+)
', html) for b in blocks: name = b[0].replace(" - 玩偶哥哥", "").strip() pan_url = b[1].strip() if pan_url: tracks.append({"name": name, "pan": pan_url}) self._json({"tracks": tracks}) except Exception as e: self._json({"error": str(e), "tracks": []}) elif parsed.path == "/api/quark-play": # 夸克网盘完整播放流程：解析分享链接 → 转存 → 获取m3u8地址 qs = urllib.parse.parse_qs(parsed.query) share_url = qs.get("share_url", [""])[0] fid = qs.get("fid", [""])[0] if not share_url or not fid: self._json({"error": "missing share_url or fid"}) return try: # 从分享链接提取pwd_id m = re.search(r'/s/([a-zA-Z0-9]+)', share_url) if not m: self._json({"error": "invalid share url"}) return pwd_id = m.group(1) # 读取cookies cookies_file = r"D:\doubao\tmp\酱腌菜电商看板\pan_cookies.json" try: with open(cookies_file, "r", encoding="utf-8") as f: cookies = json.load(f) except: cookies = {} quark_cookie = cookies.get("quark", "") if not quark_cookie: self._json({"error": "quark cookies not set"}) return headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Content-Type": "application/json", "Referer": "https://pan.quark.cn/", "Cookie": quark_cookie, } # 第一步：获取stoken token_url = "https://drive-h.quark.cn/1/clouddrive/share/sharepage/token?pr=ucpro&fr=pc" token_body = {"pwd_id": pwd_id, "passcode": ""} req = urllib.request.Request(token_url, data=json.dumps(token_body).encode("utf-8"), headers=headers, method="POST") with urllib.request.urlopen(req, timeout=10) as resp: token_result = json.loads(resp.read().decode("utf-8")) stoken = token_result.get("data", {}).get("stoken", "") if not stoken: self._json({"error": "failed to get stoken"}) return # 第二步：暂时跳过转存，直接用传入的fid # （假设fid已经是用户网盘里的fid） saved_fid = fid # 第三步：调用播放API获取m3u8地址 play_url = "https://drive-pc.quark.cn/1/clouddrive/file/v2/play?pr=ucpro&fr=pc&uc_param_str=" play_body = {"fid": saved_fid} req = urllib.request.Request(play_url, data=json.dumps(play_body).encode("utf-8"), headers=headers, method="POST") with urllib.request.urlopen(req, timeout=10) as resp: play_result = json.loads(resp.read().decode("utf-8")) # 提取m3u8地址 m3u8_url = "" video_list = play_result.get("data", {}).get("video_list", []) for v in video_list: if v.get("accessable") and v.get("video_info", {}).get("url"): m3u8_url = v.get("video_info", {}).get("url", "") break if not m3u8_url: self._json({"error": "failed to get m3u8 url", "detail": play_result}) return self._json({ "m3u8_url": m3u8_url, "file_name": play_result.get("data", {}).get("file_name", ""), "size": play_result.get("data", {}).get("size", 0), }) except Exception as e: self._json({"error": str(e)}) elif parsed.path == "/api/aliyun-qrcode": # 生成阿里云盘扫码登录二维码 try: import http.cookiejar cj = http.cookiejar.CookieJar() opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj)) headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", } # 第一步：访问authorize获取SESSIONID req1 = urllib.request.Request( "https://auth.aliyundrive.com/v2/oauth/authorize?login_type=custom&response_type=code&redirect_uri=https%3A%2F%2Fwww.aliyundrive.com%2Fsign%2Fcallback&client_id=25dzX3vbYqktVxyX&state=%7B%22origin%22%3A%22file%3A%2F%2F%22%7D", headers=headers ) opener.open(req1, timeout=10).close() # 第二步：生成二维码 req2 = urllib.request.Request( "https://passport.aliyundrive.com/newlogin/qrcode/generate.do?appName=aliyun_drive", headers=headers ) resp2 = opener.open(req2, timeout=10) data2 = json.loads(resp2.read().decode("utf-8")) qr_data = data2.get("content", {}).get("data", {}) # 保存cookie和qr_data到session（全局存储） import threading import uuid if not hasattr(Handler, '_qr_sessions'): Handler._qr_sessions = {} Handler._qr_lock = threading.Lock() with Handler._qr_lock: sid = "aliyun_" + str(uuid.uuid4()) Handler._qr_sessions[sid] = { "cookies": "; ".join([f"{c.name}={c.value}" for c in cj]), "qr_data": qr_data, } self._json({ "qr_content": qr_data.get("codeContent", ""), "sid": sid, }) except Exception as e: self._json({"error": str(e)}) elif parsed.path == "/api/aliyun-qrcode-status": # 查询扫码状态 qs = urllib.parse.parse_qs(parsed.query) sid = qs.get("sid", [""])[0] if not sid: self._json({"error": "missing sid"}) return try: if not hasattr(Handler, '_qr_sessions') or sid not in Handler._qr_sessions: self._json({"error": "session expired"}) return session = Handler._qr_sessions[sid] headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Content-Type": "application/x-www-form-urlencoded", "Cookie": session["cookies"], } # POST查询状态 post_data = urllib.parse.urlencode(session["qr_data"]).encode("utf-8") req = urllib.request.Request( "https://passport.aliyundrive.com/newlogin/qrcode/query.do?appName=aliyun_drive", data=post_data, headers=headers, method="POST" ) resp = urllib.request.urlopen(req, timeout=10) result = json.loads(resp.read().decode("utf-8")) login_data = result.get("content", {}).get("data", {}) status = login_data.get("qrCodeStatus", "") if status == "CONFIRMED": # 扫码成功，提取refresh_token biz_ext = login_data.get("bizExt", "") if biz_ext: try: biz = json.loads(biz_ext) refresh_token = biz.get("pds_login_result", {}).get("refreshToken", "") if refresh_token: # 保存到cookies文件 cookies_file = r"D:\doubao\tmp\酱腌菜电商看板\pan_cookies.json" try: with open(cookies_file, "r", encoding="utf-8") as f: cookies = json.load(f) except: cookies = {} cookies["aliyun_refresh_token"] = refresh_token with open(cookies_file, "w", encoding="utf-8") as f: json.dump(cookies, f, ensure_ascii=False, indent=2) self._json({"status": "confirmed", "refresh_token": refresh_token}) return except: pass self._json({"status": "confirmed"}) elif status == "SCANED": self._json({"status": "scanned"}) else: self._json({"status": "waiting"}) except Exception as e: self._json({"error": str(e)}) elif parsed.path == "/api/quark-qrcode": # 生成夸克网盘扫码登录二维码 try: import uuid # 从文件里读取我们从夸克页面提取的实时token和request_id session_file = r"D:\doubao\tmp\酱腌菜电商看板\quark_session.json" try: with open(session_file, "r", encoding="utf-8") as f: session_data = json.load(f) token = session_data.get("token", "") request_id = session_data.get("request_id", "") timestamp = session_data.get("timestamp", 0) except: token = "" request_id = "" timestamp = 0 if not token: # 如果没有session文件，就自己调用API生成 request_id = str(uuid.uuid4()) headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Referer": "https://pan.quark.cn/", } req = urllib.request.Request( f"https://uop.quark.cn/cas/ajax/getTokenForQrcodeLogin?client_id=532&v=1.2&request_id={request_id}", headers=headers ) resp = urllib.request.urlopen(req, timeout=10) data = json.loads(resp.read().decode("utf-8")) token = data.get("data", {}).get("members", {}).get("token", "") timestamp = time.time() if not token: self._json({"error": "获取token失败: " + str(data)}) return # 保存session import threading if not hasattr(self, '_qr_sessions'): Handler._qr_sessions = {} Handler._qr_lock = threading.Lock() with Handler._qr_lock: sid = "quark_" + str(uuid.uuid4()) Handler._qr_sessions[sid] = { "type": "quark", "token": token, "request_id": request_id, "timestamp": timestamp, } # 直接返回二维码图片的URL和sid self._json({ "qr_image": "/real_quark_qr.png?v=1", "sid": sid, }) except Exception as e: self._json({"error": str(e)}) elif parsed.path == "/api/quark-qrcode-status": # 查询夸克扫码状态 qs = urllib.parse.parse_qs(parsed.query) sid = qs.get("sid", [""])[0] if not sid: self._json({"error": "missing sid"}) return try: if not hasattr(Handler, '_qr_sessions') or sid not in Handler._qr_sessions: self._json({"error": "session expired"}) return session = Handler._qr_sessions[sid] headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Referer": "https://pan.quark.cn/", } req = urllib.request.Request( f"https://uop.quark.cn/cas/ajax/getServiceTicketByQrcodeToken?client_id=532&v=1.2&request_id={session['request_id']}&token={session['token']}", headers=headers ) resp = urllib.request.urlopen(req, timeout=10) result = json.loads(resp.read().decode("utf-8")) # 解析状态 data = result.get("data", {}) members = data.get("members", {}) status = members.get("status", 0) ticket = members.get("ticket", "") if ticket: # 扫码成功 cookies_file = r"D:\doubao\tmp\酱腌菜电商看板\pan_cookies.json" try: with open(cookies_file, "r", encoding="utf-8") as f: cookies = json.load(f) except: cookies = {} cookies["quark_ticket"] = ticket with open(cookies_file, "w", encoding="utf-8") as f: json.dump(cookies, f, ensure_ascii=False, indent=2) self._json({"status": "confirmed", "ticket": ticket}) elif status == 1: self._json({"status": "scanned"}) else: self._json({"status": "waiting"}) except Exception as e: self._json({"error": str(e)}) elif parsed.path == "/api/baidu-qrcode": # 生成百度网盘扫码登录二维码 try: import time as _time t = int(_time.time() * 1000) headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Referer": "https://pan.baidu.com/", } req = urllib.request.Request( f"https://passport.baidu.com/v2/api/getqrcode?lp=pc&client=0&login_version=2&u=https%3A%2F%2Fpan.baidu.com%2F&t={t}", headers=headers ) resp = urllib.request.urlopen(req, timeout=10) data = json.loads(resp.read().decode("utf-8")) imgurl = data.get("imgurl", "") sign = data.get("sign", "") if not sign: self._json({"error": "获取sign失败: " + str(data)}) return # 保存session import threading import uuid if not hasattr(Handler, '_qr_sessions'): Handler._qr_sessions = {} Handler._qr_lock = threading.Lock() with Handler._qr_lock: sid = "baidu_" + str(uuid.uuid4()) Handler._qr_sessions[sid] = { "type": "baidu", "sign": sign, "imgurl": "https://" + imgurl, } self._json({ "qr_img": "https://" + imgurl, "sid": sid, }) except Exception as e: self._json({"error": str(e)}) elif parsed.path == "/api/baidu-qrcode-status": # 查询百度扫码状态 qs = urllib.parse.parse_qs(parsed.query) sid = qs.get("sid", [""])[0] if not sid: self._json({"error": "missing sid"}) return try: if not hasattr(Handler, '_qr_sessions') or sid not in Handler._qr_sessions: self._json({"error": "session expired"}) return session = Handler._qr_sessions[sid] headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Referer": "https://pan.baidu.com/", } req = urllib.request.Request( f"https://passport.baidu.com/channel/unicast?channel_id={session['sign']}&callback=this_is_callback&t={int(__import__('time').time()*1000)}", headers=headers ) try: resp = urllib.request.urlopen(req, timeout=5) result_text = resp.read().decode("utf-8") # 解析JSONP m = re.search(r'this_is_callback\((.*)\)', result_text) if m: result = json.loads(m.group(1)) channel_v = result.get("channel_v", "{}") try: cv = json.loads(channel_v) status = cv.get("status", 0) if status == 0: self._json({"status": "waiting"}) elif status == 1: self._json({"status": "scanned"}) else: # 扫码成功 bduss = cv.get("v", "") cookies_file = r"D:\doubao\tmp\酱腌菜电商看板\pan_cookies.json" try: with open(cookies_file, "r", encoding="utf-8") as f: cookies = json.load(f) except: cookies = {} cookies["baidu_bduss"] = bduss with open(cookies_file, "w", encoding="utf-8") as f: json.dump(cookies, f, ensure_ascii=False, indent=2) self._json({"status": "confirmed", "bduss": bduss}) except: self._json({"status": "waiting"}) else: self._json({"status": "waiting"}) except Exception as te: # 超时是正常的，说明还在等待扫码 self._json({"status": "waiting"}) except Exception as e: self._json({"error": str(e)}) elif parsed.path == "/api/pan-cookies": # 获取或设置网盘cookies cookies_file = r"D:\doubao\tmp\酱腌菜电商看板\pan_cookies.json" if self.command == "GET": try: with open(cookies_file, "r", encoding="utf-8") as f: self._json(json.load(f)) except: self._json({"aliyun": "", "quark": "", "uc": "", "baidu": ""}) else: try: length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(length) data = json.loads(body.decode("utf-8")) with open(cookies_file, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) self._json({"ok": True}) except Exception as e: self._json({"error": str(e)}, code=500) elif parsed.path == "/api/pan-parse": # 解析网盘分享链接 qs = urllib.parse.parse_qs(parsed.query) pan_url = qs.get("url", [""])[0] if not pan_url: self._json({"error": "missing url"}, code=400) return # 读取cookies cookies_file = r"D:\doubao\tmp\酱腌菜电商看板\pan_cookies.json" try: with open(cookies_file, "r", encoding="utf-8") as f: cookies = json.load(f) except: cookies = {} try: if "alipan.com" in pan_url or "aliyundrive.com" in pan_url: result = self._parse_aliyun(pan_url, cookies.get("aliyun", "")) elif "quark.cn" in pan_url or "quark.com" in pan_url: result = self._parse_quark(pan_url, cookies.get("quark", "")) elif "uc.cn" in pan_url: result = self._parse_uc(pan_url, cookies.get("uc", "")) elif "pan.baidu.com" in pan_url or "baidu.com" in pan_url: result = self._parse_baidu(pan_url, cookies.get("baidu", "")) else: result = {"error": "不支持的网盘类型"} self._json(result) except Exception as e: self._json({"error": str(e), "files": []}) else: self._json({"error": "not found"}, code=404) def do_POST(self): parsed = urllib.parse.urlparse(self.path) if parsed.path.startswith("/lxmusic/"): if _lx is None: self.send_error(503, "lxmusic not loaded") return self._json = lambda status, payload: _lx.Handler._json(self, status, payload) self._text = lambda t, ct="text/plain; charset=utf-8": _lx.Handler._text(self, t, ct) _lx.Handler.do_POST(self) return elif parsed.path == "/api/pan-cookies": # 设置网盘cookies cookies_file = r"D:\doubao\tmp\酱腌菜电商看板\pan_cookies.json" try: length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(length) data = json.loads(body.decode("utf-8")) with open(cookies_file, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) self._json({"ok": True}) except Exception as e: self._json({"error": str(e)}, code=500) return self._json({"error": "not found"}, code=404) def _json(self, obj, code=200): body = json.dumps(obj, ensure_ascii=False).encode("utf-8") self.send_response(code) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def log_message(self, *args): pass # 静默 def _parse_aliyun(self, pan_url, cookie): """解析阿里云盘分享链接""" # 提取share_id m = re.search(r'/s/([A-Za-z0-9]+)', pan_url) if not m: return {"error": "无法解析分享ID", "files": []} share_id = m.group(1) # 获取share_token headers = { "User-Agent": HEADERS["User-Agent"], "Content-Type": "application/json", "Origin": "https://www.alipan.com", "Referer": "https://www.alipan.com/", } if cookie: headers["Cookie"] = cookie req = urllib.request.Request( "https://api.aliyundrive.com/v2/share_link/get_share_token", data=json.dumps({"share_id": share_id, "share_pwd": ""}).encode("utf-8"), headers=headers, method="POST" ) with urllib.request.urlopen(req, timeout=10) as resp: token_data = json.loads(resp.read().decode("utf-8")) share_token = token_data.get("share_token", "") if not share_token: return {"error": "获取分享token失败: " + str(token_data), "files": []} # 列出文件 headers["x-share-token"] = share_token req2 = urllib.request.Request( "https://api.aliyundrive.com/adrive/v1.0/share_link/list", data=json.dumps({ "share_id": share_id, "parent_file_id": token_data.get("share_file_id", ""), "limit": 100, "order_by": "name", "order_direction": "ASC" }).encode("utf-8"), headers=headers, method="POST" ) with urllib.request.urlopen(req2, timeout=10) as resp: list_data = json.loads(resp.read().decode("utf-8")) files = [] for f in list_data.get("items", []): files.append({ "name": f.get("name", ""), "file_id": f.get("file_id", ""), "type": f.get("type", ""), "size": f.get("size", 0), "category": f.get("category", ""), }) return {"files": files, "share_id": share_id, "share_token": share_token} def _parse_quark(self, pan_url, cookie): """解析夸克网盘分享链接""" import time m = re.search(r'/s/([A-Za-z0-9]+)', pan_url) if not m: return {"error": "无法解析分享ID", "files": []} pwd_id = m.group(1) headers = { "User-Agent": HEADERS["User-Agent"], "Content-Type": "application/json", "Origin": "https://pan.quark.cn", "Referer": f"https://pan.quark.cn/s/{pwd_id}", } if cookie: headers["Cookie"] = cookie # 获取stoken ts = int(time.time() * 1000) params1 = f"?pr=ucpro&fr=pc&__dt={ts%1000}&__t={ts}" req = urllib.request.Request( f"https://drive-h.quark.cn/1/clouddrive/share/sharepage/token{params1}", data=json.dumps({"pwd_id": pwd_id, "passcode": ""}).encode("utf-8"), headers=headers, method="POST" ) try: with urllib.request.urlopen(req, timeout=10) as resp: token_data = json.loads(resp.read().decode("utf-8")) except Exception as e: return {"error": f"获取stoken失败: {e}", "files": []} stoken = token_data.get("data", {}).get("stoken", "") if not stoken: return {"error": "获取stoken失败: " + str(token_data), "files": []} # 列出文件 ts2 = int(time.time() * 1000) params2 = f"?pr=ucpro&fr=pc&pwd_id={pwd_id}&pdir_fid=0&force=0&_page=1&_size=50&_fetch_banner=1&_fetch_share=1&_fetch_total=1&_sort=file_type:asc,file_name:asc&stoken={urllib.parse.quote(stoken)}&__dt={ts2%1000}&__t={ts2}" headers2 = { "User-Agent": HEADERS["User-Agent"], "Referer": f"https://pan.quark.cn/s/{pwd_id}", } if cookie: headers2["Cookie"] = cookie req2 = urllib.request.Request( f"https://drive-h.quark.cn/1/clouddrive/share/sharepage/detail{params2}", headers=headers2, method="GET" ) try: with urllib.request.urlopen(req2, timeout=10) as resp: list_data = json.loads(resp.read().decode("utf-8")) except Exception as e: return {"error": f"获取文件列表失败: {e}", "files": []} files = [] for f in list_data.get("data", {}).get("list", []): files.append({ "name": f.get("file_name", ""), "file_id": f.get("fid", ""), "type": "folder" if f.get("dir") else "file", "size": f.get("size", 0), "category": f.get("file_category", ""), }) return {"files": files, "pwd_id": pwd_id} def _parse_uc(self, pan_url, cookie): """解析UC网盘分享链接""" m = re.search(r'/s/([A-Za-z0-9]+)', pan_url) if not m: return {"error": "无法解析分享ID", "files": []} share_id = m.group(1) return {"error": "UC网盘解析暂未实现", "files": [], "share_id": share_id} def _parse_baidu(self, pan_url, cookie): """解析百度网盘分享链接""" m = re.search(r'/s/([A-Za-z0-9_-]+)', pan_url) if not m: return {"error": "无法解析分享ID", "files": []} share_id = m.group(1) # 提取提取码 pwd = "" pwd_match = re.search(r'pwd=([A-Za-z0-9]+)', pan_url) if pwd_match: pwd = pwd_match.group(1) headers = { "User-Agent": HEADERS["User-Agent"], "Content-Type": "application/x-www-form-urlencoded", "Referer": "https://pan.baidu.com/", } if cookie: headers["Cookie"] = cookie # 验证分享链接（带提取码） post_data = f"pwd={pwd}".encode("utf-8") req = urllib.request.Request( f"https://pan.baidu.com/share/verify?surl={share_id}&t=0&channel=chunlei&web=1&app_id=250528&clienttype=0", data=post_data, headers=headers, method="POST" ) with urllib.request.urlopen(req, timeout=10) as resp: verify_data = json.loads(resp.read().decode("utf-8")) sign = verify_data.get("data", {}).get("sign", "") timestamp = verify_data.get("data", {}).get("timestamp", "") if not sign: return {"error": f"百度网盘验证失败（errno: {verify_data.get('errno')}）", "files": []} # 列出文件 req2 = urllib.request.Request( f"https://pan.baidu.com/share/list?uk=&shareid=&surl={share_id}&sign={sign}×tamp={timestamp}&page=1&num=100&dir=/&order=name&desc=0&clienttype=0&app_id=250528&web=1", headers=headers, method="GET" ) with urllib.request.urlopen(req2, timeout=10) as resp: list_data = json.loads(resp.read().decode("utf-8")) files = [] for f in list_data.get("list", []): files.append({ "name": f.get("server_filename", ""), "file_id": f.get("fs_id", ""), "type": "folder" if f.get("isdir") else "file", "size": f.get("size", 0), "category": f.get("category", ""), "dlink": f.get("dlink", ""), }) return {"files": files, "share_id": share_id, "pwd": pwd} if __name__ == "__main__": server = ThreadingHTTPServer(("0.0.0.0", 8000), Handler) print("SOTVLA proxy on http://0.0.0.0:8000") server.serve_forever()