diff --git a/capture/live.py b/capture/live.py
deleted file mode 100644
index e892f76..0000000
--- a/capture/live.py
+++ /dev/null
@@ -1,195 +0,0 @@
-"""把 Dart 发出的明文 HTTP 显示在本机页面。
-
-Charles 看不到这些请求：它们不走系统代理，在进系统套接字之前就已经加密。
-这里直接从进程内存里取出加密前的报文。
-"""
-import json
-import threading
-from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
-
-import frida
-
-DEVICE = "emulator-5556"
-PORT = 8791
-APP_NAMES = ("RaptDrama",)
-
-JS = r"""
-function extract(addr) {
- var back = addr;
- for (var i = 0; i < 160; i++) {
- var b = 0;
- try { b = back.sub(1).readU8(); } catch (e) { break; }
- if (b !== 0x0a && b !== 0x0d && (b < 0x20 || b > 0x7e)) break;
- back = back.sub(1);
- }
- var s = "";
- for (var j = 0; j < 2500; j++) {
- var c = 0;
- try { c = back.add(j).readU8(); } catch (e) { break; }
- if (c === 0x0d) continue;
- if (c === 0x0a) { s += "\n"; continue; }
- if (c < 0x20 || c > 0x7e) break;
- s += String.fromCharCode(c);
- }
- return s;
-}
-
-var seen = {};
-var scanning = false;
-var lastScan = 0;
-
-function scan() {
- var now = Date.now();
- if (scanning || now - lastScan < 1200) return;
- scanning = true;
- lastScan = now;
- var ranges = Process.enumerateRanges("rw-");
- var fresh = [];
- for (var i = 0; i < ranges.length; i++) {
- var r = ranges[i];
- if (r.size > 24 * 1024 * 1024) continue;
- var matches;
- try { matches = Memory.scanSync(r.base, r.size, "48 54 54 50 2f 31 2e 31"); }
- catch (e) { continue; }
- for (var j = 0; j < matches.length; j++) {
- var text = extract(matches[j].address);
- if (!text || text.length < 16 || seen[text]) continue;
- if (text.indexOf("package:") === 0) continue;
- seen[text] = true;
- fresh.push(text);
- }
- }
- if (fresh.length) send(JSON.stringify(fresh));
- scanning = false;
-}
-
-var writePtr = Process.getModuleByName("libc.so").getExportByName("write");
-Interceptor.attach(writePtr, {
- onEnter: function (args) {
- try {
- if (args[1].readU8() === 0x17) scan();
- } catch (e) {}
- }
-});
-scan();
-send("ready");
-"""
-
-FLOWS = []
-LOCK = threading.Lock()
-
-
-def on_message(message, _data):
- if message.get("type") != "send":
- print(message, flush=True)
- return
- payload = message["payload"]
- if payload == "ready":
- print("已挂上进程，等待请求", flush=True)
- return
- items = json.loads(payload)
- with LOCK:
- for text in items:
- first = text.strip().split("\n", 1)[0][:180]
- FLOWS.append({"id": len(FLOWS) + 1, "summary": first, "text": text})
- print(first, flush=True)
-
-
-def attach():
- device = frida.get_device(DEVICE)
- pid = None
- for proc in device.enumerate_processes():
- if proc.name in APP_NAMES or "Rapt" in proc.name:
- pid = proc.pid
- break
- if pid is None:
- raise SystemExit("RaptDrama 没在运行。先打开 App，再启动这个页面。")
- print(f"attach {pid}", flush=True)
- session = device.attach(pid)
- script = session.create_script(JS)
- script.on("message", on_message)
- script.load()
- return session, script
-
-
-PAGE = """
-
-
-
- 明文请求
-
-
-
-
-
-
- 打开 App 并滑动页面，新的请求会出现在左边。
-
-
-
-
-"""
-
-
-class Handler(BaseHTTPRequestHandler):
- def do_GET(self):
- if self.path.startswith("/api/flows"):
- with LOCK:
- payload = json.dumps(FLOWS, ensure_ascii=False).encode("utf-8")
- self.send_response(200)
- self.send_header("Content-Type", "application/json; charset=utf-8")
- self.send_header("Content-Length", str(len(payload)))
- self.end_headers()
- self.wfile.write(payload)
- return
- body = PAGE.encode("utf-8")
- self.send_response(200)
- self.send_header("Content-Type", "text/html; charset=utf-8")
- self.send_header("Content-Length", str(len(body)))
- self.end_headers()
- self.wfile.write(body)
-
- def log_message(self, fmt, *args):
- return
-
-
-def main():
- attach()
- server = ThreadingHTTPServer(("127.0.0.1", PORT), Handler)
- print(f"打开 http://127.0.0.1:{PORT}", flush=True)
- server.serve_forever()
-
-
-if __name__ == "__main__":
- main()
diff --git a/rapt_scraper.py b/rapt_scraper.py
deleted file mode 100644
index da7c567..0000000
--- a/rapt_scraper.py
+++ /dev/null
@@ -1,660 +0,0 @@
-"""
-RaptDrama 首页目录抓取。
-
-抓取顶部分类（Comedy、Passion、Newest、Western、Comic drama）、
-New Releases，以及 Recommend 的全部分页。同一部剧只拉一次播放地址，
-并用 sections 标记它出现在哪些区块。
-
-首页 Trending Now 没有单独的 open 接口，安装包里的 extra_broadcast
-对不上截图标题，因此不写入。/app/open/foryou 按要求不调用。
-
-默认用游客 autologin。也可以用邮箱和密码调用 /app/open/emailLogin 换取 token，
-或加上 --from-device，用项目内置 adb 读取设备里已登录账号的 accessToken。
-凭证放在 apasstk 头里，正文是 JSON。已购买或已解锁的集会返回 m3u8。
-拉完目录后会询问是否把这些视频下载到项目的 video 目录，并按剧名分文件夹。
-
-重复运行会重新拉最新目录。邮箱和密码用参数或环境变量传入，不要写进源码。
-"""
-import argparse
-import gzip
-import html
-import json
-import os
-import re
-import socket
-import subprocess
-import time
-import urllib.error
-import urllib.request
-from urllib.parse import urljoin
-
-socket.setdefaulttimeout(20)
-
-BASE = "https://apis.raptdrama.com"
-OUT_NAME = "raptdrama_catalog.json"
-# 顶部分类标签。id 来自 classifyv2，列表接口的参数名是 type。
-CLASSIFY_SECTIONS = (
- ("comedy", "Comedy", 18),
- ("passion", "Passion", 20),
- ("newest", "Newest", 21),
- ("western", "Western", 22),
- ("comic", "Comic drama", 23),
-)
-CLASSIFY_MAX_PAGES = 50
-RECOMMEND_MAX_PAGES = 80
-PREFS = "/data/data/com.leivideo.raptdrama/shared_prefs/FlutterSharedPreferences.xml"
-
-HEADERS = {
- "Content-Type": "application/json",
- "Accept": "application/json",
- "User-Agent": "RaptDrama/1.1.81 (Android 12; 23113RKC6C; Redmi)",
- "x-device-fingerprint": "0a4e4b212cbb0a2434cf804c8e30ab58",
- "x-os-version": "12",
- "x-device-physical": "true",
- "x-device-model": "23113RKC6C",
- "accept-language": "en-US",
- "accept-language-custom": "en-US",
- "x-os-name": "Android",
- "x-device-brand": "Redmi",
- "x-device-id": "rd_c9f0fa17a39f96aa",
- "x-app-build": "139",
- "x-device-type": "phone",
- "x-app-version": "1.1.81",
-}
-
-
-def account_label(user):
- name = user.get("displayName") or user.get("nickname") or "device"
- uid = user.get("uid") or user.get("id") or ""
- if user.get("isGuest"):
- provider = "guest"
- else:
- provider = user.get("loginProvider") or "account"
- vip = "VIP" if user.get("isVip") else "非VIP"
- return f"{name} uid={uid} {provider} {vip}"
-
-
-def bundled_adb():
- name = "adb.exe" if os.name == "nt" else "adb"
- path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "tools", "platform-tools", name)
- if not os.path.isfile(path):
- raise RuntimeError(f"项目内没有 adb：{path}")
- return path
-
-
-def find_adb(explicit):
- if explicit:
- if not os.path.isfile(explicit):
- raise RuntimeError(f"找不到 adb：{explicit}")
- return explicit
- return bundled_adb()
-
-
-def adb_exec(adb_path, args, timeout=20):
- return subprocess.run([adb_path, *args], capture_output=True, timeout=timeout)
-
-
-def list_ready_devices(adb_path):
- proc = adb_exec(adb_path, ["devices"])
- serials = []
- for line in proc.stdout.decode("utf-8", "replace").splitlines()[1:]:
- parts = line.split()
- if len(parts) >= 2 and parts[1] == "device":
- serials.append(parts[0])
- return serials
-
-
-def resolve_serial(adb_path, serial):
- serial = (serial or "").strip()
- if serial:
- if ":" in serial:
- adb_exec(adb_path, ["connect", serial], timeout=15)
- ready = list_ready_devices(adb_path)
- if serial not in ready:
- shown = "、".join(ready) or "无"
- raise RuntimeError(f"设备 {serial} 未就绪。当前已连接：{shown}")
- return serial
- ready = list_ready_devices(adb_path)
- if len(ready) == 1:
- return ready[0]
- if not ready:
- raise RuntimeError(
- "没有检测到设备。手机请打开 USB 调试并在手机上允许这台电脑；"
- "模拟器请加上 --serial IP:端口。"
- )
- raise RuntimeError("检测到多台设备，请用 --serial 指定其一：\n" + "\n".join(ready))
-
-
-def shell_text(adb_path, serial, shell_args, timeout=15):
- try:
- proc = adb_exec(adb_path, ["-s", serial, "shell", *shell_args], timeout=timeout)
- except subprocess.TimeoutExpired:
- return 1, "", "timeout"
- return proc.returncode, proc.stdout.decode("utf-8", "replace"), proc.stderr.decode("utf-8", "replace")
-
-
-def read_prefs_text(adb_path, serial):
- def usable(text):
- return "flutter.user_data" in text
-
- _, text, _ = shell_text(adb_path, serial, ["cat", PREFS])
- if usable(text):
- return text
-
- adb_exec(adb_path, ["-s", serial, "root"], timeout=15)
- if ":" in serial:
- adb_exec(adb_path, ["connect", serial], timeout=15)
- for _ in range(10):
- time.sleep(0.5)
- _, text, _ = shell_text(adb_path, serial, ["cat", PREFS])
- if usable(text):
- return text
-
- for shell_args in (["su", "-c", f"cat {PREFS}"], ["su", "0", "cat", PREFS]):
- _, text, _ = shell_text(adb_path, serial, shell_args, timeout=8)
- if usable(text):
- return text
-
- raise RuntimeError(
- "读不到应用私有目录。此应用是正式包，未 root 的手机无法读取其中的登录凭证。"
- "请使用已 root 的手机，或带 root 的模拟器。"
- )
-
-
-def load_token_from_device(adb_path, serial):
- serial = resolve_serial(adb_path, serial)
- text = read_prefs_text(adb_path, serial)
- match = re.search(r'name="flutter\.user_data"[^>]*>(.*?)', text, re.S)
- if not match:
- raise RuntimeError("登录信息里没有 user_data，请先在 App 里登录")
- user = json.loads(html.unescape(match.group(1)))
- token = user.get("accessToken") or ""
- if not token:
- raise RuntimeError("当前账号没有 accessToken，请先在 App 里登录")
- return token, f"{account_label(user)} device={serial}"
-
-
-def load_guest_token():
- data = api("/app/open/autologin", None, "POST", {})
- token = (data or {}).get("token") or ""
- if not token:
- raise RuntimeError("游客登录没有返回 token")
- return token, (data or {}).get("nickname") or "guest"
-
-
-def load_email_token(email, password):
- data = api("/app/open/emailLogin", None, "POST", {
- "email": email,
- "password": password,
- })
- token = (data or {}).get("token") or ""
- if not token:
- raise RuntimeError("邮箱登录没有返回 token")
- nickname = (data or {}).get("nickname") or email
- uid = (data or {}).get("id") or ""
- return token, f"{nickname} uid={uid} email"
-
-
-def load_token(from_device, adb_path, serial, email, password):
- if from_device:
- return load_token_from_device(find_adb(adb_path), serial)
- email = (email or "").strip()
- password = password or ""
- if email or password:
- if not email or not password:
- raise RuntimeError("邮箱登录需要同时提供邮箱和密码")
- return load_email_token(email, password)
- env = os.environ.get("RAPTD_TOKEN", "").strip()
- if env:
- return env, "RAPTD_TOKEN"
- email = (os.environ.get("RAPTD_EMAIL") or "").strip()
- password = os.environ.get("RAPTD_PASSWORD") or ""
- if email or password:
- if not email or not password:
- raise RuntimeError("邮箱登录需要同时提供 RAPTD_EMAIL 和 RAPTD_PASSWORD")
- return load_email_token(email, password)
- return load_guest_token()
-
-
-def _read_response(req):
- last_error = None
- for attempt in range(6):
- try:
- with urllib.request.urlopen(req, timeout=25) as resp:
- return resp.read()
- except (urllib.error.URLError, TimeoutError, OSError) as exc:
- last_error = exc
- time.sleep(min(10, 1.5 * (attempt + 1)))
- raise RuntimeError(f"请求失败: {last_error}")
-
-
-def api(path, token, method="GET", body=None):
- headers = dict(HEADERS)
- if token:
- headers["apasstk"] = token
- data = None if body is None else json.dumps(body).encode()
- req = urllib.request.Request(BASE + path, data=data, headers=headers, method=method)
- raw = _read_response(req)
- if raw[:2] == b"\x1f\x8b":
- raw = gzip.decompress(raw)
- payload = json.loads(raw.decode("utf-8"))
- if payload.get("code") != 200:
- raise RuntimeError(f"{method} {path} -> {payload.get('code')} {payload.get('msg')}")
- return payload.get("data")
-
-
-def http_text(url):
- req = urllib.request.Request(url, headers={"User-Agent": "ExoPlayer"})
- raw = _read_response(req)
- return raw.decode("utf-8", "replace")
-
-
-def _as_drama_list(data):
- if isinstance(data, list):
- return [item for item in data if isinstance(item, dict)]
- if isinstance(data, dict):
- items = data.get("list")
- if isinstance(items, list):
- return [item for item in items if isinstance(item, dict)]
- return []
-
-
-def list_paged(token, path, method, make_body, max_pages):
- """按页拉取，直到空页或整页都是已经见过的 id。max_pages 防止死循环。"""
- dramas = []
- seen = set()
- page = 1
- stop_page = page
- reason = "达到页数上限"
- while page <= max_pages:
- stop_page = page
- data = api(path, token, method, make_body(page))
- items = _as_drama_list(data)
- if not items:
- reason = "空页"
- break
- fresh = []
- for item in items:
- vid = item.get("id")
- if vid in seen:
- continue
- seen.add(vid)
- fresh.append(item)
- if not fresh:
- reason = "整页都是重复 id"
- break
- dramas.extend(fresh)
- page += 1
- time.sleep(0.05)
- return dramas, stop_page, reason
-
-
-def list_classify(token, type_id):
- return list_paged(
- token,
- "/app/open/classify_video",
- "POST",
- lambda page: {"page": page, "limit": 10, "type": type_id},
- CLASSIFY_MAX_PAGES,
- )
-
-
-def list_new_releases(token):
- items = _as_drama_list(api("/app/open/orgin", token, "GET"))
- return items, 1, "单次返回"
-
-
-def list_recommend(token):
- return list_paged(
- token,
- "/app/open/recommend",
- "POST",
- lambda page: {"page": page, "limit": 10},
- RECOMMEND_MAX_PAGES,
- )
-
-
-def add_section(catalog, order, section, items):
- for item in items:
- vid = item.get("id")
- if vid is None:
- continue
- entry = catalog.get(vid)
- if entry is None:
- entry = dict(item)
- entry["sections"] = []
- catalog[vid] = entry
- order.append(vid)
- if section not in entry["sections"]:
- entry["sections"].append(section)
-
-
-def collect_home(token):
- """合并首页区块。同一 id 只保留一份剧目，sections 记录出现过的区块。"""
- catalog = {}
- order = []
- for key, label, type_id in CLASSIFY_SECTIONS:
- items, stop_page, reason = list_classify(token, type_id)
- add_section(catalog, order, key, items)
- print(f" {label} {len(items)} 部，停在第 {stop_page} 页（{reason}）", flush=True)
- items, _, reason = list_new_releases(token)
- add_section(catalog, order, "new_releases", items)
- print(f" New Releases {len(items)} 部（{reason}）", flush=True)
- items, stop_page, reason = list_recommend(token)
- add_section(catalog, order, "recommend", items)
- print(f" Recommend {len(items)} 部，停在第 {stop_page} 页（{reason}）", flush=True)
- print(f" 去重后 {len(order)} 部", flush=True)
- return [catalog[vid] for vid in order]
-
-
-def list_chapters(token, vid):
- chapters = []
- start = 0
- while start < 1000:
- data = api(
- f"/app/video/getchapters?vid={vid}&start={start}&limit=200",
- token,
- )
- items = (data or {}).get("list") or []
- chapters.extend(items)
- if len(items) < 200:
- break
- start += 200
- return chapters
-
-
-def _playlist_items(data):
- if isinstance(data, list):
- return data, len(data)
- items = (data or {}).get("list") or []
- total = (data or {}).get("count")
- return items, total
-
-
-def _playlist_complete(items, total):
- if not items:
- return False
- if total is None:
- return True
- return len(items) >= int(total)
-
-
-def list_playlist_v2(token, vid):
- episodes = []
- page = 1
- total = None
- while page <= 40:
- data = api("/app/video/playlistv2", token, "POST", {
- "vid": vid,
- "page": page,
- "before": 1 if page == 1 else 0,
- })
- items, page_total = _playlist_items(data)
- if page_total is not None:
- total = page_total
- if not items:
- break
- episodes.extend(items)
- if total is not None and len(episodes) >= int(total):
- break
- page += 1
- time.sleep(0.05)
- return episodes
-
-
-def list_playlist(token, vid):
- try:
- data = api("/app/video/playlist", token, "POST", {
- "vid": vid, "page": 1, "before": 1,
- })
- items, total = _playlist_items(data)
- if _playlist_complete(items, total):
- return items
- except Exception as exc:
- print(f" playlist 失败，改用 playlistv2：{exc}", flush=True)
- return list_playlist_v2(token, vid)
- print(f" playlist 结果不完整，改用 playlistv2", flush=True)
- return list_playlist_v2(token, vid)
-
-
-def expand_hls(master_url):
- """把主 m3u8 展开成子播放列表和 ts 绝对地址。失败时只保留主地址。"""
- if not master_url:
- return None, []
- try:
- master = http_text(master_url)
- except Exception:
- return None, []
- variant = None
- for line in master.splitlines():
- line = line.strip()
- if line and not line.startswith("#"):
- variant = urljoin(master_url, line)
- break
- if not variant:
- return None, []
- try:
- media = http_text(variant)
- except Exception:
- return variant, []
- segments = []
- for line in media.splitlines():
- line = line.strip()
- if line and not line.startswith("#"):
- segments.append(urljoin(variant, line))
- return variant, segments
-
-
-def safe_name(text, fallback):
- cleaned = re.sub(r'[<>:"/\\|?*\x00-\x1f]', " ", text or "")
- cleaned = re.sub(r"\s+", " ", cleaned).strip().rstrip(". ")
- return (cleaned[:80] or fallback)
-
-
-def download_bytes(url):
- req = urllib.request.Request(url, headers={"User-Agent": "ExoPlayer"})
- return _read_response(req)
-
-
-def localize_playlist(text, local_name_for_uri):
- lines = []
- for line in text.splitlines():
- stripped = line.strip()
- if stripped and not stripped.startswith("#"):
- lines.append(local_name_for_uri(stripped))
- else:
- lines.append(line)
- return "\n".join(lines) + "\n"
-
-
-def download_episode(episode, episode_dir):
- master_url = episode.get("cdn_url")
- if not master_url:
- return False
- os.makedirs(episode_dir, exist_ok=True)
- master_text = download_bytes(master_url).decode("utf-8", "replace")
- variant_rel = next(
- (line.strip() for line in master_text.splitlines() if line.strip() and not line.startswith("#")),
- None,
- )
- if not variant_rel:
- return False
- variant_url = urljoin(master_url, variant_rel)
- media_text = download_bytes(variant_url).decode("utf-8", "replace")
- segment_rels = [
- line.strip() for line in media_text.splitlines()
- if line.strip() and not line.startswith("#")
- ]
- if not segment_rels:
- return False
- names = [os.path.basename(urljoin(variant_url, rel).split("?", 1)[0]) for rel in segment_rels]
- for rel, name in zip(segment_rels, names):
- target = os.path.join(episode_dir, name)
- if os.path.isfile(target) and os.path.getsize(target) > 0:
- continue
- payload = download_bytes(urljoin(variant_url, rel))
- with open(target, "wb") as handle:
- handle.write(payload)
- with open(os.path.join(episode_dir, "video.m3u8"), "w", encoding="utf-8", newline="\n") as handle:
- handle.write(localize_playlist(media_text, lambda uri: os.path.basename(urljoin(variant_url, uri).split("?", 1)[0])))
- with open(os.path.join(episode_dir, "playlist.m3u8"), "w", encoding="utf-8", newline="\n") as handle:
- handle.write(localize_playlist(master_text, lambda _uri: "video.m3u8"))
- return all(os.path.isfile(os.path.join(episode_dir, name)) and os.path.getsize(os.path.join(episode_dir, name)) > 0 for name in names)
-
-
-def download_videos(results, video_root):
- used_names = {}
- saved = 0
- failed = 0
- for drama in results:
- episodes = [ep for ep in drama["episodes"] if ep.get("cdn_url")]
- if not episodes:
- continue
- title = safe_name(drama.get("title"), f"drama-{drama.get('id')}")
- if title in used_names:
- title = safe_name(f"{title} {drama.get('id')}", title)
- used_names[title] = True
- drama_dir = os.path.join(video_root, title)
- for episode in episodes:
- number = episode.get("number") or 0
- ep_title = safe_name(episode.get("title"), f"EP.{number}")
- episode_dir = os.path.join(drama_dir, f"{int(number):02d} {ep_title}")
- label = f"{title}/{int(number):02d}"
- try:
- ok = download_episode(episode, episode_dir)
- except Exception as exc:
- ok = False
- print(f" 下载失败 {label}: {exc}", flush=True)
- if ok:
- saved += 1
- print(f" 已保存 {label}", flush=True)
- else:
- failed += 1
- return saved, failed
-
-
-def ask_download(count, video_root):
- print(f"\n可下载 {count} 集，目录：{video_root}", flush=True)
- answer = input("是否立即下载全部视频？输入 y 下载，其他键跳过: ").strip().lower()
- return answer in ("y", "yes")
-
-
-def parse_args():
- parser = argparse.ArgumentParser(description="抓取 RaptDrama 首页目录和播放地址")
- parser.add_argument(
- "--from-device",
- action="store_true",
- help="从已连接设备读取当前登录账号的 accessToken",
- )
- parser.add_argument(
- "--serial",
- default=os.environ.get("RAPTD_SERIAL", ""),
- help="设备序列号。只连着一台时可以省略；模拟器填写 IP:端口",
- )
- parser.add_argument(
- "--adb",
- default=os.environ.get("RAPTD_ADB", ""),
- help="adb 路径。默认使用项目内 tools/platform-tools",
- )
- parser.add_argument(
- "--email",
- default="",
- help="邮箱登录账号。也可设置环境变量 RAPTD_EMAIL",
- )
- parser.add_argument(
- "--password",
- default="",
- help="邮箱登录密码。也可设置环境变量 RAPTD_PASSWORD",
- )
- return parser.parse_args()
-
-
-def main():
- args = parse_args()
- out_dir = os.path.dirname(os.path.abspath(__file__))
- out_file = os.path.join(out_dir, OUT_NAME)
- if args.from_device:
- mode = "从设备读取登录凭证"
- elif args.email or args.password:
- mode = "邮箱登录"
- elif os.environ.get("RAPTD_TOKEN"):
- mode = "使用 RAPTD_TOKEN"
- elif os.environ.get("RAPTD_EMAIL") or os.environ.get("RAPTD_PASSWORD"):
- mode = "邮箱登录"
- else:
- mode = "游客登录"
- print(mode, flush=True)
- token, user = load_token(args.from_device, args.adb, args.serial, args.email, args.password)
- print(f" {user}", flush=True)
-
- print("获取首页目录", flush=True)
- dramas = collect_home(token)
-
- results = []
- for index, info in enumerate(dramas, 1):
- vid = info["id"]
- title = (info.get("title") or "").strip()
- sections = info.get("sections") or []
- print(f" [{index}/{len(dramas)}] {title} [{', '.join(sections)}]", flush=True)
- chapters = list_chapters(token, vid)
- vip_by_id = {ch.get("id"): str(ch.get("isvip")) == "1" for ch in chapters}
- title_by_id = {ch.get("id"): ch.get("title") or "" for ch in chapters}
- playlist = list_playlist(token, vid)
-
- episodes = []
- for item in playlist:
- cid = item.get("cid")
- src = item.get("src") or ""
- variant, segments = expand_hls(src)
- episodes.append({
- "id": cid,
- "number": item.get("idx"),
- "title": title_by_id.get(cid) or item.get("msg") or "",
- "is_vip": vip_by_id.get(cid, not bool(src)),
- "cdn_url": src or None,
- "media_playlist": variant,
- "segments": segments,
- })
-
- free = sum(1 for ep in episodes if not ep["is_vip"])
- with_cdn = sum(1 for ep in episodes if ep["cdn_url"])
- with_ts = sum(1 for ep in episodes if ep["segments"])
- print(
- f" [{index}/{len(dramas)}] {title}: {len(episodes)} 集, 免费 {free}, 有地址 {with_cdn}, 已展开 {with_ts}",
- flush=True,
- )
-
- results.append({
- "id": vid,
- "title": title,
- "image": info.get("image") or "",
- "desc": (info.get("desc") or "").strip(),
- "score": info.get("score") or "",
- "view": info.get("view") or 0,
- "status": info.get("forstausen") or "",
- "classify": info.get("classify") or [],
- "sections": info.get("sections") or [],
- "total_episodes": len(episodes),
- "episodes": episodes,
- })
- with open(out_file, "w", encoding="utf-8") as handle:
- json.dump(results, handle, ensure_ascii=False, indent=2)
-
- total_eps = sum(item["total_episodes"] for item in results)
- cdn_eps = sum(1 for item in results for ep in item["episodes"] if ep.get("cdn_url"))
- ts_eps = sum(1 for item in results for ep in item["episodes"] if ep.get("segments"))
- print(f"完成: {len(results)} 部, {total_eps} 集, 有 m3u8 {cdn_eps}, 已展开 ts {ts_eps}", flush=True)
- print(f"保存至 {out_file}", flush=True)
-
- video_root = os.path.join(out_dir, "video")
- if cdn_eps and ask_download(cdn_eps, video_root):
- print("开始下载", flush=True)
- saved, failed = download_videos(results, video_root)
- print(f"下载结束: 成功 {saved} 集, 失败 {failed} 集", flush=True)
- print(f"文件在 {video_root}", flush=True)
- elif cdn_eps:
- print("已跳过下载", flush=True)
-
-
-if __name__ == "__main__":
- main()
diff --git a/src-tauri/splash/index.html b/src-tauri/splash/index.html
index e176682..d7c830e 100644
--- a/src-tauri/splash/index.html
+++ b/src-tauri/splash/index.html
@@ -14,7 +14,7 @@
RaptDrama
- 后台程序正在启动
+ 程序正在启动