class MemoryRetrievalEngine:
    """记忆检索引擎"""
    
    def __init__(self, workspace_path: str):
        self.workspace = Path(workspace_path)
        self.memory_index = {}
        
    def retrieve(self, query: str, context: dict) -> MemoryContext:
        """
        检索记忆
        
        Args:
            query: 用户查询
            context: 当前上下文（会话类型、项目等）
            
        Returns:
            MemoryContext: 融合后的记忆上下文
        """
        # 1. 确定检索范围
        scope = self._determine_scope(context)
        
        # 2. 并行检索三层记忆
        l1_memory = self._retrieve_l1(scope) if scope.l1 else None
        l2_memory = self._retrieve_l2(scope) if scope.l2 else None
        l3_memory = self._retrieve_l3(scope) if scope.l3 else None
        
        # 3. 语义搜索（可选）
        if context.get('semantic_search'):
            semantic_results = self._semantic_search(query)
            l2_memory = self._merge_semantic(l2_memory, semantic_results)
        
        # 4. 融合记忆
        fused = self._fuse_memories(l1_memory, l2_memory, l3_memory)
        
        # 5. 应用遗忘策略
        filtered = self._apply_forgetting(fused, context)
        
        return filtered
    
    def _determine_scope(self, context: dict) -> MemoryScope:
        """确定记忆检索范围"""
        chat_type = context.get('chat_type', 'direct')
        project = context.get('project')
        
        # 群聊不加载全局记忆（安全隔离）
        if chat_type == 'group':
            return MemoryScope(l1=False, l2=True, l3=project is not None)
        
        # 主会话加载全部
        return MemoryScope(l1=True, l2=True, l3=True)

def _semantic_search(self, query: str, top_k: int = 5) -> List[MemorySnippet]:
    """语义搜索记忆片段"""
    
    # 1. 向量化查询
    query_embedding = self.embedding_model.encode(query)
    
    # 2. 搜索记忆索引
    results = self.memory_index.search(
        query_embedding, 
        top_k=top_k,
        threshold=0.7  # 相似度阈值
    )
    
    # 3. 过滤过期记忆
    valid_results = [
        r for r in results 
        if not self._is_expired(r.metadata)
    ]
    
    # 4. 排序并返回
    return sorted(valid_results, key=lambda x: x.score, reverse=True)

class MemoryWriter:
    """记忆写入器"""
    
    def __init__(self, workspace_path: str):
        self.workspace = Path(workspace_path)
        self.today = datetime.now().strftime('%Y-%m-%d')
        
    def write_session_summary(self, session: Session) -> None:
        """写入会话总结到每日记忆"""
        
        memory_file = self.workspace / 'memory' / f'{self.today}.md'
        
        # 1. 读取现有内容
        existing = self._read_memory(memory_file)
        
        # 2. 生成新条目
        new_entry = self._generate_entry(session)
        
        # 3. 合并内容
        updated = self._merge_entries(existing, new_entry)
        
        # 4. 写回文件
        self._write_memory(memory_file, updated)
        
    def _generate_entry(self, session: Session) -> str:
        """生成记忆条目"""
        
        return f"""
### {session.start_time.strftime('%H:%M')} - {session.topic}

**完成工作**：
{self._format_tasks(session.completed_tasks)}

**关键决策**：
{self._format_decisions(session.decisions)}

**待办事项**：
{self._format_todos(session.pending_tasks)}
"""

def extract_long_term_memory(self, daily_files: List[Path]) -> List[MemoryItem]:
    """
    从每日记忆中提炼长期记忆
    
    规则：
    1. 重复出现 3 次+ 的主题 → 重要模式
    2. 用户明确标记"记住" → 长期记忆
    3. 项目里程碑事件 → 长期记忆
    4. 错误教训 → 长期记忆
    """
    
    candidates = []
    
    for file in daily_files:
        content = file.read_text()
        
        # 提取关键事件
        events = self._extract_events(content)
        candidates.extend(events)
        
        # 提取用户指令
        commands = self._extract_user_commands(content)
        candidates.extend(commands)
        
    # 聚类相似事件
    clusters = self._cluster_events(candidates)
    
    # 筛选高价值记忆
    long_term = [
        cluster for cluster in clusters
        if cluster.frequency >= 3 or cluster.importance_score > 0.8
    ]
    
    return long_term

