190 lines
7.0 KiB
Python
190 lines
7.0 KiB
Python
#!/usr/bin/env python3
|
||
"""把新版 短剧库.zip 与本地源码对比，展示变更并更新本地旧代码。
|
||
|
||
用法:
|
||
python scripts/update_from_zip.py # 从默认地址下载最新包并预览变更
|
||
python scripts/update_from_zip.py --apply # 下载并真正更新本地代码
|
||
python scripts/update_from_zip.py 某个路径/短剧库.zip # 用本地压缩包
|
||
python scripts/update_from_zip.py --url https://... # 指定其他下载地址
|
||
|
||
行为:
|
||
1. 读取压缩包（自动处理中文文件名 GBK/UTF-8、自动剥离压缩包内公共顶层目录）
|
||
2. 与本地源码逐文件比对（忽略 .git、data/、日志、压缩包等运行时文件）
|
||
3. 列出 新增 / 修改 / 删除 三类变更；加 --apply 后写入本地
|
||
4. 更新完成后用 git status / git diff 自查，再提交
|
||
"""
|
||
import argparse
|
||
import hashlib
|
||
import sys
|
||
import tempfile
|
||
import urllib.parse
|
||
import urllib.request
|
||
import zipfile
|
||
from pathlib import Path
|
||
|
||
DEFAULT_URL = "https://www.252035.xyz/短剧库.zip"
|
||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||
IGNORE_DIRS = {".git", "data", "__pycache__", "dist", "短剧下载"}
|
||
IGNORE_SUFFIX = {".log", ".zip", ".tar", ".pyc"}
|
||
IGNORE_NAMES = {".DS_Store", "Thumbs.db"}
|
||
PROTECT = {"scripts/update_from_zip.py"} # 更新工具自身，永远不删
|
||
|
||
|
||
def download(url: str) -> Path:
|
||
quoted = urllib.parse.quote(url, safe=":/?&=%")
|
||
print(f"下载最新压缩包：{url}")
|
||
request = urllib.request.Request(quoted, headers={"User-Agent": "juku-update/1.0"})
|
||
with urllib.request.urlopen(request, timeout=60) as resp:
|
||
total = int(resp.headers.get("Content-Length") or 0)
|
||
target = Path(tempfile.gettempdir()) / "短剧库-update.zip"
|
||
done = 0
|
||
with target.open("wb") as f:
|
||
while True:
|
||
chunk = resp.read(1 << 16)
|
||
if not chunk:
|
||
break
|
||
f.write(chunk)
|
||
done += len(chunk)
|
||
if total:
|
||
print(f"\r {done}/{total} 字节", end="", flush=True)
|
||
print(f"\n 已保存到 {target}")
|
||
return target
|
||
|
||
|
||
def norm_name(info: zipfile.ZipInfo) -> str:
|
||
"""修正 zip 内中文文件名。
|
||
|
||
打包工具有两种常见错误：用 UTF-8 编码但不打 0x800 标志（跨平台工具），
|
||
或用 GBK 编码且不打标志（中文 Windows 右键压缩）。依次尝试恢复：
|
||
UTF-8 优先（其字节序列极少与 GBK 混淆），失败再试 GBK。
|
||
"""
|
||
name = info.filename
|
||
if not info.flag_bits & 0x800:
|
||
try:
|
||
raw = name.encode("cp437")
|
||
except UnicodeEncodeError:
|
||
return name.replace("\\", "/")
|
||
for encoding in ("utf-8", "gbk"):
|
||
try:
|
||
name = raw.decode(encoding)
|
||
break
|
||
except UnicodeDecodeError:
|
||
continue
|
||
return name.replace("\\", "/")
|
||
|
||
|
||
def ignored(rel: str) -> bool:
|
||
parts = rel.split("/")
|
||
return (
|
||
any(p in IGNORE_DIRS for p in parts[:-1])
|
||
or Path(rel).name in IGNORE_NAMES
|
||
or Path(rel).suffix in IGNORE_SUFFIX
|
||
)
|
||
|
||
|
||
def common_prefix(names: list[str]) -> str:
|
||
"""所有条目共享同一顶层目录时返回该前缀（如 '短剧库/'），否则返回空串。"""
|
||
if not names:
|
||
return ""
|
||
tops = {n.split("/", 1)[0] for n in names}
|
||
if len(tops) == 1 and all("/" in n for n in names):
|
||
return next(iter(tops)) + "/"
|
||
return ""
|
||
|
||
|
||
def md5_of(path: Path) -> str:
|
||
h = hashlib.md5()
|
||
with path.open("rb") as f:
|
||
for chunk in iter(lambda: f.read(1 << 16), b""):
|
||
h.update(chunk)
|
||
return h.hexdigest()
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="用新版 zip 更新本地短剧库源码")
|
||
parser.add_argument("zipfile", nargs="?", default=None,
|
||
help="本地压缩包路径；省略则从 --url 地址下载最新包")
|
||
parser.add_argument("--url", default=DEFAULT_URL, help=f"新版下载地址（默认 {DEFAULT_URL}）")
|
||
parser.add_argument("--apply", action="store_true", help="真正写入本地（默认只预览）")
|
||
args = parser.parse_args()
|
||
|
||
if args.zipfile:
|
||
zip_path = Path(args.zipfile)
|
||
if not zip_path.exists():
|
||
print(f"找不到压缩包：{args.zipfile}")
|
||
return 1
|
||
else:
|
||
zip_path = download(args.url)
|
||
|
||
with zipfile.ZipFile(zip_path) as zf:
|
||
infos = [i for i in zf.infolist() if not i.is_dir()]
|
||
names = [norm_name(i) for i in infos]
|
||
# 先丢弃 macOS 压缩附带的资源文件，避免干扰顶层目录识别
|
||
pairs = [
|
||
(info, name)
|
||
for info, name in zip(infos, names)
|
||
if "__MACOSX" not in name.split("/") and not Path(name).name.startswith("._")
|
||
]
|
||
prefix = common_prefix([name for _, name in pairs])
|
||
if prefix:
|
||
print(f"检测到压缩包顶层目录：{prefix}（比对时已忽略）")
|
||
incoming: dict[str, bytes] = {}
|
||
for info, name in pairs:
|
||
rel = name[len(prefix):] if prefix else name
|
||
if rel and not ignored(rel):
|
||
incoming[rel] = zf.read(info)
|
||
|
||
added, modified, unchanged = [], [], []
|
||
for rel, blob in sorted(incoming.items()):
|
||
local = REPO_ROOT / rel
|
||
if not local.exists():
|
||
added.append(rel)
|
||
elif md5_of(local) != hashlib.md5(blob).hexdigest():
|
||
modified.append(rel)
|
||
else:
|
||
unchanged.append(rel)
|
||
|
||
local_files: set[str] = set()
|
||
for path in REPO_ROOT.rglob("*"):
|
||
if not path.is_file():
|
||
continue
|
||
rel = path.relative_to(REPO_ROOT).as_posix()
|
||
if ignored(rel) or rel in IGNORE_NAMES:
|
||
continue
|
||
local_files.add(rel)
|
||
deleted = sorted(local_files - set(incoming) - PROTECT)
|
||
|
||
print(f"\n对比完成：压缩包 {len(incoming)} 个文件，本地 {len(local_files)} 个文件")
|
||
print(f" 新增 {len(added)} 个")
|
||
for rel in added:
|
||
print(f" + {rel}")
|
||
print(f" 修改 {len(modified)} 个")
|
||
for rel in modified:
|
||
print(f" ~ {rel}")
|
||
print(f" 删除 {len(deleted)} 个")
|
||
for rel in deleted:
|
||
print(f" - {rel}")
|
||
print(f" 未变化 {len(unchanged)} 个")
|
||
|
||
if not (added or modified or deleted):
|
||
print("本地代码与压缩包一致，无需更新。")
|
||
return 0
|
||
if not args.apply:
|
||
print("\n当前为预览模式。确认无误后追加 --apply 执行更新，然后 git add -A && git commit。")
|
||
return 0
|
||
|
||
for rel in added + modified:
|
||
target = REPO_ROOT / rel
|
||
target.parent.mkdir(parents=True, exist_ok=True)
|
||
target.write_bytes(incoming[rel])
|
||
print(f"已写入 {rel}")
|
||
for rel in deleted:
|
||
(REPO_ROOT / rel).unlink(missing_ok=True)
|
||
print(f"已删除 {rel}")
|
||
print("\n更新完成。请执行 git status / git diff 复核后提交。")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|