Jenkins-py/artifacts.py

"""
artifacts.py - 构建产物管理：日志、快照、QAC、FTP准备、发布基线。
对应原 Shell 脚本的 Collecting commit log, qac report, ftp upload,
commit_snapshot_manifest 等部分。
"""
from __future__ import annotations          # 允许延迟类型注解求值
import os
import shutil
import subprocess
import logging
import tempfile
import xml.etree.ElementTree as ET
from pathlib import Path
from typing import Tuple, List

from gerrit import _run_ssh                  # 直接使用 gerrit 提供的 SSH 工具

logger = logging.getLogger(__name__)


class ArtifactError(Exception):
    """产物处理异常"""
    pass


# ---------------------------------------------------------------------------
# 1. 提交日志收集
# ---------------------------------------------------------------------------
def collect_logs(repo_dir: str, output_csv: str) -> None:
    """收集近 7 天的提交日志，保存为 CSV 格式"""
    repo_dir = Path(repo_dir)
    if not (repo_dir / '.repo').is_dir():
        logger.warning("No .repo dir in %s, skip collect_logs", repo_dir)
        return

    script_content = '''#!/bin/sh
git log --no-merges --after="$(date +"%Y-%m-%d %H:%M:%S" -d "-7 day")" \
    --date=format:'%Y-%m-%d %H:%M:%S' \
    --pretty=format:'"'$(basename $(pwd))'","%H","%an","%ae","%ad","%cd","%s"%n'
'''
    _run_repo_forall_with_script(repo_dir, script_content, output_csv)


def collect_newest_one_commit(repo_dir: str, output_csv: str) -> None:
    """收集每个项目的最新一次提交，保存为 CSV 格式"""
    repo_dir = Path(repo_dir)
    if not (repo_dir / '.repo').is_dir():
        logger.warning("No .repo dir in %s, skip collect_newest_one_commit", repo_dir)
        return

    script_content = '''#!/bin/sh
git log --no-merges -n 1 --date=format:'%Y-%m-%d %H:%M:%S' \
    --pretty=format:'"'$(basename $(pwd))'","%H","%an","%ae","%ad","%cd","%s"%n'
'''
    _run_repo_forall_with_script(repo_dir, script_content, output_csv)


def _run_repo_forall_with_script(repo_dir: Path, script: str, output_csv: str) -> None:
    """创建临时脚本，用 repo forall -c 执行，结果追加到指定 CSV 文件"""
    with tempfile.NamedTemporaryFile(mode='w', suffix='.sh', delete=False) as f:
        f.write(script)
        f.flush()
        script_path = f.name

    try:
        cmd = ['repo', 'forall', '-c', f'sh {script_path}']
        with open(output_csv, 'a') as out:
            subprocess.run(cmd, cwd=str(repo_dir), stdout=out, check=True)
    finally:
        os.unlink(script_path)


# ---------------------------------------------------------------------------
# 2. 审查记录 & 快照
# ---------------------------------------------------------------------------
def collect_review_records(repo_dir: str, output_dir: str) -> None:
    """
    从 Gerrit 获取已合并的变更历史，并生成当前代码快照。
    对应原 Shell 的 get_repo_history_from_gerrit + create_snapshot。
    """
    repo_dir = Path(repo_dir)
    repo_name = repo_dir.name
    output_dir = Path(output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)

    # 获取所有 project 名称和分支
    manifest = repo_dir / '.repo' / 'manifests' / 'default.xml'
    if not manifest.is_file():
        logger.warning("Manifest not found in %s, skipping review records", repo_dir)
        return

    projects, branch = _parse_manifest(manifest)

    # Gerrit 连接信息 —— 直接从环境变量读取（与 config.py 保持一致的默认值）
    host = os.environ.get('GERRIT_HOST', '10.2.90.253')
    port = int(os.environ.get('GERRIT_PORT', '29418'))
    user = os.environ.get('GERRIT_NAME', 'jenkins')

    # 生成 Gerrit 历史文件
    history_file = output_dir / f"{repo_name}_merged_history.txt"
    with open(history_file, 'w') as f:
        for project in projects:
            try:
                result = _get_history_to_str(host, port, user, project, branch)
                f.write(result)
            except Exception as e:
                logger.warning("Failed to get history for %s: %s", project, e)

    # 生成快照 XML
    snapshot_xml = output_dir / f"{repo_name}_snapshot.xml"
    subprocess.run(['repo', 'manifest', '-r', '-o', str(snapshot_xml)],
                   cwd=str(repo_dir), check=True)