def apply_time_decay(self, memories: List[Memory]) -> List[Memory]:
    """
    应用时间衰减
    
    衰减公式：
    score = base_score * e^(-λ * days_old)
    
    λ = 0.1 (半衰期约 7 天)
    """
    
    lambda_decay = 0.1
    today = datetime.now()
    
    for memory in memories:
        days_old = (today - memory.created_at).days
        decay_factor = math.exp(-lambda_decay * days_old)
        memory.current_score = memory.base_score * decay_factor
        
    return memories

def archive_old_memories(self, days_threshold: int = 7) -> None:
    """归档旧记忆"""
    
    memory_dir = self.workspace / 'memory'
    archive_dir = self.workspace / 'memory' / 'archive'
    archive_dir.mkdir(exist_ok=True)
    
    cutoff_date = datetime.now() - timedelta(days=days_threshold)
    
    for file in memory_dir.glob('*.md'):
        file_date = self._parse_date(file.stem)
        
        if file_date < cutoff_date:
            # 移动到归档目录
            archive_path = archive_dir / file.name
            shutil.move(str(file), str(archive_path))
            
            # 记录归档日志
            self._log_archive(file, archive_path)

def generate_greeting(self, user_id: str) -> str:
    # 1. 读取用户信息
    user_info = self.memory.retrieve_l1(user_id)
    name = user_info.get('name', '用户')
    
    # 2. 读取昨日待办
    yesterday_todos = self.memory.retrieve_l2(
        user_id, 
        date='yesterday',
        filter='pending_tasks'
    )
    
    # 3. 读取宠物信息（个性化细节）
    pet_info = self.memory.retrieve_l1(user_id, key='pet')
    
    # 4. 生成个性化问候
    greeting = f"早上好，{name}！☀️\n\n"
    
    if yesterday_todos:
        greeting += "根据昨晚的记录：\n"
        for todo in yesterday_todos[:3]:  # 最多显示 3 个
            greeting += f"- {todo}\n"
    
    if pet_info:
        greeting += f"\n{pet_info.get('name')}最近怎么样？\n"
    
    return greeting

def handle_project_switch(self, user_id: str, new_project: str):
    # 1. 保存当前项目上下文
    current_project = self.context.get('current_project')
    if current_project:
        self.memory.save_context(current_project)
    
    # 2. 加载新项目上下文
    self.context['current_project'] = new_project
    project_memory = self.memory.load_project(new_project)
    
    # 3. 更新 Agent 认知
    self.agent.update_context({
        'project': new_project,
        'version': project_memory.version,
        'status': project_memory.status,
        'key_metrics': project_memory.metrics
    })
    
    # 4. 生成上下文感知的响应
    return self._generate_project_response(new_project, project_memory)

def check_safety_rules(self, action: str, context: dict) -> SafetyCheck:
    # 1. 检索相关安全规则
    safety_rules = self.memory.retrieve_l1(
        filter='safety_rules',
        tags=['file_deletion', 'confirmation_required']
    )
    
    # 2. 检查历史教训
    past_incidents = self.memory.semantic_search(
        query=f"delete file {context['file']}",
        filter='incidents'
    )
    
    # 3. 生成安全检查
    check = SafetyCheck(
        action=action,
        risk_level=self._assess_risk(action, safety_rules),
        requires_confirmation=self._needs_confirmation(action, safety_rules),
        historical_context=past_incidents
    )
    
    return check

def handle_group_message(self, message: Message, chat_id: str):
    # 1. 检测聊天类型
    chat_type = self._get_chat_type(chat_id)
    
    # 2. 群聊不加载全局记忆（安全隔离）
    if chat_type == 'group':
        memory_scope = MemoryScope(
            l1=False,  # ❌ 不加载个人信息
            l2=False,  # ❌ 不加载工作日志
            l3=True    # ✅ 仅加载项目公开信息
        )
    else:
        memory_scope = MemoryScope(l1=True, l2=True, l3=True)
    
    # 3. 生成响应（不泄露隐私）
    response = self.agent.generate_response(
        message,
        memory_scope=memory_scope
    )
    
    return response

