#!/usr/bin/env python3 """ 个人项目与数字资产管理器 ========================= 使用SQLite本地数据库管理项目进度、技术决策、踩坑记录、知识卡和训练语料索引。 纯标准库实现，命令行交互，单文件可运行。 数据库文件: digital_assets.db（与脚本同目录） """ import argparse import json import os import sqlite3 import textwrap from datetime import datetime # ── 全局配置 ────────────────────────────────────────────── DB_FILENAME = "digital_assets.db" DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), DB_FILENAME) # ═══════════════════════════════════════════════════════════ # 数据库核心 # ═══════════════════════════════════════════════════════════ def get_conn(): """获取数据库连接。 启用外键约束(FOREIGN KEYS)确保级联删除正确工作， 启用WAL模式提升并发读写性能。row_factory=sqlite3.Row 让查询结果支持 dict-style 列名访问(row["col"])。 """ conn = sqlite3.connect(DB_PATH) conn.execute("PRAGMA foreign_keys = ON") conn.execute("PRAGMA journal_mode = WAL") conn.row_factory = sqlite3.Row # 让查询结果支持按列名访问 return conn def init_db(): """初始化数据库建表（幂等操作，已存在的表不重建）。 创建6张核心表: - projects: 项目信息 - decisions: 技术决策 - pitfalls: 踩坑记录 - knowledge_cards: 知识卡(项目文件/工作日志) - training_corpus: 训练语料索引 - import_history: 导入历史记录 同时创建7个索引加速常用查询(project_id/date/status/category/type)。 """ conn = get_conn() cursor = conn.cursor() # ── 项目表 ── cursor.execute(""" CREATE TABLE IF NOT EXISTS projects ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL UNIQUE, status TEXT NOT NULL DEFAULT 'active' CHECK(status IN ('active','paused','archived')), description TEXT DEFAULT '', key_milestone TEXT DEFAULT '', related_files TEXT DEFAULT '[]', -- JSON数组 created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); """) # ── 决策表 ── cursor.execute(""" CREATE TABLE IF NOT EXISTS decisions ( id INTEGER PRIMARY KEY AUTOINCREMENT, project_id INTEGER NOT NULL, date TEXT NOT NULL, title TEXT NOT NULL, content TEXT DEFAULT '', basis TEXT DEFAULT '', conclusion TEXT DEFAULT '', tags TEXT DEFAULT '', created_at TEXT NOT NULL, updated_at TEXT NOT NULL, FOREIGN KEY (project_id) REFERENCES projects(id) ON DELETE CASCADE ); """) # ── 踩坑记录表 ── cursor.execute(""" CREATE TABLE IF NOT EXISTS pitfalls ( id INTEGER PRIMARY KEY AUTOINCREMENT, project_id INTEGER NOT NULL, date TEXT NOT NULL, problem TEXT NOT NULL, root_cause TEXT DEFAULT '', solution TEXT DEFAULT '', tags TEXT DEFAULT '', created_at TEXT NOT NULL, updated_at TEXT NOT NULL, FOREIGN KEY (project_id) REFERENCES projects(id) ON DELETE CASCADE ); """) # ── 知识卡表 ── cursor.execute(""" CREATE TABLE IF NOT EXISTS knowledge_cards ( id INTEGER PRIMARY KEY AUTOINCREMENT, topic TEXT NOT NULL, category TEXT DEFAULT '', file_path TEXT DEFAULT '', summary TEXT DEFAULT '', version TEXT DEFAULT '1.0', created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); """) # ── 训练语料索引表 ── cursor.execute(""" CREATE TABLE IF NOT EXISTS training_corpus ( id INTEGER PRIMARY KEY AUTOINCREMENT, date TEXT NOT NULL, type TEXT NOT NULL CHECK(type IN ('qa','decision','pitfall')), tags TEXT DEFAULT '', file_path TEXT DEFAULT '', record_count INTEGER DEFAULT 0, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); """) # ── 导入历史表 ── cursor.execute(""" CREATE TABLE IF NOT EXISTS import_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, import_date TEXT NOT NULL, import_type TEXT NOT NULL, folder_path TEXT DEFAULT '', project_name TEXT DEFAULT '', file_count INTEGER DEFAULT 0, summary TEXT DEFAULT '', created_at TEXT NOT NULL ); """) # ── 索引：加速按外键、标签、日期、状态查询 ── cursor.execute( "CREATE INDEX IF NOT EXISTS idx_decisions_project ON decisions(project_id);" ) cursor.execute( "CREATE INDEX IF NOT EXISTS idx_pitfalls_project ON pitfalls(project_id);" ) cursor.execute( "CREATE INDEX IF NOT EXISTS idx_decisions_date ON decisions(date);" ) cursor.execute( "CREATE INDEX IF NOT EXISTS idx_pitfalls_date ON pitfalls(date);" ) cursor.execute( "CREATE INDEX IF NOT EXISTS idx_projects_status ON projects(status);" ) cursor.execute( "CREATE INDEX IF NOT EXISTS idx_knowledge_category ON knowledge_cards(category);" ) cursor.execute( "CREATE INDEX IF NOT EXISTS idx_corpus_type ON training_corpus(type);" ) conn.commit() conn.close() print(f"[OK] 数据库初始化完成: {DB_PATH}") # ═══════════════════════════════════════════════════════════ # 工具函数 # ═══════════════════════════════════════════════════════════ def now(): """返回当前时间字符串，ISO格式精确到秒。 格式: "YYYY-MM-DD HH:MM:SS" 所有数据库记录的 created_at/updated_at 统一使用此函数。 """ return datetime.now().strftime("%Y-%m-%d %H:%M:%S") def print_rows(rows, columns=None): """以表格形式打印查询结果。 Args: rows: sqlite3.Row 列表，每个 row 支持 dict-style 访问 columns: 要显示的列名列表，默认取第一行的 keys() 输出格式: 表头 | 分隔线 | N行数据 | 总条数 无结果时打印"(无结果)"。 """ if not rows: print("（无结果）") return if columns is None: columns = rows[0].keys() # 表头 header = " | ".join(columns) print(header) print("-" * len(header)) for row in rows: vals = [] for c in columns: v = row.get(c, "") vals.append(str(v) if v is not None else "") print(" | ".join(vals)) print(f"（共 {len(rows)} 条）") def resolve_project_id(conn, project_ref): """将名称或ID解析为 project_id。 Args: conn: 数据库连接 project_ref: 纯数字字符串→按ID查，否则→按项目名查 Returns: int: project_id, 找不到返回 None """ cursor = conn.cursor() if project_ref.isdigit(): cursor.execute("SELECT id FROM projects WHERE id = ?", (int(project_ref),)) else: cursor.execute("SELECT id FROM projects WHERE name = ?", (project_ref,)) row = cursor.fetchone() return row["id"] if row else None def timestamp_fields(): """返回 (created_at, updated_at) 的当前时间元组。 用于 INSERT 语句一次性获取两个时间戳字段的值。 """ t = now() return (t, t) def validate_date(date_str): """校验日期是否严格符合 YYYY-MM-DD 格式。 Args: date_str: 待校验字符串 Returns: bool: 合法且可解析则 True 用于所有 --date 参数的输入验证，防止非法日期写入数据库。 """ if date_str is None: return False try: datetime.strptime(date_str, "%Y-%m-%d") return True except ValueError: return False def check_db_initialized(): """检测数据库是否已初始化。 通过查询 sqlite_master 确认 projects 表存在。 未初始化时打印提示信息指导用户运行 init 命令。 Returns: bool: 已初始化返回 True """ try: conn = sqlite3.connect(DB_PATH) cursor = conn.cursor() cursor.execute( "SELECT name FROM sqlite_master WHERE type='table' AND name='projects'" ) exists = cursor.fetchone() is not None conn.close() if not exists: print( "[ERR] 数据库未初始化，请先运行: python digital_assets_manager.py init" ) return exists except Exception: print("[ERR] 数据库未初始化，请先运行: python digital_assets_manager.py init") return False # ═══════════════════════════════════════════════════════════ # 命令处理 # ═══════════════════════════════════════════════════════════ # ────────── 项目 (projects) ────────── def cmd_project_add(args): """添加新项目""" conn = get_conn() t = now() try: conn.execute( """INSERT INTO projects (name, status, description, key_milestone, related_files, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)""", ( args.name, args.status or "active", args.desc or "", args.milestone or "", json.dumps(args.files or [], ensure_ascii=False), t, t, ), ) conn.commit() print(f"[OK] 项目 '{args.name}' 已添加") except sqlite3.IntegrityError as e: print(f"[ERR] 添加失败: {e}") finally: conn.close() def cmd_project_list(args): """列出项目，支持按状态过滤""" conn = get_conn() if args.status: rows = conn.execute( "SELECT id, name, status, key_milestone, updated_at FROM projects WHERE status = ? ORDER BY updated_at DESC", (args.status,), ).fetchall() else: rows = conn.execute( "SELECT id, name, status, key_milestone, updated_at FROM projects ORDER BY updated_at DESC" ).fetchall() print_rows(rows) conn.close() def cmd_project_update(args): """更新项目字段""" conn = get_conn() pid = resolve_project_id(conn, args.project) if pid is None: print(f"[ERR] 未找到项目: {args.project}") conn.close() return # 构建动态UPDATE updates = {} if args.status: if args.status not in ("active", "paused", "archived"): print("[ERR] status 必须是 active / paused / archived") conn.close() return updates["status"] = args.status if args.desc is not None: updates["description"] = args.desc if args.milestone is not None: updates["key_milestone"] = args.milestone if args.files is not None: updates["related_files"] = json.dumps(args.files, ensure_ascii=False) if not updates: print("[WARN]️ 未指定要更新的字段") conn.close() return updates["updated_at"] = now() set_clause = ", ".join(f"{k} = ?" for k in updates) values = list(updates.values()) + [pid] conn.execute(f"UPDATE projects SET {set_clause} WHERE id = ?", values) conn.commit() print(f"[OK] 项目 '{args.project}' 已更新") conn.close() def cmd_project_delete(args): """删除项目（级联删除关联的决策和踩坑记录）""" conn = get_conn() pid = resolve_project_id(conn, args.project) if pid is None: print(f"[ERR] 未找到项目: {args.project}") conn.close() return if not args.force: confirm = ( input( f"[WARN] 确定删除项目 '{args.project}' 及其关联的所有决策/踩坑记录？(y/N): " ) .strip() .lower() ) if confirm != "y": print("已取消") conn.close() return conn.execute("DELETE FROM projects WHERE id = ?", (pid,)) conn.commit() print(f"[OK] 项目 '{args.project}' 已删除") conn.close() # ────────── 决策 (decisions) ────────── def cmd_decision_add(args): """添加决策记录""" if not validate_date(args.date): print("[ERR] 日期格式错误，需要 YYYY-MM-DD") return conn = get_conn() pid = resolve_project_id(conn, args.project) if pid is None: print(f"[ERR] 未找到项目: {args.project}") conn.close() return t = now() conn.execute( """INSERT INTO decisions (project_id, date, title, content, basis, conclusion, tags, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( pid, args.date, args.title, args.content or "", args.basis or "", args.conclusion or "", args.tags or "", t, t, ), ) conn.commit() print(f"[OK] 决策 '{args.title}' 已添加到项目 '{args.project}'") conn.close() def cmd_decision_list(args): """列出决策，支持按项目过滤""" conn = get_conn() if args.project: pid = resolve_project_id(conn, args.project) if pid is None: print(f"[ERR] 未找到项目: {args.project}") conn.close() return rows = conn.execute( """SELECT d.id, p.name AS project, d.date, d.title, d.tags, d.updated_at FROM decisions d JOIN projects p ON d.project_id = p.id WHERE d.project_id = ? ORDER BY d.date DESC""", (pid,), ).fetchall() else: rows = conn.execute( """SELECT d.id, p.name AS project, d.date, d.title, d.tags, d.updated_at FROM decisions d JOIN projects p ON d.project_id = p.id ORDER BY d.date DESC""" ).fetchall() print_rows(rows) conn.close() def cmd_decision_update(args): """更新决策 — 通过ID""" conn = get_conn() row = conn.execute("SELECT id FROM decisions WHERE id = ?", (args.id,)).fetchone() if not row: print(f"[ERR] 未找到决策 ID={args.id}") conn.close() return updates = {} if args.title is not None: updates["title"] = args.title if args.content is not None: updates["content"] = args.content if args.basis is not None: updates["basis"] = args.basis if args.conclusion is not None: updates["conclusion"] = args.conclusion if args.tags is not None: updates["tags"] = args.tags if args.date is not None: if not validate_date(args.date): print("[ERR] 日期格式错误，需要 YYYY-MM-DD") conn.close() return updates["date"] = args.date if not updates: print("[WARN] 未指定要更新的字段") conn.close() return updates["updated_at"] = now() set_clause = ", ".join(f"{k} = ?" for k in updates) values = list(updates.values()) + [args.id] conn.execute(f"UPDATE decisions SET {set_clause} WHERE id = ?", values) conn.commit() print(f"[OK] 决策 ID={args.id} 已更新") conn.close() def cmd_decision_delete(args): """删除决策""" conn = get_conn() conn.execute("DELETE FROM decisions WHERE id = ?", (args.id,)) conn.commit() print(f"[OK] 决策 ID={args.id} 已删除（如存在）") conn.close() def cmd_decision_show(args): """查看单条决策完整内容""" conn = get_conn() row = conn.execute( """SELECT d.*, p.name AS project_name FROM decisions d JOIN projects p ON d.project_id = p.id WHERE d.id = ?""", (args.id,), ).fetchone() if not row: print(f"[ERR] 未找到决策 ID={args.id}") else: print(f"ID: {row['id']}") print(f"项目: {row['project_name']}") print(f"日期: {row['date']}") print(f"标题: {row['title']}") print(f"内容: {row['content']}") print(f"依据: {row['basis']}") print(f"结论: {row['conclusion']}") print(f"标签: {row['tags']}") print(f"更新时间: {row['updated_at']}") conn.close() # ────────── 踩坑记录 (pitfalls) ────────── def cmd_pitfall_add(args): """添加踩坑记录""" if not validate_date(args.date): print("[ERR] 日期格式错误，需要 YYYY-MM-DD") return conn = get_conn() pid = resolve_project_id(conn, args.project) if pid is None: print(f"[ERR] 未找到项目: {args.project}") conn.close() return t = now() conn.execute( """INSERT INTO pitfalls (project_id, date, problem, root_cause, solution, tags, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", ( pid, args.date, args.problem, args.root_cause or "", args.solution or "", args.tags or "", t, t, ), ) conn.commit() print(f"[OK] 踩坑记录已添加到项目 '{args.project}'") conn.close() def cmd_pitfall_list(args): """列出踩坑记录""" conn = get_conn() if args.project: pid = resolve_project_id(conn, args.project) if pid is None: print(f"[ERR] 未找到项目: {args.project}") conn.close() return rows = conn.execute( """SELECT pf.id, p.name AS project, pf.date, pf.problem, pf.tags, pf.updated_at FROM pitfalls pf JOIN projects p ON pf.project_id = p.id WHERE pf.project_id = ? ORDER BY pf.date DESC""", (pid,), ).fetchall() else: rows = conn.execute( """SELECT pf.id, p.name AS project, pf.date, pf.problem, pf.tags, pf.updated_at FROM pitfalls pf JOIN projects p ON pf.project_id = p.id ORDER BY pf.date DESC""" ).fetchall() print_rows(rows) conn.close() def cmd_pitfall_update(args): """更新踩坑记录""" conn = get_conn() row = conn.execute("SELECT id FROM pitfalls WHERE id = ?", (args.id,)).fetchone() if not row: print(f"[ERR] 未找到踩坑记录 ID={args.id}") conn.close() return updates = {} if args.problem is not None: updates["problem"] = args.problem if args.root_cause is not None: updates["root_cause"] = args.root_cause if args.solution is not None: updates["solution"] = args.solution if args.tags is not None: updates["tags"] = args.tags if args.date is not None: if not validate_date(args.date): print("[ERR] 日期格式错误，需要 YYYY-MM-DD") conn.close() return updates["date"] = args.date if not updates: print("[WARN] 未指定要更新的字段") conn.close() return updates["updated_at"] = now() set_clause = ", ".join(f"{k} = ?" for k in updates) values = list(updates.values()) + [args.id] conn.execute(f"UPDATE pitfalls SET {set_clause} WHERE id = ?", values) conn.commit() print(f"[OK] 踩坑记录 ID={args.id} 已更新") conn.close() def cmd_pitfall_delete(args): """删除踩坑记录""" conn = get_conn() conn.execute("DELETE FROM pitfalls WHERE id = ?", (args.id,)) conn.commit() print(f"[OK] 踩坑记录 ID={args.id} 已删除（如存在）") conn.close() def cmd_pitfall_show(args): """查看单条踩坑记录完整内容""" conn = get_conn() row = conn.execute( """SELECT pf.*, p.name AS project_name FROM pitfalls pf JOIN projects p ON pf.project_id = p.id WHERE pf.id = ?""", (args.id,), ).fetchone() if not row: print(f"[ERR] 未找到踩坑记录 ID={args.id}") else: print(f"ID: {row['id']}") print(f"项目: {row['project_name']}") print(f"日期: {row['date']}") print(f"问题: {row['problem']}") print(f"根因: {row['root_cause']}") print(f"解决方案: {row['solution']}") print(f"标签: {row['tags']}") print(f"更新时间: {row['updated_at']}") conn.close() # ────────── 知识卡 (knowledge_cards) ────────── def cmd_knowledge_add(args): """添加知识卡""" conn = get_conn() t = now() conn.execute( """INSERT INTO knowledge_cards (topic, category, file_path, summary, version, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)""", ( args.topic, args.category or "", args.file or "", args.summary or "", args.version or "1.0", t, t, ), ) conn.commit() print(f"[OK] 知识卡 '{args.topic}' 已添加") conn.close() def cmd_knowledge_list(args): """列出知识卡，支持按分类过滤""" conn = get_conn() if args.category: rows = conn.execute( """SELECT id, topic, category, version, updated_at FROM knowledge_cards WHERE category = ? ORDER BY updated_at DESC""", (args.category,), ).fetchall() else: rows = conn.execute( "SELECT id, topic, category, version, updated_at FROM knowledge_cards ORDER BY updated_at DESC" ).fetchall() print_rows(rows) conn.close() def cmd_knowledge_update(args): """更新知识卡""" conn = get_conn() row = conn.execute( "SELECT id FROM knowledge_cards WHERE id = ?", (args.id,) ).fetchone() if not row: print(f"[ERR] 未找到知识卡 ID={args.id}") conn.close() return updates = {} if args.topic is not None: updates["topic"] = args.topic if args.category is not None: updates["category"] = args.category if args.file is not None: updates["file_path"] = args.file if args.summary is not None: updates["summary"] = args.summary if args.version is not None: updates["version"] = args.version if not updates: print("[WARN]️ 未指定要更新的字段") conn.close() return updates["updated_at"] = now() set_clause = ", ".join(f"{k} = ?" for k in updates) values = list(updates.values()) + [args.id] conn.execute(f"UPDATE knowledge_cards SET {set_clause} WHERE id = ?", values) conn.commit() print(f"[OK] 知识卡 ID={args.id} 已更新") conn.close() def cmd_knowledge_delete(args): """删除知识卡""" conn = get_conn() conn.execute("DELETE FROM knowledge_cards WHERE id = ?", (args.id,)) conn.commit() print(f"[OK] 知识卡 ID={args.id} 已删除（如存在）") conn.close() def cmd_knowledge_show(args): """查看单条知识卡完整内容""" conn = get_conn() row = conn.execute( "SELECT * FROM knowledge_cards WHERE id = ?", (args.id,) ).fetchone() if not row: print(f"[ERR] 未找到知识卡 ID={args.id}") else: print(f"ID: {row['id']}") print(f"主题: {row['topic']}") print(f"分类: {row['category']}") print(f"文件路径: {row['file_path']}") print(f"摘要: {row['summary']}") print(f"版本: {row['version']}") print(f"更新时间: {row['updated_at']}") conn.close() # ────────── 训练语料 (training_corpus) ────────── def cmd_corpus_add(args): """添加训练语料索引记录""" if not validate_date(args.date): print("[ERR] 日期格式错误，需要 YYYY-MM-DD") return conn = get_conn() if args.type not in ("qa", "decision", "pitfall"): print("[ERR] type 必须是 qa / decision / pitfall") conn.close() return t = now() conn.execute( """INSERT INTO training_corpus (date, type, tags, file_path, record_count, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)""", (args.date, args.type, args.tags or "", args.file or "", args.count or 0, t, t), ) conn.commit() print("[OK] 训练语料记录已添加") conn.close() def cmd_corpus_list(args): """列出训练语料""" conn = get_conn() query = """SELECT id, date, type, tags, file_path, record_count, updated_at FROM training_corpus WHERE 1=1""" params = [] if args.type: query += " AND type = ?" params.append(args.type) query += " ORDER BY date DESC" rows = conn.execute(query, params).fetchall() print_rows(rows) conn.close() def cmd_corpus_update(args): """更新训练语料记录""" conn = get_conn() row = conn.execute( "SELECT id FROM training_corpus WHERE id = ?", (args.id,) ).fetchone() if not row: print(f"[ERR] 未找到语料记录 ID={args.id}") conn.close() return updates = {} if args.type is not None: if args.type not in ("qa", "decision", "pitfall"): print("[ERR] type 必须是 qa / decision / pitfall") conn.close() return updates["type"] = args.type if args.tags is not None: updates["tags"] = args.tags if args.file is not None: updates["file_path"] = args.file if args.count is not None: updates["record_count"] = args.count if args.date is not None: if not validate_date(args.date): print("[ERR] 日期格式错误，需要 YYYY-MM-DD") conn.close() return updates["date"] = args.date if not updates: print("[WARN] 未指定要更新的字段") conn.close() return updates["updated_at"] = now() set_clause = ", ".join(f"{k} = ?" for k in updates) values = list(updates.values()) + [args.id] conn.execute(f"UPDATE training_corpus SET {set_clause} WHERE id = ?", values) conn.commit() print(f"[OK] 语料记录 ID={args.id} 已更新") conn.close() def cmd_corpus_delete(args): """删除训练语料记录""" conn = get_conn() conn.execute("DELETE FROM training_corpus WHERE id = ?", (args.id,)) conn.commit() print(f"[OK] 语料记录 ID={args.id} 已删除（如存在）") conn.close() def cmd_corpus_show(args): """查看单条训练语料索引完整内容""" conn = get_conn() row = conn.execute( "SELECT * FROM training_corpus WHERE id = ?", (args.id,) ).fetchone() if not row: print(f"[ERR] 未找到语料记录 ID={args.id}") else: print(f"ID: {row['id']}") print(f"日期: {row['date']}") print(f"类型: {row['type']}") print(f"标签: {row['tags']}") print(f"文件路径: {row['file_path']}") print(f"记录数: {row['record_count']}") print(f"创建时间: {row['created_at']}") print(f"更新时间: {row['updated_at']}") conn.close() def cmd_import(args): """批量导入文件夹，自动按子目录创建项目并分类文件。 核心流程: 1. 扫描目录(可选递归) → 子文件夹名→项目名 2. 文件类型判断: .jsonl(含Q&A)→语料库, 含日志关键词→工作日志, 其余→项目资产 3. 二进制文件只存元数据(路径+类型标签)，文本文件读全文(超10000字截断) 4. 自动创建不存在的项目，状态设为 active 5. 写入 import_history 记录 Args: args.dir: 要导入的目录路径 args.recursive: 是否递归子目录(每个子文件夹=一个项目) args.project: 手动指定项目名(覆盖自动检测) """ import_dir = os.path.abspath(args.dir) if not os.path.isdir(import_dir): print(f"[ERR] 目录不存在: {import_dir}") return recursive = getattr(args, "recursive", False) project_name = args.project if args.project else None conn = get_conn() cursor = conn.cursor() # 收集目录（os.walk 真递归，单次遍历整棵树，按所属顶级目录分配项目名） top_dirs = {} if recursive: for entry in os.listdir(import_dir): epath = os.path.join(import_dir, entry) if os.path.isdir(epath): top_dirs[os.path.abspath(epath)] = project_name or entry scan_groups = [("未归类", import_dir)] else: scan_groups = [(project_name or "未归类", import_dir)] # 文件类型 text_ext = ( ".md", ".txt", ".jsonl", ".py", ".c", ".h", ".cpp", ".hpp", ".json", ".csv", ".log", ".yaml", ".yml", ".ini", ".cfg", ".toml", ) binary_ext = ( ".pdf", ".docx", ".xlsx", ".pptx", ".png", ".jpg", ".jpeg", ".gif", ".bmp", ".svg", ".bin", ".hex", ".elf", ".uf2", ".sch", ".brd", ".kicad_pcb", ".kicad_sch", ".zip", ".rar", ".7z", ".gz", ) supported_ext = text_ext + binary_ext all_files = [] for pname, dpath in scan_groups: if not os.path.isdir(dpath): continue for root, dirs, files in os.walk(dpath): dirs.sort() if not recursive: dirs.clear() abs_root = os.path.abspath(root) owner = None if recursive and top_dirs: for td, td_name in top_dirs.items(): if abs_root == td or abs_root.startswith(td + os.sep): owner = td_name break fn_pname = owner or pname for fn in sorted(files): fp = os.path.join(root, fn) fn_low = fn.lower() if fn_low.endswith(supported_ext) and os.path.isfile(fp): is_bin = fn_low.endswith(binary_ext) is_jsonl = fn_low.endswith(".jsonl") all_files.append((fn_pname, root, fn, fp, is_bin, is_jsonl)) if not all_files: print("[WARN] 目录下未找到可导入的文件") conn.close() return # 确保项目存在 project_cache = {} for pname, _, _, _, _, _ in all_files: if pname in project_cache: continue cursor.execute("SELECT id FROM projects WHERE name=?", (pname,)) row = cursor.fetchone() if row: project_cache[pname] = row["id"] else: t = now() cursor.execute( "INSERT INTO projects (name,status,description,created_at,updated_at) VALUES(?,'active','封存导入',?,?)", (pname, t, t), ) conn.commit() project_cache[pname] = cursor.lastrowid print(f"[OK] 创建项目: {pname}") # 逐文件导入 stats = {"imported": 0, "skipped": 0, "projects": 0, "logs": 0, "corpus": 0} total = len(all_files) processed = set() for idx, (pname, dpath, filename, filepath, is_binary, is_jsonl) in enumerate( all_files, 1 ): if dpath not in processed: print(f"\n 扫描 [{pname}] {dpath}") processed.add(dpath) project_cache[pname] title = os.path.splitext(filename)[0] # JSONL -> 训练语料（一个文件一条记录，带去重） if is_jsonl: existing = conn.execute("SELECT id FROM training_corpus WHERE file_path=?", (filepath,)).fetchone() if existing: stats["skipped"] += 1 print(f" [{idx}/{total}] [SKIP] {filename}") continue with open(filepath, "r", encoding="utf-8", errors="replace") as fh: raw = fh.read() qa_n = dec_n = pf_n = other_n = 0 tags_set = set() dates = [] for line in raw.strip().split("\n"): line = line.strip() if not line: continue try: obj = json.loads(line) except json.JSONDecodeError: continue if "question" in obj and "answer" in obj: qa_n += 1 elif "decision" in obj or "conclusion" in obj: dec_n += 1 elif "problem" in obj or "solution" in obj: pf_n += 1 else: other_n += 1 _tags = obj.get("tags") if _tags: if isinstance(_tags, list): tags_set.update(str(t) for t in _tags) elif isinstance(_tags, str): tags_set.update( t.strip() for t in _tags.split(",") if t.strip() ) else: tags_set.add(str(_tags)) if obj.get("date"): dates.append(obj["date"]) total_valid = qa_n + dec_n + pf_n + other_n if total_valid == 0: stats["skipped"] += 1 print(f" [{idx}/{total}] [SKIP] {filename} (无有效行)") continue tags = ",".join(sorted(tags_set)[:5]) if tags_set else "" date = dates[0] if dates else now()[:10] # 按多数类型 if qa_n >= dec_n and qa_n >= pf_n and qa_n > 0: best_type = "qa" elif dec_n >= pf_n and dec_n > 0: best_type = "decision" elif pf_n > 0: best_type = "pitfall" else: best_type = "qa" t = now() conn.execute( "INSERT INTO training_corpus (date,type,tags,file_path,record_count,created_at,updated_at) VALUES(?,?,?,?,?,?,?)", (date, best_type, tags, filepath, total_valid, t, t), ) stats["imported"] += 1 stats["corpus"] += 1 parts = [f"QA {qa_n}"] if qa_n else [] if dec_n: parts.append(f"决策 {dec_n}") if pf_n: parts.append(f"踩坑 {pf_n}") if other_n: parts.append(f"其他 {other_n}") print( f" [{idx}/{total}] [OK] {filename} -> 语料库 ({total_valid}条, {', '.join(parts)})" ) continue # 工作日志判定（按分词匹配，避免 logo/catalog 含 log 子串被误判） stem = os.path.splitext(filename)[0].lower() _toks = stem.replace("-", " ").replace("_", " ").replace(".", " ").split() is_worklog = any(kw in stem for kw in ["日志", "工作记录", "周报", "日报"]) or any( t in ("log", "diary", "journal") for t in _toks ) category = "工作日志" if is_worklog else pname # 二进制文件 if is_binary: existing = conn.execute( "SELECT id FROM knowledge_cards WHERE topic=? AND file_path=?", (title, filepath), ).fetchone() if existing: stats["skipped"] += 1 print(f" [{idx}/{total}] [SKIP] {filename}") continue t = now() ext = os.path.splitext(filename)[1].lower() cat_map = { ".pdf": "PDF", ".docx": "Word", ".xlsx": "Excel", ".png": "图片", ".jpg": "图片", ".jpeg": "图片", ".gif": "图片", ".bmp": "图片", ".svg": "矢量图", ".bin": "固件", ".hex": "固件", ".elf": "固件", ".uf2": "固件", ".sch": "原理图", ".brd": "PCB", ".kicad_pcb": "KiCad", ".kicad_sch": "KiCad", ".zip": "压缩包", ".rar": "压缩包", ".7z": "压缩包", ".gz": "压缩包", } cn = cat_map.get(ext, "文档") file_mtime = datetime.fromtimestamp(os.path.getmtime(filepath)).strftime("%Y-%m-%d") conn.execute( "INSERT INTO knowledge_cards (topic,category,file_path,summary,version,created_at,updated_at) VALUES(?,?,?,?,?,?,?)", (title, category, filepath, f"[{cn} 文件] {filepath}", file_mtime, t, t), ) stats["imported"] += 1 if is_worklog: stats["logs"] += 1 else: stats["projects"] += 1 print( f" [{idx}/{total}] [OK] {filename} -> {'工作日志' if is_worklog else '项目资产'} ({cn})" ) continue # 文本文件 with open(filepath, "r", encoding="utf-8", errors="replace") as fh: raw = fh.read() existing = conn.execute( "SELECT id FROM knowledge_cards WHERE topic=? AND file_path=?", (title, filepath), ).fetchone() if existing: stats["skipped"] += 1 print(f" [{idx}/{total}] [SKIP] {filename}") continue t = now() file_mtime = datetime.fromtimestamp(os.path.getmtime(filepath)).strftime( "%Y-%m-%d" ) conn.execute( "INSERT INTO knowledge_cards (topic,category,file_path,summary,version,created_at,updated_at) VALUES(?,?,?,?,?,?,?)", (title, category, filepath, raw[:10000], file_mtime, t, t), ) conn.commit() stats["imported"] += 1 if is_worklog: stats["logs"] += 1 else: stats["projects"] += 1 print( f" [{idx}/{total}] [OK] {filename} -> {'工作日志' if is_worklog else '项目资产'}" ) conn.commit() conn.close() print() print("=" * 50) print(f" 导入完成: 成功 {stats['imported']} 条, 跳过 {stats['skipped']} 条") print( f" 项目资产: {stats['projects']} | 工作日志: {stats['logs']} | 训练语料: {stats['corpus']}" ) print(f" 涉及项目: {len(project_cache)} 个") print("=" * 50) # 写入导入历史（使 UI 导入历史面板可见） try: hconn = get_conn() hconn.execute( "INSERT INTO import_history (import_date, import_type, folder_path, project_name, file_count, summary, created_at) VALUES (?,?,?,?,?,?,?)", ( now()[:10], "CLI导入", import_dir, project_name or "", stats["imported"], f"成功{stats['imported']}条, 跳过{stats['skipped']}条", now(), ), ) hconn.commit() hconn.close() except Exception as e: print(f"[WARN] 导入历史写入失败: {e}") # ═══════════════════════════════════════════════════════════ # 搜索功能 # ═══════════════════════════════════════════════════════════ def highlight_keyword(text, keywords, context=30): """在文本中高亮关键词。 找到关键词在文本中最早出现的位置，截取前后各 context 个字符， 用【】包裹每个关键词。跨行不区分大小写。 Args: text: 原始文本 keywords: 关键词列表(AND逻辑) context: 前后截取字符数，默认30 Returns: str: 高亮后的文本片段，前后加 ... 标记截断 """ if not text or not keywords: return text[: context * 2] if text else "" import re best_pos = len(text) best_kw = keywords[0] # 找到最早出现的关键词位置 for kw in keywords: pos = text.lower().find(kw.lower()) if pos != -1 and pos < best_pos: best_pos = pos best_kw = kw if best_pos >= len(text): return text[: context * 2] if text else "" start = max(0, best_pos - context) end = min(len(text), best_pos + len(best_kw) + context) snippet = text[start:end] # 用【】包裹关键词（不区分大小写） for kw in keywords: snippet = re.sub(f"({re.escape(kw)})", r"【\1】", snippet, flags=re.IGNORECASE) return ("..." if start > 0 else "") + snippet + ("..." if end < len(text) else "") def cmd_search(args): """全局搜索，支持关键词AND逻辑和结果片段高亮。 搜索范围: decisions(title/content/basis/conclusion/tags) + pitfalls(problem/root_cause/solution/tags) + knowledge_cards(topic/summary/category) 关键词用空格分隔 → AND逻辑(每条记录必须同时包含所有关键词) 结果显示: 关键词前后30字上下文，用【】包裹关键词 支持叠加筛选: --project / --tag / --date-from / --date-to Args: args.keyword: 空格分隔的关键词(可选) args.project: 项目名模糊匹配(可选) args.tag: 标签模糊匹配(可选) args.date_from/date_to: 日期范围(可选) """ conn = get_conn() # 解析关键词（空格分隔，AND逻辑） keywords = [] if hasattr(args, "keyword") and args.keyword: keywords = [kw.strip() for kw in args.keyword.split() if kw.strip()] # 构建过滤条件 project_clause = "" project_params = [] if args.project: project_clause = "AND p.name LIKE ?" project_params = [f"%{args.project}%"] tag_clause = "" tag_params = [] if args.tag: tag_clause = "AND d.tags LIKE ?" tag_params = [f"%{args.tag}%"] date_clause = "" date_params = [] if args.date_from: date_clause += " AND d.date >= ?" date_params.append(args.date_from) if args.date_to: date_clause += " AND d.date <= ?" date_params.append(args.date_to) # ── 关键词搜索子句：在多个字段中模糊匹配，多个关键词AND逻辑 ── def build_kw_clause(fields, table_alias=""): """为指定字段列表构建关键词LIKE子句，返回(clause_sql, params_list)""" if not keywords: return "", [] prefix = f"{table_alias}." if table_alias else "" clauses = [] params = [] for kw in keywords: or_parts = " OR ".join([f"{prefix}{f} LIKE ?" for f in fields]) clauses.append(f"({or_parts})") params.extend([f"%{kw}%"] * len(fields)) return " AND " + " AND ".join(clauses), params kw_d_clause, kw_d_params = build_kw_clause( ["title", "content", "basis", "conclusion", "tags"], "d" ) kw_pf_clause, kw_pf_params = build_kw_clause( ["problem", "root_cause", "solution", "tags"], "pf" ) kw_kc_clause, kw_kc_params = build_kw_clause(["topic", "summary", "category"]) results = [] # 搜索决策 d_query = f"""SELECT 'decision' AS source, d.id, p.name AS project, d.date, d.title AS summary, d.tags, d.title AS t_title, d.content AS t_content, d.basis AS t_basis, d.conclusion AS t_conclusion FROM decisions d JOIN projects p ON d.project_id = p.id WHERE 1=1 {project_clause} {tag_clause} {date_clause} {kw_d_clause} ORDER BY d.date DESC LIMIT 100""" d_params = project_params + tag_params + date_params + kw_d_params results.extend([dict(r) for r in conn.execute(d_query, d_params).fetchall()]) # 搜索踩坑记录 d_tag_clause = tag_clause.replace("d.", "pf.") pf_query = f"""SELECT 'pitfall' AS source, pf.id, p.name AS project, pf.date, pf.problem AS summary, pf.tags, pf.problem AS t_problem, pf.root_cause AS t_root_cause, pf.solution AS t_solution FROM pitfalls pf JOIN projects p ON pf.project_id = p.id WHERE 1=1 {project_clause} {d_tag_clause} {date_clause} {kw_pf_clause} ORDER BY pf.date DESC LIMIT 100""" pf_params = project_params + tag_params + date_params + kw_pf_params results.extend([dict(r) for r in conn.execute(pf_query, pf_params).fetchall()]) # 搜索知识卡 kc_tag_clause = "" kc_tag_params = [] if args.tag: kc_tag_clause = "AND (category LIKE ? OR topic LIKE ?)" kc_tag_params = [f"%{args.tag}%", f"%{args.tag}%"] kc_query = f"""SELECT 'knowledge' AS source, id, topic AS label, updated_at AS date, summary, category, topic AS t_topic, summary AS t_summary FROM knowledge_cards WHERE 1=1 {kc_tag_clause} {kw_kc_clause} ORDER BY updated_at DESC LIMIT 100""" kc_params = kc_tag_params + kw_kc_params results.extend([dict(r) for r in conn.execute(kc_query, kc_params).fetchall()]) if results: print(f"\n[Search] 搜索结果（共 {len(results)} 条）:\n") for r in results: source = r["source"] if source == "knowledge": line1 = f"[{source}] ID={r['id']} | {r['label']} | {r['date']}" line2 = f" 摘要: {r['summary'][:80]}" print(line1) print(line2) else: line1 = f"[{source}] ID={r['id']} | {r['project']} | {r['date']} | {r['summary'][:60]}" print(line1) # 关键词高亮：在内容字段中查找并显示片段 if keywords: text_fields = { "decision": ["t_title", "t_content", "t_basis", "t_conclusion"], "pitfall": ["t_problem", "t_root_cause", "t_solution"], "knowledge": ["t_topic", "t_summary"], }.get(source, []) for tf in text_fields: val = r.get(tf, "") if val: snippet = highlight_keyword(val, keywords) print(f" [{tf.replace('t_', '')}]: {snippet}") print() else: print("[Search] 无匹配结果") conn.close() # ═══════════════════════════════════════════════════════════ # 导出功能 # ═══════════════════════════════════════════════════════════ def cmd_export(args): """将指定项目的全部数据导出为 Markdown 文件。 导出内容: 项目信息 + 技术决策 + 踩坑记录 + 知识卡 + 训练语料索引 输出路径: 脚本同目录下 {项目名}_export.md Args: args.project: 项目名称或ID """ conn = get_conn() pid = resolve_project_id(conn, args.project) if pid is None: print(f"[ERR] 未找到项目: {args.project}") conn.close() return # 获取项目基本信息 proj = conn.execute("SELECT * FROM projects WHERE id = ?", (pid,)).fetchone() # 获取该项目的决策 decisions = conn.execute( "SELECT * FROM decisions WHERE project_id = ? ORDER BY date DESC", (pid,) ).fetchall() # 获取该项目的踩坑记录 pitfalls = conn.execute( "SELECT * FROM pitfalls WHERE project_id = ? ORDER BY date DESC", (pid,) ).fetchall() # 组装Markdown内容 md = [] md.append(f"# 项目: {proj['name']}\n") md.append(f"- **状态**: {proj['status']}") md.append(f"- **描述**: {proj['description']}") md.append(f"- **关键里程碑**: {proj['key_milestone']}") md.append(f"- **关联文件**: {proj['related_files']}") md.append(f"- **创建时间**: {proj['created_at']}") md.append(f"- **最后更新**: {proj['updated_at']}") md.append("") md.append("---\n") md.append(f"## 技术决策（{len(decisions)} 条）\n") for i, d in enumerate(decisions, 1): md.append(f"### 决策 {i}: {d['title']}") md.append(f"- **日期**: {d['date']}") md.append(f"- **标签**: {d['tags']}") md.append(f"\n**内容**:\n\n{d['content']}\n") md.append(f"**依据**:\n\n{d['basis']}\n") md.append(f"**结论**:\n\n{d['conclusion']}\n") md.append("") md.append("---\n") md.append(f"## 踩坑记录（{len(pitfalls)} 条）\n") for i, pf in enumerate(pitfalls, 1): md.append(f"### 踩坑 {i}: {pf['problem'][:50]}") md.append(f"- **日期**: {pf['date']}") md.append(f"- **标签**: {pf['tags']}") md.append(f"\n**问题**:\n\n{pf['problem']}\n") md.append(f"**根因分析**:\n\n{pf['root_cause']}\n") md.append(f"**解决方案**:\n\n{pf['solution']}\n") md.append("") # ── 关联知识卡：按项目名模糊匹配 category，或者遍历全部知识卡 ── # 获取全部知识卡，显示与项目名可能相关的（category包含项目关键词，或输出全部由用户自行筛选） all_knowledge = conn.execute( "SELECT * FROM knowledge_cards ORDER BY updated_at DESC" ).fetchall() md.append("---\n") md.append(f"## 知识卡（{len(all_knowledge)} 条）\n") if all_knowledge: for i, kc in enumerate(all_knowledge, 1): md.append(f"### 知识卡 {i}: {kc['topic']}") md.append(f"- **分类**: {kc['category']}") md.append(f"- **版本**: {kc['version']}") md.append(f"- **关联文件**: {kc['file_path']}") md.append(f"- **更新时间**: {kc['updated_at']}") md.append(f"\n**摘要**:\n\n{kc['summary']}\n") md.append("") else: md.append("（无）\n") # ── 训练语料索引：按项目名模糊匹配 tags ── training_corpus_rows = conn.execute( "SELECT * FROM training_corpus ORDER BY date DESC" ).fetchall() md.append("---\n") md.append(f"## 训练语料索引（{len(training_corpus_rows)} 条）\n") if training_corpus_rows: for i, tc in enumerate(training_corpus_rows, 1): md.append(f"### 语料 {i}") md.append(f"- **日期**: {tc['date']}") md.append(f"- **类型**: {tc['type']}") md.append(f"- **标签**: {tc['tags']}") md.append(f"- **文件路径**: {tc['file_path']}") md.append(f"- **记录数**: {tc['record_count']}") md.append(f"- **更新时间**: {tc['updated_at']}") md.append("") else: md.append("（无）\n") md.append(f"> 导出时间: {now()}") # 写入文件 filename = f"{proj['name'].replace(' ', '_')}_export.md" output_path = os.path.join(os.path.dirname(DB_PATH), filename) with open(output_path, "w", encoding="utf-8") as f: f.write("\n".join(md)) print(f"[OK] 已导出到: {output_path}") conn.close() # ═══════════════════════════════════════════════════════════ # 统计功能 # ═══════════════════════════════════════════════════════════ def cmd_stats(args): """显示数据库统计信息概览。 统计内容: 项目数量+状态分布, 各项目决策/踩坑数, 全局总决策/总踩坑/总知识卡/总语料, 最近更新的10条记录。 """ conn = get_conn() print("=" * 60) print(" [STATS] 数据库统计") print("=" * 60) # 项目概览 projects = conn.execute( "SELECT name, status, updated_at FROM projects ORDER BY status, updated_at DESC" ).fetchall() print(f"\n[Projects] 项目概览（共 {len(projects)} 个）:\n") for p in projects: # 统计该项目下的决策和踩坑数 d_count = conn.execute( "SELECT COUNT(*) AS cnt FROM decisions WHERE project_id = (SELECT id FROM projects WHERE name = ?)", (p["name"],), ).fetchone()["cnt"] pf_count = conn.execute( "SELECT COUNT(*) AS cnt FROM pitfalls WHERE project_id = (SELECT id FROM projects WHERE name = ?)", (p["name"],), ).fetchone()["cnt"] status_icon = {"active": "[A]", "paused": "[P]", "archived": "[X]"}.get( p["status"], "[?]" ) print(f" {status_icon} {p['name']}") print( f" 状态: {p['status']} | 决策: {d_count} | 踩坑: {pf_count} | 更新: {p['updated_at']}" ) # 全局统计 total_decisions = conn.execute("SELECT COUNT(*) AS cnt FROM decisions").fetchone()[ "cnt" ] total_pitfalls = conn.execute("SELECT COUNT(*) AS cnt FROM pitfalls").fetchone()[ "cnt" ] total_knowledge = conn.execute( "SELECT COUNT(*) AS cnt FROM knowledge_cards" ).fetchone()["cnt"] total_corpus = conn.execute( "SELECT COUNT(*) AS cnt FROM training_corpus" ).fetchone()["cnt"] print("\n[Summary] 全局汇总:") print(f" 总决策数: {total_decisions}") print(f" 总踩坑记录: {total_pitfalls}") print(f" 总知识卡: {total_knowledge}") print(f" 总语料索引: {total_corpus}") # 最近更新 print("\n[Recent] 最近更新的10条记录:") recent = conn.execute(""" SELECT 'decision' AS kind, title AS label, updated_at FROM decisions UNION ALL SELECT 'pitfall', problem, updated_at FROM pitfalls UNION ALL SELECT 'knowledge', topic, updated_at FROM knowledge_cards ORDER BY updated_at DESC LIMIT 10 """).fetchall() for r in recent: print(f" [{r['kind']}] {r['label'][:50]} — {r['updated_at']}") conn.close() # ═══════════════════════════════════════════════════════════ # 命令行界面 # ═══════════════════════════════════════════════════════════ def build_parser(): """构建 argparse 命令行解析器，注册所有子命令和参数。 命令树: init | project {add,list,update,delete} | decision {add,list,update,delete,show} | pitfall {add,list,update,delete,show} | knowledge {add,list,update,delete,show} | corpus {add,list,update,delete,show} | search [--keyword] [--project] [--tag] [--date-*] | import --dir [--recursive] [--project] | export --project | stats Returns: argparse.ArgumentParser: 配置好的解析器 """ parser = argparse.ArgumentParser( prog="python digital_assets_manager.py", description="个人项目与数字资产管理器 — SQLite本地数据库 + CLI", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=textwrap.dedent("""\ 使用示例: # 初始化数据库 python digital_assets_manager.py init # 项目管理 python digital_assets_manager.py project add --name "电子宠物V1.0" --desc "ESP32驱动项目" python digital_assets_manager.py project list python digital_assets_manager.py project list --status active python digital_assets_manager.py project update "电子宠物V1.0" --status paused python digital_assets_manager.py project delete "电子宠物V1.0" # 决策管理 python digital_assets_manager.py decision add --project "电子宠物V1.0" --date 2026-07-01 --title "选择主控芯片" --content "对比了ESP32-S3和ESP32-C3" --basis "需要USB OTG和WiFi+BLE共存" --conclusion "选用ESP32-S3-PICO-1" --tags "芯片选型,ESP32" python digital_assets_manager.py decision list python digital_assets_manager.py decision list --project "电子宠物V1.0" python digital_assets_manager.py decision show 1 python digital_assets_manager.py decision update 1 --conclusion "改用ESP32-S3-WROOM-1" python digital_assets_manager.py decision delete 1 # 踩坑记录 python digital_assets_manager.py pitfall add --project "电子宠物V1.0" --date 2026-07-15 --problem "TFT屏幕花屏" --root_cause "SPI时钟过高且杜邦线过长" --solution "降频到40MHz，缩短接线并加磁环" --tags "硬件,SPI,TFT" python digital_assets_manager.py pitfall list python digital_assets_manager.py pitfall show 1 # 知识卡 python digital_assets_manager.py knowledge add --topic "ESP32-S3 GPIO12上拉问题" --category "ESP32" --summary "GPIO12为内部flash电压选择引脚，启动时必须为高电平" --file "docs/esp32_gpio12.md" python digital_assets_manager.py knowledge list --category ESP32 # 训练语料 python digital_assets_manager.py corpus add --date 2026-07-20 --type qa --tags "硬件调试" --file "corpus/qa_20260720.jsonl" --count 150 # 搜索 python digital_assets_manager.py search --tag "ESP32" python digital_assets_manager.py search --project "电子宠物" --date-from 2026-06-01 # 导出 python digital_assets_manager.py export --project "电子宠物V1.0" # 统计 python digital_assets_manager.py stats """), ) subparsers = parser.add_subparsers(dest="command", help="可用命令") # ── init ── subparsers.add_parser("init", help="初始化数据库（建表+索引）") # ── project ── p_project = subparsers.add_parser("project", help="项目管理") p_project_subs = p_project.add_subparsers(dest="action") p_add = p_project_subs.add_parser("add", help="添加项目") p_add.add_argument("--name", required=True, help="项目名称") p_add.add_argument( "--status", default="active", choices=["active", "paused", "archived"], help="状态", ) p_add.add_argument("--desc", default="", help="描述") p_add.add_argument("--milestone", default="", help="关键里程碑") p_add.add_argument("--files", nargs="*", default=[], help="关联文件路径列表") p_list = p_project_subs.add_parser("list", help="列出项目") p_list.add_argument( "--status", choices=["active", "paused", "archived"], help="按状态过滤" ) p_update = p_project_subs.add_parser("update", help="更新项目") p_update.add_argument("project", help="项目名称或ID") p_update.add_argument( "--status", choices=["active", "paused", "archived"], help="新状态" ) p_update.add_argument("--desc", help="新描述") p_update.add_argument("--milestone", help="新里程碑") p_update.add_argument("--files", nargs="*", help="新关联文件列表") p_delete = p_project_subs.add_parser("delete", help="删除项目") p_delete.add_argument("project", help="项目名称或ID") p_delete.add_argument("--force", action="store_true", help="跳过确认直接删除") # ── decision ── p_dec = subparsers.add_parser("decision", help="决策管理") p_dec_subs = p_dec.add_subparsers(dest="action") p_da = p_dec_subs.add_parser("add", help="添加决策") p_da.add_argument("--project", required=True, help="项目名称或ID") p_da.add_argument("--date", required=True, help="决策日期 (YYYY-MM-DD)") p_da.add_argument("--title", required=True, help="决策标题") p_da.add_argument("--content", default="", help="决策内容") p_da.add_argument("--basis", default="", help="决策依据") p_da.add_argument("--conclusion", default="", help="结论") p_da.add_argument("--tags", default="", help="标签（逗号分隔）") p_dl = p_dec_subs.add_parser("list", help="列出决策") p_dl.add_argument("--project", help="按项目过滤") p_du = p_dec_subs.add_parser("update", help="更新决策") p_du.add_argument("id", type=int, help="决策ID") p_du.add_argument("--date", help="新日期") p_du.add_argument("--title", help="新标题") p_du.add_argument("--content", help="新内容") p_du.add_argument("--basis", help="新依据") p_du.add_argument("--conclusion", help="新结论") p_du.add_argument("--tags", help="新标签") p_dd = p_dec_subs.add_parser("delete", help="删除决策") p_dd.add_argument("id", type=int, help="决策ID") p_ds = p_dec_subs.add_parser("show", help="查看决策详情") p_ds.add_argument("id", type=int, help="决策ID") # ── pitfall ── p_pf = subparsers.add_parser("pitfall", help="踩坑记录管理") p_pf_subs = p_pf.add_subparsers(dest="action") p_pa = p_pf_subs.add_parser("add", help="添加踩坑记录") p_pa.add_argument("--project", required=True, help="项目名称或ID") p_pa.add_argument("--date", required=True, help="日期 (YYYY-MM-DD)") p_pa.add_argument("--problem", required=True, help="问题描述") p_pa.add_argument("--root_cause", default="", help="根因分析") p_pa.add_argument("--solution", default="", help="解决方案") p_pa.add_argument("--tags", default="", help="标签") p_pl = p_pf_subs.add_parser("list", help="列出踩坑记录") p_pl.add_argument("--project", help="按项目过滤") p_pu = p_pf_subs.add_parser("update", help="更新踩坑记录") p_pu.add_argument("id", type=int, help="记录ID") p_pu.add_argument("--date", help="新日期") p_pu.add_argument("--problem", help="新问题描述") p_pu.add_argument("--root_cause", help="新根因分析") p_pu.add_argument("--solution", help="新解决方案") p_pu.add_argument("--tags", help="新标签") p_pd = p_pf_subs.add_parser("delete", help="删除踩坑记录") p_pd.add_argument("id", type=int, help="记录ID") p_ps = p_pf_subs.add_parser("show", help="查看踩坑记录详情") p_ps.add_argument("id", type=int, help="记录ID") # ── knowledge ── p_kc = subparsers.add_parser("knowledge", help="知识卡管理") p_kc_subs = p_kc.add_subparsers(dest="action") p_ka = p_kc_subs.add_parser("add", help="添加知识卡") p_ka.add_argument("--topic", required=True, help="主题") p_ka.add_argument("--category", default="", help="分类") p_ka.add_argument("--file", default="", help="关联文件路径") p_ka.add_argument("--summary", default="", help="摘要") p_ka.add_argument("--version", default="1.0", help="版本号") p_kl = p_kc_subs.add_parser("list", help="列出知识卡") p_kl.add_argument("--category", help="按分类过滤") p_ku = p_kc_subs.add_parser("update", help="更新知识卡") p_ku.add_argument("id", type=int, help="知识卡ID") p_ku.add_argument("--topic", help="新主题") p_ku.add_argument("--category", help="新分类") p_ku.add_argument("--file", help="新文件路径") p_ku.add_argument("--summary", help="新摘要") p_ku.add_argument("--version", help="新版本号") p_kd = p_kc_subs.add_parser("delete", help="删除知识卡") p_kd.add_argument("id", type=int, help="知识卡ID") p_ks = p_kc_subs.add_parser("show", help="查看知识卡详情") p_ks.add_argument("id", type=int, help="知识卡ID") # ── corpus ── p_cp = subparsers.add_parser("corpus", help="训练语料管理") p_cp_subs = p_cp.add_subparsers(dest="action") p_ca = p_cp_subs.add_parser("add", help="添加语料索引") p_ca.add_argument("--date", required=True, help="日期 (YYYY-MM-DD)") p_ca.add_argument( "--type", required=True, choices=["qa", "decision", "pitfall"], help="语料类型" ) p_ca.add_argument("--tags", default="", help="标签") p_ca.add_argument("--file", default="", help="JSONL文件路径") p_ca.add_argument("--count", type=int, default=0, help="记录数") p_cl = p_cp_subs.add_parser("list", help="列出语料索引") p_cl.add_argument( "--type", choices=["qa", "decision", "pitfall"], help="按类型过滤" ) p_cu = p_cp_subs.add_parser("update", help="更新语料索引") p_cu.add_argument("id", type=int, help="记录ID") p_cu.add_argument("--date", help="新日期") p_cu.add_argument("--type", choices=["qa", "decision", "pitfall"], help="新类型") p_cu.add_argument("--tags", help="新标签") p_cu.add_argument("--file", help="新文件路径") p_cu.add_argument("--count", type=int, help="新记录数") p_cd = p_cp_subs.add_parser("delete", help="删除语料索引") p_cd.add_argument("id", type=int, help="记录ID") p_cs = p_cp_subs.add_parser("show", help="查看语料索引详情") p_cs.add_argument("id", type=int, help="记录ID") # ── search ── p_search = subparsers.add_parser("search", help="全局搜索") p_search.add_argument("--keyword", help="全文搜索关键词（空格分隔，AND逻辑）") p_search.add_argument("--project", help="项目名（模糊匹配）") p_search.add_argument("--tag", help="标签（模糊匹配）") p_search.add_argument("--date-from", help="起始日期 (YYYY-MM-DD)") p_search.add_argument("--date-to", help="截止日期 (YYYY-MM-DD)") # ── import ── p_import = subparsers.add_parser("import", help="批量导入文件") p_import.add_argument("--dir", required=True, help="扫描目录路径") p_import.add_argument( "--project", help="指定项目名（覆盖自动检测，全部归入该项目）" ) p_import.add_argument( "--recursive", action="store_true", help="递归扫描子文件夹，每个子文件夹作为一个项目", ) # ── export ── p_export = subparsers.add_parser("export", help="导出项目为Markdown") p_export.add_argument("--project", required=True, help="项目名称或ID") # ── stats ── subparsers.add_parser("stats", help="显示统计信息") return parser def main(): parser = build_parser() args = parser.parse_args() if args.command is None: parser.print_help() return # 除了 init 之外，所有数据库操作前检查是否已初始化 if args.command != "init" and not check_db_initialized(): return # 路由到对应处理函数 if args.command == "init": init_db() elif args.command == "project": if args.action == "add": cmd_project_add(args) elif args.action == "list": cmd_project_list(args) elif args.action == "update": cmd_project_update(args) elif args.action == "delete": cmd_project_delete(args) else: print("[ERR] 未知 project 子命令，可用: add / list / update / delete") elif args.command == "decision": if args.action == "add": cmd_decision_add(args) elif args.action == "list": cmd_decision_list(args) elif args.action == "update": cmd_decision_update(args) elif args.action == "delete": cmd_decision_delete(args) elif args.action == "show": cmd_decision_show(args) else: print( "[ERR] 未知 decision 子命令，可用: add / list / update / delete / show" ) elif args.command == "pitfall": if args.action == "add": cmd_pitfall_add(args) elif args.action == "list": cmd_pitfall_list(args) elif args.action == "update": cmd_pitfall_update(args) elif args.action == "delete": cmd_pitfall_delete(args) elif args.action == "show": cmd_pitfall_show(args) else: print( "[ERR] 未知 pitfall 子命令，可用: add / list / update / delete / show" ) elif args.command == "knowledge": if args.action == "add": cmd_knowledge_add(args) elif args.action == "list": cmd_knowledge_list(args) elif args.action == "update": cmd_knowledge_update(args) elif args.action == "delete": cmd_knowledge_delete(args) elif args.action == "show": cmd_knowledge_show(args) else: print( "[ERR] 未知 knowledge 子命令，可用: add / list / update / delete / show" ) elif args.command == "corpus": if args.action == "add": cmd_corpus_add(args) elif args.action == "list": cmd_corpus_list(args) elif args.action == "update": cmd_corpus_update(args) elif args.action == "delete": cmd_corpus_delete(args) elif args.action == "show": cmd_corpus_show(args) else: print("[ERR] 未知 corpus 子命令，可用: add / list / update / delete / show") elif args.command == "search": cmd_search(args) elif args.command == "import": cmd_import(args) elif args.command == "export": cmd_export(args) elif args.command == "stats": cmd_stats(args) else: print(f"[ERR] 未知命令: {args.command}") parser.print_help() if __name__ == "__main__": main()