版本 {config.get('VERSION', '')}
安装 App" "", encoding="utf-8", ) def _remote_paths(config: dict) -> tuple[str, str, str, str]: stem = _artifact_stem(config) folder = config.get("OSS_FLODER", "ios-builds").strip("/") or "ios-builds" root = posixpath.join(folder, "iOS") return ( posixpath.join(root, f"{stem}.ipa"), posixpath.join(root, f"{stem}.plist"), posixpath.join(root, f"{stem}.html"), posixpath.join(root, f"{stem}.png"), ) def _delete_webdav(config: dict, remote_paths: list[str]): webdav = config.get("webdav", {}) server_url = webdav.get("server_url", "").rstrip("/") if not server_url or not webdav.get("username"): raise DistributionError("WebDAV 配置不完整") base_path = webdav.get("base_path", "/ios-builds").strip("/") auth = (webdav.get("username", ""), webdav.get("password", "")) with httpx.Client(auth=auth, timeout=120, follow_redirects=True) as client: for relative_path in remote_paths: remote_path = posixpath.join(base_path, relative_path) response = client.delete(f"{server_url}/{remote_path}") # 404 代表文件已不存在，可视为清理完成。 if response.status_code not in (200, 202, 204, 404): raise DistributionError(f"WebDAV 删除失败: {relative_path} ({response.status_code})") def _delete_oss(config: dict, remote_paths: list[str]): oss = config.get("oss", {}) required = ["access_key_id", "access_key_secret", "endpoint", "bucket_name"] if any(not oss.get(key) for key in required): raise DistributionError("OSS 配置不完整") try: import oss2 except ImportError as exc: raise DistributionError("未安装 oss2，请重新执行 ./deploy.sh build") from exc bucket = oss2.Bucket( oss2.Auth(oss["access_key_id"], oss["access_key_secret"]), oss["endpoint"], oss["bucket_name"], connect_timeout=30, ) for relative_path in remote_paths: result = bucket.delete_object(relative_path) if result.status // 100 != 2: raise DistributionError(f"OSS 删除失败: {relative_path}") def _remote_paths_from_published_url( build_config: dict, upload_config: dict, published_url: str, ) -> list[str] | None: """从已保存的公开链接还原实际对象键，兼容历史命名规则。""" parsed = urlparse(published_url) if not parsed.path: return None remote_path = unquote(parsed.path).lstrip("/") # base_url 允许带路径前缀；该前缀是公开地址的一部分，不属于 OSS 对象键。 base_url = upload_config.get("oss", {}).get("base_url", "") base = urlparse(base_url) if base_url and base.netloc == parsed.netloc: base_path = unquote(base.path).strip("/") if base_path and remote_path.startswith(f"{base_path}/"): remote_path = remote_path[len(base_path) + 1:] path = Path(remote_path) if path.suffix not in {".ipa", ".plist", ".html", ".png"}: return None if build_config.get("BUILD_TYPE") == "App_Store": return [str(path.with_suffix(".ipa"))] return [str(path.with_suffix(suffix)) for suffix in (".ipa", ".plist", ".html", ".png")] def delete_published_artifacts( build_config: dict, upload_config: dict, published_url: str = "", ): """删除任务对应的远端分发产物。 优先按任务保存的下载链接还原对象键。这样即使后续升级了文件命名 规则，或历史快照含有新的分支字段，仍会删除当时实际上传的文件。 """ mode = upload_config.get("mode", "") if mode not in {"oss", "webdav"}: raise DistributionError("请选择 OSS 或 WebDAV 上传方式") remote_paths = None if mode == "oss" and published_url: remote_paths = _remote_paths_from_published_url( build_config, upload_config, published_url, ) if not remote_paths: ipa_remote, manifest_remote, html_remote, qr_remote = _remote_paths(build_config) remote_paths = [ipa_remote] if build_config.get("BUILD_TYPE") != "App_Store": remote_paths.extend([manifest_remote, html_remote, qr_remote]) deleter = _delete_oss if mode == "oss" else _delete_webdav deleter(upload_config, remote_paths) def _ensure_webdav_dirs(client: httpx.Client, server_url: str, remote_path: str): path = "" for segment in remote_path.strip("/").split("/")[:-1]: path = f"{path}/{segment}" response = client.request("MKCOL", f"{server_url.rstrip('/')}{path}") if response.status_code not in (200, 201, 204, 301, 405): raise DistributionError(f"创建 WebDAV 目录失败: {path} ({response.status_code})") def _upload_webdav(config: dict, files: list[tuple[Path, str]]) -> dict[str, str]: webdav = config.get("webdav", {}) server_url = webdav.get("server_url", "").rstrip("/") public_url = (webdav.get("public_url") or server_url).rstrip("/") if not server_url or not webdav.get("username"): raise DistributionError("WebDAV 配置不完整") base_path = webdav.get("base_path", "/ios-builds").strip("/") auth = (webdav.get("username", ""), webdav.get("password", "")) urls = {} with httpx.Client(auth=auth, timeout=120, follow_redirects=True) as client: for local_path, relative_path in files: remote_path = posixpath.join(base_path, relative_path) _ensure_webdav_dirs(client, server_url, remote_path) with local_path.open("rb") as f: response = client.put(f"{server_url}/{remote_path}", content=f.read()) if response.status_code not in (200, 201, 204): raise DistributionError(f"WebDAV 上传失败: {local_path.name} ({response.status_code})") urls[local_path.suffix] = f"{public_url}/{quote(remote_path)}" return urls def _upload_oss(config: dict, files: list[tuple[Path, str]]) -> dict[str, str]: oss = config.get("oss", {}) required = ["access_key_id", "access_key_secret", "endpoint", "bucket_name"] if any(not oss.get(key) for key in required): raise DistributionError("OSS 配置不完整") try: import oss2 except ImportError as exc: raise DistributionError("未安装 oss2，请重新执行 ./deploy.sh build") from exc bucket = oss2.Bucket( oss2.Auth(oss["access_key_id"], oss["access_key_secret"]), oss["endpoint"], oss["bucket_name"], connect_timeout=30, ) base_url = (oss.get("base_url") or f"https://{oss['bucket_name']}.{oss['endpoint']}").rstrip("/") urls = {} for local_path, relative_path in files: result = bucket.put_object_from_file(relative_path, str(local_path)) if result.status // 100 != 2: raise DistributionError(f"OSS 上传失败: {local_path.name}") urls[local_path.suffix] = f"{base_url}/{quote(relative_path)}" return urls def publish_ipa(config: dict, ipa_path: Path, build_dir: Path) -> tuple[str, str]: """发布 IPA，返回下载 URL 与二维码公开 URL。""" upload = config.get("_upload_config", {}) mode = upload.get("mode", "") if mode not in {"oss", "webdav"}: raise DistributionError("请选择 OSS 或 WebDAV 上传方式") output_dir = build_dir / "distribution" ipa_file, manifest, html = _write_distribution_files(config, ipa_path, output_dir) ipa_remote, manifest_remote, html_remote, qr_remote = _remote_paths(config) uploader = _upload_oss if mode == "oss" else _upload_webdav # App Store 包只需要提供 IPA 下载地址，不能生成 itms-services 安装页。 if config.get("BUILD_TYPE") == "App_Store": urls = uploader(upload, [(ipa_file, ipa_remote)]) return urls[".ipa"], "" # Ad Hoc manifest 依赖 IPA URL，先发布 IPA。 urls = uploader(upload, [(ipa_file, ipa_remote)]) _write_manifest(config, manifest, urls[".ipa"]) urls.update(uploader(upload, [(manifest, manifest_remote)])) _write_download_page(config, html, urls[".plist"]) urls.update(uploader(upload, [(html, html_remote)])) qr_path = output_dir / f"{_artifact_stem(config)}.png" try: import qrcode except ImportError as exc: raise DistributionError("未安装 qrcode，请重新执行 ./deploy.sh build") from exc qrcode.make(urls[".html"]).save(qr_path) urls.update(uploader(upload, [(qr_path, qr_remote)])) return urls[".html"], urls[".png"]