def _parse_manifest(manifest_file: Path) -> Tuple[List[str], str]:
    """从 default.xml 提取所有 project name 和 default revision"""
    tree = ET.parse(manifest_file)
    root = tree.getroot()
    projects = [elem.get('name') for elem in root.findall('project')]
    branch = root.find('default').get('revision', 'master')
    return projects, branch


def _get_history_to_str(host: str, port: int, user: str, project: str, branch: str) -> str:
    """获取 Gerrit 历史并返回字符串（内部使用）"""
    query = f"query --format=TEXT status:merged project:{project} branch:{branch} after:2020-01-01"
    return _run_ssh(host, port, user, query)


# ---------------------------------------------------------------------------
# 3. QAC 报告收集
# ---------------------------------------------------------------------------
def collect_qac_report(mcu_sdk_dir: str, output_dir: str) -> None:
    """复制 MCU SDK 中的 QAC 报告到输出目录"""
    qac_source = Path(mcu_sdk_dir) / '..' / 'qac_report' / 'qac_output' / 'soc' / 'sdk_la'
    if not qac_source.is_dir():
        logger.info("No QAC report found at %s, skipping", qac_source)
        return

    output_dir = Path(output_dir)
    if output_dir.exists():
        shutil.rmtree(output_dir)
    output_dir.mkdir(parents=True)

    # 复制所有 html 和 xlsx 文件
    for root, dirs, files in os.walk(qac_source):
        for file in files:
            if any(file.lower().endswith(ext) for ext in ['.html', '.xlsx']):
                src_path = Path(root) / file
                rel_path = src_path.relative_to(qac_source)
                dest_path = output_dir / rel_path
                dest_path.parent.mkdir(parents=True, exist_ok=True)
                shutil.copy2(src_path, dest_path)
    logger.info("QAC report copied to %s", output_dir)


# ---------------------------------------------------------------------------
# 4. FTP 上传准备
# ---------------------------------------------------------------------------
def prepare_ftp_upload(cfg, collect_log_file, newest_log_file,
                       review_records_dir, qac_report_dir) -> None:
    """
    将所有构建产物移动到本地 FTP 目录，并生成 env.properties。
    对应原 Shell 的 prepare_for_upload_ftp()。
    """
    # 确定 SOC 部署输出目录
    soc_dir = Path(cfg.soc_dir)
    deploy_root = soc_dir / 'out' / 'deploy'
    if not deploy_root.is_dir():
        raise ArtifactError("No deploy directory found in SOC output")
    deploy_dirs = list(deploy_root.iterdir())
    if not deploy_dirs:
        raise ArtifactError("No deploy subdirectories found")
    deploy_dir_name = deploy_dirs[0].name

    image_deploy_local = Path(cfg.deploy_image_local_dir) / cfg.repo_branch / deploy_dir_name
    image_deploy_local.mkdir(parents=True, exist_ok=True)

    # 移动整个 deploy 目录
    src = deploy_dirs[0]
    for item in os.listdir(src):
        src_item = src / item
        dst_item = image_deploy_local / item
        if src_item.is_dir():
            shutil.move(str(src_item), str(dst_item))
        else:
            shutil.move(str(src_item), str(dst_item))
    logger.info("Deploy files moved to %s", image_deploy_local)

    # 决定是否需要测试提示
    test_msg = "无需测试"
    commit_log_filename = "no_commit"
    if Path(collect_log_file).is_file():
        with open(collect_log_file, 'r') as f:
            lines = f.readlines()
            # 排除 auto commit 行
            non_auto = [line for line in lines if 'auto commit by server' not in line]
            if len(non_auto) > 1:  # 至少有一行有效信息（不含表头）
                test_msg = "需要测试"
                commit_log_filename = Path(collect_log_file).name
                shutil.copy2(collect_log_file, image_deploy_local)

    # 复制其他文件
    for src_file, dst_name in [
        (newest_log_file, Path(newest_log_file).name) if Path(newest_log_file).is_file() else None,
        (review_records_dir, Path(review_records_dir).name) if Path(review_records_dir).is_dir() else None,
        (qac_report_dir, Path(qac_report_dir).name) if Path(qac_report_dir).is_dir() else None,
    ]:
        if src_file is None:
            continue
        src_path = Path(src_file)
        dst_path = image_deploy_local / dst_name
        if src_path.is_dir():
            shutil.copytree(src_path, dst_path, dirs_exist_ok=True)
        else:
            shutil.copy2(src_path, dst_path)

    # 构建时间戳
    if '[' in deploy_dir_name:
        image_build_time = deploy_dir_name.split('[')[1].split(']')[0].replace('-', '')
    else:
        from datetime import datetime
        image_build_time = datetime.now().strftime('%Y%m%d%H%M%S')
    cfg.image_build_time = image_build_time  # 存回 config 备用

    # 生成 env.properties
    ftp_user = os.environ.get('ftpUser', 'ftpuser')
    ftp_password = os.environ.get('ftpPasword', '123456')
    ftp_addr = os.environ.get('ftpAddr', '10.2.90.252')
    ftp_port = os.environ.get('ftpPort', '21')
    ftp_root = f"DB-{datetime.now().year}/{cfg.ftp_root}"
    deploy_rel = f"{cfg.repo_branch}/{deploy_dir_name}"
    ftp_deploy_url = f"ftp://{ftp_user}:{ftp_password}@{ftp_addr}:{ftp_port}/{ftp_root}/{deploy_rel}"
    ftp_commit_log_url = f"{ftp_deploy_url}/{commit_log_filename}" if commit_log_filename != "no_commit" else ""

    env_content = f"""store.repo_branch={cfg.repo_branch}
store.deploy_dir={ftp_deploy_url}
store.commit_log={ftp_commit_log_url}
store.test_msg={test_msg}
"""
    env_file = image_deploy_local / 'env.properties'
    with open(env_file, 'w') as f:
        f.write(env_content)
    logger.info("env.properties generated at %s", env_file)


