project_memory_loader

#!/usr/bin/env python3
"""
项目记忆加载器 v2.0
用于调用项目记忆 API 获取项目上下文信息

功能：
- 加载项目上下文（PROJECT.md + TODO + 记忆数据）
- 离线模式支持
- 记忆质量评分
- 重复记忆合并
- 过期记忆清理
- 自动维护

API 端点：http://localhost:5000/api/project-memory/context
"""

import argparse
import json
import os
import re
import sys
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional

try:
    import requests
except ImportError:
    requests = None
    print("Warning: requests module not installed. Offline mode will be used.", file=sys.stderr)

# 配置
DEFAULT_API_URL = "http://localhost:5000/api/project-memory/context"
PROJECT_ROOT = os.environ.get("COZE_WORKSPACE_PATH", "/workspace/projects")
OFFLINE_MODE = os.environ.get("OFFLINE_MODE", "").lower() in ("true", "1", "yes")

class ProjectContextLoader:
    """项目上下文加载器"""

def __init__(
        self,
        api_url: str = DEFAULT_API_URL,
        offline: bool = False,
        timeout: int = 10
    ):
        self.api_url = api_url
        self.offline = offline or OFFLINE_MODE or requests is None
        self.timeout = timeout
        self.project_root = Path(PROJECT_ROOT)

def load_context(
        self,
        format: str = "json",
        refresh: bool = False,
        compact: bool = False
    ) -> Dict[str, Any]:
        """
        加载项目上下文

Args:
            format: 返回格式 (json/markdown)
            refresh: 是否强制刷新缓存
            compact: 是否使用精简模式

Returns:
            项目上下文数据
        """
        if self.offline:
            return self._load_offline()

return self._load_from_api(format, refresh, compact)

def _load_from_api(
        self,
        format: str,
        refresh: bool,
        compact: bool
    ) -> Dict[str, Any]:
        """从 API 加载上下文"""
        params = {}
        if format == "markdown":
            params["format"] = "markdown"
        if refresh:
            params["refresh"] = "true"
        if compact:
            params["compact"] = "true"

try:
            response = requests.get(
                self.api_url,
                params=params,
                headers={"Accept": "application/json"},
                timeout=self.timeout
            )

if response.status_code == 200:
                if format == "markdown":
                    return {
                        "success": True,
                        "format": "markdown",
                        "content": response.text
                    }
                return response.json()
            elif response.status_code == 404:
                return {
                    "success": False,
                    "error": "API 端点不存在，请确认 API 服务已正确配置",
                    "fallback": self._load_offline()
                }
            else:
                return {
                    "success": False,
                    "error": f"API 请求失败，状态码: {response.status_code}",
                    "fallback": self._load_offline()
                }

except requests.exceptions.ConnectionError:
            return {
                "success": False,
                "error": "无法连接到 API 服务，使用离线模式",
                "offline": True,
                "fallback": self._load_offline()
            }
        except requests.exceptions.Timeout:
            return {
                "success": False,
                "error": "API 请求超时",
                "fallback": self._load_offline()
            }
        except Exception as e:
            return {
                "success": False,
                "error": f"API 请求异常: {str(e)}",
                "fallback": self._load_offline()
            }

def _load_offline(self) -> Dict[str, Any]:
        """离线模式：直接读取本地文件"""
        result = {
            "success": True,
            "offline": True,
            "loadedAt": datetime.now().isoformat(),
            "project": {},
            "todos": {"pending": [], "inProgress": [], "completed": []},
            "memories": {"recent": [], "highPriority": []},
            "redLines": [],
            "commonErrors": [],
            "devGuide": {"quickCommands": [], "coreFiles": []}
        }

# 读取 PROJECT.md
        project_md = self._read_project_md()
        if project_md:
            result["project"] = project_md["project"]
            result["todos"] = project_md["todos"]

# 读取压缩摘要
        summary = self._read_summary_md()
        if summary:
            result["todos"].update(summary.get("todos", {}))
            result["redLines"] = summary.get("redLines", [])
            result["commonErrors"] = summary.get("commonErrors", [])
            result["devGuide"] = summary.get("devGuide", {})

return result

def _read_project_md(self) -> Optional[Dict[str, Any]]:
        """读取 PROJECT.md"""
        project_path = self.project_root / "PROJECT.md"
        if not project_path.exists():
            return None

try:
            content = project_path.read_text(encoding="utf-8")
            return self._parse_project_md(content)
        except Exception as e:
            return {"error": str(e)}

def _parse_project_md(self, content: str) -> Dict[str, Any]:
        """解析 PROJECT.md 内容"""
        result = {
            "project": {},
            "todos": {"pending": [], "inProgress": [], "completed": []}
        }

# 解析基本信息
        name_match = re.search(r"\*\*名称\*\*\s*\|\s*(.+)", content)
        version_match = re.search(r"\*\*版本\*\*\s*\|\s*(.+)", content)
        tech_match = re.search(r"\*\*技术栈\*\*\s*\|\s*(.+)", content)