class MemoryCache:
    """记忆缓存层"""
    
    def __init__(self, ttl_seconds: int = 300):
        self.cache = {}
        self.ttl = ttl_seconds
        
    def get(self, key: str) -> Optional[MemoryContext]:
        """获取缓存"""
        if key in self.cache:
            entry = self.cache[key]
            if time.time() - entry.timestamp < self.ttl:
                return entry.data
            else:
                del self.cache[key]
        return None
    
    def set(self, key: str, value: MemoryContext) -> None:
        """设置缓存"""
        self.cache[key] = CacheEntry(
            data=value,
            timestamp=time.time()
        )

def retrieve_with_lazy_loading(self, query: str) -> MemoryContext:
    """增量加载记忆"""
    
    # 1. 先加载核心记忆（L1）
    context = MemoryContext()
    context.l1 = self._retrieve_l1_fast()
    
    # 2. 异步加载每日记忆（L2）
    future_l2 = self.executor.submit(self._retrieve_l2, query)
    
    # 3. 异步加载项目记忆（L3）
    future_l3 = self.executor.submit(self._retrieve_l3, query)
    
    # 4. 等待 L2/L3 完成（超时 100ms）
    try:
        context.l2 = future_l2.result(timeout=0.1)
    except TimeoutError:
        context.l2 = MemorySnippet(partial=True)
    
    try:
        context.l3 = future_l3.result(timeout=0.1)
    except TimeoutError:
        context.l3 = MemorySnippet(partial=True)
    
    return context

def compress_old_memories(self, days_threshold: int = 30) -> None:
    """压缩旧记忆文件"""
    
    memory_dir = self.workspace / 'memory'
    
    for file in memory_dir.glob('*.md'):
        file_date = self._parse_date(file.stem)
        days_old = (datetime.now() - file_date).days
        
        if days_old > days_threshold:
            content = file.read_text()
            
            # 1. 移除冗余空白
            compressed = self._remove_redundant_whitespace(content)
            
            # 2. 压缩重复模式
            compressed = self._compress_patterns(compressed)
            
            # 3. 写回文件
            file.write_text(compressed)
            
            # 4. 记录压缩率
            original_size = len(content)
            compressed_size = len(compressed)
            ratio = (1 - compressed_size / original_size) * 100
            
            logger.info(f"压缩 {file.name}: {ratio:.1f}%")

class ShardedMemoryStorage:
    """分片记忆存储"""
    
    def __init__(self, num_shards: int = 10):
        self.num_shards = num_shards
        
    def _get_shard(self, memory_id: str) -> int:
        """计算记忆所属分片"""
        hash_value = hashlib.md5(memory_id.encode()).hexdigest()
        return int(hash_value[:8], 16) % self.num_shards
    
    def store(self, memory: Memory) -> None:
        """存储记忆到对应分片"""
        shard_id = self._get_shard(memory.id)
        shard_path = self.base_path / f'shard_{shard_id}'
        
        # 写入分片文件
        (shard_path / f'{memory.id}.md').write_text(memory.content)

class MemoryIndex:
    """记忆索引"""
    
    def __init__(self):
        self.inverted_index = defaultdict(set)  # term -> memory_ids
        self.embedding_index = None  # 向量索引
        
    def build_inverted_index(self, memories: List[Memory]) -> None:
        """构建倒排索引"""
        
        for memory in memories:
            tokens = self._tokenize(memory.content)
            
            for token in tokens:
                self.inverted_index[token].add(memory.id)
    
    def build_embedding_index(self, memories: List[Memory]) -> None:
        """构建向量索引"""
        
        embeddings = []
        for memory in memories:
            embedding = self.embedding_model.encode(memory.content)
            embeddings.append(embedding)
        
        # 使用 FAISS 构建索引
        self.embedding_index = faiss.IndexFlatIP(len(embeddings[0]))
        self.embedding_index.add(np.array(embeddings))

def rewrite_query(self, query: str, context: MemoryContext) -> str:
    """查询重写"""
    
    # 1. 提取实体
    entities = self._extract_entities(query)
    
    # 2. 扩展同义词
    expanded = self._expand_synonyms(query)
    
    # 3. 添加上下文过滤
    if context.project:
        expanded += f" project:{context.project}"
    
    if context.date_range:
        expanded += f" date:{context.date_range}"
    
    return expanded

