LQAgentPlatform/core/debug/prompt_manager.py

"""
提示词版本管理器

读写/版本化 config/prompt/ 下的 YAML 文件（实际路径为
core/construction_review/component/reviewers/prompt/）。
管理提示词的版本历史、激活状态和差异对比，兼容 prompt_loader.py 的加载格式。
"""

import os
import re
import difflib
import logging
from datetime import datetime
from typing import Optional, List

import yaml

logger = logging.getLogger(__name__)

# ─── 路径常量 ───────────────────────────────────────────────
# prompt_manager.py 位于 core/debug/prompt_manager.py
# 项目根目录 = 当前文件上 2 级
_PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..'))
# 与 prompt_loader.py 使用相同的 prompt 目录
PROMPT_DIR = os.path.join(
    _PROJECT_ROOT,
    'core', 'construction_review', 'component', 'reviewers', 'prompt',
)
VERSIONS_DIR = os.path.join(PROMPT_DIR, 'prompt_versions')

# ─── 映射表 ─────────────────────────────────────────────────

# 主文件映射：file_key -> 文件名
FILE_MAP = {
    'basic': 'basic_reviewers.yaml',
    'technical': 'technical_reviewers.yaml',
    'rag': 'rag_reviewers.yaml',
    'ai': 'ai_suggestion.yaml',
    'outline': 'outline_reviewers.yaml',
    'query_extract': 'query_extract.yaml',
}

# 提示词名称 -> file_key 映射（共 19 个提示词）
PROMPT_FILE_MAP = {
    # basic_reviewers.yaml
    'completeness_check': 'basic',
    'grammar_check': 'basic',
    'semantic_logic_check': 'basic',
    'sensitive_word_check': 'basic',
    'timeliness_check': 'basic',
    'reference_check': 'basic',
    # technical_reviewers.yaml
    'non_parameter_compliance_check': 'technical',
    'parameter_compliance_check': 'technical',
    # rag_reviewers.yaml
    'rag_enhanced_review': 'rag',
    'vector_search_review': 'rag',
    'hybrid_search_review': 'rag',
    # ai_suggestion.yaml
    'professional_suggestion': 'ai',
    'standardization_suggestion': 'ai',
    'completeness_suggestion': 'ai',
    'readability_suggestion': 'ai',
    # outline_reviewers.yaml
    'outline_completeness_classifier': 'outline',
    'outline_completeness_review': 'outline',
    'overall_outline_completeness_review': 'outline',
    # query_extract.yaml
    'query_extract': 'query_extract',
}

# 提示词名称 -> 链路中文名称映射
PROMPT_CHAIN_MAP = {
    'completeness_check': '完整性',
    'grammar_check': '语法',
    'semantic_logic_check': '语义逻辑',
    'sensitive_word_check': '敏感词',
    'timeliness_check': '时效性',
    'reference_check': '规范性',
    'non_parameter_compliance_check': '专业性',
    'parameter_compliance_check': '专业性',
    'rag_enhanced_review': '专业性',
    'vector_search_review': '专业性',
    'hybrid_search_review': '专业性',
    'professional_suggestion': '专业性',
    'standardization_suggestion': '规范性',
    'completeness_suggestion': '完整性',
    'readability_suggestion': '完整性',
    'outline_completeness_classifier': '完整性',
    'outline_completeness_review': '完整性',
    'overall_outline_completeness_review': '完整性',
    'query_extract': '专业性',
}

# 可筛选的链路列表（7 条）
CHAINS = ['完整性', '时效性', '规范性', '敏感词', '语义逻辑', '语法', '专业性']


# ─── 工具函数 ───────────────────────────────────────────────

def _extract_variables(text: str) -> List[str]:
    """从文本中提取 {var} 占位符变量"""
    return re.findall(r'\{(\w+)\}', text)


def _parse_version(ver: str) -> tuple:
    """解析版本号字符串为可比较的元组，如 'v1.2' -> (1, 2)"""
    try:
        parts = ver.lstrip('v').split('.')
        return tuple(int(p) for p in parts)
    except (ValueError, IndexError):
        return (0, 0)


