#!/usr/bin/env python3 """Mirror an HTTP directory tree. Python 3.9+, Linux/macOS, no dependencies.""" import argparse import csv import fcntl import hashlib from http.client import HTTPException import json import os import re import shutil import sys import tempfile import time import uuid from collections import deque from datetime import datetime, timezone from html.parser import HTMLParser from pathlib import Path from urllib.error import URLError from urllib.parse import unquote, urljoin, urlsplit, urlunsplit from urllib.request import HTTPRedirectHandler, Request, build_opener DEFAULT_SOURCE = "https://fastly.mirror.pkgbuild.com/iso/latest/" MARKER = ".arch-mirror.json" HTML_EXTENSIONS = {".html", ".htm", ".xhtml"} IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".gif", ".svg", ".webp", ".avif", ".ico", ".bmp", ".tif", ".tiff", ".apng"} HTML_TYPES = {"text/html", "application/xhtml+xml"} CHUNK = 1024 * 1024 def now(): return datetime.now(timezone.utc).isoformat(timespec="seconds") class MirrorError(Exception): pass def write_json(path, value): temporary = path.with_name(path.name + ".tmp") with temporary.open("w", encoding="utf-8") as stream: json.dump(value, stream, ensure_ascii=False, indent=2) stream.write("\n") os.replace(temporary, path) def safe_path(path): path = Path(os.path.abspath(os.path.expanduser(str(path)))) for component in [path, *path.parents]: if component.is_symlink(): raise MirrorError(f"路径不得包含符号链接：{component}") return path class Scope: def __init__(self, source): parts = urlsplit(source) if parts.scheme not in {"http", "https"} or not parts.hostname or parts.username or parts.password: raise MirrorError("源地址必须是无凭据的 HTTP(S) 目录 URL") if parts.query or parts.fragment: raise MirrorError("源地址不能包含查询参数或片段") self.base = urlunsplit((parts.scheme.lower(), parts.netloc.lower(), parts.path.rstrip("/") + "/", "", "")) self.parts = urlsplit(self.base) self.prefix = unquote(self.parts.path, errors="strict") def candidate(self, page, reference): parts = urlsplit(urljoin(page, reference)) if parts.scheme.lower() != self.parts.scheme or parts.netloc.lower() != self.parts.netloc or parts.query: return None try: decoded = unquote(parts.path, errors="strict") except UnicodeError: return None if not decoded.startswith(self.prefix): return None relative = decoded[len(self.prefix):] segments = relative.rstrip("/").split("/") if relative else [] if any(s in {"", ".", "..", MARKER} or "\\" in s or any(ord(c) < 32 or ord(c) == 127 for c in s) for s in segments): return None local = "/".join(segments) if decoded.endswith("/"): local = (local + "/" if local else "") + "index.html" return urlunsplit((parts.scheme, parts.netloc, parts.path, "", "")), local class ScopedRedirect(HTTPRedirectHandler): def __init__(self, scope): self.scope = scope def redirect_request(self, req, fp, code, msg, headers, newurl): if self.scope.candidate(req.full_url, newurl) is None: raise MirrorError(f"拒绝跳转到镜像范围之外：{newurl}") return super().redirect_request(req, fp, code, msg, headers, newurl) class Links(HTMLParser): def __init__(self): super().__init__(convert_charrefs=True) self.references = [] def handle_starttag(self, tag, attributes): attrs = dict(attributes) if tag in {"a", "area"} and attrs.get("href"): self.references.append(attrs["href"]) if tag in {"img", "source"} and attrs.get("src"): self.references.append(attrs["src"]) if tag in {"img", "source"} and attrs.get("srcset"): self.references.extend(item.strip().split()[0] for item in attrs["srcset"].split(",") if item.strip()) if tag == "video" and attrs.get("poster"): self.references.append(attrs["poster"]) if tag == "link" and "icon" in attrs.get("rel", "").lower() and attrs.get("href"): self.references.append(attrs["href"]) handle_startendtag = handle_starttag def download(opener, url, destination, timeout, retries, expected): destination.parent.mkdir(parents=True, exist_ok=True) last_error = None for attempt in range(retries + 1): try: request = Request(url, headers={"User-Agent": "ArchMirrorKit/1.0", "Accept-Encoding": "identity"}) with opener.open(request, timeout=timeout) as response: content_type = response.headers.get_content_type() if expected == "html" and content_type not in HTML_TYPES: raise MirrorError(f"目录或 HTML 链接返回了 {content_type}：{url}") if expected == "image" and not content_type.startswith("image/"): raise MirrorError(f"图片链接返回了 {content_type}：{url}") size, digest = 0, hashlib.sha256() with destination.open("wb") as stream: while True: chunk = response.read(CHUNK) if not chunk: break stream.write(chunk) digest.update(chunk) size += len(chunk) length = response.headers.get("Content-Length") if length is not None and size != int(length): raise MirrorError(f"文件未完整下载：{url}（收到 {size} / {length} 字节）") return {"source_url": url, "final_url": response.url, "content_type": content_type, "size": size, "sha256": digest.hexdigest(), "downloaded_at": now(), "transfer": "http", "transferred_bytes": size, "last_modified": response.headers.get("Last-Modified"), "etag": response.headers.get("ETag")}, response.headers.get_content_charset() or "utf-8" except (OSError, URLError, HTTPException, MirrorError, ValueError) as error: last_error = error if attempt < retries: time.sleep(min(2 ** attempt, 5)) raise MirrorError(f"下载失败：{url}：{last_error}") def checksum_inventory(filename): """Use fresh upstream checksums only as a download reuse hint, never as trust.""" if filename.stat().st_size > 2 * CHUNK: return {} checksums = {} try: for line in filename.read_text(encoding="ascii").splitlines(): if not line.strip(): continue match = re.fullmatch(r"([0-9a-fA-F]{64}) [ *](.+)", line) if not match: return {} name = match[2].removeprefix("./") if (name in checksums or name.startswith("/") or "\\" in name or any(part in {"", ".", ".."} for part in name.split("/"))): return {} checksums[name] = match[1].lower() except (OSError, UnicodeError): return {} return checksums def link_duplicate(stage, destination, target): """Only link to a regular file already downloaded in this private snapshot.""" if Path(target).is_absolute() or any(part in {"", ".", ".."} for part in target.split("/")): raise MirrorError("去重目标不得离开本轮镜像目录") canonical = stage / target safe_path(canonical) if canonical.is_symlink() or not canonical.is_file(): raise MirrorError("去重目标必须是本轮镜像内的普通文件") destination.parent.mkdir(parents=True, exist_ok=True) temporary = destination.with_name(destination.name + ".link-" + uuid.uuid4().hex) try: temporary.symlink_to(os.path.relpath(canonical, destination.parent)) os.replace(temporary, destination) finally: if temporary.is_symlink(): temporary.unlink() def crawl(scope, stage, args, report): opener = build_opener(ScopedRedirect(scope)) queue, visited, local_paths = deque([scope.base]), set(), {} checksums, canonical_files = {}, {} while queue: url = queue.popleft() if url in visited: continue visited.add(url) candidate = scope.candidate(scope.base, url) if candidate is None: continue url, relative = candidate suffix = Path(relative).suffix.lower() is_html = urlsplit(url).path.endswith("/") or suffix in HTML_EXTENSIONS is_image = suffix in IMAGE_EXTENSIONS if args.assets_only and not (is_html or is_image): report["skipped_urls"].append(url) continue if relative in local_paths and local_paths[relative] != url: raise MirrorError(f"多个 URL 对应同一本地路径：{relative}") local_paths[relative] = url destination = stage / relative expected_digest = checksums.get(relative) if not is_html else None canonical = next((file for (digest, _), file in canonical_files.items() if digest == expected_digest), None) if canonical is not None: print(f"链接 {relative} -> {canonical['path']}（SHA256 一致）", flush=True) metadata = {**canonical, "source_url": url, "reused_from_url": canonical["source_url"], "linked_at": now(), "link_target": canonical["path"], "transfer": "checksum-reuse", "transferred_bytes": 0} charset = "utf-8" else: print(f"下载 {relative}", flush=True) expected = "html" if is_html else "image" if args.assets_only else None metadata, charset = download(opener, url, destination, args.timeout, args.retries, expected) metadata["path"] = relative identity = (metadata["sha256"], metadata["size"]) canonical = canonical_files.get(identity) if canonical is not None: link_duplicate(stage, destination, canonical["path"]) metadata["link_target"] = canonical["path"] else: canonical_files[identity] = metadata report["files"].append(metadata) if relative == "sha256sums.txt" and not args.assets_only: checksums = checksum_inventory(destination) if metadata["content_type"] in HTML_TYPES: parser = Links() parser.feed(destination.read_bytes().decode(charset, errors="replace")) for reference in sorted(set(parser.references)): link = scope.candidate(metadata["final_url"], reference) if link is not None and link[0] not in visited: # Obtain the small, fresh root checksum file before ISO aliases. if link[1] == "sha256sums.txt" and not args.assets_only: queue.appendleft(link[0]) else: queue.append(link[0]) def save_report(directory, report): directory.mkdir(parents=True, exist_ok=True) write_json(directory / "manifest.json", report) fields = ["path", "source_url", "size", "sha256", "content_type", "downloaded_at", "link_target", "transfer", "transferred_bytes"] with (directory / "files.csv").open("w", newline="", encoding="utf-8-sig") as stream: writer = csv.DictWriter(stream, fieldnames=fields, extrasaction="ignore") writer.writeheader() writer.writerows(sorted(report["files"], key=lambda file: file["path"])) lines = [f"运行：{report['run_id']}", f"状态：{report['status']}", f"源：{report['source']}", f"模式：{report['mode']}", f"文件数：{len(report['files'])}", f"逻辑字节数：{sum(file['size'] for file in report['files'])}", f"HTTP 正文下载字节数：{sum(file.get('transferred_bytes', file['size']) for file in report['files'])}", f"去重后内容字节数：{sum(file['size'] for file in report['files'] if not file.get('link_target'))}", "", "文件列表："] lines += [f"{file['size']:>12} {file['sha256']} {file['path']}" + (f" -> {file['link_target']}" if file.get('link_target') else "") for file in sorted(report["files"], key=lambda file: file["path"])] for key, title in [("added", "新增清单"), ("changed", "内容变更清单"), ("removed", "移除清单（是否生效见运行状态）")]: lines += ["", title + "：", *report.get(key, [])] if report.get("error"): lines += ["", "错误：" + report["error"]] (directory / "summary.txt").write_text("\n".join(lines) + "\n", encoding="utf-8") def run(args): scope = Scope(args.source) output = safe_path(args.output) records = safe_path(args.records or output.with_name(output.name + ".records")) cwd = Path.cwd().resolve() if output == Path(output.anchor) or output == Path.home().resolve() or output == cwd or output in cwd.parents: raise MirrorError("请使用专用子目录作为镜像目录") if output == records or output in records.parents or records in output.parents: raise MirrorError("记录目录必须与镜像目录相互独立") output.parent.mkdir(parents=True, exist_ok=True) lock_path = output.with_name("." + output.name + ".lock") if lock_path.is_symlink(): raise MirrorError("锁文件不得是符号链接") with lock_path.open("a") as lock: try: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: raise MirrorError("同一镜像目录已有任务正在运行") previous = {"files": []} if output.exists(): if not output.is_dir(): raise MirrorError("镜像路径必须是目录") marker = output / MARKER if marker.is_symlink(): raise MirrorError("镜像状态文件不得是符号链接") if marker.is_file(): previous = json.loads(marker.read_text(encoding="utf-8")) if previous.get("schema_version") != 1 or previous.get("source") != scope.base: raise MirrorError("现有镜像的来源或状态不匹配，请选择新的输出目录") elif any(output.iterdir()): raise MirrorError("拒绝清理非空且未经本脚本管理的目录，请选择新的输出目录") records.mkdir(parents=True, exist_ok=True) run_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ-") + uuid.uuid4().hex[:8] report = {"schema_version": 1, "run_id": run_id, "status": "running", "source": scope.base, "output": str(output), "mode": "assets-only" if args.assets_only else "full", "started_at": now(), "previous_run_id": previous.get("run_id"), "files": [], "skipped_urls": []} stage = Path(tempfile.mkdtemp(prefix="." + output.name + ".stage-", dir=output.parent)) backup = output.with_name("." + output.name + ".previous-" + run_id) published = False try: crawl(scope, stage, args, report) old = {file["path"]: file for file in previous["files"]} new = {file["path"]: file for file in report["files"]} report.update(added=sorted(new.keys() - old.keys()), removed=sorted(old.keys() - new.keys()), changed=sorted(path for path in old.keys() & new.keys() if old[path]["sha256"] != new[path]["sha256"])) # Persist the inventory before the directory switch; incomplete runs stay identifiable. report["status"] = "prepared" save_report(records / run_id, report) state = {key: report[key] for key in ["schema_version", "run_id", "source", "mode", "files"]} write_json(stage / MARKER, state) if output.exists(): os.replace(output, backup) try: os.replace(stage, output) published = True except BaseException: if backup.exists(): os.replace(backup, output) raise if backup.exists(): shutil.rmtree(backup) report.update(status="success", finished_at=now()) save_report(records / run_id, report) write_json(records / "latest.json", {"run_id": run_id, "manifest": str(records / run_id / "manifest.json")}) print(f"完成：{len(new)} 个文件；清理 {len(report['removed'])} 个旧文件") print(f"镜像：{output}\n备查：{records / run_id}") return 0 except BaseException as error: report.update(status="published_with_error" if published else "failed", finished_at=now(), error=str(error) or type(error).__name__) try: save_report(records / run_id, report) except OSError as report_error: print(f"无法写入失败记录：{report_error}", file=sys.stderr) if published: print(f"镜像已切换，但后续整理失败；检查记录及备份目录：{backup}", file=sys.stderr) raise finally: if stage.exists(): shutil.rmtree(stage) def main(argv=None): parser = argparse.ArgumentParser(description="镜像 Arch Linux ISO 目录，成功后清理旧文件并保存下载清单。") parser.add_argument("--source", default=DEFAULT_SOURCE, help="镜像源目录 URL") parser.add_argument("--output", default="./mirror", help="专用镜像目录（默认 ./mirror）") parser.add_argument("--assets-only", action="store_true", help="仅下载目录 HTML、HTML 页面及图片，不下载 ISO 等其他文件") parser.add_argument("--records", help="备查记录目录（默认 <镜像目录>.records）") parser.add_argument("--timeout", type=float, default=60, help="单次网络读写超时秒数（默认 60）") parser.add_argument("--retries", type=int, default=2, help="失败重试次数（默认 2）") args = parser.parse_args(argv) if args.timeout <= 0 or args.retries < 0: parser.error("timeout 必须大于 0，retries 必须大于等于 0") try: return run(args) except (MirrorError, OSError, ValueError) as error: print(f"错误：{error}", file=sys.stderr) return 1 except KeyboardInterrupt: print("已中断；尚未切换时旧镜像保持不变。", file=sys.stderr) return 130 if __name__ == "__main__": sys.exit(main())