Jenkins-py/repo_manager.py

import os
import subprocess
import logging
from pathlib import Path
from typing import Optional, List

logger = logging.getLogger(__name__)


class RepoError(Exception):
    """Repo 操作异常"""
    pass


def _run(cmd: List[str], cwd: Optional[str] = None) -> None:
    """执行命令，失败时抛出异常，同时输出 stdout/stderr 到控制台"""
    logger.debug("Running: %s", ' '.join(cmd))
    subprocess.run(cmd, cwd=cwd, check=True)


def _repo_init(manifest_url: str, branch: str, repo_url: str,
               manifest_file: Optional[str] = None, depth: int = 7,
               cwd: Optional[str] = None):
    """
    执行 `repo init`。
    可选参数 `manifest_file` 用于指定 release manifest（相对路径，如 'release.xml' 或 'release/release_xxx.xml'）
    """
    cmd = [
        'repo', 'init',
        f'--depth={depth}',
        '-u', manifest_url,
        '-b', branch,
        '--repo-url', repo_url,
    ]
    if manifest_file:
        cmd += ['-m', manifest_file]
    _run(cmd, cwd=cwd)


def _repo_sync(jobs: int = 4, cwd: Optional[str] = None):
    """
    执行 `repo sync`，参数与 Shell 版本完全一致：
    -c: 只下载当前分支
    -d: 分离到 manifest 指定版本
    -f: 即使某个项目失败也继续
    --force-sync: 覆盖错误的 object 目录
    --no-tags: 不下载 tags
    --prune: 删除不存在远端分支的本地分支
    """
    cmd = [
        'repo', 'sync',
        '-c', '-d', '-f',
        '--force-sync',
        '--no-tags',
        '--prune',
        f'-j{jobs}',
    ]
    _run(cmd, cwd=cwd)


def _get_release_manifest_path(project_dir: str, version: str, snapshot: str) -> Optional[str]:
    """
    根据快照配置返回需要使用的 manifest 文件参数（相对路径）。
    返回 None 表示不需要额外 manifest。
    逻辑完全对应原 Shell 中的 update_repo_extra_para()。
    """
    if not version:
        return None

    manifests_dir = Path(project_dir) / '.repo' / 'manifests'
    release_dir = manifests_dir / 'release'

    # 默认使用 release.xml（原脚本：当 version 非空时，先设置 -m release.xml）
    manifest_file = 'release.xml'

    if snapshot != 'default':
        specific_file = f"release_{version}.xml"
        specific_path = release_dir / specific_file
        if specific_path.is_file():
            # 使用具体快照文件
            manifest_file = f"release/{specific_file}"
        else:
            raise RepoError(f"Snapshot manifest not found: {specific_path}")
    return manifest_file


def download_project(project_dir: str, manifest_url: str, branch: str,
                     repo_url: str, version: Optional[str] = None,
                     snapshot: str = 'default', sync_jobs: int = 4):
    """
    下载一个 repo 工程到指定目录。
    对应原 Shell 中的 download_mcu / download_soc。
    """
    target = Path(project_dir)
    target.mkdir(parents=True, exist_ok=True)

    # 第一次 repo init（不带 -m）
    _repo_init(manifest_url, branch, repo_url, depth=7, cwd=str(target))

    # 是否需要额外的 manifest（release.xml 或 release/release_xxx.xml）
    try:
        extra_manifest = _get_release_manifest_path(str(target), version, snapshot)
        if extra_manifest:
            logger.info("Using extra manifest: %s", extra_manifest)
            # 第二次 repo init 带上 -m 参数（覆盖）
            _repo_init(manifest_url, branch, repo_url,
                       manifest_file=extra_manifest, depth=7, cwd=str(target))
    except RepoError:
        logger.exception("Failed to determine release manifest")
        raise

    # repo sync
    _repo_sync(jobs=sync_jobs, cwd=str(target))