Jenkins-py/gerrit.py

import subprocess
import logging
import os
from typing import List, Dict, Optional

logger = logging.getLogger(__name__)


class GerritError(Exception):
    """Gerrit 操作异常"""
    pass


def _run_ssh(host: str, port: int, user: str, command: str) -> str:
    """执行 Gerrit SSH 命令并返回标准输出"""
    cmd = [
        'ssh',
        '-p', str(port),
        f'{user}@{host}',
        'gerrit', *command.split()
    ]
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, check=True)
        return result.stdout
    except subprocess.CalledProcessError as e:
        error_msg = e.stderr.strip()
        logger.error("Gerrit SSH command failed: %s", error_msg)
        raise GerritError(error_msg) from e


def query_changes(host: str, port: int, user: str,
                  change_ids: List[str]) -> List[Dict[str, str]]:
    """
    查询 Gerrit 上的变更信息。

    Args:
        host: Gerrit 主机
        port: SSH 端口
        user: Gerrit 用户名
        change_ids: 变更 ID 列表（如 ['12345', '12346']）

    Returns:
        每个变更的信息字典列表，包含 change_id, project, revision, ref
    """
    changes_info = []
    for cid in change_ids:
        output = _run_ssh(host, port, user,
                          f"query --current-patch-set status:open change:{cid}")

        project = revision = ref = None
        for line in output.splitlines():
            line = line.strip()
            if line.startswith('project:'):
                project = line.split(':', 1)[1].strip()
            elif line.startswith('revision:'):
                revision = line.split(':', 1)[1].strip()
            elif line.startswith('ref:') and cid in line:
                # ref 行格式：ref: refs/changes/xx/12345/1
                ref = line.split('ref:', 1)[1].strip()
                # 确保提取到 refs/changes/... 部分
                if 'refs/' in ref:
                    ref = ref[ref.index('refs/'):]

        if not project or not revision:
            raise GerritError(f"Cannot parse project/revision for change {cid}")

        changes_info.append({
            'change_id': cid,
            'project': project,
            'revision': revision,
            'ref': ref or ''
        })
        logger.info("Change %s: project=%s, revision=%s", cid, project, revision)

    return changes_info


def _find_repo_root(project: str, mcu_dir: str, soc_dir: str) -> Optional[str]:
    """
    根据 project 名称，找到它属于 MCU 还是 SOC 工程。
    优先检查 manifest/default.xml 中是否包含该项目。
    """
    for repo_dir in [soc_dir, mcu_dir]:
        manifest = os.path.join(repo_dir, '.repo', 'manifests', 'default.xml')
        if os.path.isfile(manifest):
            try:
                with open(manifest, 'r') as f:
                    if project in f.read():
                        return repo_dir
            except Exception:
                continue
    # 备选：使用 repo list 命令查找
    for repo_dir in [soc_dir, mcu_dir]:
        try:
            result = subprocess.run(
                ['repo', 'list'], cwd=repo_dir,
                capture_output=True, text=True, check=False
            )
            if project in result.stdout:
                return repo_dir
        except Exception:
            continue

    return None


def apply_changes(changes_info: List[Dict], mcu_dir: str, soc_dir: str):
    """
    将 Gerrit 变更通过 repo download 应用到对应的工程。

    Args:
        changes_info: query_changes 的返回结果
        mcu_dir: MCU 工程的根目录
        soc_dir: SOC 工程的根目录
    """
    for info in changes_info:
        project = info['project']
        ref = info['ref']

        repo_root = _find_repo_root(project, mcu_dir, soc_dir)
        if repo_root is None:
            raise GerritError(f"Project {project} not found in MCU or SOC manifest")

        # ref 可能带前缀，确保为 refs/changes/... 的形式
        if not ref.startswith('refs/'):
            ref = 'refs/changes/' + ref.lstrip('/')
        # repo download 需要 'project_path change/ref' 的形式
        # 如果 ref 已包含 'refs/changes/...'，直接用；否则拼接
        download_ref = f"{ref}" if '/' in ref else f"refs/changes/{ref}"
        # 原脚本中 download_gerrit_project 使用了 change$change_index，即直接使用查询到的 ref 部分
        # 这里我们直接使用 ref 即可（通常是 refs/changes/<NN>/<change>/<patchset>）

        cmd = ['repo', 'download', project, ref]
        logger.info("Applying change %s: %s", info['change_id'], ' '.join(cmd))
        try:
            subprocess.run(cmd, cwd=repo_root, check=True)
        except subprocess.CalledProcessError as e:
            raise GerritError(f"repo download failed for {info['change_id']}: {e}") from e


def label_verified(host: str, port: int, user: str,
                   changes_info: List[Dict]):
    """
    给所有变更打上 Verified +1 标签。

    Args:
        changes_info: query_changes 的返回结果
    """
    for info in changes_info:
        revision = info['revision']
        _run_ssh(host, port, user, f"review --verified +1 {revision}")
        logger.info("Verified +1 on change %s", info['change_id'])


def get_history(host: str, port: int, user: str,
                project: str, branch: str, output_file: str):
    """
    查询已合并的评审记录，并保存到指定文件。

    Args:
        project: Git 项目名（如 INVO/MCU/APP/canbus_mng）
        branch: 分支名
        output_file: 输出文件路径
    """
    # 查询已合并的状态，起始时间定为较早的时间，确保覆盖所有历史
    query = f"query --format=TEXT status:merged project:{project} branch:{branch} after:2020-01-01"
    output = _run_ssh(host, port, user, query)
    with open(output_file, 'w', encoding='utf-8') as f:
        f.write(output)
    logger.info("Saved merged history for %s to %s", project, output_file)