def _make_diff_lines(a_text: str, b_text: str) -> list:
    """生成行级差异列表，使用 difflib.unified_diff"""
    lines = []
    for line in difflib.unified_diff(
        a_text.splitlines(keepends=False),
        b_text.splitlines(keepends=False),
        lineterm='',
        n=999,  # 显示所有行（不折叠）
    ):
        if line.startswith('---') or line.startswith('+++') or line.startswith('@@'):
            continue  # 跳过 unified diff 头部
        if line.startswith('+'):
            lines.append({'type': 'add', 'text': line[1:]})
        elif line.startswith('-'):
            lines.append({'type': 'del', 'text': line[1:]})
        else:
            # 上下文行以空格开头
            text = line[1:] if line.startswith(' ') else line
            lines.append({'type': 'ctx', 'text': text})
    return lines


# ─── PromptManager ──────────────────────────────────────────

class PromptManager:
    """提示词版本管理器

    提供提示词的查询、版本管理、对比、激活和回滚功能。
    版本历史存储在 PROMPT_DIR/prompt_versions/{prompt_name}/{version}.yaml。
    """

    def __init__(self):
        os.makedirs(VERSIONS_DIR, exist_ok=True)
        self._init_versions()

    # ==================== 首次运行初始化 ====================

    def _init_versions(self):
        """首次运行时，从当前主文件为所有提示词创建 v1.0 初始版本"""
        for file_key, filename in FILE_MAP.items():
            filepath = os.path.join(PROMPT_DIR, filename)
            if not os.path.exists(filepath):
                continue
            with open(filepath, 'r', encoding='utf-8') as f:
                config = yaml.safe_load(f) or {}
            for prompt_name, entry in config.items():
                prompt_dir = os.path.join(VERSIONS_DIR, prompt_name)
                if os.path.exists(prompt_dir):
                    continue  # 已有版本目录，跳过
                os.makedirs(prompt_dir, exist_ok=True)
                # 兼容 query_extract 的特殊字段名
                if 'system' in entry and 'user_template' in entry:
                    system_prompt = entry.get('system', '')
                    user_prompt = entry.get('user_template', '')
                else:
                    system_prompt = entry.get('system_prompt', '')
                    user_prompt = entry.get('user_prompt_template', '')
                version_data = {
                    'version': 'v1.0',
                    'created_at': datetime.now().isoformat(),
                    'based_on': None,
                    'is_current': True,
                    'note': '初始版本',
                    'system_prompt': system_prompt,
                    'user_prompt_template': user_prompt,
                }
                with open(os.path.join(prompt_dir, 'v1.0.yaml'), 'w', encoding='utf-8') as f:
                    yaml.dump(version_data, f, allow_unicode=True, default_flow_style=False)

    # ==================== 内部辅助方法 ====================

    def _get_main_file_path(self, prompt_name: str) -> Optional[str]:
        """获取提示词对应的主 YAML 文件路径"""
        file_key = PROMPT_FILE_MAP.get(prompt_name)
        if not file_key:
            return None
        filename = FILE_MAP.get(file_key)
        if not filename:
            return None
        return os.path.join(PROMPT_DIR, filename)

    def _read_current_from_main(self, prompt_name: str) -> Optional[dict]:
        """从主文件读取当前激活的提示词内容"""
        filepath = self._get_main_file_path(prompt_name)
        if not filepath or not os.path.exists(filepath):
            return None
        with open(filepath, 'r', encoding='utf-8') as f:
            config = yaml.safe_load(f) or {}
        entry = config.get(prompt_name)
        if not entry:
            return None
        # 兼容 query_extract 的特殊字段名
        if 'system' in entry and 'user_template' in entry:
            return {
                'system_prompt': entry.get('system', ''),
                'user_prompt': entry.get('user_template', ''),
            }
        return {
            'system_prompt': entry.get('system_prompt', ''),
            'user_prompt': entry.get('user_prompt_template', ''),
        }

    def _get_version_dir(self, prompt_name: str) -> str:
        return os.path.join(VERSIONS_DIR, prompt_name)

    def _get_version_file(self, prompt_name: str, version: str) -> str:
        return os.path.join(self._get_version_dir(prompt_name), f'{version}.yaml')

    def _load_version_file(self, prompt_name: str, version: str) -> Optional[dict]:
        """加载指定版本文件，失败时返回 None"""
        filepath = self._get_version_file(prompt_name, version)
        if not os.path.exists(filepath):
            return None
        try:
            with open(filepath, 'r', encoding='utf-8') as f:
                return yaml.safe_load(f)
        except Exception as e:
            logger.error('读取版本文件失败: %s, 错误: %s', filepath, e)
            return None

    def _list_version_files(self, prompt_name: str) -> List[str]:
        """列出某提示词的所有版本文件名，按版本号降序"""
        version_dir = self._get_version_dir(prompt_name)
        if not os.path.exists(version_dir):
            return []
        files = [f for f in os.listdir(version_dir) if f.endswith('.yaml')]
        files.sort(key=lambda x: _parse_version(x.replace('.yaml', '')), reverse=True)
        return files

    def _get_current_main_version_info(self, prompt_name: str) -> dict:
        """获取当前激活版本的元信息"""
        version_dir = self._get_version_dir(prompt_name)
        if os.path.exists(version_dir):
            for fname in sorted(os.listdir(version_dir), reverse=True):
                if not fname.endswith('.yaml'):
                    continue
                data = self._load_version_file(prompt_name, fname.replace('.yaml', ''))
                if data and data.get('is_current'):
                    return data
        # 降级：从主文件读取
        current = self._read_current_from_main(prompt_name)
        if current:
            return {
                'version': 'current',
                'is_current': True,
                'system_prompt': current['system_prompt'],
                'user_prompt_template': current['user_prompt'],
                'created_at': '',
                'note': '',
            }
        return {}

    def _update_is_current_flag(self, prompt_name: str, active_version: str):
        """更新某提示词的 is_current 标记，清除其他版本的标记"""
        version_dir = self._get_version_dir(prompt_name)
        if not os.path.exists(version_dir):
            return
        for fname in os.listdir(version_dir):
            if not fname.endswith('.yaml'):
                continue
            ver = fname.replace('.yaml', '')
            filepath = os.path.join(version_dir, fname)
            data = self._load_version_file(prompt_name, ver)
            if data is None:
                continue
            data['is_current'] = (ver == active_version)
            with open(filepath, 'w', encoding='utf-8') as f:
                yaml.dump(data, f, allow_unicode=True, default_flow_style=False)

    def _sync_main_to_versions(self, prompt_name: str):
        """确保主文件当前内容在版本目录中有对应版本（兜底逻辑）"""
        version_files = self._list_version_files(prompt_name)
        if version_files:
            return  # 已有版本记录，无需同步
        current = self._read_current_from_main(prompt_name)
        if not current:
            return
        prompt_dir = self._get_version_dir(prompt_name)
        os.makedirs(prompt_dir, exist_ok=True)
        version_data = {
            'version': 'v1.0',
            'created_at': datetime.now().isoformat(),
            'based_on': None,
            'is_current': True,
            'note': '初始版本（同步自主文件）',
            'system_prompt': current['system_prompt'],
            'user_prompt_template': current['user_prompt'],
        }
        with open(os.path.join(prompt_dir, 'v1.0.yaml'), 'w', encoding='utf-8') as f:
            yaml.dump(version_data, f, allow_unicode=True, default_flow_style=False)

    # ==================== 公开方法 ====================

    def get_all_prompts(self, chain_filter: str = None, search: str = None) -> list:
        """获取所有提示词及其版本信息列表

        Args:
            chain_filter: 链路名称筛选
            search: 提示词名称搜索关键字

        Returns:
            list[dict]: 每项包含 name, version, time, chain, is_current, note
        """
        results = []
        for prompt_name in PROMPT_FILE_MAP:
            if search and search.lower() not in prompt_name.lower():
                continue
            chain = PROMPT_CHAIN_MAP.get(prompt_name, '')
            if chain_filter and chain != chain_filter:
                continue

            self._sync_main_to_versions(prompt_name)
            version_files = self._list_version_files(prompt_name)

            if not version_files:
                # 没有版本记录，直接从主文件读取
                current = self._read_current_from_main(prompt_name)
                if current:
                    results.append({
                        'name': prompt_name,
                        'version': 'current',
                        'time': '',
                        'chain': chain,
                        'is_current': True,
                        'note': '',
                    })
            else:
                for fname in version_files:
                    ver = fname.replace('.yaml', '')
                    data = self._load_version_file(prompt_name, ver)
                    if data is None:
                        continue
                    results.append({
                        'name': prompt_name,
                        'version': data.get('version', ver),
                        'time': data.get('created_at', ''),
                        'chain': chain,
                        'is_current': data.get('is_current', False),
                        'note': data.get('note', ''),
                    })
        return results

    def get_prompt_detail(self, name: str, version: str = None) -> Optional[dict]:
        """获取提示词详情

        Args:
            name: 提示词名称
            version: 版本号，不指定则返回当前激活版本

        Returns:
            dict | None: 包含完整提示词详情，不存在时返回 None
        """
        filepath = self._get_main_file_path(name)
        if not filepath:
            return None
        if name not in PROMPT_FILE_MAP:
            return None

        chain = PROMPT_CHAIN_MAP.get(name, '')

        if version:
            # 从指定版本文件读取
            data = self._load_version_file(name, version)
            if not data:
                return None
            system_prompt = data.get('system_prompt', '')
            user_prompt = data.get('user_prompt_template', '')
            return {
                'name': name,
                'version': data.get('version', version),
                'time': data.get('created_at', ''),
                'chain': chain,
                'is_current': data.get('is_current', False),
                'system_prompt': system_prompt,
                'user_prompt': user_prompt,
                'note': data.get('note', ''),
                'variables': _extract_variables(user_prompt),
                'based_on': data.get('based_on'),
                'file_path': filepath,
            }
        else:
            # 读取当前激活版本
            version_info = self._get_current_main_version_info(name)
            if not version_info:
                return None
            system_prompt = version_info.get('system_prompt', '')
            user_prompt = version_info.get('user_prompt_template', '')
            return {
                'name': name,
                'version': version_info.get('version', 'current'),
                'time': version_info.get('created_at', ''),
                'chain': chain,
                'is_current': True,
                'system_prompt': system_prompt,
                'user_prompt': user_prompt,
                'note': version_info.get('note', ''),
                'variables': _extract_variables(user_prompt),
                'based_on': version_info.get('based_on'),
                'file_path': filepath,
            }

    def save_new_version(self, name: str, system_prompt: str, user_prompt: str,
                         note: str = '', set_current: bool = True) -> dict:
        """保存新版本

        Args:
            name: 提示词名称
            system_prompt: 系统提示词内容
            user_prompt: 用户提示词模板
            note: 版本说明
            set_current: 是否设为当前激活版本

        Returns:
            dict: {version, name, time}

        Raises:
            ValueError: 提示词名称不存在
        """
        if name not in PROMPT_FILE_MAP:
            raise ValueError(f'未知的提示词: {name}')

        # 同步主文件到版本目录（兜底）
        self._sync_main_to_versions(name)

        # 计算版本号：递增 major 版本
        version_files = self._list_version_files(name)
        if version_files:
            latest_ver = version_files[0].replace('.yaml', '')
            latest_num = _parse_version(latest_ver)
            new_ver = f'v{latest_num[0] + 1}.0'
            based_on = latest_ver
        else:
            new_ver = 'v1.0'
            based_on = None

        # 写入版本文件
        prompt_dir = self._get_version_dir(name)
        os.makedirs(prompt_dir, exist_ok=True)
        version_data = {
            'version': new_ver,
            'created_at': datetime.now().isoformat(),
            'based_on': based_on,
            'is_current': False,
            'note': note,
            'system_prompt': system_prompt,
            'user_prompt_template': user_prompt,
        }
        with open(self._get_version_file(name, new_ver), 'w', encoding='utf-8') as f:
            yaml.dump(version_data, f, allow_unicode=True, default_flow_style=False)

        # 更新主文件
        if set_current:
            self.activate_version(name, new_ver)

        return {
            'version': new_ver,
            'name': name,
            'time': version_data['created_at'],
        }

    def get_versions(self, name: str) -> List[dict]:
        """获取某提示词的所有历史版本列表

        Args:
            name: 提示词名称

        Returns:
            list[dict]: 每项包含 version, time, is_current, note,
                         system_prompt_preview, user_prompt_preview
        """
        self._sync_main_to_versions(name)
        version_files = self._list_version_files(name)
        versions = []
        for fname in version_files:
            ver = fname.replace('.yaml', '')
            data = self._load_version_file(name, ver)
            if data is None:
                continue
            sys_prompt = data.get('system_prompt', '')
            user_prompt = data.get('user_prompt_template', '')
            versions.append({
                'version': data.get('version', ver),
                'time': data.get('created_at', ''),
                'is_current': data.get('is_current', False),
                'note': data.get('note', ''),
                'system_prompt_preview': (sys_prompt[:50] + '...'
                                          if len(sys_prompt) > 50 else sys_prompt),
                'user_prompt_preview': (user_prompt[:50] + '...'
                                        if len(user_prompt) > 50 else user_prompt),
            })
        return versions

    def compare_versions(self, name: str, base_version: str, target_version: str) -> dict:
        """对比两个版本的差异（行级 Diff）

        Args:
            name: 提示词名称
            base_version: 基准版本号
            target_version: 目标版本号

        Returns:
            dict: {name, base_version, target_version, diffs}

        Raises:
            FileNotFoundError: 版本文件不存在
        """
        base_data = self._load_version_file(name, base_version)
        target_data = self._load_version_file(name, target_version)

        if not base_data or not target_data:
            missing = []
            if not base_data:
                missing.append(base_version)
            if not target_data:
                missing.append(target_version)
            raise FileNotFoundError(f'版本文件不存在: {name} [{", ".join(missing)}]')

        diffs = []
        for section_key, display_section in [
            ('system_prompt', 'system_prompt'),
            ('user_prompt_template', 'user_prompt'),
        ]:
            base_text = base_data.get(section_key, '')
            target_text = target_data.get(section_key, '')
            diff_lines = _make_diff_lines(base_text, target_text)
            diffs.append({
                'section': display_section,
                'type': 'text_diff',
                'lines': diff_lines,
            })

        return {
            'name': name,
            'base_version': base_version,
            'target_version': target_version,
            'diffs': diffs,
        }

    def activate_version(self, name: str, version: str) -> dict:
        """激活指定版本（将版本内容覆盖写入主 YAML 文件）

        Args:
            name: 提示词名称
            version: 版本号

        Returns:
            dict: {success, name, version}

        Raises:
            ValueError: 提示词名称不存在
            FileNotFoundError: 版本文件或主文件不存在
        """
        if name not in PROMPT_FILE_MAP:
            raise ValueError(f'未知的提示词: {name}')

        version_data = self._load_version_file(name, version)
        if not version_data:
            raise FileNotFoundError(f'版本文件不存在: {name}/{version}.yaml')

        # 读取主文件
        filepath = self._get_main_file_path(name)
        if not filepath or not os.path.exists(filepath):
            raise FileNotFoundError(f'主文件不存在: {filepath}')

        with open(filepath, 'r', encoding='utf-8') as f:
            config = yaml.safe_load(f) or {}

        # 保留主文件原有字段名（兼容 query_extract 等特殊格式）
        original_entry = config.get(name, {})
        new_system_prompt = version_data.get('system_prompt', '')
        new_user_prompt = version_data.get('user_prompt_template', '')

        if name not in config:
            config[name] = {}

        # 根据原字段名决定写入方式
        if 'system' in original_entry and 'user_template' in original_entry:
            config[name]['system'] = new_system_prompt
            config[name]['user_template'] = new_user_prompt
        else:
            config[name]['system_prompt'] = new_system_prompt
            config[name]['user_prompt_template'] = new_user_prompt

        # 写回主文件
        with open(filepath, 'w', encoding='utf-8') as f:
            yaml.dump(config, f, allow_unicode=True, default_flow_style=False)

        # 更新版本文件的 is_current 标记
        self._update_is_current_flag(name, version)

        # 清除 prompt_loader 缓存
        self._clear_prompt_loader_cache()

        return {
            'success': True,
            'name': name,
            'version': version,
        }

    def rollback_version(self, name: str, version: str) -> dict:
        """回滚到指定历史版本（等同于激活该版本）

        Args:
            name: 提示词名称
            version: 版本号

        Returns:
            dict: {success, name, version}
        """
        return self.activate_version(name, version)

    @staticmethod
    def _clear_prompt_loader_cache():
        """清除 prompt_loader 缓存，使下一次读取使用最新激活的内容"""
        try:
            from core.construction_review.component.reviewers.utils.prompt_loader import (
                prompt_loader,
            )
            prompt_loader.manage_cache(action='clear')
        except ImportError:
            logger.warning('无法导入 prompt_loader，跳过缓存清理')
        except Exception as e:
            logger.error('清理 prompt_loader 缓存失败: %s', e)