from __future__ import annotations import csv import datetime as dt import json import re import shutil from collections import Counter, defaultdict from pathlib import Path from typing import Any REPO_ROOT = Path(__file__).resolve().parents[2] HOME = Path.home() CLAUDE_HOME = HOME / ".claude" CODEX_HOME = HOME / ".codex" CURSOR_HOME = HOME / ".cursor" CURSOR_USER_SETTINGS = HOME / "AppData" / "Roaming" / "Cursor" / "User" / "settings.json" VSCODE_INSIDERS_ROOT = Path("C:/Dev/VSCodeInsiders") PROJECT_CURSOR = REPO_ROOT / ".cursor" PROJECT_RULES = PROJECT_CURSOR / "rules" PROJECT_SKILLS = PROJECT_CURSOR / "skills" PROJECT_HOOKS = PROJECT_CURSOR / "hooks" USER_CURSOR_SKILLS = CURSOR_HOME / "skills" USER_CURSOR_AGENTS = CURSOR_HOME / "agents" HISTORY_ROOT = REPO_ROOT / "docs" / "ai-env-history" HISTORY_SUMMARIES = HISTORY_ROOT / "sessions" MANIFEST_PATH = HISTORY_ROOT / "cursor_migration_manifest.json" STAMP = dt.datetime.now(dt.timezone(dt.timedelta(hours=8))).strftime("%Y%m%d-%H%M%S") BACKUP_ROOT = CURSOR_HOME / "backups" / "neozqyy-cursor-migration" / STAMP SKILLS_TO_COPY = [ "agent-introspection-debugging", "claude-api", "code-tour", "codebase-onboarding", "repo-scan", "rules-distill", "search-first", "security-review", "strategic-compact", "tdd-workflow", "claude-agent-roles", "claude-rules-reference", ] EXPLICIT_ONLY_SKILLS = { "agent-introspection-debugging", "claude-agent-roles", "claude-rules-reference", "code-tour", "codebase-onboarding", "repo-scan", "rules-distill", "search-first", "security-review", "tdd-workflow", } NATURAL_TRIGGER_SKILLS = { "claude-api", "neozqyy-cursor-migration", "strategic-compact", } TOPIC_RULES = [ ("AI 开发环境迁移", ["迁移", "cursor", "codex", "claude", "agents", "skills", "mcp", ".cursor", ".claude"]), ("AI 应用与 Prompt", ["app2", "app3", "prompt", "dashscope", "ai_run", "ai_run_logs", "app/ai", "百炼", "千问"]), ("Admin AI 管理后台", ["admin-web", "ai 管理", "ai management", "ai triggers", "airunlogs", "aioperations"]), ("运行时上下文", ["runtime_context", "runtime context", "sandbox", "admin_runtime_context"]), ("后端 API 与服务", ["apps/backend", "fastapi", "router", "service", "schemas", "websocket"]), ("ETL 与财务口径", ["apps/etl", "dwd", "dws", "finance", "consume_money", "revenue", "财务"]), ("数据库与 RLS", ["db/", "migrations", "schema", "rls", "fdw", "view", "postgres"]), ("小程序体验", ["apps/miniprogram", "wxml", "wxss", "微信", "小程序", "board-finance"]), ("任务引擎与触发器", ["task_engine", "task_generator", "trigger", "scheduler", "任务"]), ("审计与文档", ["docs/audit", "audit", "审计", "文档", "docs/"]), ] SECRET_PATTERNS = [ re.compile(r"sk-[A-Za-z0-9_-]{12,}"), re.compile(r"sk-proj-[A-Za-z0-9_-]{12,}"), re.compile(r"(?i)(password|passwd|pwd|token|secret|api[_-]?key)\s*[:=]\s*['\"]?[^'\"\s,;]+"), re.compile(r"postgresql://[^\s'\"`]+"), re.compile(r"mysql://[^\s'\"`]+"), re.compile(r"mongodb(?:\+srv)?://[^\s'\"`]+"), ] FILE_PATTERN = re.compile( r"(?:(?:[A-Za-z]:)?[\\/](?:Project|c)[\\/]NeoZQYY[\\/])?" r"((?:apps|db|docs|tools|scripts|packages|tests|infra|\.cursor|\.claude)[\\/][^\"'\s,;)\]}]+)", re.IGNORECASE, ) manifest: dict[str, Any] = { "generated_at": STAMP, "repo_root": str(REPO_ROOT), "backup_root": str(BACKUP_ROOT), "actions": [], "history_sources": [], } def record(action: str, target: Path, source: Path | None = None, note: str = "") -> None: manifest["actions"].append( { "action": action, "target": str(target), "source": str(source) if source else None, "note": note, } ) def sanitize(text: str, limit: int | None = None) -> str: value = text.replace("\r\n", "\n").replace("\r", "\n") for pattern in SECRET_PATTERNS: value = pattern.sub("[已脱敏]", value) value = re.sub(r"\s+", " ", value).strip() if limit and len(value) > limit: return value[: limit - 1].rstrip() + "…" return value def read_text(path: Path) -> str: return path.read_text(encoding="utf-8", errors="replace") def write_text(path: Path, text: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(text, encoding="utf-8", newline="\n") record("write", path) def backup(path: Path) -> None: if not path.exists(): return safe = re.sub(r"[^A-Za-z0-9_.-]+", "_", str(path).replace(":", "")) dst = BACKUP_ROOT / safe dst.parent.mkdir(parents=True, exist_ok=True) if path.is_dir(): if dst.exists(): shutil.rmtree(dst) shutil.copytree(path, dst) else: dst.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(path, dst) record("backup", dst, path) def copytree(src: Path, dst: Path) -> None: if dst.exists(): backup(dst) shutil.rmtree(dst) shutil.copytree(src, dst) record("copytree", dst, src) def parse_frontmatter(text: str) -> tuple[dict[str, str], str]: if not text.startswith("---"): return {}, text end = text.find("\n---", 3) if end == -1: return {}, text raw = text[3:end].strip() body = text[end + len("\n---") :].lstrip("\n") meta: dict[str, str] = {} for line in raw.splitlines(): if ":" not in line: continue key, value = line.split(":", 1) meta[key.strip()] = value.strip().strip('"').strip("'") return meta, body def render_frontmatter(meta: dict[str, Any], body: str) -> str: lines = ["---"] for key, value in meta.items(): if isinstance(value, bool): rendered = "true" if value else "false" else: rendered = str(value) lines.append(f"{key}: {rendered}") lines.append("---") return "\n".join(lines) + "\n\n" + body.lstrip("\n") def normalize_file_path(value: str) -> str: value = value.replace("\\\\", "\\").replace("\\", "/").strip() value = re.sub(r"^[A-Za-z]:/Project/NeoZQYY/", "", value, flags=re.IGNORECASE) value = re.sub(r"^/c/Project/NeoZQYY/", "", value, flags=re.IGNORECASE) for marker in ["\\n", "/n", "\\r", "/r", "`", "<", ">", "|"]: value = value.split(marker, 1)[0] ext_match = re.match( r"(.+?\.(?:py|ts|tsx|js|jsx|md|mdc|sql|json|toml|yaml|yml|wxml|wxss|css|scss|html|png|jpg|jpeg|xlsx|csv|txt|bat|ps1))", value, flags=re.IGNORECASE, ) if ext_match: value = ext_match.group(1) value = value.strip("/").rstrip("`").strip() lower = value.lower() if not value or len(value) > 240: return "" if "postgresql://" in lower or "=" in value: return "" if lower.endswith("/.env") or lower.endswith(".env") or ".env.local" in lower: return "" return value def extract_files(text: str) -> set[str]: return {path for match in FILE_PATTERN.finditer(text) if (path := normalize_file_path(match.group(1)))} def extract_message_text(message: Any) -> str: if isinstance(message, str): return message if isinstance(message, dict): content = message.get("content") if isinstance(content, str): return content if isinstance(content, list): parts: list[str] = [] for item in content: if isinstance(item, dict): if item.get("type") in {"text", "input_text", "output_text"}: parts.append(str(item.get("text", ""))) elif "content" in item and isinstance(item["content"], str): parts.append(item["content"]) elif isinstance(item, str): parts.append(item) return "\n".join(parts) return "" def extract_json_value(data: dict[str, Any], *path: str) -> Any: node: Any = data for key in path: if not isinstance(node, dict): return None node = node.get(key) return node def write_rule(name: str, description: str, globs: str | None, always: bool, body: str) -> None: frontmatter = ["---", f"description: {description}"] if globs: frontmatter.append(f"globs: {globs}") frontmatter.append(f"alwaysApply: {'true' if always else 'false'}") frontmatter.append("---") write_text(PROJECT_RULES / f"{name}.mdc", "\n".join(frontmatter) + "\n\n" + body.strip() + "\n") def create_project_rules() -> None: backup(PROJECT_RULES) write_rule( "neozqyy-core", "NeoZQYY 核心工作规范：中文、调研、验证、审计、dirty tree 保护。", None, True, """ # NeoZQYY 核心规范 - 始终使用中文交流、解释、审计和文档；命令、API 字段、变量名保持原文。 - 以 `AGENTS.md` 为当前权威规则，`CLAUDE.md` 作为历史兼容来源。 - 逻辑改动前先做需求审问和前置调研；用户明确跳过时除外。 - 逻辑改动后运行相关验证，输出 diff 摘要和未覆盖风险。 - 不回滚用户已有改动，不使用破坏性 git 命令，除非用户明确要求。 - 审计记录统一写入 `docs/audit/changes/`，`docs/audit/audit_dashboard.md` 只由脚本生成。 - 历史追溯优先查 `docs/ai-env-history/`、`docs/claude-history/`、`docs/audit/`，再查原始对话。 - 用户偏好模型为 GPT 5.5 与 Claude 4.7；模型选择由 Cursor UI/会话设置控制，规则只保留偏好。 - CLI / Shell 中文处理必须优先确保 UTF-8：Python 用 `encoding="utf-8"` / `PYTHONUTF8=1`，CSV 给 Excel 用 `utf-8-sig`，PowerShell/Node 避免依赖系统默认 ANSI 编码。 - 遇到中文乱码时，不要把乱码输出当作事实；先调整编码重跑，或明确说明终端编码异常并转述可确认的信息。 - Shell 路径和参数含中文、空格或特殊字符时必须正确加引号；复杂中文输出优先用脚本或结构化 API，避免手写脆弱转义。 """, ) write_rule( "backend-fastapi", "FastAPI 后端规则：响应包装、认证、AI 集成、RLS 与测试库。", "apps/backend/**", False, """ # 后端规则 - 2xx 响应经 `ResponseWrapperMiddleware` 包装为 `{ "code": 0, "data": }`。 - 后端内部使用 snake_case，JSON 输出通过 `CamelModel` 转 camelCase。 - admin、miniapp、tenant-admin 三类 JWT aud 不可混用。 - 访问 ETL FDW/RLS 视图前必须设置 `app.current_site_id`。 - AI 集成涉及 DashScope、熔断、限流、预算、缓存和运行日志，改动后必须查审计历史。 - 后端验证默认在 `apps/backend` 下运行，使用测试库，禁止连正式库。 """, ) write_rule( "etl-feiqiu", "飞球 ETL 规则：DWD-DOC 优先、金额口径、DWS 优先、禁止归档目录。", "apps/etl/connectors/feiqiu/**", False, """ # ETL 飞球规则 - 金额、支付、消费链路、字段语义优先参考 `apps/etl/connectors/feiqiu/docs/reports/DWD-DOC/`。 - `consume_money` 禁止直接用于计算，使用 `items_sum` 拆分字段。 - 助教费用必须区分 `assistant_pd_money` 和 `assistant_cx_money`。 - 正向结算使用 `settle_type IN (1, 3)`，禁止随意读取 ODS 做业务计算。 - DWS/DWD 汇总默认保持幂等，禁止 `TRUNCATE`。 - 所有 `_archived/` 目录禁止读取或参考。 """, ) write_rule( "database", "数据库规则：schema 变更、RLS 双 schema、文档同步和验证 SQL。", "db/**,docs/database/**", False, """ # 数据库规则 - 任何 PostgreSQL schema/迁移/DDL/ORM 结构变更必须同步 `docs/database/`。 - 新建 DWS/DWD RLS 视图必须同时创建原 schema 和 `app` schema 视图。 - 回滚需逆序 DROP/ALTER，并提供至少 3 条验证 SQL。 - 默认使用 `TEST_DB_DSN` / `TEST_APP_DB_DSN`，禁止连正式库。 - DDL 基线变更后运行 `tools/db/gen_consolidated_ddl.py` 并同步权威 DDL。 """, ) write_rule( "admin-web", "admin-web 规则：React/Vite/AntD、AI 管理套件与前端验证。", "apps/admin-web/**", False, """ # admin-web 规则 - `admin-web` 是开发/运维后台，不是租户后台。 - 遵循 React + Vite + Ant Design 现有页面和 API 封装风格。 - AI 管理相关改动先查 2026-04-21、2026-04-30 审计记录。 - 前端逻辑改动后优先运行 `pnpm test` / `pnpm lint`，无法运行需说明原因。 """, ) write_rule( "miniprogram", "微信小程序规则：生产小程序、Donut/TDesign、demo 标杆对齐。", "apps/miniprogram/**", False, """ # 小程序规则 - `apps/miniprogram` 是生产小程序，数据来自后端 API。 - UI 样式和展示格式需要参考 `apps/demo-miniprogram` 的 MOCK 标杆。 - 改动关键交互、鉴权、API 字段或页面跳转后必须说明验证路径。 - 涉及微信开发者工具时优先使用 `weixin-devtools-mcp` 或记录手工验证步骤。 """, ) write_rule( "demo-miniprogram-protect", "demo-miniprogram 保护规则：假数据标杆，不删除不迁移到 _DEL。", "apps/demo-miniprogram/**", False, """ # demo-miniprogram 保护 - 本目录是假数据 MOCK 版小程序，用于页面样式和展示格式标杆校对。 - 禁止删除、移入 `_DEL/` 或改造成真实 API 驱动。 - 只有在用户明确要求校正 demo 标杆时才修改。 """, ) write_rule( "audit-history", "历史追溯规则：优先使用精简索引、审计记录和当前代码。", "docs/**", False, """ # 历史追溯规则 - 日常追溯先查 `docs/ai-env-history/README.md`、`docs/claude-history/`、`docs/audit/changes/`。 - 历史摘要只解释来龙去脉，编码前仍以当前文件、当前 diff、当前测试为准。 - 原始 JSONL 可用于追查细节，但不要把密钥、DSN、token 原文写入仓库文档。 """, ) def create_user_skills() -> None: USER_CURSOR_SKILLS.mkdir(parents=True, exist_ok=True) for name in SKILLS_TO_COPY: src = CODEX_HOME / "skills" / name if not src.exists(): src = CLAUDE_HOME / "skills" / name if src.exists(): target = USER_CURSOR_SKILLS / name copytree(src, target) skill_md = target / "SKILL.md" if name in EXPLICIT_ONLY_SKILLS and skill_md.exists(): meta, body = parse_frontmatter(read_text(skill_md)) meta["disable-model-invocation"] = True write_text(skill_md, render_frontmatter(meta, body)) skill = """--- name: neozqyy-cursor-migration description: NeoZQYY AI 开发环境迁移与历史追溯工作流。Use when 需要理解 Claude Code、Codex、Cursor 迁移关系，查历史对话，或恢复用户 AI 开发习惯。 --- # NeoZQYY Cursor 迁移 ## 快速入口 1. 当前权威项目规则：`AGENTS.md` 与 `.cursor/rules/`。 2. 迁移说明：`docs/cursor_migration.md`。 3. 历史索引：`docs/ai-env-history/README.md`。 4. Claude 历史摘要：`docs/claude-history/`。 5. 审计记录：`docs/audit/changes/`。 ## 使用原则 - 先查精简索引，再读原始对话。 - 历史用于理解前因后果，不替代当前代码和测试。 - GPT 5.5 / Claude 4.7 是用户偏好模型；具体模型选择由 Cursor 当前会话或 UI 控制。 """ write_text(USER_CURSOR_SKILLS / "neozqyy-cursor-migration" / "SKILL.md", skill) def create_user_agents() -> None: USER_CURSOR_AGENTS.mkdir(parents=True, exist_ok=True) src_root = CLAUDE_HOME / "agents" if not src_root.exists(): return for src in sorted(src_root.glob("*.md")): text = read_text(src) meta, body = parse_frontmatter(text) name = re.sub(r"[^a-z0-9-]", "-", meta.get("name", src.stem).lower()).strip("-") or src.stem description = meta.get("description") or f"从 Claude Code 迁移的 {name} 子代理。Use proactively when relevant." target = USER_CURSOR_AGENTS / f"{name}.md" write_text(target, f"---\nname: {name}\ndescription: {description}\n---\n\n{body.strip()}\n") def create_project_skills() -> None: backup(PROJECT_SKILLS) commands_dir = REPO_ROOT / ".claude" / "commands" for src in sorted(commands_dir.glob("*.md")): body = read_text(src) title = body.splitlines()[0].lstrip("# ").strip() if body.splitlines() else src.stem description = f"{title}。从 Claude Code 命令迁移为 Cursor project skill；用户要求执行 {src.stem}、/{src.stem} 或相关流程时使用。" content = ( "---\n" f"name: {src.stem}\n" f"description: {description}\n" "disable-model-invocation: true\n" "---\n\n" + body.strip() + "\n" ) write_text(PROJECT_SKILLS / src.stem / "SKILL.md", content) def create_project_hooks() -> None: backup(PROJECT_CURSOR / "hooks.json") backup(PROJECT_HOOKS) hook_config = { "version": 1, "hooks": { "preToolUse": [ { "command": ".venv\\Scripts\\python.exe .cursor\\hooks\\ai_env_guard.py preToolUse", "matcher": "Read|Glob|Edit|Write|ApplyPatch", "timeout": 5, "failClosed": True, } ], "beforeReadFile": [ { "command": ".venv\\Scripts\\python.exe .cursor\\hooks\\ai_env_guard.py beforeReadFile", "matcher": "_archived", "timeout": 5, "failClosed": True, } ], "beforeMCPExecution": [ { "command": ".venv\\Scripts\\python.exe .cursor\\hooks\\ai_env_guard.py beforeMCPExecution", "matcher": "pg-etl|pg-app", "timeout": 5, "failClosed": False, } ], "beforeShellExecution": [ { "command": ".venv\\Scripts\\python.exe .cursor\\hooks\\ai_env_guard.py beforeShellExecution", "matcher": "git\\s+(reset|checkout)|--no-verify|psql|mcp-postgres|PG_DSN|APP_DB_DSN", "timeout": 5, "failClosed": False, } ], "postToolUse": [ { "command": ".venv\\Scripts\\python.exe .cursor\\hooks\\ai_env_guard.py postToolUse", "matcher": "ApplyPatch|Edit|Write", "timeout": 5, "failClosed": False, } ], }, } write_text(PROJECT_CURSOR / "hooks.json", json.dumps(hook_config, ensure_ascii=False, indent=2) + "\n") hook_script = r'''from __future__ import annotations import json import re import sys from pathlib import Path ARCHIVED_RE = re.compile(r"(^|[/\\])_archived([/\\]|$)", re.IGNORECASE) PROD_MCP_RE = re.compile(r"\b(pg-etl|pg-app)\b", re.IGNORECASE) TEST_MCP_RE = re.compile(r"\b(pg-etl-test|pg-app-test)\b", re.IGNORECASE) WRITE_SQL_RE = re.compile( r"\b(insert|update|delete|truncate|drop|alter|create|grant|revoke|merge|copy|call|vacuum|reindex)\b", re.IGNORECASE, ) def load_input() -> dict: try: return json.load(sys.stdin) except Exception: return {} def target_text(data: dict) -> str: chunks = [json.dumps(data, ensure_ascii=False)] for key in ("command", "file_path", "path"): value = data.get(key) if isinstance(value, str): chunks.append(value) tool_input = data.get("tool_input") if isinstance(tool_input, dict): chunks.append(json.dumps(tool_input, ensure_ascii=False)) return "\n".join(chunks) def allow() -> None: print(json.dumps({"permission": "allow"}, ensure_ascii=False)) def before_shell(data: dict) -> None: text = target_text(data) dangerous = [ r"git\s+reset\s+--hard", r"git\s+checkout\s+--", r"--no-verify", r"\bPG_DSN\b", r"\bAPP_DB_DSN\b", r"\bpsql\b", ] if any(re.search(pattern, text, re.IGNORECASE) for pattern in dangerous): print(json.dumps({ "permission": "ask", "user_message": "命令命中 NeoZQYY 高风险规则：请确认是否需要执行。", "agent_message": "执行前说明风险、目标库/文件、回滚方式；生产库默认不执行。", }, ensure_ascii=False)) return allow() def guard_archived(data: dict) -> None: text = target_text(data) if ARCHIVED_RE.search(text): print(json.dumps({ "permission": "deny", "user_message": "已阻断：`_archived/` 是废弃归档目录，禁止读取、搜索或作为实现参考。", "agent_message": "改用当前版本文件；如确需考古，请让用户明确授权并说明目的。", }, ensure_ascii=False)) return allow() def before_mcp(data: dict) -> None: text = target_text(data) if not PROD_MCP_RE.search(text) or TEST_MCP_RE.search(text): allow() return if WRITE_SQL_RE.search(text): print(json.dumps({ "permission": "deny", "user_message": "已阻断：生产库 MCP 写入/DDL 类操作默认禁止。请改用测试库验证，或让用户给出单次明确授权。", "agent_message": "生产库仅允许人工确认后的只读排查；DDL/写入需走单独变更流程。", }, ensure_ascii=False)) return print(json.dumps({ "permission": "ask", "user_message": "检测到生产库 MCP 只读调用。请确认本次查询目的、SQL 是否只读、是否会暴露敏感数据。", "agent_message": "说明查询目的、目标库、SQL 摘要和风险后等待用户确认。", }, ensure_ascii=False)) def post_tool_use(data: dict) -> None: text = target_text(data).replace("\\", "/") hints = [] if "/_archived/" in text or text.endswith("/_archived"): hints.append("检测到 `_archived/` 路径：该目录内容已废弃，禁止作为实现参考。") if "apps/demo-miniprogram/" in text: hints.append("检测到 demo-miniprogram：这是 MOCK 标杆目录，修改前需确认目的。") if re.search(r"(db/|docs/database/|migrations/|schemas/)", text): hints.append("检测到数据库相关改动：如影响 schema，需同步 `docs/database/` 并提供 3 条验证 SQL。") if re.search(r"(apps/backend/|apps/etl/|app/ai/|prompts/)", text): hints.append("检测到后端/ETL/AI 逻辑改动：完成后需运行相关测试并写审计记录。") if hints: print(json.dumps({"additional_context": "\n".join(f"[NeoZQYY] {hint}" for hint in hints)}, ensure_ascii=False)) def main() -> None: mode = sys.argv[1] if len(sys.argv) > 1 else "" data = load_input() if mode in {"preToolUse", "beforeReadFile"}: guard_archived(data) elif mode == "beforeMCPExecution": before_mcp(data) elif mode == "beforeShellExecution": before_shell(data) elif mode == "postToolUse": post_tool_use(data) if __name__ == "__main__": main() ''' write_text(PROJECT_HOOKS / "ai_env_guard.py", hook_script) def summarize_jsonl(path: Path, source: str) -> dict[str, Any]: session_id = path.stem timestamps: list[str] = [] cwd_values: Counter[str] = Counter() user_prompts: list[str] = [] assistant_notes: list[str] = [] tools: Counter[str] = Counter() files: set[str] = set() line_count = 0 with path.open("r", encoding="utf-8", errors="replace") as fh: for line in fh: line_count += 1 raw = line.strip() if not raw: continue files.update(extract_files(raw)) try: data = json.loads(raw) except Exception: continue ts = data.get("timestamp") or extract_json_value(data, "payload", "timestamp") if isinstance(ts, str): timestamps.append(ts) sid = data.get("sessionId") or extract_json_value(data, "payload", "id") if isinstance(sid, str): session_id = sid cwd = data.get("cwd") or extract_json_value(data, "payload", "cwd") if isinstance(cwd, str): cwd_values[cwd] += 1 role = data.get("type") or data.get("role") or extract_json_value(data, "payload", "role") if role == "user": prompt = extract_message_text(data.get("message") or data) if prompt: user_prompts.append(sanitize(prompt, 600)) elif role == "assistant": note = extract_message_text(data.get("message") or data) if note: assistant_notes.append(sanitize(note, 400)) payload = data.get("payload") if isinstance(data.get("payload"), dict) else {} if payload: if payload.get("type") == "function_call": name = payload.get("name") if isinstance(name, str): tools[name] += 1 args = payload.get("arguments") if isinstance(args, str): files.update(extract_files(args)) if payload.get("type") == "message": for item in payload.get("content", []) or []: if isinstance(item, dict): if item.get("type") == "tool_use": name = item.get("name") if isinstance(name, str): tools[name] += 1 files.update(extract_files(json.dumps(item, ensure_ascii=False))) message = data.get("message") if isinstance(message, dict): for item in message.get("content", []) or []: if isinstance(item, dict): if item.get("type") == "tool_use": name = item.get("name") if isinstance(name, str): tools[name] += 1 files.update(extract_files(json.dumps(item, ensure_ascii=False))) attachment = data.get("attachment") if isinstance(attachment, dict): atype = attachment.get("type") if isinstance(atype, str): tools[atype] += 1 first_ts = min(timestamps) if timestamps else "" last_ts = max(timestamps) if timestamps else "" prompts = user_prompts[:12] notes = assistant_notes[:8] risk_flags = [] joined_files = " ".join(sorted(files)).lower() if "db/" in joined_files or "migrations/" in joined_files or "schemas/" in joined_files: risk_flags.append("数据库") if "apps/backend/" in joined_files: risk_flags.append("后端") if "apps/etl/" in joined_files: risk_flags.append("ETL") if "apps/miniprogram/" in joined_files: risk_flags.append("小程序") if any("_archived" in item.lower() for item in files): risk_flags.append("涉及归档目录") return { "source": source, "session_id": session_id, "path": str(path), "first_ts": first_ts, "last_ts": last_ts, "line_count": line_count, "cwd": cwd_values.most_common(1)[0][0] if cwd_values else "", "user_prompts": prompts, "assistant_notes": notes, "tools": dict(tools.most_common(20)), "files": sorted(files), "risk_flags": risk_flags, "topics": classify_topics( "\n".join(user_prompts + assistant_notes + sorted(files) + list(tools.keys())) ), } def classify_topics(text: str) -> list[str]: lower = text.lower() topics = [] for topic, keywords in TOPIC_RULES: if any(keyword.lower() in lower for keyword in keywords): topics.append(topic) return topics or ["未分类"] def write_session_summary(summary: dict[str, Any]) -> None: source = summary["source"] sid = re.sub(r"[^A-Za-z0-9_.-]+", "_", summary["session_id"])[:120] target = HISTORY_SUMMARIES / source / f"{sid}.md" prompts = "\n".join(f"- {item}" for item in summary["user_prompts"]) or "- 未识别" notes = "\n".join(f"- {item}" for item in summary["assistant_notes"]) or "- 未识别" files = "\n".join(f"- `{item}`" for item in summary["files"][:80]) or "- 未识别" tools = "\n".join(f"- `{name}`: {count}" for name, count in summary["tools"].items()) or "- 未识别" risks = "、".join(summary["risk_flags"]) if summary["risk_flags"] else "未识别" body = f"""# {source} 会话 {summary['session_id']} ## 基本信息 - 来源：{source} - 时间：{summary['first_ts']} -> {summary['last_ts']} - 行数：{summary['line_count']} - cwd：`{summary['cwd']}` - 风险标签：{risks} - 主题：{", ".join(summary.get("topics", ["未分类"]))} - 原始记录：`{summary['path']}` ## 用户需求线索 {prompts} ## 助手回应线索 {notes} ## 工具与事件 {tools} ## 文件影响线索 {files} ## 使用说明 本摘要由原始 JSONL 逐行分析生成，用于理解对话前因后果。编码前仍需读取当前文件、git diff、审计记录和测试结果。 """ write_text(target, body) def collect_history_paths() -> dict[str, list[Path]]: return { "claude": sorted((CLAUDE_HOME / "projects" / "C--Project-NeoZQYY").glob("*.jsonl")), "codex": sorted((CODEX_HOME / "sessions").glob("**/*.jsonl")) + sorted((CODEX_HOME / "archived_sessions").glob("*.jsonl")), "cursor": sorted((CURSOR_HOME / "projects" / "c-Project-NeoZQYY" / "agent-transcripts").glob("*/*.jsonl")), } def analyze_histories() -> None: backup(HISTORY_ROOT) sources = collect_history_paths() all_summaries: list[dict[str, Any]] = [] file_index: dict[str, list[dict[str, str]]] = defaultdict(list) for source, paths in sources.items(): manifest["history_sources"].append({"source": source, "count": len(paths)}) for path in paths: summary = summarize_jsonl(path, source) all_summaries.append(summary) write_session_summary(summary) for file_path in summary["files"]: file_index[file_path].append( { "source": source, "session_id": summary["session_id"], "first_ts": summary["first_ts"], "last_ts": summary["last_ts"], "summary": str(HISTORY_SUMMARIES / source / f"{re.sub(r'[^A-Za-z0-9_.-]+', '_', summary['session_id'])[:120]}.md"), } ) HISTORY_ROOT.mkdir(parents=True, exist_ok=True) with (HISTORY_ROOT / "conversation_index.csv").open("w", encoding="utf-8-sig", newline="") as fh: writer = csv.DictWriter( fh, fieldnames=["source", "session_id", "first_ts", "last_ts", "line_count", "cwd", "file_count", "risk_flags", "summary_path", "raw_path"], ) writer.writeheader() for summary in sorted(all_summaries, key=lambda item: (item["first_ts"], item["source"])): sid = re.sub(r"[^A-Za-z0-9_.-]+", "_", summary["session_id"])[:120] writer.writerow( { "source": summary["source"], "session_id": summary["session_id"], "first_ts": summary["first_ts"], "last_ts": summary["last_ts"], "line_count": summary["line_count"], "cwd": summary["cwd"], "file_count": len(summary["files"]), "risk_flags": ";".join(summary["risk_flags"]), "summary_path": str(HISTORY_SUMMARIES / summary["source"] / f"{sid}.md"), "raw_path": summary["path"], } ) record("write", HISTORY_ROOT / "conversation_index.csv") with (HISTORY_ROOT / "file_impact_index.csv").open("w", encoding="utf-8-sig", newline="") as fh: writer = csv.DictWriter(fh, fieldnames=["file", "source", "session_id", "first_ts", "last_ts", "summary"]) writer.writeheader() for file_path, rows in sorted(file_index.items()): for row in rows: writer.writerow({"file": file_path, **row}) record("write", HISTORY_ROOT / "file_impact_index.csv") with (HISTORY_ROOT / "topic_timeline.csv").open("w", encoding="utf-8-sig", newline="") as fh: writer = csv.DictWriter( fh, fieldnames=["topic", "source", "session_id", "first_ts", "last_ts", "risk_flags", "prompt", "summary_path", "raw_path"], ) writer.writeheader() for summary in sorted(all_summaries, key=lambda item: (item["first_ts"], item["source"])): sid = re.sub(r"[^A-Za-z0-9_.-]+", "_", summary["session_id"])[:120] prompt = summary["user_prompts"][0] if summary["user_prompts"] else "" for topic in summary.get("topics", ["未分类"]): writer.writerow( { "topic": topic, "source": summary["source"], "session_id": summary["session_id"], "first_ts": summary["first_ts"], "last_ts": summary["last_ts"], "risk_flags": ";".join(summary["risk_flags"]), "prompt": prompt, "summary_path": str(HISTORY_SUMMARIES / summary["source"] / f"{sid}.md"), "raw_path": summary["path"], } ) record("write", HISTORY_ROOT / "topic_timeline.csv") grouped: dict[str, list[dict[str, Any]]] = defaultdict(list) for summary in all_summaries: for topic in summary.get("topics", ["未分类"]): grouped[topic].append(summary) parts = ["# AI 开发历史主题时间线", ""] for topic in sorted(grouped): parts.append(f"## {topic}") for summary in sorted(grouped[topic], key=lambda item: item["first_ts"])[:80]: sid = re.sub(r"[^A-Za-z0-9_.-]+", "_", summary["session_id"])[:120] prompt = summary["user_prompts"][0] if summary["user_prompts"] else "未识别用户需求" risks = "、".join(summary["risk_flags"]) if summary["risk_flags"] else "无显式风险标签" parts.append( f"- {summary['first_ts']} -> {summary['last_ts']} " f"`{summary['source']}` `{summary['session_id']}`：{prompt} " f"（风险：{risks}；摘要：`docs/ai-env-history/sessions/{summary['source']}/{sid}.md`）" ) parts.append("") write_text(HISTORY_ROOT / "topic_timeline.md", "\n".join(parts).rstrip() + "\n") claude_history = CLAUDE_HOME / "history.jsonl" plans = sorted((CLAUDE_HOME / "plans").glob("*.md")) file_history_count = len(list((CLAUDE_HOME / "file-history").glob("*/*"))) if (CLAUDE_HOME / "file-history").exists() else 0 checkpoints = sorted((HOME / "AppData" / "Roaming" / "Cursor" / "User" / "globalStorage" / "anysphere.cursor-commits" / "checkpoints").glob("*/metadata.json")) local_history = sorted((HOME / "AppData" / "Roaming" / "Cursor" / "User" / "History").glob("**/*.json")) catalog = f"""# AI 开发环境历史索引 ## 覆盖范围 - Claude 原始项目会话：{len(sources['claude'])} 个，逐一生成摘要。 - Codex 本地/归档会话：{len(sources['codex'])} 个，逐一生成摘要。 - Cursor 父会话 transcripts：{len(sources['cursor'])} 个，逐一生成摘要。 - Claude 全局 history：`{claude_history}`，存在：{claude_history.exists()}。 - Claude plans：{len(plans)} 份。 - Claude file-history 快照：约 {file_history_count} 个。 - Cursor 本地文件历史 JSON：{len(local_history)} 个。 - Cursor checkpoints：{len(checkpoints)} 个。 ## 主要入口 - 会话索引：`docs/ai-env-history/conversation_index.csv` - 文件影响索引：`docs/ai-env-history/file_impact_index.csv` - 主题时间线：`docs/ai-env-history/topic_timeline.md` - 会话摘要：`docs/ai-env-history/sessions/` - Claude 旧摘要：`docs/claude-history/` - 审计记录：`docs/audit/changes/` ## 使用策略 1. 查“某文件为什么改”：先查 `file_impact_index.csv`，再读对应会话摘要和审计记录。 2. 查“某次对话做了什么”：先查 `conversation_index.csv`，再读摘要；必要时读原始 JSONL。 3. 恢复误改：再查 Claude file-history、Cursor local History 或 Cursor checkpoints。 4. 原始记录可能包含敏感信息；摘要会脱敏常见密钥和 DSN 形态，原文只在本机按需读取。 """ write_text(HISTORY_ROOT / "README.md", catalog) def create_cursor_migration_doc() -> None: doc = """# Cursor AI 开发环境迁移说明 ## 状态 本轮把 NeoZQYY 的 AI 开发环境从 Claude Code / Codex 迁移到 Cursor 原生资产： - 项目规则：`.cursor/rules/` - 项目流程技能：`.cursor/skills/` - 用户技能：`C:\\Users\\Administrator\\.cursor\\skills` - 用户子代理：`C:\\Users\\Administrator\\.cursor\\agents` - 轻量 hooks：`.cursor/hooks.json` 与 `.cursor/hooks/` - 历史索引：`docs/ai-env-history/` ## 去重原则 1. 当前仓库 `AGENTS.md` 是项目权威规则。 2. `CLAUDE.md` 保留为历史兼容来源。 3. `.cursor/rules` 只放短规则和入口，不复制长文。 4. skills 使用“短入口 + reference”，避免每次会话加载过多旧内容。 5. 原始对话逐一分析为摘要和索引，不把原始 JSONL 全文写入仓库。 ## 模型偏好 用户主要使用 GPT 5.5 与 Claude 4.7。Cursor 的实际模型选择由当前会话或 UI 控制；本迁移只把偏好写入规则和说明，不强写不可控的内部模型字段。 ## 内置流程与规则边界 Cursor、GPT 5.5、Claude 4.7 都具备调研、审计、验证、子代理和工具调用能力，但它们不知道 NeoZQYY 的项目事实、历史踩坑、数据库口径、RLS 双 schema、demo 标杆保护和审计路径。因此仍需要项目规则和技能作为触发器与约束。 规则不重复实现模型能力，只固化三类内容： - 项目事实：目录职责、数据库、ETL、后端、前端、小程序约定。 - 用户习惯：中文、先调研、再改动、再验证、再审计。 - 风险边界：生产库禁用、敏感信息不扩散、危险 git/SQL/归档目录保护。 ## 历史入口 - 新索引：`docs/ai-env-history/README.md` - Claude 摘要：`docs/claude-history/README.md` - 审计一览：`docs/audit/audit_dashboard.md` - Codex 迁移说明：`docs/codex_migration.md` ## 用户级资产触发策略 ### 保留自然触发 - `neozqyy-cursor-migration`：NeoZQYY 迁移和历史追溯入口。 - `strategic-compact`：长会话上下文压缩建议。 - `claude-api`：仅在 Anthropic / Claude API 场景触发。 ### 改为显式触发 - `claude-agent-roles`：旧 Claude agents 参考；日常使用 Cursor subagents。 - `claude-rules-reference`：旧 Claude rules 档案；当前权威规则是 `AGENTS.md` 与 `.cursor/rules/`。 - `tdd-workflow`：详细 TDD 参考；日常入口是 `tdd-guide` subagent。 - `security-review`：详细安全 checklist；日常入口是 `security-reviewer` subagent。 - `search-first`：通用搜索优先流程；NeoZQYY 内优先使用 `.cursor/skills/pre-change`。 - `rules-distill`、`repo-scan`、`codebase-onboarding`、`code-tour`、`agent-introspection-debugging`：低频或重型工具，按需显式使用。 保留全部 8 个用户级 subagents：`planner`、`architect`、`code-reviewer`、`security-reviewer`、`database-reviewer`、`python-reviewer`、`tdd-guide`、`refactor-cleaner`。 ## Hooks 强度判断 当前 hooks 保持轻量提醒和 ask，不默认强阻断。建议： - `_archived/` 读取/搜索：已升级为 `failClosed` 强阻断。依据：项目规则明确禁止参考归档目录，误用概率高。 - 危险 git 命令：保持 ask，不建议 `failClosed`。依据：极少数维护场景可能需要，必须由用户明确确认。 - 生产库 MCP / 生产 DSN：只读调用走 ask；写入/DDL 类调用 deny。依据：测试库足够覆盖日常验证，生产库写操作风险高。 - demo-miniprogram：保持提醒，不建议强阻断。依据：它是标杆目录，偶尔仍可能需要按用户明确要求修改。 - DB schema 改动提醒：保持提醒。依据：是否需要文档同步要结合实际变更判断。 ## MCP 策略 延续既有用途： - `pg-etl-test`、`pg-app-test`：测试库，默认可用。 - `pg-etl`、`pg-app`：生产库，默认禁用或 ask，不自动执行。 - `weixin-devtools-mcp`：小程序调试和截图验证。 - `playwright` / `cursor-ide-browser`：Web 前端验证、截图、交互检查。 - `openapi`：后端 API 合同查询。 ## VSCode Insiders 迁移 - 用户数据目录：`C:\\Users\\Administrator\\AppData\\Roaming\\Code - Insiders\\User` - 扩展清单：`C:\\Users\\Administrator\\.vscode-insiders\\extensions\\extensions.json` - Cursor 扩展对照表：`docs/ai-env-history/vscode_insiders_extensions.csv` - 补装脚本：`tools/cursor/install_vscode_insiders_extensions.ps1` - 旧环境 disabled 状态已记录；Cursor 没有稳定 CLI 持久禁用接口，补装后需要在 Cursor UI 中按清单确认禁用状态。 ## 主题时间线 - 主题时间线：`docs/ai-env-history/topic_timeline.md` - 主题 CSV：`docs/ai-env-history/topic_timeline.csv` ## 验证 ```powershell .venv\\Scripts\\python.exe tools\\cursor\\migrate_ai_environment.py --check .venv\\Scripts\\python.exe scripts\\audit\\prescan.py --files "tools/cursor/migrate_ai_environment.py,docs/cursor_migration.md" ``` ## 回滚 迁移脚本会把覆盖前的 Cursor 用户资产和项目 `.cursor` 资产备份到： `C:\\Users\\Administrator\\.cursor\\backups\\neozqyy-cursor-migration\\` 如需回滚，按 manifest 中的源/目标路径恢复即可。 """ write_text(REPO_ROOT / "docs" / "cursor_migration.md", doc) def create_audit_record() -> None: now = dt.datetime.now(dt.timezone(dt.timedelta(hours=8))).strftime("%Y-%m-%d %H:%M:%S") record = f"""# 变更审计记录：Cursor AI 开发环境迁移 | 字段 | 值 | |------|-----| | 日期 | {now} | ## 操作摘要 本轮将 Claude Code 与 Codex 中已经沉淀的 AI 开发习惯迁移到 Cursor 原生结构。迁移目标是让 Cursor 继承中文沟通、前置调研、验证、审计、子代理审查、MCP 测试库优先和历史追溯习惯，同时避免把长规则和原始对话全文堆入常规上下文。 ## 变更文件 ### 新增/修改 - `tools/cursor/migrate_ai_environment.py`：可重复迁移脚本。 - `.cursor/rules/`：Cursor 项目级短规则。 - `.cursor/skills/`：审计、前置调研、文档同步和 spec 收尾流程技能。 - `.cursor/hooks.json`、`.cursor/hooks/ai_env_guard.py`：轻量提醒/保护 hooks。 - `docs/cursor_migration.md`：Cursor 迁移说明。 - `docs/ai-env-history/`：Claude/Codex/Cursor 原始对话精简索引和逐会话摘要。 - `C:\\Users\\Administrator\\.cursor\\skills`：用户级技能迁移。 - `C:\\Users\\Administrator\\.cursor\\agents`：用户级子代理迁移。 ## 数据库变更 无数据库 schema 变更。未执行 DDL，未修改迁移 SQL。 ## 风险与回滚 - 中：Cursor hooks 与 Claude hooks 事件字段不完全一致，本轮只启用轻量提醒和 ask，不启用强阻断。 - 中：历史摘要从原始 JSONL 自动分析，仍可能遗漏自然语言里的隐含决策；关键判断需回看原文和当前代码。 - 低：用户级 Cursor 资产覆盖前已备份，可从 manifest 恢复。 ## 验证 - 运行 `python tools/cursor/migrate_ai_environment.py --check` 校验生成结构。 - 运行 `python scripts/audit/prescan.py --files "<本轮迁移文件>"` 做审计预扫描。 - 解析 `.mcp.json`、`.cursor/hooks.json`、skill/subagent frontmatter。 ## 合规检查 - 文档同步：已新增 `docs/cursor_migration.md` 和 `docs/ai-env-history/README.md`。 - 数据库文档：不适用，本轮无 schema 变更。 - 审计记录：已落盘。 """ write_text(REPO_ROOT / "docs" / "audit" / "changes" / "2026-05-01__cursor_migration.md", record) def update_cursor_cli_config() -> None: path = CURSOR_HOME / "cli-config.json" if path.exists(): backup(path) try: data = json.loads(read_text(path)) except Exception: data = {} else: data = {} data.setdefault("display", {}) data["display"].setdefault("showStatusIndicators", True) data.setdefault("network", {}) data["network"].setdefault("useHttp1ForAgent", True) data.setdefault("permissions", {}) data["permissions"].setdefault("allow", []) data["permissions"].setdefault("deny", []) write_text(path, json.dumps(data, ensure_ascii=False, indent=2) + "\n") def collect_dev_environment_note() -> None: cursor_extensions = CURSOR_HOME / "extensions" / "extensions.json" vscode_user = HOME / "AppData" / "Roaming" / "Code - Insiders" / "User" vscode_settings = vscode_user / "settings.json" vscode_keybindings = vscode_user / "keybindings.json" vscode_extensions = HOME / ".vscode-insiders" / "extensions" / "extensions.json" sync_extensions = sorted((vscode_user / "sync" / "extensions").glob("*.json")) cursor_ids = read_extension_ids(cursor_extensions) vscode_ids = read_extension_ids(vscode_extensions) disabled_ids = read_disabled_synced_extension_ids(sync_extensions) missing_all = sorted(vscode_ids - cursor_ids) disabled_missing = sorted((disabled_ids & vscode_ids) - cursor_ids) install_script = [ "# 从 VSCode Insiders 迁移到 Cursor 的扩展补装脚本", "# 自动生成；安装旧 VSCode Insiders 中已安装但 Cursor 当前缺失的扩展。", "# 旧环境 disabled 状态见 docs/ai-env-history/vscode_insiders_extensions.csv；Cursor 无稳定 CLI 持久禁用接口，需在 UI 中按清单确认。", "$ErrorActionPreference = 'Continue'", "$cursor = 'cursor'", "$timeoutSec = 90", "$extensions = @(", ] for extension_id in missing_all: install_script.append(f' "{extension_id}"') install_script.extend( [ ")", "foreach ($ext in $extensions) {", " Write-Host \"Installing $ext ...\"", " $p = Start-Process -FilePath $cursor -ArgumentList @('--install-extension', $ext) -NoNewWindow -PassThru -Wait:$false", " if (-not $p.WaitForExit($timeoutSec * 1000)) {", " Write-Warning \"Timeout installing $ext; killing process and continuing.\"", " try { Stop-Process -Id $p.Id -Force } catch {}", " continue", " }", " Write-Host \"Finished $ext exit=$($p.ExitCode)\"", "}", "", ] ) if disabled_missing: install_script.append("") install_script.append("# 以下扩展在 VSCode Insiders 同步状态中为 disabled，安装后建议在 Cursor UI 中禁用：") for extension_id in disabled_missing: install_script.append(f"# disabled-before: {extension_id}") write_text(REPO_ROOT / "tools" / "cursor" / "install_vscode_insiders_extensions.ps1", "\n".join(install_script) + "\n") with (HISTORY_ROOT / "vscode_insiders_extensions.csv").open("w", encoding="utf-8-sig", newline="") as fh: writer = csv.DictWriter(fh, fieldnames=["extension_id", "in_vscode_insiders", "in_cursor", "disabled_in_sync", "recommended_action"]) writer.writeheader() for extension_id in sorted(vscode_ids | cursor_ids | disabled_ids): in_vscode = extension_id in vscode_ids in_cursor = extension_id in cursor_ids disabled = extension_id in disabled_ids if in_cursor: action = "已在 Cursor" elif disabled: action = "安装到 Cursor 后按旧环境状态禁用" elif in_vscode: action = "建议安装到 Cursor" else: action = "仅 Cursor 当前存在" writer.writerow( { "extension_id": extension_id, "in_vscode_insiders": in_vscode, "in_cursor": in_cursor, "disabled_in_sync": disabled, "recommended_action": action, } ) record("write", HISTORY_ROOT / "vscode_insiders_extensions.csv") note = f"""# Cursor / VSCode Insiders 开发环境盘点 ## Cursor - 用户设置：`{CURSOR_USER_SETTINGS}`，存在：{CURSOR_USER_SETTINGS.exists()} - 扩展清单：`{cursor_extensions}`，存在：{cursor_extensions.exists()} - 当前 Cursor 扩展数量：{len(cursor_ids)} ## 旧 VSCode Insiders - 根目录：`{VSCODE_INSIDERS_ROOT}`，存在：{VSCODE_INSIDERS_ROOT.exists()} - 用户数据目录：`{vscode_user}`，存在：{vscode_user.exists()} - 设置文件：`{vscode_settings}`，存在：{vscode_settings.exists()} - 键位文件：`{vscode_keybindings}`，存在：{vscode_keybindings.exists()} - 扩展清单：`{vscode_extensions}`，存在：{vscode_extensions.exists()} - VSCode Insiders 扩展数量：{len(vscode_ids)} - 同步状态中禁用扩展数量：{len(disabled_ids)} - Cursor 缺失且建议安装数量：{len(missing_all)} ## 扩展迁移 - 扩展对照表：`docs/ai-env-history/vscode_insiders_extensions.csv` - 安装脚本：`tools/cursor/install_vscode_insiders_extensions.ps1` - 旧环境 disabled 状态已写入扩展对照表；补装后需在 Cursor UI 中按表确认禁用状态。 - 未发现旧环境中已安装 Playwright、PostgreSQL/SQL、微信开发者工具扩展；这些能力当前主要由 MCP 和项目命令承接。 """ write_text(HISTORY_ROOT / "dev_environment_inventory.md", note) def read_extension_ids(path: Path) -> set[str]: if not path.exists(): return set() try: data = json.loads(read_text(path)) except Exception: return set() ids = set() if isinstance(data, list): for item in data: identifier = item.get("identifier") if isinstance(item, dict) else None if isinstance(identifier, dict) and isinstance(identifier.get("id"), str): ids.add(identifier["id"].lower()) return ids def read_disabled_synced_extension_ids(paths: list[Path]) -> set[str]: disabled: set[str] = set() for path in paths: try: wrapper = json.loads(read_text(path)) except Exception: continue contents: list[str] = [] if isinstance(wrapper.get("content"), str): contents.append(wrapper["content"]) sync_data = wrapper.get("syncData") if isinstance(sync_data, dict) and isinstance(sync_data.get("content"), str): contents.append(sync_data["content"]) for content in contents: try: extensions = json.loads(content) except Exception: continue if not isinstance(extensions, list): continue for item in extensions: if not isinstance(item, dict) or not item.get("disabled"): continue identifier = item.get("identifier") if isinstance(identifier, dict) and isinstance(identifier.get("id"), str): disabled.add(identifier["id"].lower()) return disabled def run_migration() -> None: for path in [ PROJECT_CURSOR, PROJECT_RULES, PROJECT_SKILLS, USER_CURSOR_SKILLS, USER_CURSOR_AGENTS, HISTORY_ROOT, CURSOR_USER_SETTINGS, ]: backup(path) create_project_rules() create_project_skills() create_project_hooks() create_user_skills() create_user_agents() update_cursor_cli_config() analyze_histories() collect_dev_environment_note() create_cursor_migration_doc() create_audit_record() write_text(MANIFEST_PATH, json.dumps(manifest, ensure_ascii=False, indent=2) + "\n") def check_generated() -> int: required = [ PROJECT_RULES / "neozqyy-core.mdc", PROJECT_SKILLS / "audit" / "SKILL.md", PROJECT_CURSOR / "hooks.json", USER_CURSOR_SKILLS / "neozqyy-cursor-migration" / "SKILL.md", USER_CURSOR_AGENTS / "planner.md", HISTORY_ROOT / "conversation_index.csv", HISTORY_ROOT / "file_impact_index.csv", REPO_ROOT / "docs" / "cursor_migration.md", REPO_ROOT / "docs" / "audit" / "changes" / "2026-05-01__cursor_migration.md", ] missing = [path for path in required if not path.exists()] for json_path in [PROJECT_CURSOR / "hooks.json", MANIFEST_PATH, CURSOR_HOME / "cli-config.json"]: if json_path.exists(): json.loads(read_text(json_path)) if missing: print("缺失文件:") for path in missing: print(f"- {path}") return 1 print("Cursor 迁移生成文件校验通过。") return 0 if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="迁移 NeoZQYY AI 开发环境到 Cursor。") parser.add_argument("--check", action="store_true", help="只校验生成结果。") args = parser.parse_args() if args.check: raise SystemExit(check_generated()) run_migration() raise SystemExit(check_generated())