Jenkins-py/utils.py

"""
utils.py - 通用工具函数，包括：
  - 带重试的命令执行（解决 License 不足等临时错误）
  - 简单命令执行
  - CSV 文件清理与表头添加
"""
import subprocess
import time
import re
import logging
from pathlib import Path
from typing import List, Optional

logger = logging.getLogger(__name__)


class CommandError(Exception):
    """命令执行失败异常"""
    pass


# ---------------------------------------------------------------------------
# 1. 带重试的命令执行
# ---------------------------------------------------------------------------
def run_command_with_retry(
    cmd_parts: List[str],
    cwd: Optional[str] = None,
    retry_count: int = 5,
    retry_pattern: Optional[str] = None,
    log_file: Optional[str] = None
) -> None:
    """
    执行命令，支持基于输出内容的重试。

    Args:
        cmd_parts: 命令及参数列表，如 ['make', 'all']
        cwd: 工作目录
        retry_count: 最大重试次数（尝试总次数 = 1 + retry_count？）
                     本函数实现为：最多重试 retry_count 次，
                     即总共执行 retry_count+1 次。
        retry_pattern: 用于匹配 stdout/stderr 的正则表达式；
                       如果匹配成功且命令失败，则重试。
        log_file: 可选路径，命令输出同时写入此文件。
    """
    attempt = 0
    max_attempts = retry_count + 1   # 总共尝试次数，至少1次

    while attempt < max_attempts:
        attempt += 1
        logger.debug("Running (attempt %d/%d): %s", attempt, max_attempts, ' '.join(cmd_parts))
        result = subprocess.run(cmd_parts, cwd=cwd, capture_output=True, text=True)

        # 输出到控制台（可实时显示，这里偷懒了，捕获完后统一打印）
        if result.stdout:
            logger.info(result.stdout.rstrip())
        if result.stderr:
            logger.error(result.stderr.rstrip())

        # 写入日志文件
        if log_file:
            Path(log_file).parent.mkdir(parents=True, exist_ok=True)
            with open(log_file, 'w', encoding='utf-8') as f:
                f.write(result.stdout)
                if result.stderr:
                    f.write('\n[STDERR]\n')
                    f.write(result.stderr)

        # 成功则直接返回
        if result.returncode == 0:
            logger.debug("Command succeeded on attempt %d", attempt)
            return

        # 失败处理
        # 检查是否需要重试
        if retry_pattern and re.search(retry_pattern, result.stdout + result.stderr):
            if attempt < max_attempts:
                logger.warning("Pattern '%s' found, retrying (%d/%d)...",
                               retry_pattern, attempt, retry_count)
                time.sleep(1)
                continue
            else:
                raise CommandError(
                    f"Command failed after {retry_count} retries (pattern match): "
                    f"exit code {result.returncode}"
                )
        else:
            # 非重试类错误，直接失败
            raise CommandError(
                f"Command failed (exit code {result.returncode}):\n"
                f"STDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}"
            )

    # 不应到达这里，但为安全起见
    raise CommandError(f"Command failed after {retry_count} retries")


# ---------------------------------------------------------------------------
# 2. 简单命令执行（无重试）
# ---------------------------------------------------------------------------
def run_command(cmd_parts: List[str], cwd: Optional[str] = None) -> None:
    """
    执行命令，失败抛出 CommandError。
    """
    result = subprocess.run(cmd_parts, cwd=cwd, capture_output=True, text=True)
    if result.returncode != 0:
        raise CommandError(
            f"Command '{' '.join(cmd_parts)}' failed with exit code {result.returncode}\n"
            f"STDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}"
        )


# ---------------------------------------------------------------------------
# 3. CSV 文件清理与表头添加
# ---------------------------------------------------------------------------
def clean_csv_add_header(file_path: Path, header_columns: List[str]) -> None:
    """
    删除 CSV 文件中的空白行，并在第一行插入指定的表头列。
    注意：原 Shell 使用 sed 在文件最前面插入表头，这意味着原文件没有表头，
          只有数据行。本函数也按此逻辑处理。
    """
    if not file_path.is_file():
        logger.warning("CSV file not found for header cleaning: %s", file_path)
        return

    # 读取所有行，过滤空行
    lines = []
    with open(file_path, 'r', encoding='utf-8') as f:
        for line in f:
            stripped = line.strip()
            if stripped:  # 保留非空行
                lines.append(stripped)

    # 重新写入：先写入表头，再写入数据行
    with open(file_path, 'w', encoding='utf-8') as f:
        f.write(','.join(header_columns) + '\n')
        for line in lines:
            f.write(line + '\n')

    logger.debug("Cleaned CSV %s: %d data rows added header", file_path, len(lines))