AI-Check-Test/.gitea/checker/change_logger.py

"""
变更日志持久化模块。
受 log.enabled 开关控制，默认关闭；仅记录接口参数变更及 LLM 审核结果。
"""

import json
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional

from comparator import EndpointChangeReport
from git_utils import CommitInfo


def is_log_enabled(config: Dict[str, Any]) -> bool:
    """判断日志总开关是否开启。"""
    return config.get("log", {}).get("enabled", False)


def _serialize_reports(reports: List[EndpointChangeReport]) -> List[dict]:
    """将参数变更报告序列化为 JSON 结构。"""
    result = []
    for r in reports:
        result.append(
            {
                "uri": r.uri,
                "http_method": r.http_method,
                "controller_class": r.controller_class,
                "method_name": r.method_name,
                "is_new_endpoint": r.is_new_endpoint,
                "is_removed_endpoint": r.is_removed_endpoint,
                "parameter_changes": [
                    {
                        "change_type": c.change_type.value,
                        "param_name": c.param_name,
                        "param_type": c.param_type,
                        "old_name": c.old_name,
                        "old_type": c.old_type,
                        "required": c.required,
                        "detail": c.detail,
                    }
                    for c in r.parameter_changes
                ],
            }
        )
    return result


def save_to_file(
    reports: List[EndpointChangeReport],
    commit_info: CommitInfo,
    log_dir: str,
    llm_review: Optional[str] = None,
) -> Path:
    """写入 JSON 日志文件。"""
    log_path = Path(log_dir)
    log_path.mkdir(parents=True, exist_ok=True)

    short_sha = commit_info.sha[:8]
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    output = log_path / f"{timestamp}_{short_sha}.json"

    record = {
        "commit_sha": commit_info.sha,
        "author": commit_info.author,
        "commit_time": commit_info.commit_time,
        "message": commit_info.message,
        "detected_at": datetime.now().isoformat(),
        "change_count": len(reports),
        "parameter_changes": _serialize_reports(reports),
        "llm_review": llm_review,
    }

    with open(output, "w", encoding="utf-8") as f:
        json.dump(record, f, ensure_ascii=False, indent=2)

    print(f"[日志] 参数变更记录已写入: {output}")
    return output


def save_to_mysql(
    reports: List[EndpointChangeReport],
    commit_info: CommitInfo,
    mysql_config: Dict[str, Any],
    llm_review: Optional[str] = None,
) -> bool:
    """写入 MySQL。"""
    try:
        import pymysql
    except ImportError:
        print("[错误] MySQL 模式需要: pip install pymysql")
        return False

    host = mysql_config.get("host", "")
    if not host or host == "YOUR_MYSQL_HOST":
        print("[警告] 未配置 MySQL，跳过写入。")
        return False

    try:
        conn = pymysql.connect(
            host=host,
            port=int(mysql_config.get("port", 3306)),
            user=mysql_config.get("user"),
            password=mysql_config.get("password"),
            database=mysql_config.get("database"),
            charset="utf8mb4",
        )
        table = mysql_config.get("table", "api_change_logs")
        payload = json.dumps(_serialize_reports(reports), ensure_ascii=False)

        with conn.cursor() as cursor:
            sql = f"""
                INSERT INTO `{table}`
                (commit_sha, author, commit_time, commit_message, change_count, reports_json, llm_review, created_at)
                VALUES (%s, %s, %s, %s, %s, %s, %s, NOW())
            """
            cursor.execute(
                sql,
                (
                    commit_info.sha,
                    commit_info.author,
                    commit_info.commit_time,
                    commit_info.message,
                    len(reports),
                    payload,
                    llm_review,
                ),
            )
        conn.commit()
        conn.close()
        print(f"[日志] 已写入 MySQL: {table}")
        return True
    except Exception as exc:
        print(f"[错误] MySQL 写入失败: {exc}")
        return False


def persist_change_log(
    reports: List[EndpointChangeReport],
    commit_info: CommitInfo,
    config: Dict[str, Any],
    llm_review: Optional[str] = None,
) -> None:
    """
    根据 log.enabled 决定是否持久化接口参数变更日志。

    :param reports:     参数变更报告
    :param commit_info: 提交信息
    :param config:      完整配置
    :param llm_review:  LLM 参数变更审核结论
    """
    if not is_log_enabled(config):
        print("[日志] 日志开关已关闭（log.enabled=false），跳过写入。")
        return

    log_cfg = config.get("log", {})
    if log_cfg.get("storage") == "mysql":
        save_to_mysql(reports, commit_info, log_cfg.get("mysql", {}), llm_review)
    else:
        save_to_file(
            reports,
            commit_info,
            log_cfg.get("file_dir", ".gitea/logs/api-changes"),
            llm_review,
        )