result["project"] = {
            "name": name_match.group(1).strip() if name_match else "工资表SaaS系统",
            "version": version_match.group(1).strip() if version_match else "0.5.0",
            "techStack": [t.strip() for t in tech_match.group(1).split(",")] if tech_match else [],
            "port": 5000
        }

# 解析待开发功能
        pending_section = re.search(r"## &#128679; 待开发功能[\s\S]*?(?=##|$)", content)
        if pending_section:
            lines = pending_section.group(0).split("\n")
            current_priority = "normal"

for line in lines:
                priority_match = re.match(r"### (P\d+):", line)
                if priority_match:
                    p_num = int(priority_match.group(1)[1:])
                    current_priority = "critical" if p_num <= 1 else ("high" if p_num == 2 else "normal")

item_match = re.match(r"- \[ \] (.+)", line)
                if item_match:
                    result["todos"]["pending"].append({
                        "content": item_match.group(1).strip(),
                        "priority": current_priority,
                        "status": "pending"
                    })

# 解析已完成功能
        completed_section = re.search(r"## ✅ 已完成功能[\s\S]*?(?=##|$)", content)
        if completed_section:
            lines = completed_section.group(0).split("\n")
            for line in lines:
                item_match = re.match(r"### (P\d+): (.+)", line)
                if item_match:
                    result["todos"]["completed"].append({
                        "content": item_match.group(2).strip(),
                        "priority": "normal",
                        "status": "completed"
                    })

return result

def _read_summary_md(self) -> Optional[Dict[str, Any]]:
        """读取压缩摘要"""
        summary_path = self.project_root / "docs/17-其他/压缩摘要.md"
        if not summary_path.exists():
            return None

try:
            content = summary_path.read_text(encoding="utf-8")
            return self._parse_summary_md(content)
        except Exception as e:
            return {"error": str(e)}

def _parse_summary_md(self, content: str) -> Dict[str, Any]:
        """解析压缩摘要内容"""
        result = {
            "todos": {"pending": [], "inProgress": [], "completed": []},
            "redLines": [],
            "commonErrors": [],
            "devGuide": {"quickCommands": [], "coreFiles": []}
        }

# 解析 TODO
        todo_section = re.search(r"## TODO[\s\S]*?(?=##|$)", content)
        if todo_section:
            lines = todo_section.group(0).split("\n")
            for line in lines:
                item_match = re.match(r"- (.+)", line)
                if item_match and "TODO" not in item_match.group(1):
                    result["todos"]["pending"].append({
                        "content": item_match.group(1).strip(),
                        "priority": "normal",
                        "status": "pending"
                    })

# 解析常见错误
        error_section = re.search(r"## &#128027; 常见错误速查[\s\S]*?(?=##|$)", content)
        if error_section:
            lines = error_section.group(0).split("\n")
            for line in lines:
                match = re.match(r"\|\s*(.+?)\s*\|\s*(.+?)\s*\|\s*(.+?)\s*\|", line)
                if match and match.group(1) != "错误":
                    result["commonErrors"].append({
                        "error": match.group(1).strip(),
                        "cause": match.group(2).strip(),
                        "solution": match.group(3).strip()
                    })

# 解析快速命令
        cmd_section = re.search(r"## ⚡ 快速命令[\s\S]*?(?=##|$)", content)
        if cmd_section:
            lines = cmd_section.group(0).split("\n")
            for line in lines:
                match = re.match(r"\|\s*`?([^`|\n]+)`?\s*\|\s*(.+?)\s*\|\s*(.+?)\s*\|", line)
                if match and match.group(1) != "命令":
                    result["devGuide"]["quickCommands"].append({
                        "command": match.group(1).strip(),
                        "purpose": match.group(2).strip(),
                        "when": match.group(3).strip()
                    })

return result

def generate_markdown_summary(self, context: Dict[str, Any]) -> str:
        """生成 Markdown 格式的摘要"""
        md = "# &#128203; 项目状态概览\n\n"
        md += f"> 加载时间: {datetime.now().strftime('%Y/%m/%d %H:%M:%S')}\n"
        md += f"> API 状态: {'❌ 离线模式' if context.get('offline') else '✅ 在线'}\n\n"

# 项目信息
        project = context.get("project", {})
        if project:
            md += "## &#128230; 项目信息\n"
            md += "| 项目 | 内容 |\n"
            md += "|------|------|\n"
            md += f"| **名称** | {project.get('name', 'N/A')} |\n"
            md += f"| **版本** | {project.get('version', 'N/A')} |\n"
            md += f"| **技术栈** | {', '.join(project.get('techStack', []))} |\n"
            md += f"| **端口** | {project.get('port', 5000)} |\n\n"

# TODO 列表
        todos = context.get("todos", {})
        pending = todos.get("pending", [])
        if pending:
            md += "## &#128221; 待办事项\n"
            for todo in pending[:10]:
                priority = todo.get("priority", "normal")
                icon = "&#128308;" if priority == "critical" else ("&#128993;" if priority == "high" else "⚪")
                md += f"- {icon} {todo.get('content', '')}\n"
            md += "\n"