class MemoryIsolation:
    """记忆隔离机制"""
    
    def check_access(self, session: Session, memory_id: str) -> bool:
        """检查访问权限"""
        
        memory = self.storage.get(memory_id)
        
        # 1. 检查会话类型
        if session.chat_type == 'group':
            # 群聊只能访问公开记忆
            return memory.visibility == 'public'
        
        # 2. 检查用户所有权
        if memory.owner_id != session.user_id:
            return memory.shared_with and session.user_id in memory.shared_with
        
        # 3. 检查记忆敏感度
        if memory.sensitivity == 'private':
            # 私密记忆仅限主会话访问
            return session.is_main_session
        
        return True

class EncryptedMemoryStorage:
    """加密记忆存储"""
    
    def __init__(self, encryption_key: str):
        self.cipher = Fernet(encryption_key.encode())
        
    def store(self, memory: Memory) -> None:
        """加密存储"""
        
        # 1. 序列化
        data = json.dumps(memory.to_dict()).encode()
        
        # 2. 加密
        encrypted_data = self.cipher.encrypt(data)
        
        # 3. 存储
        self.storage.write(memory.id, encrypted_data)
    
    def retrieve(self, memory_id: str) -> Memory:
        """解密读取"""
        
        # 1. 读取
        encrypted_data = self.storage.read(memory_id)
        
        # 2. 解密
        data = self.cipher.decrypt(encrypted_data)
        
        # 3. 反序列化
        memory_dict = json.loads(data.decode())
        
        return Memory.from_dict(memory_dict)

def log_memory_access(self, session: Session, memory_id: str, action: str):
    """记录记忆访问日志"""
    
    log_entry = {
        'timestamp': datetime.now().isoformat(),
        'user_id': session.user_id,
        'session_id': session.id,
        'memory_id': memory_id,
        'action': action,  # read/write/delete
        'chat_type': session.chat_type,
        'ip_address': session.ip_address
    }
    
    # 写入审计日志
    self.audit_log.append(log_entry)
    
    # 异常访问告警
    if self._is_anomalous(log_entry):
        self._send_alert(log_entry)

# 错误代码 ❌
def generate_response(self, message: Message):
    # 没有检查聊天类型，直接加载所有记忆
    context = self.memory.retrieve_all()
    return self._generate(message, context)

# 正确代码 ✅
def generate_response(self, message: Message):
    chat_type = self._get_chat_type(message.chat_id)
    
    if chat_type == 'group':
        # 群聊仅加载公开项目记忆
        context = self.memory.retrieve(scope=MemoryScope(l1=False, l2=False, l3=True))
    else:
        # 私聊加载全部记忆
        context = self.memory.retrieve(scope=MemoryScope(l1=True, l2=True, l3=True))
    
    return self._generate(message, context)

def maintain_memory_files(self):
    """记忆文件维护"""
    
    # 1. 去重
    self._deduplicate_entries()
    
    # 2. 压缩
    self._compress_old_entries(days_threshold=7)
    
    # 3. 归档
    self._archive_old_files(days_threshold=30)
    
    # 4. 提炼长期记忆
    self._extract_long_term_memories()

def fit_context_window(self, context: MemoryContext, max_tokens: int = 28000):
    """适配上下文窗口"""
    
    # 1. 计算当前 token 数
    current_tokens = self._count_tokens(context)
    
    # 2. 如果超出，按优先级裁剪
    if current_tokens > max_tokens:
        # 优先保留 L1（核心事实）
        # 其次保留 L2（近期记忆）
        # 最后保留 L3（项目上下文）
        
        while current_tokens > max_tokens:
            if context.l3:
                context.l3 = self._trim_l3(context.l3)
            elif context.l2:
                context.l2 = self._trim_l2(context.l2)
            else:
                break
            
            current_tokens = self._count_tokens(context)
    
    return context