# ---------------------------------------------------------------------------
# 5. 发布基线 manifest 提交
# ---------------------------------------------------------------------------
def commit_snapshot_manifest(repo_dir: str, version: str) -> None:
    """生成 release_<version>.xml 并提交到 manifest 仓库的相应分支"""
    if not version:
        logger.info("No release version specified, skip commit manifest")
        return

    repo_dir = Path(repo_dir)
    manifests_repo = repo_dir / '.repo' / 'manifests'
    if not (manifests_repo / '.git').is_dir():
        logger.warning("No manifests git repo in %s, skip commit manifest", repo_dir)
        return

    release_dir = manifests_repo / 'release'
    release_dir.mkdir(parents=True, exist_ok=True)

    # 生成快照 XML
    snapshot_file = release_dir / f'release_{version}.xml'
    subprocess.run(['repo', 'manifest', '-r', '-o', str(snapshot_file)],
                   cwd=str(repo_dir), check=True)

    # 提交并推送
    current_branch = _get_current_branch(manifests_repo)
    subprocess.run(['git', 'add', f'release/release_{version}.xml'],
                   cwd=str(manifests_repo), check=True)
    try:
        subprocess.run(['git', 'commit', '-m', f'Auto commit by server: add release_{version}.xml'],
                       cwd=str(manifests_repo), check=True, capture_output=True)
    except subprocess.CalledProcessError as e:
        if 'nothing to commit' in e.stderr.decode().lower():
            logger.info("Nothing to commit, skipping push for release manifest")
            return
        raise

    # 推送到当前分支（通常为 REPO_BRANCH 或默认分支）
    push_branch = os.environ.get('REPO_BRANCH', current_branch)
    subprocess.run(['git', 'push', 'origin', f'HEAD:{push_branch}'],
                   cwd=str(manifests_repo), check=True)
    logger.info("Release manifest %s pushed to %s", snapshot_file.name, push_branch)


def _get_current_branch(repo_dir: Path) -> str:
    """获取仓库当前分支"""
    result = subprocess.run(['git', 'rev-parse', '--abbrev-ref', 'HEAD'],
                            cwd=str(repo_dir), capture_output=True, text=True, check=True)
    return result.stdout.strip()