Maas-Model-platform/skills/prompt-storage-skill.py

#!/usr/bin/env python3
"""
提示词存储技能 - 用于存储和检索用户的提示词

使用方法:
1. 存储提示词: /store-prompt 您的提示词内容
2. 查看所有提示词: /view-prompts
3. 搜索提示词: /search-prompt 关键词
4. 删除提示词: /delete-prompt ID

存储位置: d:/UGit/AIGC-Space/memory/prompts.json
"""

import json
import os
import uuid
from datetime import datetime
from pathlib import Path


class PromptStorage:
    """提示词存储管理器"""

    def __init__(self):
        self.memory_dir = Path("d:/UGit/AIGC-Space/memory")
        self.prompts_file = self.memory_dir / "prompts.json"

        # 确保memory目录存在
        self.memory_dir.mkdir(exist_ok=True)

        # 初始化或加载提示词文件
        self.prompts = self._load_prompts()

    def _load_prompts(self):
        """加载存储的提示词"""
        if self.prompts_file.exists():
            try:
                with open(self.prompts_file, 'r', encoding='utf-8') as f:
                    return json.load(f)
            except (json.JSONDecodeError, IOError):
                return []
        return []

    def _save_prompts(self):
        """保存提示词到文件"""
        try:
            with open(self.prompts_file, 'w', encoding='utf-8') as f:
                json.dump(self.prompts, f, ensure_ascii=False, indent=2)
        except IOError as e:
            print(f"保存提示词失败: {e}")

    def store_prompt(self, content, category="general", tags=None):
        """存储提示词"""
        if not content.strip():
            return False, "提示词内容不能为空"

        prompt = {
            "id": str(uuid.uuid4()),
            "content": content,
            "category": category,
            "tags": tags or [],
            "created_at": datetime.now().isoformat(),
            "updated_at": datetime.now().isoformat()
        }

        self.prompts.append(prompt)
        self._save_prompts()

        return True, f"提示词已存储，ID: {prompt['id'][:8]}"

    def get_all_prompts(self):
        """获取所有提示词"""
        return self.prompts

    def search_prompts(self, keyword):
        """搜索提示词"""
        keyword = keyword.lower()
        results = []

        for prompt in self.prompts:
            # 在内容中搜索
            if keyword in prompt["content"].lower():
                results.append(prompt)
            # 在标签中搜索
            elif any(keyword in tag.lower() for tag in prompt["tags"]):
                results.append(prompt)
            # 在分类中搜索
            elif keyword in prompt["category"].lower():
                results.append(prompt)

        return results

    def get_prompt_by_id(self, prompt_id):
        """根据ID获取提示词"""
        for prompt in self.prompts:
            if prompt["id"].startswith(prompt_id):
                return prompt
        return None

    def delete_prompt(self, prompt_id):
        """删除提示词"""
        for i, prompt in enumerate(self.prompts):
            if prompt["id"].startswith(prompt_id):
                del self.prompts[i]
                self._save_prompts()
                return True, "提示词已删除"

        return False, "未找到指定ID的提示词"

    def get_categories(self):
        """获取所有分类"""
        categories = set()
        for prompt in self.prompts:
            categories.add(prompt["category"])
        return sorted(list(categories))

    def get_tags(self):
        """获取所有标签"""
        tags = set()
        for prompt in self.prompts:
            tags.update(prompt["tags"])
        return sorted(list(tags))

    def get_stats(self):
        """获取统计信息"""
        return {
            "total": len(self.prompts),
            "categories": len(self.get_categories()),
            "tags": len(self.get_tags()),
            "by_category": self._count_by_category()
        }

    def _count_by_category(self):
        """按分类统计"""
        counts = {}
        for prompt in self.prompts:
            category = prompt["category"]
            counts[category] = counts.get(category, 0) + 1
        return counts


# 全局实例
prompt_storage = PromptStorage()


def handle_store_prompt(content):
    """处理存储提示词请求"""
    if not content:
        return "请提供要存储的提示词内容"

    # 自动分类（简单启发式）
    category = "general"
    if "代码" in content or "code" in content.lower():
        category = "code"
    elif "架构" in content or "architecture" in content.lower():
        category = "architecture"
    elif "设计" in content or "design" in content.lower():
        category = "design"
    elif "数据库" in content or "database" in content.lower():
        category = "database"
    elif "前端" in content or "frontend" in content.lower():
        category = "frontend"
    elif "后端" in content or "backend" in content.lower():
        category = "backend"

    # 自动提取标签
    tags = []
    if len(content) > 200:
        tags.append("长文本")
    if "?" in content:
        tags.append("问题")

    success, message = prompt_storage.store_prompt(content, category, tags)
    return message


def handle_view_prompts(limit=10):
    """处理查看提示词请求"""
    prompts = prompt_storage.get_all_prompts()

    if not prompts:
        return "暂无存储的提示词"

    # 按创建时间倒序排列
    prompts = sorted(prompts, key=lambda x: x["created_at"], reverse=True)

    # 限制返回数量
    if limit:
        prompts = prompts[:int(limit)]

    result = []
    for prompt in prompts:
        result.append(f"""
ID: {prompt['id'][:8]}
分类: {prompt['category']}
标签: {', '.join(prompt['tags'])}
创建时间: {prompt['created_at'][:10]}
内容: {prompt['content'][:100]}...
""")

    return "\n".join(result)


def handle_search_prompts(keyword):
    """处理搜索提示词请求"""
    if not keyword:
        return "请提供搜索关键词"

    results = prompt_storage.search_prompts(keyword)

    if not results:
        return f"未找到包含 '{keyword}' 的提示词"

    result = []
    for prompt in results:
        result.append(f"""
ID: {prompt['id'][:8]}
分类: {prompt['category']}
标签: {', '.join(prompt['tags'])}
创建时间: {prompt['created_at'][:10]}
内容: {prompt['content'][:200]}
""")

    return f"找到 {len(results)} 个结果:\n\n" + "\n".join(result)


def handle_delete_prompt(prompt_id):
    """处理删除提示词请求"""
    if not prompt_id:
        return "请提供要删除的提示词ID"

    success, message = prompt_storage.delete_prompt(prompt_id)
    return message


def handle_prompt_stats():
    """处理获取提示词统计信息"""
    stats = prompt_storage.get_stats()

    result = f"""
提示词统计:
总数量: {stats['total']}
分类数量: {stats['categories']}
标签数量: {stats['tags']}

按分类统计:
"""
    for category, count in stats['by_category'].items():
        result += f"  {category}: {count}\n"

    return result


# 技能函数
def skill_store_prompt(prompt_content):
    """存储提示词"""
    return handle_store_prompt(prompt_content)


def skill_view_prompts(limit="10"):
    """查看所有提示词"""
    return handle_view_prompts(limit)


def skill_search_prompt(keyword):
    """搜索提示词"""
    return handle_search_prompts(keyword)


def skill_delete_prompt(prompt_id):
    """删除提示词"""
    return handle_delete_prompt(prompt_id)


def skill_prompt_stats():
    """获取提示词统计信息"""
    return handle_prompt_stats()


# 导出技能函数供外部使用
__all__ = [
    'skill_store_prompt',
    'skill_view_prompts',
    'skill_search_prompt',
    'skill_delete_prompt',
    'skill_prompt_stats'
]