# 高效检索模式
def retrieve_memory(self, query: str, context: dict):
    # 1. 先查缓存
    cached = self.cache.get(query)
    if cached:
        return cached
    
    # 2. 确定范围
    scope = self._determine_scope(context)
    
    # 3. 并行检索
    results = await asyncio.gather(
        self._retrieve_l1(scope),
        self._retrieve_l2(scope),
        self._retrieve_l3(scope)
    )
    
    # 4. 融合排序
    fused = self._fuse_and_rank(results)
    
    # 5. 缓存结果
    self.cache.set(query, fused)
    
    return fused

# skills/image-to-xmind/generate_xmind.py

import zipfile
import xml.etree.ElementTree as ET
from pathlib import Path

class XMindGenerator:
    """XMind 文件生成器"""
    
    def __init__(self, template_dir: str = 'templates'):
        self.template_dir = Path(template_dir)
        self.content_template = self._load_template('content.xml')
        self.styles_template = self._load_template('styles.xml')
    
    def generate(self, content: dict, output_path: str) -> str:
        """
        生成 XMind 文件
        
        Args:
            content: 思维导图内容（嵌套字典）
            output_path: 输出文件路径
            
        Returns:
            生成的文件路径
        """
        import tempfile
        
        # 1. 创建临时目录
        with tempfile.TemporaryDirectory() as tmpdir:
            tmpdir = Path(tmpdir)
            
            # 2. 生成 content.xml
            content_xml = self._generate_content(content)
            (tmpdir / 'content.xml').write_text(content_xml)
            
            # 3. 生成 styles.xml
            styles_xml = self._generate_styles(content)
            (tmpdir / 'styles.xml').write_text(styles_xml)
            
            # 4. 生成 manifest.xml
            manifest = self._generate_manifest()
            (tmpdir / 'META-INF' / 'manifest.xml').write_text(manifest)
            
            # 5. 打包为 ZIP（XMind 本质是 ZIP）
            output = Path(output_path)
            with zipfile.ZipFile(output, 'w', zipfile.ZIP_STORED) as zf:
                for file in tmpdir.rglob('*'):
                    arcname = file.relative_to(tmpdir)
                    zf.write(file, arcname)
        
        return str(output)
    
    def _generate_content(self, content: dict) -> str:
        """生成 content.xml"""
        
        # 使用模板替换
        root_topic = self._build_topic_xml(content)
        
        return self.content_template.replace(
            '{{ROOT_TOPIC}}',
            ET.tostring(root_topic, encoding='unicode')
        )
    
    def _build_topic_xml(self, node: dict) -> ET.Element:
        """构建主题 XML"""
        
        topic = ET.Element('topic')
        topic.set('id', node.get('id', 'root'))
        
        # 标题
        title = ET.SubElement(topic, 'title')
        title.text = node['title']
        
        # 样式
        if 'color' in node:
            props = ET.SubElement(topic, 'topic-properties')
            props.set('svg:fill', node['color'])
        
        # 子主题
        if 'children' in node:
            children = ET.SubElement(topic, 'children')
            for child in node['children']:
                child_topic = self._build_topic_xml(child)
                children.append(child_topic)
        
        return topic

# skills/image-to-xmind/generate_xmind-expert.py
# 专家脚本：经过多次迭代优化的最佳实践

"""
XMind 生成专家脚本

使用方式：
1. 修改 content_xml 中的节点内容
2. 修改 styles_xml 中的颜色值（可选）
3. 运行：python3 generate_xmind-expert.py

验证清单：
- [ ] XMind 可正常打开
- [ ] 颜色显示正确
- [ ] 层级结构正确
- [ ] 无乱码/格式错误
"""

import zipfile
from pathlib import Path

# 内容模板（关键：根元素必须是 map，命名空间必须正确）
CONTENT_XML = """<?xml version="1.0" encoding="UTF-8" standalone="no"?>
<map xmlns="urn:x-mind:map:1.0" xmlns:svg="urn:x-mind:svg:1.0">
  <topic id="root">
    <title>中心主题</title>
    <topic-properties svg:fill="#FFD700"/>
    <children>
      <topic id="topic-1">
        <title>分支主题 1</title>
        <topic-properties svg:fill="#FF6B6B"/>
      </topic>
      <topic id="topic-2">
        <title>分支主题 2</title>
        <topic-properties svg:fill="#4ECDC4"/>
      </topic>
    </children>
  </topic>
</map>
"""

