nixevol/RaptDramaDump
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: remove unused rapt_scraper and live capture scripts

Browse Source
This commit is contained in:
Nixevol
2026-09-27 06:29:33 +08:00
parent 7811e6dbaa
commit 3449adb721
3 changed files with 1 additions and 856 deletions
Download Patch File Download Diff File Expand all files Collapse all files
capture/live.py
-195
View File
@@ -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 = """<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<title>明文请求</title>
<style>
html, body { height: 100%; margin: 0; background: #161616; color: #eee; font: 14px/1.45 "Segoe UI", sans-serif; }
header { position: sticky; top: 0; display: flex; justify-content: space-between; align-items: center;
padding: 10px 16px; background: #111; border-bottom: 1px solid #333; z-index: 2; }
main { display: flex; height: calc(100% - 49px); }
#list { width: 42%; overflow: auto; border-right: 1px solid #333; }
button.row { display: block; width: 100%; text-align: left; padding: 8px 12px; border: 0;
border-bottom: 1px solid #2a2a2a; background: transparent; color: #ddd; cursor: pointer; }
button.row.on { background: #2c3a28; }
#detail { flex: 1; margin: 0; padding: 12px 16px; overflow: auto; white-space: pre-wrap; word-break: break-all; }
.muted { color: #9aa; }
</style>
</head>
<body>
<header>
<strong>Dart 明文请求</strong>
<span id="count" class="muted">0 条</span>
</header>
<main>
<div id="list"></div>
<pre id="detail">打开 App 并滑动页面，新的请求会出现在左边。</pre>
</main>
<script>
let selected = 0;
async function refresh() {
const rows = await (await fetch("/api/flows")).json();
document.getElementById("count").textContent = rows.length + " 条";
const list = document.getElementById("list");
list.innerHTML = "";
rows.slice().reverse().forEach((row) => {
const button = document.createElement("button");
button.className = "row" + (row.id === selected ? " on" : "");
button.textContent = row.summary;
button.onclick = () => { selected = row.id; document.getElementById("detail").textContent = row.text; refresh(); };
list.appendChild(button);
});
}
refresh();
setInterval(refresh, 1000);
</script>
</body>
</html>
"""
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()
rapt_scraper.py
-660
View File
@@ -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"[^>]*>(.*?)</string>', 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()
src-tauri/splash/index.html
+1 -1
View File
@@ -14,7 +14,7 @@
<body> <body>
<main> <main>
<h1>RaptDrama</h1> <h1>RaptDrama</h1>
<p id="status">后台程序正在启动</p> <p id="status">程序正在启动</p>
</main> </main>
<script> <script>
window.setStatus = (text) => { window.setStatus = (text) => {