/** * GraphStore — 本地知识图谱存储层 * * 使用 Node.js 内置 node:sqlite(Node 22.5+),无需额外依赖。 * * 节点类型:Person | Organization | Project | Event | Asset * 边类型: KNOWS | BELONGS_TO | PARTICIPATED | CONTAINS | MENTIONED_IN | ASSOCIATED */ const { DatabaseSync } = require('node:sqlite'); const { randomUUID } = require('crypto'); const path = require('path'); const fs = require('fs'); const DB_PATH = path.resolve(__dirname, 'lemon_crm.db'); const SCHEMA = ` CREATE TABLE IF NOT EXISTS nodes ( id TEXT PRIMARY KEY, type TEXT NOT NULL, name TEXT, data TEXT NOT NULL DEFAULT '{}', analysis_status TEXT NOT NULL DEFAULT 'none', created_at INTEGER NOT NULL DEFAULT (unixepoch()), updated_at INTEGER NOT NULL DEFAULT (unixepoch()) ); CREATE INDEX IF NOT EXISTS idx_nodes_type ON nodes(type); CREATE INDEX IF NOT EXISTS idx_nodes_name ON nodes(name COLLATE NOCASE); CREATE TABLE IF NOT EXISTS edges ( id TEXT PRIMARY KEY, from_id TEXT NOT NULL, to_id TEXT NOT NULL, relation TEXT NOT NULL, data TEXT NOT NULL DEFAULT '{}', created_at INTEGER NOT NULL DEFAULT (unixepoch()), FOREIGN KEY (from_id) REFERENCES nodes(id), FOREIGN KEY (to_id) REFERENCES nodes(id) ); CREATE INDEX IF NOT EXISTS idx_edges_from ON edges(from_id); CREATE INDEX IF NOT EXISTS idx_edges_to ON edges(to_id); CREATE INDEX IF NOT EXISTS idx_edges_relation ON edges(relation); `; class GraphStore { constructor(dbPath = DB_PATH) { fs.mkdirSync(path.dirname(dbPath), { recursive: true }); this.db = new DatabaseSync(dbPath); this.db.exec(SCHEMA); this._migrate(); } _migrate() { // 加 name_sort 列(存全拼,用于拼音排序);幂等 try { this.db.exec('ALTER TABLE nodes ADD COLUMN name_sort TEXT'); } catch (_) { /* 列已存在,忽略 */ } try { this.db.exec('CREATE INDEX IF NOT EXISTS idx_nodes_name_sort ON nodes(name_sort COLLATE NOCASE)'); } catch (_) {} } // ─── 节点操作 ─────────────────────────────────────────────── /** * 创建或更新节点。id 存在时执行 upsert。 * @param {string} type - Person | Organization | Project | Event | Asset * @param {object} data - 节点属性 * @returns {object} 节点对象 */ createNode(type, data) { const id = data.id || `${type.toLowerCase().slice(0, 3)}-${randomUUID().slice(0, 8)}`; const name = data.name || data.title || data.summary || null; // name_sort: 用全拼(quanPin)作为排序键,无则退回 name const nameSort = data.quanPin || name || null; const payload = JSON.stringify({ ...data, id }); this.db.prepare(` INSERT INTO nodes (id, type, name, name_sort, data, analysis_status) VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET name = excluded.name, name_sort = CASE WHEN excluded.name_sort IS NOT NULL THEN excluded.name_sort ELSE name_sort END, data = excluded.data, analysis_status = CASE WHEN excluded.analysis_status != 'none' THEN excluded.analysis_status ELSE analysis_status END, updated_at = unixepoch() `).run(id, type, name, nameSort, payload, data.analysisStatus || 'none'); return this.getNode(id); } /** * 按 id 查询节点(含所有关联边和邻居节点概要) */ getNode(id) { const row = this.db.prepare('SELECT * FROM nodes WHERE id = ?').get(id); if (!row) return null; return this._formatNode(row); } /** * 按名称模糊查询节点(主要用于 Person 查找) */ findByName(name, type = null) { const q = `%${name}%`; if (type) { const rows = this.db.prepare( 'SELECT * FROM nodes WHERE type = ? AND name LIKE ? COLLATE NOCASE LIMIT 10' ).all(type, q); return rows.map(r => this._formatNode(r)); } const rows = this.db.prepare( 'SELECT * FROM nodes WHERE name LIKE ? COLLATE NOCASE LIMIT 10' ).all(q); return rows.map(r => this._formatNode(r)); } /** * 按类型列出节点(支持分页) */ listByType(type, limit = 100, offset = 0) { const rows = this.db.prepare( 'SELECT * FROM nodes WHERE type = ? ORDER BY updated_at DESC LIMIT ? OFFSET ?' ).all(type, limit, offset); return rows.map(r => this._formatNode(r)); } /** * 按指定排序方式列出节点 * @param {string} type - 节点类型 * @param {string} sort - 'name' | 'count' | 'lastChat' * @param {string} order - 'asc' | 'desc' * @param {number} limit * @param {number} offset */ listSorted(type, sort = 'name', order = 'asc', limit = 100, offset = 0) { const dir = order === 'desc' ? 'DESC' : 'ASC'; if (sort === 'count') { // 按关联 Event 节点数量排序 const rows = this.db.prepare(` SELECT n.*, COUNT(e.id) AS evt_count FROM nodes n LEFT JOIN edges e ON (e.from_id = n.id OR e.to_id = n.id) LEFT JOIN nodes ne ON ne.id = CASE WHEN e.from_id = n.id THEN e.to_id WHEN e.to_id = n.id THEN e.from_id END AND ne.type = 'Event' WHERE n.type = ? GROUP BY n.id ORDER BY evt_count ${dir}, n.name_sort COLLATE NOCASE ASC LIMIT ? OFFSET ? `).all(type, limit, offset); return rows.map(r => this._formatNode(r)); } if (sort === 'lastChat') { // 按最近一次 Event 的 created_at 排序 const rows = this.db.prepare(` SELECT n.*, MAX(ne.created_at) AS last_event FROM nodes n LEFT JOIN edges e ON (e.from_id = n.id OR e.to_id = n.id) LEFT JOIN nodes ne ON ne.id = CASE WHEN e.from_id = n.id THEN e.to_id WHEN e.to_id = n.id THEN e.from_id END AND ne.type = 'Event' WHERE n.type = ? GROUP BY n.id ORDER BY last_event ${dir} NULLS LAST, n.name_sort COLLATE NOCASE ASC LIMIT ? OFFSET ? `).all(type, limit, offset); return rows.map(r => this._formatNode(r)); } // default: sort by name_sort (pinyin / name) const rows = this.db.prepare(` SELECT * FROM nodes WHERE type = ? ORDER BY COALESCE(name_sort, name) COLLATE NOCASE ${dir} LIMIT ? OFFSET ? `).all(type, limit, offset); return rows.map(r => this._formatNode(r)); } /** * 返回某类型节点总数 */ countByType(type) { const row = this.db.prepare('SELECT count(*) as c FROM nodes WHERE type = ?').get(type); return row ? row.c : 0; } /** * 全文搜索:在 name + data JSON 中搜索关键词(支持分页) */ search(query, limit = 20, offset = 0) { const q = `%${query}%`; const rows = this.db.prepare(` SELECT * FROM nodes WHERE name LIKE ? COLLATE NOCASE OR data LIKE ? ORDER BY updated_at DESC LIMIT ? OFFSET ? `).all(q, q, limit, offset); return rows.map(r => this._formatNode(r)); } /** * 搜索结果总数 */ countSearch(query) { const q = `%${query}%`; const row = this.db.prepare(` SELECT count(*) as c FROM nodes WHERE name LIKE ? COLLATE NOCASE OR data LIKE ? `).get(q, q); return row ? row.c : 0; } /** * 更新节点的 analysisStatus */ setAnalysisStatus(id, status) { this.db.prepare( 'UPDATE nodes SET analysis_status = ?, updated_at = unixepoch() WHERE id = ?' ).run(status, id); } // ─── 边操作 ───────────────────────────────────────────────── /** * 创建边(关系)。同一 from/to/relation 组合只保留一条。 */ createEdge(fromId, toId, relation, data = {}) { const existing = this.db.prepare( 'SELECT id FROM edges WHERE from_id = ? AND to_id = ? AND relation = ?' ).get(fromId, toId, relation); if (existing) { this.db.prepare( 'UPDATE edges SET data = ? WHERE id = ?' ).run(JSON.stringify(data), existing.id); return { id: existing.id, fromId, toId, relation, data }; } const id = `edg-${randomUUID().slice(0, 8)}`; this.db.prepare( 'INSERT INTO edges (id, from_id, to_id, relation, data) VALUES (?, ?, ?, ?, ?)' ).run(id, fromId, toId, relation, JSON.stringify(data)); return { id, fromId, toId, relation, data }; } /** * 获取节点的所有边(出边 + 入边),含邻居节点概要 */ getEdges(nodeId) { const outgoing = this.db.prepare(` SELECT e.*, n.type as neighbor_type, n.name as neighbor_name FROM edges e JOIN nodes n ON n.id = e.to_id WHERE e.from_id = ? `).all(nodeId); const incoming = this.db.prepare(` SELECT e.*, n.type as neighbor_type, n.name as neighbor_name FROM edges e JOIN nodes n ON n.id = e.from_id WHERE e.to_id = ? `).all(nodeId); return { outgoing: outgoing.map(e => ({ ...e, data: JSON.parse(e.data) })), incoming: incoming.map(e => ({ ...e, data: JSON.parse(e.data) })), }; } // ─── 图算法 ───────────────────────────────────────────────── /** * BFS 寻找两节点之间的最短关系路径 * @returns {Array|null} 路径步骤数组,或 null(不可达) */ getRelationshipPath(fromId, toId, maxDepth = 6) { if (fromId === toId) return []; const visited = new Set([fromId]); const queue = [{ id: fromId, path: [] }]; while (queue.length > 0) { const { id: currentId, path } = queue.shift(); if (path.length >= maxDepth) continue; const neighbors = this.db.prepare(` SELECT e.from_id, e.to_id, e.relation, e.data, n.name as neighbor_name, n.type as neighbor_type FROM edges e JOIN nodes n ON n.id = CASE WHEN e.from_id = ? THEN e.to_id ELSE e.from_id END WHERE e.from_id = ? OR e.to_id = ? `).all(currentId, currentId, currentId); for (const edge of neighbors) { const nextId = edge.from_id === currentId ? edge.to_id : edge.from_id; if (visited.has(nextId)) continue; const step = { from: currentId, to: nextId, neighborName: edge.neighbor_name, neighborType: edge.neighbor_type, relation: edge.relation, direction: edge.from_id === currentId ? 'outgoing' : 'incoming', data: JSON.parse(edge.data), }; const newPath = [...path, step]; if (nextId === toId) return newPath; visited.add(nextId); queue.push({ id: nextId, path: newPath }); } } return null; } // ─── AI 分析算法 ───────────────────────────────────────────── /** * 简单标签传播社区发现(Label Propagation Algorithm) * * 迭代逻辑:每个 Person 节点采用邻居中出现最多的标签,直到稳定或达到 maxIterations。 * * @param {number} maxIterations - 最大迭代次数(默认 10) * @returns {{communities: Array<{id: number, nodes: Array<{id:string, name:string}>}>}} */ detectCommunities(maxIterations = 10) { // 1. 加载所有 Person 节点 const personRows = this.db.prepare( "SELECT id, name FROM nodes WHERE type = 'Person'" ).all(); if (personRows.length === 0) { return { communities: [] }; } // 2. 初始化 label:每个节点的 label = 自己的 id const labels = new Map(); for (const p of personRows) { labels.set(p.id, p.id); } // 构建邻接表(Person 节点之间的连接,通过任意边) const adjMap = new Map(); for (const p of personRows) { adjMap.set(p.id, new Set()); } // 查所有边,仅保留两端都是 Person 的 const allEdges = this.db.prepare('SELECT from_id, to_id FROM edges').all(); const personIds = new Set(personRows.map(p => p.id)); for (const e of allEdges) { if (personIds.has(e.from_id) && personIds.has(e.to_id)) { adjMap.get(e.from_id).add(e.to_id); adjMap.get(e.to_id).add(e.from_id); } } // 3. 迭代:每个节点采用邻居中最多的 label for (let iter = 0; iter < maxIterations; iter++) { let changed = false; // 随机打乱顺序(用 sort + random 近似) const order = [...personRows].sort(() => Math.random() - 0.5); for (const p of order) { const neighbors = adjMap.get(p.id); if (!neighbors || neighbors.size === 0) continue; // 统计邻居 label 频率 const freq = new Map(); for (const nid of neighbors) { const lbl = labels.get(nid); if (lbl !== undefined) { freq.set(lbl, (freq.get(lbl) || 0) + 1); } } if (freq.size === 0) continue; // 取频率最高的 label(相等时取字典序最小,保证确定性) let bestLabel = labels.get(p.id); let bestCount = 0; for (const [lbl, cnt] of freq) { if (cnt > bestCount || (cnt === bestCount && lbl < bestLabel)) { bestLabel = lbl; bestCount = cnt; } } if (bestLabel !== labels.get(p.id)) { labels.set(p.id, bestLabel); changed = true; } } if (!changed) break; } // 4. 按 label 分组,生成社区列表 const communityMap = new Map(); const nameById = new Map(personRows.map(p => [p.id, p.name])); for (const [nodeId, lbl] of labels) { if (!communityMap.has(lbl)) { communityMap.set(lbl, []); } communityMap.get(lbl).push({ id: nodeId, name: nameById.get(nodeId) || nodeId, }); } const communities = []; let cIdx = 1; for (const [, nodes] of communityMap) { communities.push({ id: cIdx++, nodes }); } // 按社区大小降序排序 communities.sort((a, b) => b.nodes.length - a.nodes.length); return { communities }; } /** * 返回超过 N 天未互动的重要联系人(analysisStatus >= partial) * * "互动"定义:该联系人通过 PARTICIPATED 或 ASSOCIATED 边关联的最新 Event 节点的 date 字段。 * * @param {number} daysSince - 超过多少天未互动(默认 30) * @param {number} limit - 最多返回多少个(默认 20) * @returns {Array<{person: object, lastInteraction: string|null, daysSince: number, suggestion: string}>} */ getFollowUpCandidates(daysSince = 30, limit = 20) { // 查所有 analysisStatus 为 partial 或 complete 的 Person const persons = this.db.prepare(` SELECT * FROM nodes WHERE type = 'Person' AND analysis_status IN ('partial', 'complete') ORDER BY updated_at DESC LIMIT 200 `).all(); const now = Date.now(); const results = []; for (const row of persons) { const person = this._formatNode(row); // 查该 Person 关联的所有 Event 节点,找最新互动时间 const relatedEvents = this.db.prepare(` SELECT n.data FROM edges e JOIN nodes n ON n.id = CASE WHEN e.from_id = ? THEN e.to_id WHEN e.to_id = ? THEN e.from_id END WHERE (e.from_id = ? OR e.to_id = ?) AND n.type = 'Event' `).all(row.id, row.id, row.id, row.id); let lastInteraction = null; let lastTs = 0; for (const ev of relatedEvents) { try { const evData = JSON.parse(ev.data); const dateStr = evData.date || evData.created_at || evData.createdAt || null; if (dateStr) { const ts = new Date(dateStr).getTime(); if (!isNaN(ts) && ts > lastTs) { lastTs = ts; lastInteraction = dateStr; } } } catch { /* skip malformed */ } } // 计算距今天数 let daysElapsed; if (lastInteraction && lastTs > 0) { daysElapsed = Math.floor((now - lastTs) / (1000 * 60 * 60 * 24)); } else { // 没有互动记录,按节点创建时间计算 daysElapsed = Math.floor((now - row.created_at * 1000) / (1000 * 60 * 60 * 24)); } if (daysElapsed >= daysSince) { results.push({ person, lastInteraction, daysSince: daysElapsed, suggestion: lastInteraction ? `已有 ${daysElapsed} 天未与 ${person.name} 互动,建议主动联系。` : `${person.name} 自录入以来从未互动(${daysElapsed} 天),建议安排沟通。`, }); } } // 按未互动天数降序(最久未联系的排前面) results.sort((a, b) => b.daysSince - a.daysSince); return results.slice(0, limit); } // ─── 内部工具 ──────────────────────────────────────────────── _formatNode(row) { return { ...JSON.parse(row.data), _type: row.type, _analysisStatus: row.analysis_status, _createdAt: row.created_at, _updatedAt: row.updated_at, }; } } // 单例:整个进程共享同一个数据库连接 let _instance = null; function getGraphStore(dbPath) { if (!_instance) { _instance = new GraphStore(dbPath); } return _instance; } module.exports = { GraphStore, getGraphStore };