# 样式模板（关键：必须包含 topic-properties 定义）
STYLES_XML = """<?xml version="1.0" encoding="UTF-8" standalone="no"?>
<styles xmlns="urn:x-mind:map:1.0" xmlns:svg="urn:x-mind:svg:1.0">
  <topic-properties svg:fill="#FFD700"/>
</styles>
"""

# Manifest（关键：不要带 manifest: 前缀）
MANIFEST_XML = """<?xml version="1.0" encoding="UTF-8"?>
<manifest:manifest xmlns:manifest="urn:oasis:names:tc:opendocument:xmlns:manifest:1.0">
 <manifest:file-entry manifest:media-type="application/vnd.xmind.workbook" manifest:full-path="/"/>
 <manifest:file-entry manifest:media-type="text/xml" manifest:full-path="content.xml"/>
 <manifest:file-entry manifest:media-type="text/xml" manifest:full-path="styles.xml"/>
</manifest:manifest>
"""

def generate_xmind(output_path: str = 'output.xmind'):
    """生成 XMind 文件"""
    
    output = Path(output_path)
    
    with zipfile.ZipFile(output, 'w', zipfile.ZIP_STORED) as zf:
        zf.writestr('content.xml', CONTENT_XML)
        zf.writestr('styles.xml', STYLES_XML)
        zf.writestr('META-INF/manifest.xml', MANIFEST_XML)
    
    print(f"✅ XMind 已生成：{output}")
    return output

if __name__ == '__main__':
    generate_xmind()

# skills/image-to-xmind/tests/test_xmind.py

import unittest
from pathlib import Path
from generate_xmind import XMindGenerator

class TestXMindGenerator(unittest.TestCase):
    
    def setUp(self):
        self.generator = XMindGenerator()
        self.test_output = '/tmp/test.xmind'
    
    def test_generate_basic(self):
        """测试基础生成"""
        
        content = {
            'id': 'root',
            'title': '测试主题',
            'children': [
                {'id': 'c1', 'title': '子主题 1'},
                {'id': 'c2', 'title': '子主题 2'}
            ]
        }
        
        output = self.generator.generate(content, self.test_output)
        
        # 验证文件存在
        self.assertTrue(Path(output).exists())
        
        # 验证可以打开
        with zipfile.ZipFile(output, 'r') as zf:
            self.assertIn('content.xml', zf.namelist())
            self.assertIn('styles.xml', zf.namelist())
    
    def test_generate_with_styles(self):
        """测试带样式生成"""
        
        content = {
            'id': 'root',
            'title': '彩色主题',
            'color': '#FF0000',
            'children': [
                {'id': 'c1', 'title': '红色子主题', 'color': '#FF0000'},
                {'id': 'c2', 'title': '蓝色子主题', 'color': '#0000FF'}
            ]
        }
        
        output = self.generator.generate(content, self.test_output)
        
        # 验证样式
        with zipfile.ZipFile(output, 'r') as zf:
            styles = zf.read('styles.xml').decode()
            self.assertIn('#FF0000', styles)
            self.assertIn('#0000FF', styles)
    
    def test_open_in_xmind(self):
        """测试在 XMind 中打开（集成测试）"""
        
        # 需要安装 XMind 命令行工具
        import subprocess
        
        content = {'id': 'root', 'title': '集成测试'}
        output = self.generator.generate(content, self.test_output)
        
        result = subprocess.run(
            ['xmind', 'open', output],
            capture_output=True
        )
        
        self.assertEqual(result.returncode, 0)

if __name__ == '__main__':
    unittest.main()

# skills/markdown-converter/SKILL.md

## 描述
将 PDF/Word/PPT/Excel/图片等转换为 Markdown，使用 markitdown 库

## 使用场景
- 用户需要处理非文本格式文档
- 提取 PDF/Word 中的文字内容
- 图片 OCR 识别
- 音频转录

## 工具函数

```python
def convert_to_markdown(file_path: str) -> str:
    """
    转换文件为 Markdown
    
    Args:
        file_path: 文件路径
        
    Returns:
        Markdown 内容
    """
    from markitdown import MarkItDown
    
    md = MarkItDown()
    result = md.convert(file_path)
    return result.text_content