# 最近记忆
        memories = context.get("memories", {})
        recent = memories.get("recent", [])
        if recent:
            md += "## &#129504; 最近记忆\n"
            for m in recent[:5]:
                md += f"- **[{m.get('type', 'unknown')}]** {m.get('title', '')}\n"
            md += "\n"

# 高优先级记忆
        high_priority = memories.get("highPriority", [])
        if high_priority:
            md += "## ⭐ 重要事项\n"
            for m in high_priority[:5]:
                importance = m.get("importance", "normal").upper()
                md += f"- **[{importance}]** {m.get('title', '')}\n"
            md += "\n"

# 红线规则
        red_lines = context.get("redLines", [])
        if red_lines:
            md += "## &#128680; 关键红线\n"
            for i, rule in enumerate(red_lines[:5], 1):
                md += f"{i}. {rule.get('title', rule.get('category', ''))}: {rule.get('description', '')[:50]}...\n"
            md += "\n"

# 快速命令
        dev_guide = context.get("devGuide", {})
        commands = dev_guide.get("quickCommands", [])
        if commands:
            md += "## ⚡ 快速命令\n"
            md += "| 命令 | 用途 |\n"
            md += "|------|------|\n"
            for cmd in commands[:5]:
                md += f"| `{cmd.get('command', '')}` | {cmd.get('purpose', '')} |\n"
            md += "\n"

return md

def call_maintenance_action(
        self,
        action: str,
        dry_run: bool = False,
        **kwargs
    ) -> Dict[str, Any]:
        """
        调用维护动作

Args:
            action: 动作类型 (quality/merge/cleanup/maintenance/report)
            dry_run: 是否试运行

Returns:
            动作结果
        """
        if self.offline:
            return {
                "success": False,
                "error": "离线模式不支持维护操作"
            }

api_url = self.api_url.replace("/context", "/maintenance")

try:
            response = requests.post(
                api_url,
                json={
                    "action": action,
                    "dryRun": dry_run,
                    **kwargs
                },
                headers={"Content-Type": "application/json"},
                timeout=self.timeout * 2
            )

if response.status_code == 200:
                return response.json()
            else:
                return {
                    "success": False,
                    "error": f"API 请求失败，状态码: {response.status_code}"
                }

except Exception as e:
            return {
                "success": False,
                "error": str(e)
            }

def main():
    """主函数"""
    parser = argparse.ArgumentParser(
        description="项目记忆加载器 v2.0 - 加载项目上下文信息",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
示例:
  # 加载项目上下文（JSON 格式）
  python project_memory_loader.py

# 加载项目上下文（Markdown 格式）
  python project_memory_loader.py --format markdown

# 精简模式
  python project_memory_loader.py --compact

# 离线模式
  python project_memory_loader.py --offline

# 获取记忆质量报告
  python project_memory_loader.py --action quality

# 预览重复记忆合并建议
  python project_memory_loader.py --action merge --dry-run

# 执行重复记忆合并
  python project_memory_loader.py --action merge

# 清理过期记忆
  python project_memory_loader.py --action cleanup

# 运行完整维护
  python project_memory_loader.py --action maintenance
        """
    )

parser.add_argument(
        "--format", "-f",
        choices=["json", "markdown"],
        default="json",
        help="输出格式 (默认: json)"
    )

parser.add_argument(
        "--refresh", "-r",
        action="store_true",
        help="强制刷新缓存"
    )

parser.add_argument(
        "--compact", "-c",
        action="store_true",
        help="精简模式，只返回核心信息"
    )

parser.add_argument(
        "--offline", "-o",
        action="store_true",
        help="离线模式，直接读取本地文件"
    )

parser.add_argument(
        "--action", "-a",
        choices=["quality", "merge", "cleanup", "maintenance", "report"],
        help="执行维护动作"
    )

parser.add_argument(
        "--dry-run",
        action="store_true",
        help="试运行，只返回结果不执行"
    )

parser.add_argument(
        "--api-url",
        default=DEFAULT_API_URL,
        help=f"API 端点 URL (默认: {DEFAULT_API_URL})"
    )

parser.add_argument(
        "--timeout",
        type=int,
        default=10,
        help="API 请求超时时间（秒）(默认: 10)"
    )

args = parser.parse_args()

loader = ProjectContextLoader(
        api_url=args.api_url,
        offline=args.offline,
        timeout=args.timeout
    )

# 执行维护动作
    if args.action:
        result = loader.call_maintenance_action(
            args.action,
            dry_run=args.dry_run
        )
        print(json.dumps(result, ensure_ascii=False, indent=2))
        sys.exit(0 if result.get("success") else 1)

# 加载上下文
    result = loader.load_context(
        format=args.format,
        refresh=args.refresh,
        compact=args.compact
    )

# 输出结果
    if args.format == "markdown" and "content" in result:
        print(result["content"])
    elif "fallback" in result and result.get("offline"):
        # 使用降级数据生成 markdown
        md = loader.generate_markdown_summary(result["fallback"])
        print(md)
    else:
        print(json.dumps(result, ensure_ascii=False, indent=2))

# 设置退出码
    if not result.get("success", True):
        sys.exit(1)

if __name__ == "__main__":
    main()