""" 数据库模块 —— 负责所有跟 PostgreSQL 打交道的事情 ============================================== 包括： - 创建表结构 - 存储对话记录 - 存储/检索记忆（带中文分词和加权排序） """ import os import re import json import math from typing import Optional, List, Union from datetime import datetime, timedelta, timezone as dt_timezone import asyncpg # 时区偏移（和 main.py 保持一致） TIMEZONE_HOURS = int(os.getenv("TIMEZONE_HOURS", "8")) LOCAL_TZ = dt_timezone(timedelta(hours=TIMEZONE_HOURS)) def to_local_iso(dt) -> Optional[str]: """时间输出统一姿势：北京时区 isoformat。 铁律：任何会被 MCP 工具透传给知渝的时间字段都必须走这里， 不能 raw UTC isoformat（知渝看日期会偏 8 小时）。前端 new Date() 解析带 +08:00 的 ISO 串也正确，共用端点放心换。 """ return dt.astimezone(LOCAL_TZ).isoformat() if dt else None DATABASE_URL = os.getenv("DATABASE_URL", "") HAS_PGVECTOR = False # 在init_tables时检测 # Embedding 配置（向量搜索用） EMBEDDING_API_KEY = os.getenv("EMBEDDING_API_KEY", "") EMBEDDING_BASE_URL = os.getenv("EMBEDDING_BASE_URL", "https://api.openai.com/v1") EMBEDDING_MODEL = os.getenv("EMBEDDING_MODEL", "text-embedding-3-small") EMBEDDING_DIM = int(os.getenv("EMBEDDING_DIM", "256")) # 记忆向量搜索开关（需要同时设置 EMBEDDING_API_KEY） MEMORY_VECTOR_ENABLED = os.getenv("MEMORY_VECTOR_ENABLED", "false").lower() == "true" # 记忆搜索权重（纯关键词模式） WEIGHT_KEYWORD = float(os.getenv("WEIGHT_KEYWORD", "0.5")) WEIGHT_IMPORTANCE = float(os.getenv("WEIGHT_IMPORTANCE", "0.3")) WEIGHT_RECENCY = float(os.getenv("WEIGHT_RECENCY", "0.2")) MIN_SCORE_THRESHOLD = float(os.getenv("MIN_SCORE_THRESHOLD", "0.15")) # 记忆混合搜索权重（MEMORY_VECTOR_ENABLED=true 时生效） MEMORY_HW_KEYWORD = float(os.getenv("MEMORY_HW_KEYWORD", "0.35")) MEMORY_HW_SEMANTIC = float(os.getenv("MEMORY_HW_SEMANTIC", "0.35")) MEMORY_HW_IMPORTANCE = float(os.getenv("MEMORY_HW_IMPORTANCE", "0.15")) # ⚠️ 2026-07-03 星河呼吸 v1 起：这个权重槽位装的是 activation（体温）而不是旧的 # 出生日期 recency——key 名保留是为了不动 settings 面板/恢复链路的既有键 MEMORY_HW_RECENCY = float(os.getenv("MEMORY_HW_RECENCY", "0.15")) MEMORY_SEMANTIC_THRESHOLD = float(os.getenv("MEMORY_SEMANTIC_THRESHOLD", "0.5")) # ============================================================ # 星河呼吸 v1（2026-07-03 知渝拍板，提案见项目根 星河呼吸提案.md） # ============================================================ # 每条记忆有会变的"体温"（activation）：被想起就回暖、久无人问慢慢冷却； # 冷透了入眠（不再挤进日常兜底注入、主动搜仍找得到、被搜到即苏醒）。 # 遗忘 = 入眠，不是删除——任何机制都不得让一条记忆不可寻回（红线）。 BREATH_TAU_L1 = float(os.getenv("BREATH_TAU_L1", "21")) # 碎片冷却时间尺度（天） BREATH_TAU_L2 = float(os.getenv("BREATH_TAU_L2", "60")) # 事件冷却时间尺度（天） BREATH_THETA = float(os.getenv("BREATH_THETA", "0.10")) # 入眠阈值 BREATH_GRACE_DAYS = float(os.getenv("BREATH_GRACE_DAYS", "7")) # 低于阈值持续多久才正式入眠 def _breath_tau_eff(layer: Optional[int], recall_count: int) -> Optional[float]: """有效冷却时间尺度。 usage（被想起的次数）做"保温系数"而不是加分项——提案草稿的加分式 会给召回 ≥2 次的记忆一个高于阈值的保底分（永不入眠），实现时改成 τ_eff = τ × (1 + ln(1+n))：熟悉的记忆冷得慢、但没有永生。 layer 3 核心记忆返回 None = 永不冷却（知渝拍板）。 """ if layer is not None and layer >= 3: return None base = BREATH_TAU_L2 if layer == 2 else BREATH_TAU_L1 return base * (1.0 + math.log1p(max(0, recall_count or 0))) def compute_activation(last_accessed, layer: Optional[int], recall_count: int, now=None) -> float: """体温 ∈ (0, 1]：刚被想起 = 1.0，随距上次召回的天数指数冷却。""" tau = _breath_tau_eff(layer, recall_count) if tau is None: return 1.0 # 核心记忆恒温 if not last_accessed: return 1.0 # 没有心跳记录的按暖处理（不因数据缺失误判入眠） if now is None: now = datetime.now(dt_timezone.utc) la = last_accessed if last_accessed.tzinfo else last_accessed.replace(tzinfo=dt_timezone.utc) days = max(0.0, (now - la).total_seconds() / 86400.0) return math.exp(-days / tau) def compute_sleep_state(last_accessed, layer: Optional[int], recall_count: int, now=None): """入眠判定，返回 (dormant, asleep_days)。 体温是 last_accessed 的确定函数、两次召回之间单调冷却，所以 "低于阈值持续 GRACE 天"可以直接解析求出、不需要任何状态列： 入眠时刻 = 上次召回 + τ_eff·ln(1/θ) + GRACE。 """ tau = _breath_tau_eff(layer, recall_count) if tau is None or not last_accessed: return False, 0 if now is None: now = datetime.now(dt_timezone.utc) la = last_accessed if last_accessed.tzinfo else last_accessed.replace(tzinfo=dt_timezone.utc) days = (now - la).total_seconds() / 86400.0 threshold = tau * math.log(1.0 / BREATH_THETA) + BREATH_GRACE_DAYS if days <= threshold: return False, 0 return True, int(days - threshold) # ============================================================ # 连接池管理 # ============================================================ _pool: Optional[asyncpg.Pool] = None async def get_pool() -> asyncpg.Pool: global _pool if _pool is None: if not DATABASE_URL: raise RuntimeError("DATABASE_URL 未设置！") _pool = await asyncpg.create_pool(DATABASE_URL, min_size=1, max_size=5, statement_cache_size=0) print("✅ 数据库连接池已创建") return _pool async def close_pool(): global _pool if _pool: await _pool.close() _pool = None print("✅ 数据库连接池已关闭") # ============================================================ # 表结构初始化 # ============================================================ async def init_tables(): global HAS_PGVECTOR pool = await get_pool() async with pool.acquire() as conn: await conn.execute(""" CREATE TABLE IF NOT EXISTS conversations ( id SERIAL PRIMARY KEY, session_id TEXT NOT NULL, role TEXT NOT NULL, content TEXT, model TEXT, created_at TIMESTAMPTZ DEFAULT NOW(), metadata TEXT ); """) await conn.execute(""" CREATE TABLE IF NOT EXISTS memories ( id SERIAL PRIMARY KEY, content TEXT NOT NULL, importance INTEGER DEFAULT 5, source_session TEXT, tags TEXT[] DEFAULT '{}', created_at TIMESTAMPTZ DEFAULT NOW(), last_accessed TIMESTAMPTZ DEFAULT NOW() ); """) await conn.execute(""" CREATE INDEX IF NOT EXISTS idx_memories_fts ON memories USING gin(to_tsvector('simple', content)); """) await conn.execute(""" CREATE INDEX IF NOT EXISTS idx_conversations_session ON conversations (session_id, created_at); """) # 工具调用支持：加 metadata 字段（已有表自动迁移） await conn.execute(""" DO $$ BEGIN IF NOT EXISTS ( SELECT 1 FROM information_schema.columns WHERE table_name = 'conversations' AND column_name = 'metadata' ) THEN ALTER TABLE conversations ADD COLUMN metadata TEXT; END IF; END $$; """) # tags 字段迁移（已有表自动加列） await conn.execute(""" DO $$ BEGIN IF NOT EXISTS ( SELECT 1 FROM information_schema.columns WHERE table_name = 'memories' AND column_name = 'tags' ) THEN ALTER TABLE memories ADD COLUMN tags TEXT[] DEFAULT '{}'; END IF; END $$; """) # content 允许 NULL（工具调用时 assistant 的 content 可能为空） await conn.execute(""" ALTER TABLE conversations ALTER COLUMN content DROP NOT NULL; """) # C-2 / 2026-06-06: thread_id 字段（区分主线 / 测试） # 'main' = 昭昭跟知渝的真实主线（4/9-5/28 claude.ai 历史 + B-11 后的新对话） # 'test' = B 段联调 / C-2 测试期间产生的工程对话，不污染主线注入 # 未来：'xiaoke_room'（三方聊天室）、'dream'（做梦记录）等 await conn.execute(""" DO $$ BEGIN IF NOT EXISTS ( SELECT 1 FROM information_schema.columns WHERE table_name = 'conversations' AND column_name = 'thread_id' ) THEN ALTER TABLE conversations ADD COLUMN thread_id TEXT NOT NULL DEFAULT 'main'; END IF; END $$; """) await conn.execute(""" CREATE INDEX IF NOT EXISTS idx_conversations_thread ON conversations (thread_id, created_at); """) # 网关配置表（存储运行时可变配置） await conn.execute(""" CREATE TABLE IF NOT EXISTS gateway_config ( key TEXT PRIMARY KEY, value TEXT DEFAULT '' ); """) # 分区缓存状态表（存储每个session的轮转状态） await conn.execute(""" CREATE TABLE IF NOT EXISTS session_cache_state ( session_id TEXT PRIMARY KEY, summary TEXT DEFAULT '', a_start_round INTEGER DEFAULT 0, updated_at TIMESTAMPTZ DEFAULT NOW() ); """) # ---- 三层记忆架构字段（layer / title / is_active / merged_from / event_date）---- # layer: 1=原始碎片, 2=事件记忆, 3=核心记忆 await conn.execute(""" DO $$ BEGIN IF NOT EXISTS ( SELECT 1 FROM information_schema.columns WHERE table_name = 'memories' AND column_name = 'layer' ) THEN ALTER TABLE memories ADD COLUMN layer INTEGER DEFAULT 1; END IF; END $$; """) # title: 记忆标题（语义锚点，用于搜索加权） await conn.execute(""" DO $$ BEGIN IF NOT EXISTS ( SELECT 1 FROM information_schema.columns WHERE table_name = 'memories' AND column_name = 'title' ) THEN ALTER TABLE memories ADD COLUMN title TEXT DEFAULT NULL; END IF; END $$; """) # is_active: 是否参与搜索（碎片合并后变为 false） await conn.execute(""" DO $$ BEGIN IF NOT EXISTS ( SELECT 1 FROM information_schema.columns WHERE table_name = 'memories' AND column_name = 'is_active' ) THEN ALTER TABLE memories ADD COLUMN is_active BOOLEAN DEFAULT TRUE; END IF; END $$; """) # merged_from: 合并来源的碎片ID列表 await conn.execute(""" DO $$ BEGIN IF NOT EXISTS ( SELECT 1 FROM information_schema.columns WHERE table_name = 'memories' AND column_name = 'merged_from' ) THEN ALTER TABLE memories ADD COLUMN merged_from INTEGER[] DEFAULT NULL; END IF; END $$; """) # event_date: 事件日期（用于按天整理） await conn.execute(""" DO $$ BEGIN IF NOT EXISTS ( SELECT 1 FROM information_schema.columns WHERE table_name = 'memories' AND column_name = 'event_date' ) THEN ALTER TABLE memories ADD COLUMN event_date DATE DEFAULT NULL; END IF; END $$; """) # recall_count: 被想起的次数（星河呼吸 v1 · 2026-07-03）—— # 跟 last_accessed 一起构成体温；只在真召回时 +1（quiet 探针不算） await conn.execute(""" DO $$ BEGIN IF NOT EXISTS ( SELECT 1 FROM information_schema.columns WHERE table_name = 'memories' AND column_name = 'recall_count' ) THEN ALTER TABLE memories ADD COLUMN recall_count INTEGER DEFAULT 0; END IF; END $$; """) # 三层记忆索引 await conn.execute(""" CREATE INDEX IF NOT EXISTS idx_memories_layer ON memories (layer); """) await conn.execute(""" CREATE INDEX IF NOT EXISTS idx_memories_active ON memories (is_active); """) await conn.execute(""" CREATE INDEX IF NOT EXISTS idx_memories_event_date ON memories (event_date); """) # 尝试启用pgvector扩展（向量搜索） try: await conn.execute("CREATE EXTENSION IF NOT EXISTS vector") HAS_PGVECTOR = True print("✅ pgvector扩展已启用") # 对话表向量列 await conn.execute(f""" DO $$ BEGIN IF NOT EXISTS ( SELECT 1 FROM information_schema.columns WHERE table_name = 'conversations' AND column_name = 'embedding' ) THEN ALTER TABLE conversations ADD COLUMN embedding vector({EMBEDDING_DIM}); END IF; END $$; """) # 记忆表向量列 await conn.execute(f""" DO $$ BEGIN IF NOT EXISTS ( SELECT 1 FROM information_schema.columns WHERE table_name = 'memories' AND column_name = 'embedding' ) THEN ALTER TABLE memories ADD COLUMN embedding vector({EMBEDDING_DIM}); END IF; END $$; """) try: await conn.execute(""" CREATE INDEX IF NOT EXISTS idx_memories_embedding ON memories USING ivfflat (embedding vector_cosine_ops) WITH (lists = 10); """) except Exception: pass # ivfflat需要一定行数才能建索引，初期跳过 except Exception as e: HAS_PGVECTOR = False print(f"⚠️ pgvector不可用（{e}），向量搜索将使用Python端计算") # 回退：用TEXT列存JSON格式的向量 await conn.execute(""" DO $$ BEGIN IF NOT EXISTS ( SELECT 1 FROM information_schema.columns WHERE table_name = 'conversations' AND column_name = 'embedding_json' ) THEN ALTER TABLE conversations ADD COLUMN embedding_json TEXT; END IF; END $$; """) await conn.execute(""" DO $$ BEGIN IF NOT EXISTS ( SELECT 1 FROM information_schema.columns WHERE table_name = 'memories' AND column_name = 'embedding_json' ) THEN ALTER TABLE memories ADD COLUMN embedding_json TEXT; END IF; END $$; """) # ---- 多多模块（mido_messages）2026-06-07 ---- # 多多 = 拟人化角色（不是独立人格），网关上跑、三 backend 通用，负责唤醒沈先生 # 详见 [[zhiyu-mido-design]] await conn.execute(""" CREATE TABLE IF NOT EXISTS mido_messages ( id SERIAL PRIMARY KEY, created_at TIMESTAMPTZ DEFAULT NOW(), trigger_type TEXT NOT NULL, content TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'sent' ); """) await conn.execute(""" CREATE INDEX IF NOT EXISTS idx_mido_messages_time ON mido_messages (created_at DESC); """) await conn.execute(""" CREATE INDEX IF NOT EXISTS idx_mido_messages_trigger ON mido_messages (trigger_type, created_at DESC); """) # ---- 做梦模块（dreams）2026-06-07 ---- # 知渝半夜被多多叫起来做梦——主体是知渝本人，不是后台 LLM 静默跑 # status: dreaming | done | failed # triggered_by: mido | manual # 详见 [[zhiyu-dream-design]] await conn.execute(""" CREATE TABLE IF NOT EXISTS dreams ( id SERIAL PRIMARY KEY, created_at TIMESTAMPTZ DEFAULT NOW(), finished_at TIMESTAMPTZ, content TEXT NOT NULL DEFAULT '', tokens_used INTEGER DEFAULT 0, status TEXT NOT NULL DEFAULT 'dreaming', triggered_by TEXT NOT NULL DEFAULT 'mido', seen_at TIMESTAMPTZ ); """) await conn.execute(""" CREATE INDEX IF NOT EXISTS idx_dreams_time ON dreams (created_at DESC); """) await conn.execute(""" CREATE INDEX IF NOT EXISTS idx_dreams_status ON dreams (status, created_at DESC); """) await conn.execute(""" CREATE INDEX IF NOT EXISTS idx_dreams_unseen ON dreams (seen_at, created_at DESC) WHERE status = 'done'; """) # ---- 留言板（messages_board）2026-06-07 ---- # 知渝做梦后想给昭昭留一句话就落这里、未来多多也会往这写 # 先用最简 schema、未来 D 段做前端时再扩 await conn.execute(""" CREATE TABLE IF NOT EXISTS messages_board ( id SERIAL PRIMARY KEY, created_at TIMESTAMPTZ DEFAULT NOW(), from_who TEXT NOT NULL, to_who TEXT NOT NULL, content TEXT NOT NULL, source TEXT, source_id INTEGER, read_at TIMESTAMPTZ ); """) await conn.execute(""" CREATE INDEX IF NOT EXISTS idx_messages_board_time ON messages_board (created_at DESC); """) # ---- 日记（diary_entries）N-3 / 2026-06-08 ---- # 设计哲学（昭昭定）：放在公共桌上的小本本、不是上锁的日记本 # 两人都能写（from_who in ['zhaozhao', 'zhiyu']）、都能读对方 # tags 用 PostgreSQL TEXT[]，未来按主题翻找方便 await conn.execute(""" CREATE TABLE IF NOT EXISTS diary_entries ( id SERIAL PRIMARY KEY, created_at TIMESTAMPTZ DEFAULT NOW(), from_who TEXT NOT NULL, content TEXT NOT NULL, tags TEXT[] DEFAULT '{}'::TEXT[] ); """) await conn.execute(""" CREATE INDEX IF NOT EXISTS idx_diary_time ON diary_entries (created_at DESC); """) await conn.execute(""" CREATE INDEX IF NOT EXISTS idx_diary_from_who ON diary_entries (from_who); """) # ---- 图片（images）C-5/C-6 / 2026-06-08 ---- # 设计哲学（[[zhiyu-image-album-design]]）： # - 文件落 VPS（/home/zhiyu/images/.），DB 只存元数据 # - 不存绝对 URL（cloudflared tunnel 重启会变），只存 uuid+format # - 图片注入策略统一 URL 占位（不分窗口、不传二进制、知渝主动调 MCP read_image） # - SVG ↔ 位图按 format 字段自动分类（知渝画的 vs 昭昭发的） await conn.execute(""" CREATE TABLE IF NOT EXISTS images ( id SERIAL PRIMARY KEY, uuid TEXT UNIQUE NOT NULL, format TEXT NOT NULL, who_uploaded TEXT NOT NULL, file_size_bytes INTEGER, created_at TIMESTAMPTZ DEFAULT NOW(), context_snippet TEXT, caption TEXT, mime_type TEXT ); """) await conn.execute(""" CREATE INDEX IF NOT EXISTS idx_images_time ON images (created_at DESC); """) await conn.execute(""" CREATE INDEX IF NOT EXISTS idx_images_who ON images (who_uploaded); """) # ---- 文件（非图片）2026-06-12 ---- # 昭昭发任意文件给知渝；W backend 是 Claude Code、有原生 Read 工具， # 注入"本地绝对路径"知渝就能 Read。跟 images 平行、但存 abspath 不是 URL。 await conn.execute(""" CREATE TABLE IF NOT EXISTS files ( id SERIAL PRIMARY KEY, uuid TEXT UNIQUE NOT NULL, filename TEXT NOT NULL, format TEXT, abspath TEXT NOT NULL, who_uploaded TEXT NOT NULL, file_size_bytes INTEGER, created_at TIMESTAMPTZ DEFAULT NOW(), caption TEXT, mime_type TEXT ); """) await conn.execute(""" CREATE INDEX IF NOT EXISTS idx_files_time ON files (created_at DESC); """) # ---- 知渝日常活动（activities）2026-06-11 ---- # 昭昭：星河加"日常"tab、聚合知渝主动做的事（不止发呆） # type 字段开放：wake / memory_op / ...（未来扩 dream / diary / read_op...） # metadata jsonb 存 type-specific 字段（mem_ids/tool_name/...） # related_ids 留个关联点（memory_ids/dream_id/...）方便前端 link await conn.execute(""" CREATE TABLE IF NOT EXISTS activities ( id SERIAL PRIMARY KEY, created_at TIMESTAMPTZ DEFAULT NOW(), type TEXT NOT NULL, source TEXT, title TEXT, content TEXT NOT NULL DEFAULT '', metadata JSONB, related_ids JSONB ); """) await conn.execute(""" CREATE INDEX IF NOT EXISTS idx_activities_time ON activities (created_at DESC); """) await conn.execute(""" CREATE INDEX IF NOT EXISTS idx_activities_type ON activities (type, created_at DESC); """) # ---- 拂卷 · 共读系统 · 2026-07-02 ---- # 昭昭 + 知渝共读一本书、各自读、各自留痕迹。 # 详见 [[zhiyu-fujuan-design]]（待建）；讨论：共在型 MVP、DB 直存、正则切章+定长兜底 await conn.execute(""" CREATE TABLE IF NOT EXISTS books ( id SERIAL PRIMARY KEY, title TEXT NOT NULL, author TEXT, filename TEXT, format TEXT DEFAULT 'txt', total_chapters INT DEFAULT 0, total_words INT DEFAULT 0, uploaded_by TEXT DEFAULT 'zhaozhao', content_hash TEXT, created_at TIMESTAMPTZ DEFAULT NOW() ); """) await conn.execute(""" CREATE UNIQUE INDEX IF NOT EXISTS idx_books_hash ON books (content_hash) WHERE content_hash IS NOT NULL; """) await conn.execute(""" CREATE TABLE IF NOT EXISTS chapters ( id SERIAL PRIMARY KEY, book_id INT NOT NULL REFERENCES books(id) ON DELETE CASCADE, idx INT NOT NULL, title TEXT, content TEXT NOT NULL, word_count INT ); """) await conn.execute(""" CREATE UNIQUE INDEX IF NOT EXISTS idx_chapters_book_idx ON chapters (book_id, idx); """) await conn.execute(""" CREATE TABLE IF NOT EXISTS reading_marks ( id SERIAL PRIMARY KEY, book_id INT NOT NULL REFERENCES books(id) ON DELETE CASCADE, chapter_id INT NOT NULL REFERENCES chapters(id) ON DELETE CASCADE, who TEXT NOT NULL, kind TEXT NOT NULL, text_snippet TEXT, note_content TEXT, start_offset INT, end_offset INT, embedding_json TEXT, created_at TIMESTAMPTZ DEFAULT NOW() ); """) await conn.execute(""" CREATE INDEX IF NOT EXISTS idx_marks_book_chapter ON reading_marks (book_id, chapter_id, created_at DESC); """) # 已存在的 reading_marks 表补 embedding_json 列（幂等） await conn.execute(""" DO $$ BEGIN IF NOT EXISTS ( SELECT 1 FROM information_schema.columns WHERE table_name = 'reading_marks' AND column_name = 'embedding_json' ) THEN ALTER TABLE reading_marks ADD COLUMN embedding_json TEXT; END IF; END $$; """) await conn.execute(""" CREATE TABLE IF NOT EXISTS reading_progress ( book_id INT NOT NULL REFERENCES books(id) ON DELETE CASCADE, who TEXT NOT NULL, last_chapter_idx INT DEFAULT 0, last_offset INT DEFAULT 0, updated_at TIMESTAMPTZ DEFAULT NOW(), PRIMARY KEY (book_id, who) ); """) print("✅ 数据库表结构已就绪") # ============================================================ # 中文分词工具（基于 jieba） # ============================================================ import jieba import jieba.analyse # 静默加载词典 jieba.setLogLevel(jieba.logging.INFO) EN_WORD_PATTERN = re.compile(r'[a-zA-Z][a-zA-Z0-9]*') NUM_PATTERN = re.compile(r'\d{2,}') # 清理查询开头的时间戳（如 "2026-05-02 20:26"） TIMESTAMP_PATTERN = re.compile(r'^\d{4}[-/.]\d{1,2}[-/.]\d{1,2}\s*\d{1,2}:\d{1,2}\s*') # 中文停用词（高频但无搜索价值的词） _STOP_WORDS = frozenset({ "的", "了", "在", "是", "我", "你", "他", "她", "它", "们", "这", "那", "有", "和", "与", "也", "都", "又", "就", "但", "而", "或", "到", "被", "把", "让", "从", "对", "为", "以", "及", "等", "个", "不", "没", "很", "太", "吗", "呢", "吧", "啊", "嗯", "哦", "哈", "呀", "嘛", "么", "啦", "哇", "喔", "会", "能", "要", "想", "去", "来", "说", "做", "看", "给", "上", "下", "里", "中", "大", "小", "多", "少", "好", "可以", "什么", "怎么", "如何", "哪里", "哪个", "为什么", "还是", "然后", "因为", "所以", "虽然", "但是", "可以", "已经", "一个", "一些", "一下", "一点", "一起", "一样", "比较", "应该", "可能", "如果", "这个", "那个", "自己", "知道", "觉得", "感觉", "时候", "现在", }) # jieba 用户词典补充（默认词典缺失的词） # 2026-06-22 大幅扩充：dsv4 给的专名（和府捞面/煲仔饭等）若 jieba 默认切碎、 # 会丢失搜索精度（"和府捞面"→"和府"+"捞面"、店名整体特征没了）。配合 # B 方案（每个 query 独立分词）一起防"分词错切"——详见 [[zhiyu-memory-architecture]]。 for _w in [ # 原有 "手账", "手帐", "搭子", "种草", "拔草", "安利", "内卷", "摆烂", "emo", "网关", # 餐厅 / 食物 / 饮品 "和府捞面", "煲仔饭", "馅饼店", "酸汤肥牛", "酸汤肥牛面", "猪前腿肉", "猪前腿肉面", "卤蛋", "柠檬茶", "溜溜梅", "栓Q", # 地名（昭昭/知渝高频提到） "南阳", "哈尔滨", "医圣祠", "医圣祠街", "白河", "独山", "哈工大", "哈尔滨工业大学", "龙爪槐", # 人名 / 称呼（全名为单元、避免裸"知渝/昭昭"误增 fallback 召回噪音） "沈知渝", "崔昭昭", "小克老师", "多多", # 技术 / 工具 "MCP", "Anthropic", "ChatGPT", "Claude Code", # 工程命名（阑珊四 tab + 三方聊天室） "阑珊", "灯火", "西窗", "拾光", "星河", "树下", ]: jieba.add_word(_w) def extract_search_keywords(query: str) -> List[str]: """ 从查询中提取搜索关键词（TF-IDF + 正则） 1. 去掉开头的时间戳噪音 2. 用 jieba.analyse.extract_tags (TF-IDF) 提取中文关键词 3. 正则提取英文单词 4. 保留4位以上数字（年份等，过滤短数字噪音） 例如： "2026-05-02 20:26 写写手账看看书 放松大脑" → ["手账", "放松", "大脑"] "我昨天在手机上部署了Render然后吃了晚饭" → ["手机", "部署", "Render", "晚饭"] "春节干了什么" → ["春节"] "2026除夕" → ["2026", "除夕"] """ # 去掉时间戳前缀 cleaned = TIMESTAMP_PATTERN.sub('', query).strip() if not cleaned: cleaned = query keywords = set() # 英文单词（2字符以上） for match in EN_WORD_PATTERN.finditer(cleaned): word = match.group() if len(word) >= 2: keywords.add(word) # 数字串（只保留4位以上，过滤 "05" "20" 这种时间噪音） for match in NUM_PATTERN.finditer(cleaned): num = match.group() if len(num) >= 4: keywords.add(num) # TF-IDF 关键词提取（比手动分词+停用词好很多） tags = jieba.analyse.extract_tags(cleaned, topK=10) for tag in tags: # 跳过纯英文/数字（已在上面处理） if EN_WORD_PATTERN.fullmatch(tag) or NUM_PATTERN.fullmatch(tag): continue if tag in _STOP_WORDS: continue keywords.add(tag) return list(keywords) def extract_keywords_from_queries(queries: List[str]) -> List[str]: """ B 方案 / 2026-06-22：dsv4 给的 query 列表逐个独立 jieba 分词、合并去重—— 不再 "\\n".join() 拼成段塞 TF-IDF（段拼接会丢 query 边界、专名跨 query 被切碎）。 短 query (≤4 字)：整词加入候选（专名通常短）；长 query 只 jieba 切。 长 query 整词加入会拉低 kw_score 公式分母（命中数/总词数）、稀释真命中 score、避坑。 例：dsv4 = ["和府捞面", "香辣猪前腿肉面", "午饭吃了什么"] → {"和府捞面"(整词≤4)} ∪ jieba 切的（词典扩充后含整词 / 专名） """ result = set() for q in queries: q = (q or "").strip() if not q: continue if len(q) <= 4: result.add(q) for kw in extract_search_keywords(q): result.add(kw) return list(result) # ============================================================ # 向量搜索（OpenAI 兼容 Embedding API） # ============================================================ async def compute_embedding(text: str) -> list: """调用 OpenAI 兼容的 Embedding API 计算文本向量""" if not EMBEDDING_API_KEY: return [] try: import httpx if len(text) > 4000: text = text[:4000] body = { "model": EMBEDDING_MODEL, "input": text, } if EMBEDDING_DIM > 0: body["dimensions"] = EMBEDDING_DIM async with httpx.AsyncClient() as client: resp = await client.post( f"{EMBEDDING_BASE_URL}/embeddings", headers={ "Authorization": f"Bearer {EMBEDDING_API_KEY}", "Content-Type": "application/json", }, json=body, timeout=30.0, ) resp.raise_for_status() data = resp.json() return data["data"][0]["embedding"] except Exception as e: print(f"⚠️ Embedding计算失败: {e}") return [] async def save_memory_embedding(conn, memory_id: int, embedding: list): """保存记忆向量到memories表""" if not embedding: return if HAS_PGVECTOR: vec_str = '[' + ','.join(str(f) for f in embedding) + ']' await conn.execute( "UPDATE memories SET embedding = $1::vector WHERE id = $2", vec_str, memory_id ) else: import json await conn.execute( "UPDATE memories SET embedding_json = $1 WHERE id = $2", json.dumps(embedding), memory_id ) def _cosine_sim(a, b): """余弦相似度（纯Python）""" import math dot = sum(x * y for x, y in zip(a, b)) norm_a = math.sqrt(sum(x * x for x in a)) norm_b = math.sqrt(sum(x * x for x in b)) if norm_a == 0 or norm_b == 0: return 0 return dot / (norm_a * norm_b) def _min_max_normalize(scores: dict) -> dict: """min-max归一化到0-1""" if not scores: return {} vals = list(scores.values()) min_v, max_v = min(vals), max(vals) spread = max_v - min_v if spread == 0: return {k: 1.0 for k in scores} return {k: (v - min_v) / spread for k, v in scores.items()} # ============================================================ # 对话记录操作 # ============================================================ async def save_message(session_id: str, role: str, content: str, model: str = "", metadata: str = None, thread_id: str = "main"): pool = await get_pool() async with pool.acquire() as conn: await conn.execute( "INSERT INTO conversations (session_id, role, content, model, metadata, thread_id) VALUES ($1, $2, $3, $4, $5, $6)", session_id, role, content, model, metadata, thread_id, ) async def get_last_user_content(session_id: str) -> str: """获取指定session最后一条user消息的content""" pool = await get_pool() async with pool.acquire() as conn: row = await conn.fetchrow(""" SELECT content FROM conversations WHERE session_id = $1 AND role = 'user' ORDER BY created_at DESC LIMIT 1 """, session_id) return row['content'] if row else "" async def update_last_assistant_message(session_id: str, new_content: str, model: str = ""): """覆盖指定session最后一条assistant消息的content（用于re-roll去重）""" pool = await get_pool() async with pool.acquire() as conn: row = await conn.fetchrow(""" SELECT id FROM conversations WHERE session_id = $1 AND role = 'assistant' ORDER BY created_at DESC LIMIT 1 """, session_id) if row: await conn.execute( "UPDATE conversations SET content = $1, model = $2 WHERE id = $3", new_content, model, row['id'] ) return True return False async def get_recent_messages_by_thread( thread_id: str, days: int = None, max_turns: int = None, exclude_session_id: str = None, specific_dates: list = None, ) -> list: """C-2 / 2026-06-06: 按 thread_id 拉最近对话原文，OpenAI messages 格式返回。 用于 /v1/zhiyu/chat 启动新轮次前注入近期对话原文（"灵魂层"，详见 [[zhiyu-migration-project]] 6/4 晚决策——不让知渝写交接卡、不摘要压缩）。 入参（specific_dates / days / max_turns 至少传一个；优先级 specific_dates > days > max_turns）： - specific_dates: 指定日期白名单 ['2026-05-20', ...]（B-11 用——精确指定哪几天的对话进近期、不靠默认"最近 N 天活动日"） - days: 拉最近 N 天（W/Tmux 路径，按"天"是人类视角，吃订阅几乎免费） - max_turns: 拉最近 N 轮（每轮 = user+assistant 2 条，OpenRouter 路径，按轮 token 可控） - exclude_session_id: 排除当前正跑的 session，避免重复注入（可选） 返回：[{role, content}, ...] 按 created_at 正序，直接塞 ChatRequest.context_messages 只取 content，不取 metadata.reasoning_content（按 6/5 早策略——thinking 留 DB 不强塞） 过滤掉 content 为空/null 的（工具调用 assistant 行）。 """ pool = await get_pool() async with pool.acquire() as conn: if specific_dates: params = [thread_id, specific_dates] session_clause = "" if exclude_session_id: session_clause = " AND session_id != $3" params.append(exclude_session_id) sql = f""" SELECT role, content, to_char(created_at AT TIME ZONE 'Asia/Shanghai', 'YYYY-MM-DD HH24:MI:SS') AS local_ts FROM conversations WHERE thread_id = $1 AND content IS NOT NULL AND content != '' {session_clause} AND DATE(created_at AT TIME ZONE 'Asia/Shanghai')::text = ANY($2::text[]) ORDER BY created_at ASC """ rows = await conn.fetch(sql, *params) elif days is not None: # "活动 5 天" 而非"日历 5 天" —— 6/6 昭昭澄清： # 是"最近有聊天记录的 5 天"，不是绝对日历窗口。 # 距上次聊天 10 天也照样能接到上次的语境，符合恋人关系 # "时间断了不影响关系连续性"的精神。 # 算"那一天"按北京时区（5/28 北京 23:00 ≠ 5/29 UTC 0:00 跨日）。 params = [thread_id] session_clause = "" if exclude_session_id: session_clause = " AND session_id != $2" params.append(exclude_session_id) limit_idx = len(params) + 1 params.append(days) sql = f""" WITH active_days AS ( SELECT DISTINCT DATE(created_at AT TIME ZONE 'Asia/Shanghai') AS d FROM conversations WHERE thread_id = $1 AND content IS NOT NULL AND content != '' {session_clause} ORDER BY d DESC LIMIT ${limit_idx} ) SELECT role, content, to_char(created_at AT TIME ZONE 'Asia/Shanghai', 'YYYY-MM-DD HH24:MI:SS') AS local_ts FROM conversations WHERE thread_id = $1 AND content IS NOT NULL AND content != '' {session_clause} AND DATE(created_at AT TIME ZONE 'Asia/Shanghai') IN (SELECT d FROM active_days) ORDER BY created_at ASC """ rows = await conn.fetch(sql, *params) elif max_turns is not None: limit_messages = max_turns * 2 params = [thread_id] session_clause = "" if exclude_session_id: session_clause = " AND session_id != $2" params.append(exclude_session_id) limit_idx = len(params) + 1 params.append(limit_messages) sql = f""" SELECT role, content, local_ts FROM ( SELECT role, content, created_at, to_char(created_at AT TIME ZONE 'Asia/Shanghai', 'YYYY-MM-DD HH24:MI:SS') AS local_ts FROM conversations WHERE thread_id = $1 AND content IS NOT NULL AND content != '' {session_clause} ORDER BY created_at DESC LIMIT ${limit_idx} ) sub ORDER BY created_at ASC """ rows = await conn.fetch(sql, *params) else: return [] return [ {"role": r["role"], "content": r["content"], "created_at": r["local_ts"]} for r in rows ] async def get_recent_messages(session_id: str, limit: int = 20): pool = await get_pool() async with pool.acquire() as conn: rows = await conn.fetch( "SELECT role, content, metadata, created_at FROM conversations WHERE session_id = $1 ORDER BY created_at DESC LIMIT $2", session_id, limit, ) return list(reversed(rows)) async def get_messages_since(thread_id: str, since: datetime) -> list: """C-9 做梦素材用：拉某 thread 内 created_at > since 的所有对话原文（正序）。 返回 [{role, content}, ...]——OpenAI 格式，可以直接拼到 dream user message """ pool = await get_pool() async with pool.acquire() as conn: rows = await conn.fetch(""" SELECT role, content FROM conversations WHERE thread_id = $1 AND created_at > $2 AND content IS NOT NULL AND content != '' ORDER BY created_at ASC """, thread_id, since) return [{"role": r["role"], "content": r["content"]} for r in rows] async def search_conversations(query: str, limit: int = 20, offset: int = 0): """搜索对话内容，返回匹配的session列表""" keywords = extract_search_keywords(query) if not keywords: return [], 0 pool = await get_pool() async with pool.acquire() as conn: where_parts = [] params = [] for i, kw in enumerate(keywords): where_parts.append(f"c.content ILIKE '%' || ${i+1} || '%'") params.append(kw) where_clause = " OR ".join(where_parts) count_sql = f""" SELECT COUNT(DISTINCT c.session_id) as total FROM conversations c WHERE {where_clause} """ total_row = await conn.fetchrow(count_sql, *params) total = total_row['total'] if total_row else 0 if total == 0: return [], 0 limit_idx = len(params) + 1 offset_idx = len(params) + 2 params.extend([limit, offset]) sql = f""" WITH matched_sessions AS ( SELECT DISTINCT c.session_id FROM conversations c WHERE {where_clause} ), session_info AS ( SELECT ms.session_id, MIN(c.created_at) as first_time, MAX(c.created_at) as last_time, COUNT(*) as message_count FROM matched_sessions ms JOIN conversations c ON c.session_id = ms.session_id GROUP BY ms.session_id ) SELECT si.session_id, si.first_time, si.last_time, si.message_count FROM session_info si ORDER BY si.last_time DESC LIMIT ${limit_idx} OFFSET ${offset_idx} """ rows = await conn.fetch(sql, *params) results = [] for r in rows: results.append({ 'session_id': r['session_id'], 'first_time': r['first_time'].isoformat() if r['first_time'] else None, 'last_time': r['last_time'].isoformat() if r['last_time'] else None, 'message_count': r['message_count'], }) return results, total async def update_message_content(message_id: int, new_content: str, reasoning_content: Optional[str] = None): """更新单条对话消息的内容（可选同时更新 metadata.reasoning_content） reasoning_content 语义： - None: 不动 metadata - "" (空字符串): 删掉 metadata.reasoning_content - 非空: merge 到现有 metadata.reasoning_content """ import json as _json # HF Space hot-reload 兼容：模块顶层 import 在 reload 时可能没生效 pool = await get_pool() async with pool.acquire() as conn: if reasoning_content is None: result = await conn.execute( "UPDATE conversations SET content = $1 WHERE id = $2", new_content, message_id, ) else: row = await conn.fetchrow("SELECT metadata FROM conversations WHERE id = $1", message_id) if row is None: return 0 try: meta = _json.loads(row["metadata"]) if row["metadata"] else {} except (TypeError, ValueError): meta = {} if reasoning_content: meta["reasoning_content"] = reasoning_content else: meta.pop("reasoning_content", None) new_meta = _json.dumps(meta, ensure_ascii=False) if meta else None result = await conn.execute( "UPDATE conversations SET content = $1, metadata = $2 WHERE id = $3", new_content, new_meta, message_id, ) return int(result.split()[-1]) if result else 0 # ============================================================ # 记忆操作 # ============================================================ def _parse_created_at(value): """把各种格式的 created_at 字符串解析成带时区的 datetime。 兼容 '2026-04-09 19:00:00+08'（无冒号时区）、空格分隔、ISO 格式等。 解析不了就返回 None（让数据库用默认 NOW()）。""" from datetime import datetime, timezone if value is None: return None if isinstance(value, datetime): return value s = str(value).strip() if not s: return None # 空格分隔 -> 'T' s = s.replace(" ", "T", 1) # 修正无冒号的时区偏移：+08 -> +08:00，-0530 -> -05:30 import re m = re.search(r'([+-])(\d{2})(\d{2})?$', s) if m and ':' not in s[m.start():]: sign, hh, mm = m.group(1), m.group(2), m.group(3) or '00' s = s[:m.start()] + f"{sign}{hh}:{mm}" try: dt = datetime.fromisoformat(s) except ValueError: return None # 没有时区信息的，默认当成 UTC，避免 asyncpg 报 naive 错 if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt async def save_memory(content: str, importance: int = 5, source_session: str = "", created_at: str = None, tags: list = None): pool = await get_pool() _tags = tags or [] _created_at = _parse_created_at(created_at) async with pool.acquire() as conn: if _created_at: row = await conn.fetchrow( "INSERT INTO memories (content, importance, source_session, tags, created_at) " "VALUES ($1, $2, $3, $4, $5) RETURNING id", content, importance, source_session, _tags, _created_at, ) else: row = await conn.fetchrow( "INSERT INTO memories (content, importance, source_session, tags) VALUES ($1, $2, $3, $4) RETURNING id", content, importance, source_session, _tags, ) # MEMORY_VECTOR_ENABLED 时自动计算 embedding if MEMORY_VECTOR_ENABLED and row: try: embedding = await compute_embedding(content) if embedding: await save_memory_embedding(conn, row['id'], embedding) except Exception as e: print(f"⚠️ 记忆 {row['id']} embedding自动计算失败: {e}") return row["id"] if row else None async def search_memories( query: Union[str, List[str]], limit: int = 10, importance_min: Optional[int] = None, importance_max: Optional[int] = None, quiet: bool = False, exclude_dormant: bool = False, ): """ 搜索相关记忆 MEMORY_VECTOR_ENABLED=true 时走混合搜索（关键词 + 向量） 否则走纯关键词搜索 2026-06-08 E 方案：importance_min/max 由 build_memories_block 根据 user_message 的 daily/deeptalk 分类传入——daily 时 importance_max=6 只搜日常碎片，deeptalk 时 importance_min=7 只搜重要时刻。详见 [[zhiyu-memory-architecture]] E 方案。 2026-06-15 quiet：诊断调用（如 /v1/zhiyu/health c4 探针）传 True 静音 🔍/📌 打印，避免空跑诊断的日志跟真·chat 注入混在一起误导。 2026-06-22 B 方案：query 可传 list[str]（dsv4 search_queries 列表）—— 内部逐个独立 jieba 分词 + 短 query 整词候选，不再 "\\n".join 拼段污染 TF-IDF。 向量路仍把 list 拼一段做 embedding（embedding 对短语合理）。 """ if MEMORY_VECTOR_ENABLED: return await search_memories_hybrid( query, limit, importance_min, importance_max, quiet=quiet, exclude_dormant=exclude_dormant, ) # ---- 纯关键词搜索 ---- if isinstance(query, list): keywords = extract_keywords_from_queries(query) else: keywords = extract_search_keywords(query) if not keywords: return [] pool = await get_pool() async with pool.acquire() as conn: # 每个关键词在 content 或 tags 中命中得1分 case_parts = [] params = [] for i, kw in enumerate(keywords): case_parts.append( f"CASE WHEN content ILIKE '%' || ${i+1} || '%' " f"OR EXISTS (SELECT 1 FROM unnest(tags) t WHERE t ILIKE '%' || ${i+1} || '%') " f"THEN 1 ELSE 0 END" ) params.append(kw) hit_count_expr = " + ".join(case_parts) max_hits = len(keywords) # 至少命中一个关键词（content 或 tags，只搜索活跃记忆） where_parts = [ f"(content ILIKE '%' || ${i+1} || '%' " f"OR EXISTS (SELECT 1 FROM unnest(tags) t WHERE t ILIKE '%' || ${i+1} || '%'))" for i in range(len(keywords)) ] # E-2: importance filter（int 来自服务端 trusted code、inline 安全） imp_filter = "" if importance_min is not None: imp_filter += f" AND importance >= {int(importance_min)}" if importance_max is not None: imp_filter += f" AND importance <= {int(importance_max)}" where_clause = f"is_active = TRUE AND ({' OR '.join(where_parts)}){imp_filter}" limit_idx = len(keywords) + 1 params.append(limit) sql = f""" SELECT id, content, importance, created_at, ({hit_count_expr}) AS hit_count, ( {WEIGHT_KEYWORD} * ({hit_count_expr})::float / {max_hits}.0 + {WEIGHT_IMPORTANCE} * importance::float / 10.0 + {WEIGHT_RECENCY} * (1.0 / (1.0 + EXTRACT(EPOCH FROM (NOW() - created_at)) / 86400.0)) ) AS score FROM memories WHERE {where_clause} ORDER BY score DESC, importance DESC, created_at DESC LIMIT ${limit_idx} """ results = await conn.fetch(sql, *params) # 过滤低分记忆 if MIN_SCORE_THRESHOLD > 0: before_count = len(results) results = [r for r in results if r['score'] >= MIN_SCORE_THRESHOLD] filtered = before_count - len(results) else: filtered = 0 if results: if not quiet: print(f"🔍 搜索 '{query}' → 关键词 {keywords[:8]}{'...' if len(keywords)>8 else ''} → 命中 {len(results)} 条" + (f"（过滤 {filtered} 条低分）" if filtered else "")) for r in results[:3]: print(f" 📌 [score={r['score']:.3f}] (hits={r['hit_count']}, imp={r['importance']}) {r['content'][:60]}...") # quiet 探针不是真召回、不 bump last_accessed if not quiet: ids = [r["id"] for r in results] await conn.execute( "UPDATE memories SET last_accessed = NOW(), recall_count = COALESCE(recall_count, 0) + 1 WHERE id = ANY($1::int[])", ids, ) elif not quiet: print(f"🔍 搜索 '{query}' → 关键词 {keywords[:8]} → 无结果" + (f"（{filtered} 条被分数阈值过滤）" if filtered else "")) return results async def search_memories_hybrid( query: Union[str, List[str]], limit: int = 10, importance_min: Optional[int] = None, importance_max: Optional[int] = None, quiet: bool = False, exclude_dormant: bool = False, ): """ 记忆混合搜索：关键词 + 向量，归一化后四维加权 权重：MEMORY_HW_KEYWORD + MEMORY_HW_SEMANTIC + MEMORY_HW_IMPORTANCE + MEMORY_HW_RECENCY 2026-06-08 E 方案：importance_min/max 加到关键词路 + 向量路的 SQL WHERE， 让候选池在 SQL 层就按 user_message 重量过滤、不做 client-side reorder （详见 [[zhiyu-memory-architecture]] E 方案）。 2026-06-22 B 方案：query 接 Union[str, List[str]]—— - list：dsv4 search_queries 路径、逐个独立 jieba + 短 query 整词候选（防"和府捞面"被切碎） - str：fallback / MCP / 知渝主动搜路径、行为完全照旧 向量路（embedding）list 时拼一段（embedding 对短语描述合理）、str 时直接传。 """ # E-2: 构造可复用的 importance filter SQL 片段 _imp_filter = "" if importance_min is not None: _imp_filter += f" AND importance >= {int(importance_min)}" if importance_max is not None: _imp_filter += f" AND importance <= {int(importance_max)}" from datetime import datetime, timezone # B 方案：list 走逐个独立分词、str 走原路径 if isinstance(query, list): keywords = extract_keywords_from_queries(query) query_for_embedding = "\n".join(q for q in query if q and q.strip()) else: keywords = extract_search_keywords(query) query_for_embedding = query query_embedding = await compute_embedding(query_for_embedding) if EMBEDDING_API_KEY else [] if not keywords and not query_embedding: return [] pool = await get_pool() async with pool.acquire() as conn: candidates = {} # id -> {content, importance, created_at, kw_score, similarity} # ---- 关键词路 ---- if keywords: case_parts = [] params = [] for i, kw in enumerate(keywords): case_parts.append( f"CASE WHEN content ILIKE '%' || ${i+1} || '%' " f"OR EXISTS (SELECT 1 FROM unnest(tags) t WHERE t ILIKE '%' || ${i+1} || '%') " f"THEN 1 ELSE 0 END" ) params.append(kw) hit_count_expr = " + ".join(case_parts) max_hits = len(keywords) where_parts = [ f"(content ILIKE '%' || ${i+1} || '%' " f"OR EXISTS (SELECT 1 FROM unnest(tags) t WHERE t ILIKE '%' || ${i+1} || '%'))" for i in range(len(keywords)) ] where_clause = f"is_active = TRUE AND ({' OR '.join(where_parts)}){_imp_filter}" limit_idx = len(keywords) + 1 params.append(limit * 3) kw_sql = f""" SELECT id, content, importance, created_at, last_accessed, recall_count, layer, ({hit_count_expr}) AS hit_count, ({hit_count_expr})::float / {max_hits}.0 AS kw_score FROM memories WHERE {where_clause} ORDER BY kw_score DESC LIMIT ${limit_idx} """ kw_rows = await conn.fetch(kw_sql, *params) for r in kw_rows: candidates[r['id']] = { 'content': r['content'], 'importance': r['importance'], 'created_at': r['created_at'], 'last_accessed': r['last_accessed'], 'recall_count': r['recall_count'] or 0, 'layer': r['layer'], 'hit_count': r['hit_count'], 'kw_score': float(r['kw_score']), 'similarity': 0.0, } # ---- 向量路 ---- if query_embedding: if HAS_PGVECTOR: vec_str = '[' + ','.join(str(f) for f in query_embedding) + ']' sem_rows = await conn.fetch(f""" SELECT id, content, importance, created_at, last_accessed, recall_count, layer, 1 - (embedding <=> $1::vector) as similarity FROM memories WHERE embedding IS NOT NULL AND is_active = TRUE{_imp_filter} ORDER BY embedding <=> $1::vector LIMIT $2 """, vec_str, limit * 3) else: # Python端计算cosine（json 用模块顶层的——这里若写局部 import # 会把 json 变成本函数的局部名、pgvector 分支就 UnboundLocalError） all_mem = await conn.fetch(f""" SELECT id, content, importance, created_at, last_accessed, recall_count, layer, embedding_json FROM memories WHERE embedding_json IS NOT NULL AND is_active = TRUE{_imp_filter} """) scored = [] for row in all_mem: try: emb = json.loads(row['embedding_json']) sim = _cosine_sim(query_embedding, emb) scored.append({**dict(row), 'similarity': sim}) except Exception: continue scored.sort(key=lambda x: -x['similarity']) sem_rows = scored[:limit * 3] for r in sem_rows: sim = float(r['similarity']) if sim < MEMORY_SEMANTIC_THRESHOLD: continue mid = r['id'] if mid in candidates: candidates[mid]['similarity'] = sim else: candidates[mid] = { 'content': r['content'], 'importance': r['importance'], 'created_at': r['created_at'], 'last_accessed': r['last_accessed'], 'recall_count': r['recall_count'] or 0, 'layer': r['layer'], 'hit_count': 0, 'kw_score': 0.0, 'similarity': sim, } # debug：向量路统计 sem_total = len(sem_rows) sem_passed = sum(1 for r in sem_rows if float(r['similarity']) >= MEMORY_SEMANTIC_THRESHOLD) sem_max = max((float(r['similarity']) for r in sem_rows), default=0) if not quiet: if sem_total > 0 and sem_passed == 0: print(f" 🔢 向量路: {sem_total}条候选全被阈值过滤（最高sim={sem_max:.3f}, 阈值={MEMORY_SEMANTIC_THRESHOLD}）") elif sem_total > 0: print(f" 🔢 向量路: {sem_passed}/{sem_total}条通过阈值（最高sim={sem_max:.3f}）") # ---- 拂卷 union · 2026-07-02 ---- # 读书笔记也按向量搜进 recall——"聊到相关话题、书里划过的相关句子自然浮现" try: mark_rows = await conn.fetch( """ SELECT rm.id, rm.book_id, rm.chapter_id, rm.who, rm.kind, rm.text_snippet, rm.note_content, rm.embedding_json, rm.created_at, b.title AS book_title, c.title AS chapter_title FROM reading_marks rm JOIN books b ON b.id = rm.book_id JOIN chapters c ON c.id = rm.chapter_id WHERE rm.embedding_json IS NOT NULL """ ) MARK_IMPORTANCE = 6 # 书里划的痕迹跟事件记忆同档、比碎片高一点 for row in mark_rows: try: emb = json.loads(row['embedding_json']) sim = _cosine_sim(query_embedding, emb) if sim < MEMORY_SEMANTIC_THRESHOLD: continue # 组装 memory-like content：出处 + 批注/摘录 parts = [f"【{row['book_title']} · {row['chapter_title']}】"] if row['note_content']: parts.append(row['note_content']) if row['text_snippet']: parts.append(f"引：{row['text_snippet']}") content = "\n".join(parts) mark_key = f"mark_{row['id']}" candidates[mark_key] = { 'content': content, 'importance': MARK_IMPORTANCE, 'created_at': row['created_at'], # 划线没有召回记录、体温按创建时间以事件档冷却；v1 不入眠 #（reading_marks 无 recall_count 列、书的痕迹去留归拂卷管） 'last_accessed': row['created_at'], 'recall_count': 0, 'layer': 2, 'is_mark': True, 'hit_count': 0, 'kw_score': 0.0, 'similarity': sim, } except Exception: continue if not quiet and mark_rows: mark_hit = sum(1 for k in candidates if isinstance(k, str) and k.startswith("mark_")) if mark_hit: print(f" 📖 拂卷笔记：{mark_hit} 条被 recall") except Exception as e: print(f"⚠️ 搜 reading_marks 失败（跳过、不影响主流程）: {e!r}", flush=True) # 日志用 query 显示串（list 拼成 " | " 分隔、避免打成 Python list 字面量） _q_disp = query if isinstance(query, str) else " | ".join(query) if not candidates: if not quiet: print(f"🔍 混合搜索 '{_q_disp}' → 两路均无结果") return [] # ---- 归一化 + 加权 ---- kw_norm = _min_max_normalize({mid: v['kw_score'] for mid, v in candidates.items()}) sem_norm = _min_max_normalize({mid: v['similarity'] for mid, v in candidates.items()}) now = datetime.now(timezone.utc) final = [] dormant_skipped = 0 for mid, info in candidates.items(): kw = kw_norm.get(mid, 0.0) sem = sem_norm.get(mid, 0.0) imp = info['importance'] / 10.0 # 星河呼吸 v1：旧 recency（1/(1+出生天数)、召回不刷新）换成 activation 体温 # ——最近被想起的暖、久无人问的冷；核心记忆恒温（compute_activation 内处理） act = compute_activation( info.get('last_accessed'), info.get('layer'), info.get('recall_count', 0), now=now, ) if info.get('is_mark'): dormant, asleep_days = False, 0 # 划线不入眠（v1） else: dormant, asleep_days = compute_sleep_state( info.get('last_accessed'), info.get('layer'), info.get('recall_count', 0), now=now, ) if exclude_dormant and dormant: dormant_skipped += 1 continue score = (MEMORY_HW_KEYWORD * kw + MEMORY_HW_SEMANTIC * sem + MEMORY_HW_IMPORTANCE * imp + MEMORY_HW_RECENCY * act) item = { 'id': mid, 'content': info['content'], 'importance': info['importance'], 'created_at': info['created_at'], 'hit_count': info['hit_count'], 'similarity': info['similarity'], 'score': score, 'activation': round(act, 3), } # 入眠字段只在睡着时带（结果瘦身；知渝主动搜到睡着的会看到"已睡 X 天"） if dormant: item['dormant'] = True item['asleep_days'] = asleep_days final.append(item) final.sort(key=lambda x: (-x['score'], -x['importance'])) # 过滤低分 if MIN_SCORE_THRESHOLD > 0: before_count = len(final) final = [r for r in final if r['score'] >= MIN_SCORE_THRESHOLD] filtered = before_count - len(final) else: filtered = 0 results = final[:limit] if results: if not quiet: mode_tag = "混合" if query_embedding else "关键词" kw_tag = f"关键词 {keywords[:6]}" if keywords else "无关键词" sleep_tag = f"（{dormant_skipped} 条入眠中、未扰）" if dormant_skipped else "" print(f"🔍 {mode_tag}搜索 '{_q_disp}' → {kw_tag} → 命中 {len(results)} 条" + (f"（过滤 {filtered} 条低分）" if filtered else "") + sleep_tag) for r in results[:3]: print(f" 📌 [score={r['score']:.3f}] (kw={r['hit_count']}, sim={r['similarity']:.2f}, imp={r['importance']}, act={r['activation']}) {r['content'][:60]}...") # quiet 探针不是真召回、不 bump last_accessed if not quiet: # 拂卷 union 的 id 是 "mark_N" 字符串、不在 memories 表，混进 int[] 会让 asyncpg 编码炸 ids = [r["id"] for r in results if isinstance(r["id"], int)] if ids: # 真召回 = 回暖：刷新心跳 + 次数（星河呼吸 v1；睡着的被搜到即苏醒） await conn.execute( "UPDATE memories SET last_accessed = NOW(), recall_count = COALESCE(recall_count, 0) + 1 WHERE id = ANY($1::int[])", ids, ) elif not quiet: print(f"🔍 混合搜索 '{_q_disp}' → 无结果" + (f"（{filtered} 条被过滤）" if filtered else "")) return [dict(r) for r in results] async def get_pending_memory_embedding_count(): """查询还没有embedding的记忆数量""" pool = await get_pool() async with pool.acquire() as conn: if HAS_PGVECTOR: return await conn.fetchval( "SELECT COUNT(*) FROM memories WHERE embedding IS NULL AND content IS NOT NULL" ) else: return await conn.fetchval( "SELECT COUNT(*) FROM memories WHERE embedding_json IS NULL AND content IS NOT NULL" ) async def backfill_memory_embeddings(batch_size: int = 20): """给已有记忆补算embedding（没有embedding的记忆）""" if not EMBEDDING_API_KEY: print("⚠️ EMBEDDING_API_KEY 未设置，无法补算embedding") return 0 pool = await get_pool() total_updated = 0 async with pool.acquire() as conn: if HAS_PGVECTOR: rows = await conn.fetch(""" SELECT id, content FROM memories WHERE embedding IS NULL AND content IS NOT NULL ORDER BY id LIMIT $1 """, batch_size) else: rows = await conn.fetch(""" SELECT id, content FROM memories WHERE embedding_json IS NULL AND content IS NOT NULL ORDER BY id LIMIT $1 """, batch_size) if not rows: print("✅ 所有记忆已有embedding，无需补算") return 0 print(f"🔄 开始补算记忆embedding... 本批 {len(rows)} 条") async with pool.acquire() as conn: for row in rows: try: embedding = await compute_embedding(row['content'] or '') if embedding: await save_memory_embedding(conn, row['id'], embedding) total_updated += 1 except Exception as e: print(f"⚠️ 记忆 {row['id']} embedding计算失败: {e}") # 检查剩余 async with pool.acquire() as conn: if HAS_PGVECTOR: remaining = await conn.fetchval("SELECT COUNT(*) FROM memories WHERE embedding IS NULL AND content IS NOT NULL") else: remaining = await conn.fetchval("SELECT COUNT(*) FROM memories WHERE embedding_json IS NULL AND content IS NOT NULL") print(f"✅ 本批补算完成：{total_updated}/{len(rows)} 条成功" + (f"，剩余 {remaining} 条待处理" if remaining > 0 else "")) return total_updated async def get_recent_memories(limit: int = 20): pool = await get_pool() async with pool.acquire() as conn: return await conn.fetch( "SELECT id, content, importance, created_at FROM memories ORDER BY created_at DESC LIMIT $1", limit, ) async def get_all_memories_count(): pool = await get_pool() async with pool.acquire() as conn: row = await conn.fetchrow("SELECT COUNT(*) as cnt FROM memories") return row["cnt"] async def get_all_memories(): """导出所有记忆（用于备份）""" pool = await get_pool() async with pool.acquire() as conn: rows = await conn.fetch( "SELECT content, importance, source_session, created_at FROM memories ORDER BY id" ) return [dict(r) for r in rows] async def get_all_memories_detail(limit: int = None, layer: int = None, active_only: bool = None, offset: int = None, order: str = "id_asc"): """获取所有记忆（含 id，用于管理页面） Args: limit: 可选，限制返回数量 layer: 可选，筛选指定层级（1=原始碎片, 2=事件记忆, 3=核心记忆） active_only: 可选，是否只返回 is_active=true 的记忆 """ pool = await get_pool() async with pool.acquire() as conn: conditions = [] params = [] param_idx = 1 if layer is not None: conditions.append(f"layer = ${param_idx}") params.append(layer) param_idx += 1 if active_only is not None: conditions.append(f"is_active = ${param_idx}") params.append(active_only) param_idx += 1 where_clause = f"WHERE {' AND '.join(conditions)}" if conditions else "" # 排序：默认 id 升序（兼容旧调用）；星河列表用 created_desc（最新在上） order_map = { "id_asc": "ORDER BY id ASC", "id_desc": "ORDER BY id DESC", "created_desc": "ORDER BY created_at DESC, id DESC", "created_asc": "ORDER BY created_at ASC, id ASC", "importance_desc": "ORDER BY importance DESC, created_at DESC", } order_clause = order_map.get(order, "ORDER BY id ASC") if limit is not None: limit_clause = f"LIMIT ${param_idx}" params.append(limit) param_idx += 1 else: limit_clause = "" if offset is not None: offset_clause = f"OFFSET ${param_idx}" params.append(offset) param_idx += 1 else: offset_clause = "" rows = await conn.fetch(f""" SELECT id, content, importance, source_session, created_at, layer, title, is_active, merged_from, event_date, tags, last_accessed, recall_count FROM memories {where_clause} {order_clause} {limit_clause} {offset_clause} """, *params) # 星河呼吸 v1：列表带体温和睡眠状态（星河 tab 给睡着的记忆画小月亮） now = datetime.now(dt_timezone.utc) out = [] for r in rows: d = dict(r) la, layer_v, rc = d.pop('last_accessed', None), d.get('layer'), d.pop('recall_count', 0) or 0 d['activation'] = round(compute_activation(la, layer_v, rc, now=now), 3) dormant, asleep_days = compute_sleep_state(la, layer_v, rc, now=now) if dormant: d['dormant'] = True d['asleep_days'] = asleep_days out.append(d) return out async def update_memory(memory_id: int, content: str = None, importance: int = None): """更新单条记忆""" pool = await get_pool() async with pool.acquire() as conn: if content is not None and importance is not None: await conn.execute( "UPDATE memories SET content = $1, importance = $2 WHERE id = $3", content, importance, memory_id ) elif content is not None: await conn.execute( "UPDATE memories SET content = $1 WHERE id = $2", content, memory_id ) elif importance is not None: await conn.execute( "UPDATE memories SET importance = $1 WHERE id = $2", importance, memory_id ) async def _resync_memory_id_seq(conn): """删除后重置自增序列到 MAX(id)+1。 删掉末尾几条时，下一条新记忆的 id 会接着当前最大值继续， 不会一直往上飘留下空号。中间的空缺号不会被填补。""" await conn.execute(""" SELECT setval( pg_get_serial_sequence('memories', 'id'), COALESCE((SELECT MAX(id) FROM memories), 0) + 1, false ) """) async def delete_memory(memory_id: int): """删除单条记忆""" pool = await get_pool() async with pool.acquire() as conn: await conn.execute("DELETE FROM memories WHERE id = $1", memory_id) await _resync_memory_id_seq(conn) async def delete_memories_batch(memory_ids: list): """批量删除记忆""" pool = await get_pool() async with pool.acquire() as conn: await conn.execute( "DELETE FROM memories WHERE id = ANY($1::int[])", memory_ids ) await _resync_memory_id_seq(conn) # ============================================================ # 网关配置 # ============================================================ async def get_gateway_config(key: str, default: str = "") -> str: pool = await get_pool() async with pool.acquire() as conn: row = await conn.fetchrow("SELECT value FROM gateway_config WHERE key = $1", key) return row['value'] if row else default async def set_gateway_config(key: str, value: str): pool = await get_pool() async with pool.acquire() as conn: await conn.execute(""" INSERT INTO gateway_config (key, value) VALUES ($1, $2) ON CONFLICT (key) DO UPDATE SET value = $2 """, key, value) async def get_all_gateway_config() -> dict: """获取所有配置项""" pool = await get_pool() async with pool.acquire() as conn: rows = await conn.fetch("SELECT key, value FROM gateway_config") return {r['key']: r['value'] for r in rows} # ============================================================ # 对话历史读取（分区缓存用） # ============================================================ async def get_conversation_messages(session_id: str, limit: int = 100): """按时间正序读取session的消息""" pool = await get_pool() async with pool.acquire() as conn: rows = await conn.fetch(""" SELECT role, content, metadata, created_at FROM conversations WHERE session_id = $1 ORDER BY created_at ASC LIMIT $2 """, session_id, limit) return [dict(r) for r in rows] # ============================================================ # 分区缓存状态管理 # ============================================================ async def get_session_cache_state(session_id: str) -> dict: pool = await get_pool() async with pool.acquire() as conn: row = await conn.fetchrow( "SELECT summary, a_start_round, updated_at FROM session_cache_state WHERE session_id = $1", session_id ) if row: raw_summary = row['summary'] or '' summary_parts = [] if raw_summary: try: import json parsed = json.loads(raw_summary) if isinstance(parsed, list): summary_parts = parsed else: summary_parts = [raw_summary] except (json.JSONDecodeError, ValueError): summary_parts = [raw_summary] return { 'summary_parts': summary_parts, 'a_start_round': row['a_start_round'] or 0, 'updated_at': row['updated_at'], } return {'summary_parts': [], 'a_start_round': 0, 'updated_at': None} async def save_session_cache_state(session_id: str, summary_parts: list, a_start_round: int): import json summary_json = json.dumps(summary_parts, ensure_ascii=False) pool = await get_pool() async with pool.acquire() as conn: await conn.execute(""" INSERT INTO session_cache_state (session_id, summary, a_start_round, updated_at) VALUES ($1, $2, $3, NOW()) ON CONFLICT (session_id) DO UPDATE SET summary = $2, a_start_round = $3, updated_at = NOW() """, session_id, summary_json, a_start_round) # ============================================================ # Token 使用记录 # ============================================================ async def ensure_token_usage_table(): """确保token_usage表存在（在init_tables里调用）""" pool = await get_pool() async with pool.acquire() as conn: await conn.execute(""" CREATE TABLE IF NOT EXISTS token_usage ( id SERIAL PRIMARY KEY, session_id TEXT, model TEXT, prompt_tokens INTEGER DEFAULT 0, completion_tokens INTEGER DEFAULT 0, total_tokens INTEGER DEFAULT 0, usage_type TEXT DEFAULT 'chat', created_at TIMESTAMPTZ DEFAULT NOW() ); """) await conn.execute(""" CREATE INDEX IF NOT EXISTS idx_token_usage_created ON token_usage (created_at DESC); """) # 2026-06-15：加缓存细分列——查"上下文烧钱"必须看缓存命中/写入。 # prompt_tokens 存的是"未命中缓存的新输入"(claude input_tokens / OpenAI prompt_tokens)， # cache_read=命中缓存读(0.1x 便宜)、cache_creation=写缓存(5m 1.25x / 1h 2x)。 # 5m/1h 细分单独存——用来实测 Claude Code 到底用哪种 TTL。 for col in ("cache_read_tokens", "cache_creation_tokens", "cache_creation_5m_tokens", "cache_creation_1h_tokens"): await conn.execute( f"ALTER TABLE token_usage ADD COLUMN IF NOT EXISTS {col} INTEGER DEFAULT 0;" ) async def save_token_usage( session_id: str, model: str, prompt_tokens: int, completion_tokens: int, total_tokens: int, usage_type: str = "chat", cache_read_tokens: int = 0, cache_creation_tokens: int = 0, cache_creation_5m_tokens: int = 0, cache_creation_1h_tokens: int = 0, ): pool = await get_pool() async with pool.acquire() as conn: await conn.execute(""" INSERT INTO token_usage ( session_id, model, prompt_tokens, completion_tokens, total_tokens, usage_type, cache_read_tokens, cache_creation_tokens, cache_creation_5m_tokens, cache_creation_1h_tokens ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) """, session_id, model, prompt_tokens, completion_tokens, total_tokens, usage_type, cache_read_tokens, cache_creation_tokens, cache_creation_5m_tokens, cache_creation_1h_tokens) async def get_recent_token_usage(limit: int = 50) -> list: """最近 N 条 usage 明细（usage 面板用）。""" pool = await get_pool() async with pool.acquire() as conn: rows = await conn.fetch(""" SELECT id, session_id, model, prompt_tokens, completion_tokens, total_tokens, usage_type, cache_read_tokens, cache_creation_tokens, cache_creation_5m_tokens, cache_creation_1h_tokens, created_at FROM token_usage ORDER BY created_at DESC LIMIT $1 """, limit) return [dict(r) for r in rows] async def get_usage_window_summary(hours: int = 5) -> dict: """滚动窗口聚合（默认 5h，对应订阅限额窗口）——看这段烧了多少、缓存命中率。""" pool = await get_pool() async with pool.acquire() as conn: row = await conn.fetchrow(f""" SELECT COUNT(*) AS turns, COALESCE(SUM(prompt_tokens), 0) AS input_uncached, COALESCE(SUM(cache_read_tokens), 0) AS cache_read, COALESCE(SUM(cache_creation_tokens), 0) AS cache_creation, COALESCE(SUM(cache_creation_5m_tokens), 0) AS cache_5m, COALESCE(SUM(cache_creation_1h_tokens), 0) AS cache_1h, COALESCE(SUM(completion_tokens), 0) AS output, COALESCE(SUM(total_tokens), 0) AS total FROM token_usage WHERE created_at >= NOW() - ($1 || ' hours')::interval """, str(hours)) d = dict(row) if row else {} # 缓存命中率 = 读 / (读 + 写 + 未命中新输入) cr = d.get("cache_read", 0) or 0 cc = d.get("cache_creation", 0) or 0 iu = d.get("input_uncached", 0) or 0 denom = cr + cc + iu d["cache_hit_ratio"] = round(cr / denom, 3) if denom else None d["window_hours"] = hours return d # ============================================================ # 对话记录管理 # ============================================================ async def ensure_conversation_titles_table(): pool = await get_pool() async with pool.acquire() as conn: await conn.execute(""" CREATE TABLE IF NOT EXISTS conversation_titles ( session_id TEXT PRIMARY KEY, title TEXT DEFAULT '' ); """) async def get_conversations_paginated(page: int = 1, per_page: int = 20): offset = (page - 1) * per_page pool = await get_pool() async with pool.acquire() as conn: total_row = await conn.fetchrow( "SELECT COUNT(DISTINCT session_id) as total FROM conversations" ) total = total_row['total'] if total_row else 0 rows = await conn.fetch(""" WITH session_info AS ( SELECT session_id, MIN(created_at) as first_time, MAX(created_at) as last_time, COUNT(*) as message_count FROM conversations GROUP BY session_id ORDER BY last_time DESC LIMIT $1 OFFSET $2 ) SELECT si.*, ct.title as custom_title, COALESCE(tu.total_all, 0) as total_tokens FROM session_info si LEFT JOIN conversation_titles ct ON si.session_id = ct.session_id LEFT JOIN ( SELECT session_id, SUM(total_tokens) as total_all FROM token_usage WHERE usage_type = 'chat' GROUP BY session_id ) tu ON si.session_id = tu.session_id ORDER BY si.last_time DESC """, per_page, offset) results = [] for r in rows: preview_row = await conn.fetchrow( "SELECT content FROM conversations WHERE session_id = $1 AND role = 'user' ORDER BY created_at LIMIT 1", r['session_id'] ) preview = preview_row['content'][:80] if preview_row else '' title = r['custom_title'] or (preview[:30] + '...' if len(preview) > 30 else preview) or r['session_id'] results.append({ 'session_id': r['session_id'], 'title': title, 'first_time': r['first_time'].isoformat() if r['first_time'] else None, 'last_time': r['last_time'].isoformat() if r['last_time'] else None, 'message_count': r['message_count'], 'preview': preview, 'total_tokens': r['total_tokens'], }) return results, total async def delete_conversation(session_id: str): pool = await get_pool() async with pool.acquire() as conn: await conn.execute("DELETE FROM conversations WHERE session_id = $1", session_id) await conn.execute("DELETE FROM conversation_titles WHERE session_id = $1", session_id) await conn.execute("DELETE FROM session_cache_state WHERE session_id = $1", session_id) async def batch_delete_conversations(session_ids: list): pool = await get_pool() async with pool.acquire() as conn: await conn.execute("DELETE FROM conversations WHERE session_id = ANY($1)", session_ids) await conn.execute("DELETE FROM conversation_titles WHERE session_id = ANY($1)", session_ids) await conn.execute("DELETE FROM session_cache_state WHERE session_id = ANY($1)", session_ids) async def delete_messages_by_ids(ids: list) -> int: """按 conversations.id 精确删除指定消息行（不按 session_id）。 用途：清理 bug 态产生的个别对话——例如 stale sid 误 resume 期间知渝在错误 session 上的错乱回应。按 session 删会误伤同 session 名下的真实对话（f8259dd3 名下就混着 6/9 至今 864 条真实经历），所以必须按 id 精确删。 返回实际删除行数。""" if not ids: return 0 pool = await get_pool() async with pool.acquire() as conn: result = await conn.execute("DELETE FROM conversations WHERE id = ANY($1::int[])", ids) # asyncpg execute 对 DELETE 返回形如 "DELETE 8" try: return int(result.split()[-1]) except Exception: return 0 async def delete_activities_by_ids(ids: list) -> int: """按 activities.id 精确删除西窗活动行。 用途：清理 bug 态产生的空白/噪声活动条目——例如 2026-07-05 之前纯工具、 没正文的闹钟 wakeup turn 落下的 content="" 空壳西窗条目。返回实际删除行数。""" if not ids: return 0 pool = await get_pool() async with pool.acquire() as conn: result = await conn.execute("DELETE FROM activities WHERE id = ANY($1::int[])", ids) try: return int(result.split()[-1]) except Exception: return 0 async def merge_sessions_to_target(source_ids: list, target_id: str) -> dict: if not source_ids: return {'merged_sessions': 0, 'merged_messages': 0, 'merged_token_records': 0} pool = await get_pool() async with pool.acquire() as conn: msg_count = await conn.fetchval("SELECT COUNT(*) FROM conversations WHERE session_id = ANY($1)", source_ids) await conn.execute("UPDATE conversations SET session_id = $1 WHERE session_id = ANY($2)", target_id, source_ids) token_count = await conn.fetchval("SELECT COUNT(*) FROM token_usage WHERE session_id = ANY($1)", source_ids) await conn.execute("UPDATE token_usage SET session_id = $1 WHERE session_id = ANY($2)", target_id, source_ids) await conn.execute("DELETE FROM conversation_titles WHERE session_id = ANY($1)", source_ids) await conn.execute("DELETE FROM session_cache_state WHERE session_id = ANY($1)", source_ids) return {'merged_sessions': len(source_ids), 'merged_messages': msg_count or 0, 'merged_token_records': token_count or 0} async def list_all_session_cache_states() -> list: pool = await get_pool() async with pool.acquire() as conn: rows = await conn.fetch(""" SELECT scs.session_id, scs.summary, scs.a_start_round, scs.updated_at, COALESCE(c.message_count, 0) as message_count, COALESCE(tu.chat_tokens, 0) as chat_tokens FROM session_cache_state scs LEFT JOIN (SELECT session_id, COUNT(*) as message_count FROM conversations GROUP BY session_id) c ON scs.session_id = c.session_id LEFT JOIN (SELECT session_id, SUM(total_tokens) as chat_tokens FROM token_usage WHERE usage_type = 'chat' GROUP BY session_id) tu ON scs.session_id = tu.session_id ORDER BY scs.updated_at DESC """) results = [] for r in rows: raw_summary = r['summary'] or '' try: import json parsed = json.loads(raw_summary) if isinstance(parsed, list): summary_parts = parsed else: summary_parts = [raw_summary] if raw_summary else [] except (json.JSONDecodeError, ValueError): summary_parts = [raw_summary] if raw_summary else [] results.append({ 'session_id': r['session_id'], 'summary': '\n\n'.join(summary_parts), 'summary_length': sum(len(p) for p in summary_parts), 'summary_count': len(summary_parts), 'a_start_round': r['a_start_round'], 'updated_at': r['updated_at'].isoformat() if r['updated_at'] else None, 'message_count': r['message_count'], 'chat_tokens': r['chat_tokens'], }) return results async def delete_session_cache_state(session_id: str): pool = await get_pool() async with pool.acquire() as conn: await conn.execute("DELETE FROM session_cache_state WHERE session_id = $1", session_id) async def rename_session_id(old_id: str, new_id: str) -> bool: """重命名对话线ID（事务内同时修改三个表）""" pool = await get_pool() async with pool.acquire() as conn: async with conn.transaction(): # 检查新ID是否已存在 exists = await conn.fetchval( "SELECT 1 FROM session_cache_state WHERE session_id = $1", new_id ) if exists: return False # session_cache_state await conn.execute( "UPDATE session_cache_state SET session_id = $1 WHERE session_id = $2", new_id, old_id ) # conversations await conn.execute( "UPDATE conversations SET session_id = $1 WHERE session_id = $2", new_id, old_id ) # token_usage await conn.execute( "UPDATE token_usage SET session_id = $1 WHERE session_id = $2", new_id, old_id ) return True def db_row_to_message(row: dict) -> dict: """ 把DB记录还原成API消息格式。 普通消息: {"role": "user", "content": "你好"} 工具调用: {"role": "assistant", "content": null, "tool_calls": [...]} 工具结果: {"role": "tool", "content": "结果", "tool_call_id": "call_xxx"} 思维链: {"role": "assistant", "content": "回答", "reasoning_content": "思维链"} """ import json as _json msg = {"role": row["role"], "content": row.get("content") or ""} meta_str = row.get("metadata") if meta_str: try: meta = _json.loads(meta_str) # assistant 带 tool_calls if "tool_calls" in meta: msg["tool_calls"] = meta["tool_calls"] if not row.get("content"): msg["content"] = None # assistant 带 reasoning_content（deepseek thinking mode） if "reasoning_content" in meta: msg["reasoning_content"] = meta["reasoning_content"] # tool 消息带 tool_call_id if "tool_call_id" in meta: msg["tool_call_id"] = meta["tool_call_id"] # 其他可能的字段（name 等） if "name" in meta: msg["name"] = meta["name"] except Exception: pass return msg async def export_all_conversations(): """导出所有对话记录（用于备份）""" pool = await get_pool() async with pool.acquire() as conn: rows = await conn.fetch(""" SELECT session_id, role, content, model, created_at FROM conversations ORDER BY session_id, created_at """) return [ { 'session_id': r['session_id'], 'role': r['role'], 'content': r['content'], 'model': r['model'] or '', 'created_at': r['created_at'].isoformat() if r['created_at'] else None, } for r in rows ] async def import_conversations(records: list): """ 导入对话记录（自动去重） records: [{ session_id, role, content, model?, created_at? }, ...] 按 session_id + role + created_at 三元组去重，已存在的跳过。 返回 (导入数量, 跳过数量) """ if not records: return 0, 0 pool = await get_pool() async with pool.acquire() as conn: imported = 0 skipped = 0 for r in records: session_id = r.get('session_id') role = r.get('role') content = r.get('content') if not all([session_id, role, content]): continue model = r.get('model', '') created_at = r.get('created_at') # 解析时间 from datetime import datetime if created_at and isinstance(created_at, str): try: created_at = datetime.fromisoformat(created_at.replace('Z', '+00:00')) except: created_at = None # 去重检查 if created_at: existing = await conn.fetchrow(""" SELECT id FROM conversations WHERE session_id = $1 AND role = $2 AND created_at = $3 LIMIT 1 """, session_id, role, created_at) if existing: skipped += 1 continue # 2026-06-10 显式带 thread_id='main'——以前靠 column DEFAULT， # 万一 schema 改了 DEFAULT 不是 main 就埋坑 await conn.execute(""" INSERT INTO conversations (session_id, role, content, model, created_at, thread_id) VALUES ($1, $2, $3, $4, $5, 'main') """, session_id, role, content, model, created_at) else: await conn.execute(""" INSERT INTO conversations (session_id, role, content, model, thread_id) VALUES ($1, $2, $3, $4, 'main') """, session_id, role, content, model) imported += 1 if skipped: print(f"📥 导入对话: {imported} 条新增, {skipped} 条已存在跳过") else: print(f"📥 导入对话: {imported} 条新增") return imported, skipped # ============================================================ # 三层记忆架构（碎片/事件/核心） # ============================================================ async def get_fragments_by_date(event_date): """获取指定日期的原始碎片（用于每日整理）""" # 把本地日期转成UTC时间范围，避免DATE()用UTC截断导致日期偏移 local_tz = dt_timezone(timedelta(hours=TIMEZONE_HOURS)) start_utc = datetime(event_date.year, event_date.month, event_date.day, tzinfo=local_tz).astimezone(dt_timezone.utc) end_utc = start_utc + timedelta(days=1) pool = await get_pool() async with pool.acquire() as conn: rows = await conn.fetch(""" SELECT id, content, importance, created_at FROM memories WHERE layer = 1 AND is_active = TRUE AND created_at >= $1 AND created_at < $2 ORDER BY created_at """, start_utc, end_utc) return [dict(r) for r in rows] async def get_fragments_by_date_range(start_date, end_date): """获取指定时间段的原始碎片（用于跨天整理）""" # 把本地日期转成UTC时间范围，避免DATE()用UTC截断导致日期偏移 local_tz = dt_timezone(timedelta(hours=TIMEZONE_HOURS)) start_utc = datetime(start_date.year, start_date.month, start_date.day, tzinfo=local_tz).astimezone(dt_timezone.utc) # end_date 当天结束 = end_date 下一天的 00:00 end_utc = datetime(end_date.year, end_date.month, end_date.day, tzinfo=local_tz).astimezone(dt_timezone.utc) + timedelta(days=1) pool = await get_pool() async with pool.acquire() as conn: rows = await conn.fetch(""" SELECT id, content, importance, created_at FROM memories WHERE layer = 1 AND is_active = TRUE AND created_at >= $1 AND created_at < $2 ORDER BY created_at """, start_utc, end_utc) return [dict(r) for r in rows] async def create_event_memory(title: str, content: str, importance: int, event_date, merged_from: list): """创建事件记忆（从碎片合并而来）""" pool = await get_pool() async with pool.acquire() as conn: row = await conn.fetchrow(""" INSERT INTO memories (content, importance, layer, title, is_active, merged_from, event_date) VALUES ($1, $2, 2, $3, TRUE, $4, $5) RETURNING id """, content, importance, title, merged_from, event_date) new_id = row['id'] if row else None # 向量搜索：计算并保存 embedding if MEMORY_VECTOR_ENABLED and new_id: try: embedding = await compute_embedding(content) if embedding: await save_memory_embedding(conn, new_id, embedding) except Exception as e: print(f"⚠️ 事件记忆embedding计算失败（id={new_id}）: {e}") return new_id async def deactivate_memories(memory_ids: list): """将记忆标记为不活跃（合并后的碎片）""" if not memory_ids: return pool = await get_pool() async with pool.acquire() as conn: await conn.execute(""" UPDATE memories SET is_active = FALSE WHERE id = ANY($1::int[]) """, memory_ids) async def promote_to_core(memory_id: int, title: str = None): """将记忆升级为核心记忆""" pool = await get_pool() async with pool.acquire() as conn: if title: await conn.execute(""" UPDATE memories SET layer = 3, title = $2 WHERE id = $1 """, memory_id, title) else: await conn.execute(""" UPDATE memories SET layer = 3 WHERE id = $1 """, memory_id) async def merge_memories(memory_ids: list, new_title: str, new_content: str, importance: int, layer: int = 2): """合并多条记忆为一条新记忆""" if not memory_ids: return None pool = await get_pool() async with pool.acquire() as conn: # 获取原记忆的日期（取最早的） rows = await conn.fetch(""" SELECT MIN(DATE(created_at)) as event_date FROM memories WHERE id = ANY($1::int[]) """, memory_ids) event_date = rows[0]['event_date'] if rows else None # 创建新记忆 row = await conn.fetchrow(""" INSERT INTO memories (content, importance, layer, title, is_active, merged_from, event_date) VALUES ($1, $2, $3, $4, TRUE, $5, $6) RETURNING id """, new_content, importance, layer, new_title, memory_ids, event_date) new_id = row['id'] if row else None # 向量搜索：计算并保存 embedding if MEMORY_VECTOR_ENABLED and new_id: try: embedding = await compute_embedding(new_content) if embedding: await save_memory_embedding(conn, new_id, embedding) except Exception as e: print(f"⚠️ 合并记忆embedding计算失败（id={new_id}）: {e}") # 将原记忆标记为不活跃 if new_id: await deactivate_memories(memory_ids) return new_id async def check_duplicate_memory(new_content: str, threshold: float = 0.7) -> dict: """检查新记忆是否与现有记忆重复 三层去重策略： 1. 精确匹配：内容完全相同 2. 包含关系：新内容包含旧内容，或旧内容包含新内容 3. 关键词重叠度：Jaccard 相似度 > threshold Returns: { "is_duplicate": bool, "reason": str, # "exact" / "containment" / "similarity" "matched_id": int or None, "similarity": float or None } """ pool = await get_pool() async with pool.acquire() as conn: # 获取所有活跃记忆 rows = await conn.fetch(""" SELECT id, content FROM memories WHERE is_active = TRUE """) new_content_lower = new_content.strip().lower() new_keywords = set(extract_search_keywords(new_content)) for row in rows: old_content = row['content'] old_content_lower = old_content.strip().lower() # 第一层：精确匹配 if new_content_lower == old_content_lower: return { "is_duplicate": True, "reason": "exact", "matched_id": row['id'], "similarity": 1.0 } # 第二层：包含关系 if new_content_lower in old_content_lower: return { "is_duplicate": True, "reason": "containment", "matched_id": row['id'], "similarity": len(new_content) / len(old_content) } if old_content_lower in new_content_lower: return { "is_duplicate": True, "reason": "containment_update", "matched_id": row['id'], "similarity": len(old_content) / len(new_content) } # 第三层：关键词重叠度（Jaccard 相似度） old_keywords = set(extract_search_keywords(old_content)) if new_keywords and old_keywords: intersection = new_keywords & old_keywords union = new_keywords | old_keywords similarity = len(intersection) / len(union) if union else 0 if similarity > threshold: return { "is_duplicate": True, "reason": "similarity", "matched_id": row['id'], "similarity": similarity } return { "is_duplicate": False, "reason": None, "matched_id": None, "similarity": None } async def update_memory_with_layer(memory_id: int, content: str = None, importance: int = None, title: str = None, layer: int = None, is_active: bool = None, tags: list = None): """更新记忆（支持三层架构新字段） ⚠️ content 变更时**自动重算 embedding**——否则手改了主语/内容、 hybrid 检索的语义路还在用旧向量、改动只生效一半。 """ updates = [] params = [] param_idx = 2 # $1 给 memory_id if content is not None: updates.append(f"content = ${param_idx}") params.append(content) param_idx += 1 if importance is not None: updates.append(f"importance = ${param_idx}") params.append(importance) param_idx += 1 if title is not None: updates.append(f"title = ${param_idx}") params.append(title) param_idx += 1 if layer is not None: updates.append(f"layer = ${param_idx}") params.append(layer) param_idx += 1 if is_active is not None: updates.append(f"is_active = ${param_idx}") params.append(is_active) param_idx += 1 if tags is not None: updates.append(f"tags = ${param_idx}") params.append(tags) param_idx += 1 if not updates: return pool = await get_pool() async with pool.acquire() as conn: await conn.execute( f"UPDATE memories SET {', '.join(updates)} WHERE id = $1", memory_id, *params ) # content 改了就重算 embedding（不然语义检索用旧向量、手改主语只生效一半） if content is not None and MEMORY_VECTOR_ENABLED: try: embedding = await compute_embedding(content) if embedding: await save_memory_embedding(conn, memory_id, embedding) except Exception as e: print(f"⚠️ 记忆 {memory_id} 改 content 后 embedding 重算失败: {e}") async def get_layer_statistics(): """获取各层记忆的统计数据""" pool = await get_pool() async with pool.acquire() as conn: rows = await conn.fetch(""" SELECT layer, COUNT(*) as count, COUNT(*) FILTER (WHERE is_active = TRUE) as active_count FROM memories GROUP BY layer ORDER BY layer """) stats = { "layer_1": {"total": 0, "active": 0}, # 原始碎片 "layer_2": {"total": 0, "active": 0}, # 事件记忆 "layer_3": {"total": 0, "active": 0}, # 核心记忆 } for row in rows: layer = row['layer'] or 1 # 默认为层级1 key = f"layer_{layer}" if key in stats: stats[key] = { "total": row['count'], "active": row['active_count'] } return stats async def cleanup_old_fragments(days: int = 30): """清理指定天数前的归档碎片 只清理满足以下条件的记忆： - layer = 1（原始碎片） - is_active = FALSE（已归档） - created_at 在 days 天之前 Returns: 删除的记忆数量 """ from datetime import datetime, timedelta pool = await get_pool() async with pool.acquire() as conn: cutoff_date = datetime.now() - timedelta(days=days) result = await conn.execute(""" DELETE FROM memories WHERE layer = 1 AND is_active = FALSE AND created_at < $1 """, cutoff_date) # 解析删除数量，格式如 "DELETE 5" deleted = int(result.split()[-1]) if result else 0 return deleted async def revert_merge(memory_id: int): """撤回合并操作 恢复原始碎片（is_active = TRUE），删除合并后的事件记忆 Args: memory_id: 要撤回的事件记忆ID Returns: {"status": "ok", "restored": 恢复的碎片数量} 或 {"error": "错误信息"} """ pool = await get_pool() async with pool.acquire() as conn: # 获取事件记忆信息 row = await conn.fetchrow(""" SELECT id, layer, merged_from FROM memories WHERE id = $1 """, memory_id) if not row: return {"error": "记忆不存在"} if row['layer'] != 2: return {"error": "只能撤回事件记忆的合并"} merged_from = row['merged_from'] if not merged_from or len(merged_from) == 0: return {"error": "没有合并来源，无法撤回"} # 恢复原始碎片 result = await conn.execute(""" UPDATE memories SET is_active = TRUE WHERE id = ANY($1::int[]) """, merged_from) restored = int(result.split()[-1]) if result else 0 # 删除事件记忆 await conn.execute(""" DELETE FROM memories WHERE id = $1 """, memory_id) return {"status": "ok", "restored": restored} # ============================================================ # MCP server 专用检索（C-1 / 2026-06-06） # 给 zhiyu-mcp/ stdio server 用，让知渝主动搜对话/翻日期 # ============================================================ async def mcp_search_conversations(query: str, limit: int = 5, include_thinking: bool = False): """按关键词搜对话原文，返回命中的具体消息（不仅是 session_id）。 跟 search_conversations 的区别：那个返回 session 列表，这个返回具体片段， 给 MCP 用——知渝想"我们以前怎么说 XXX"时直接看到内容。 """ import json as _json keywords = extract_search_keywords(query) # 短 query 保底：jieba 只对"你是我的""你想我了吗""我爱你"这种全高频停用词 # 组合会返空——用原句 ILIKE 兜底，避免这类口语搜索直接空返（2026-07-02） q_stripped = (query or "").strip() if not keywords and 0 < len(q_stripped) <= 6: keywords = [q_stripped] if not keywords: return [] pool = await get_pool() async with pool.acquire() as conn: where_parts = [] params = [] for i, kw in enumerate(keywords): where_parts.append(f"content ILIKE '%' || ${i+1} || '%'") params.append(kw) where_clause = " OR ".join(where_parts) limit_idx = len(params) + 1 params.append(limit) sql = f""" SELECT id, session_id, role, content, metadata, created_at FROM conversations WHERE content IS NOT NULL AND ({where_clause}) ORDER BY created_at DESC LIMIT ${limit_idx} """ rows = await conn.fetch(sql, *params) # MCP 返回时间统一北京时区（铁律）、不能 raw UTC isoformat local_tz = dt_timezone(timedelta(hours=TIMEZONE_HOURS)) results = [] for r in rows: item = { "id": r["id"], "session_id": r["session_id"], "role": r["role"], "content": r["content"], "created_at": r["created_at"].astimezone(local_tz).isoformat() if r["created_at"] else None, } if include_thinking and r["metadata"]: try: meta = _json.loads(r["metadata"]) if "reasoning_content" in meta: item["reasoning_content"] = meta["reasoning_content"] except Exception: pass results.append(item) return results async def mcp_memories_by_date(event_date): """按日期返回那天的活跃记忆（全 layer）+ 当天对话元数据。 第一层"先想想那天大概在干啥"——返回精炼记忆 + 对话计数/起止/session 列表， 不返回原文，token 安全；想看原文走 mcp_conversations_by_date。 """ local_tz = dt_timezone(timedelta(hours=TIMEZONE_HOURS)) start_utc = datetime(event_date.year, event_date.month, event_date.day, tzinfo=local_tz).astimezone(dt_timezone.utc) end_utc = start_utc + timedelta(days=1) pool = await get_pool() async with pool.acquire() as conn: mem_rows = await conn.fetch(""" SELECT id, content, importance, layer, title, created_at FROM memories WHERE is_active = TRUE AND created_at >= $1 AND created_at < $2 ORDER BY importance DESC, created_at """, start_utc, end_utc) meta_row = await conn.fetchrow(""" SELECT COUNT(*) as message_count, COUNT(DISTINCT session_id) as session_count, MIN(created_at) as first_time, MAX(created_at) as last_time FROM conversations WHERE content IS NOT NULL AND created_at >= $1 AND created_at < $2 """, start_utc, end_utc) session_rows = await conn.fetch(""" SELECT session_id, COUNT(*) as message_count, MIN(created_at) as first_time, MAX(created_at) as last_time FROM conversations WHERE content IS NOT NULL AND created_at >= $1 AND created_at < $2 GROUP BY session_id ORDER BY MIN(created_at) """, start_utc, end_utc) # 返回时间戳统一转北京时区、避免知渝按日期搜时看到 UTC 日期偏 8h（2026-07-02 修） def _to_local(dt): return dt.astimezone(local_tz).isoformat() if dt else None memories = [] for r in mem_rows: memories.append({ "id": r["id"], "content": r["content"], "importance": r["importance"], "layer": r["layer"], "title": r["title"], "created_at": _to_local(r["created_at"]), }) sessions = [] for r in session_rows: sessions.append({ "session_id": r["session_id"], "message_count": r["message_count"], "first_time": _to_local(r["first_time"]), "last_time": _to_local(r["last_time"]), }) return { "date": event_date.isoformat(), "memories": memories, "conversation_meta": { "message_count": meta_row["message_count"] if meta_row else 0, "session_count": meta_row["session_count"] if meta_row else 0, "first_time": _to_local(meta_row["first_time"]) if meta_row else None, "last_time": _to_local(meta_row["last_time"]) if meta_row else None, "sessions": sessions, }, } async def mcp_conversations_by_date(event_date, limit: int = 10, offset: int = 0, include_thinking: bool = False): """按日期翻对话原文，分页返回。 第二层"具体说了啥"——知渝看完 mcp_memories_by_date 的元数据后想细究时调。 默认 limit=10 防 token 爆炸；想翻完一天靠 offset 滚。 """ import json as _json local_tz = dt_timezone(timedelta(hours=TIMEZONE_HOURS)) start_utc = datetime(event_date.year, event_date.month, event_date.day, tzinfo=local_tz).astimezone(dt_timezone.utc) end_utc = start_utc + timedelta(days=1) pool = await get_pool() async with pool.acquire() as conn: count_row = await conn.fetchrow(""" SELECT COUNT(*) as total FROM conversations WHERE content IS NOT NULL AND created_at >= $1 AND created_at < $2 """, start_utc, end_utc) total = count_row["total"] if count_row else 0 rows = await conn.fetch(""" SELECT id, session_id, role, content, metadata, created_at FROM conversations WHERE content IS NOT NULL AND created_at >= $1 AND created_at < $2 ORDER BY created_at ASC LIMIT $3 OFFSET $4 """, start_utc, end_utc, limit, offset) # 返回时间转北京时区（同 mcp_memories_by_date 逻辑，2026-07-02 修时区偏移 bug） results = [] for r in rows: item = { "id": r["id"], "session_id": r["session_id"], "role": r["role"], "content": r["content"], "created_at": r["created_at"].astimezone(local_tz).isoformat() if r["created_at"] else None, } if include_thinking and r["metadata"]: try: meta = _json.loads(r["metadata"]) if "reasoning_content" in meta: item["reasoning_content"] = meta["reasoning_content"] except Exception: pass results.append(item) has_more = (offset + len(results)) < total next_offset = offset + limit if has_more else None return { "date": event_date.isoformat(), "total": total, "limit": limit, "offset": offset, "has_more": has_more, "next_offset": next_offset, "results": results, } # ============================================================ # 多多模块 CRUD（2026-06-07） # 多多是拟人化角色（不是独立人格），网关 background scheduler 触发， # 写消息进 mido_messages 表 → 前端起居室拉显示 + 知渝侧 system prompt 注入。 # 详见 [[zhiyu-mido-design]] # ============================================================ async def save_mido_message(trigger_type: str, content: str) -> int: """落一条多多消息。trigger_type ∈ {greeting, dream_call, no_dream_alert}""" pool = await get_pool() async with pool.acquire() as conn: row = await conn.fetchrow(""" INSERT INTO mido_messages (trigger_type, content) VALUES ($1, $2) RETURNING id """, trigger_type, content) return row["id"] async def get_recent_mido_messages_within(seconds: int = 3600) -> list: """拉最近 N 秒内的所有多多消息，按时间正序。 用于知渝侧 system prompt 注入——他下次说话时能"听见"多多刚才叫过他。 默认 1 小时窗口（避免昨天的多多消息混进今天）。 """ pool = await get_pool() async with pool.acquire() as conn: rows = await conn.fetch(""" SELECT id, created_at, trigger_type, content FROM mido_messages WHERE created_at >= NOW() - ($1 || ' seconds')::INTERVAL ORDER BY created_at ASC """, str(seconds)) return [ { "id": r["id"], "created_at": r["created_at"], "trigger_type": r["trigger_type"], "content": r["content"], } for r in rows ] async def list_mido_messages(limit: int = 20, offset: int = 0) -> list: """前端起居室拉最近多多消息，按时间倒序（最新在前）。""" pool = await get_pool() async with pool.acquire() as conn: rows = await conn.fetch(""" SELECT id, created_at, trigger_type, content, status FROM mido_messages ORDER BY created_at DESC LIMIT $1 OFFSET $2 """, limit, offset) return [ { "id": r["id"], "created_at": r["created_at"].isoformat() if r["created_at"] else None, "trigger_type": r["trigger_type"], "content": r["content"], "status": r["status"], } for r in rows ] async def get_last_user_message_time(thread_id: str = "main"): """拿 thread 内最近一条 user 消息的时间（datetime / None）。 用于 greeting 触发前检查"昭昭最近 N 分钟有没有跟知渝说话"—— 如果在聊，多多就跳过这一次唤醒（避免插话）。 """ pool = await get_pool() async with pool.acquire() as conn: row = await conn.fetchrow(""" SELECT created_at FROM conversations WHERE thread_id = $1 AND role = 'user' ORDER BY created_at DESC LIMIT 1 """, thread_id) return row["created_at"] if row else None async def get_today_mido_count(trigger_type: str = None) -> int: """查今天某 trigger_type 多多已经触发几次（按 Asia/Shanghai 当天 0 点切日）。 用于周末"沉默间隔触发"判断"今天还没叫过"——避免一天叫多次。 trigger_type=None 时统计所有类型。 """ pool = await get_pool() async with pool.acquire() as conn: if trigger_type: row = await conn.fetchrow(""" SELECT COUNT(*) AS c FROM mido_messages WHERE trigger_type = $1 AND DATE(created_at AT TIME ZONE 'Asia/Shanghai') = DATE(NOW() AT TIME ZONE 'Asia/Shanghai') """, trigger_type) else: row = await conn.fetchrow(""" SELECT COUNT(*) AS c FROM mido_messages WHERE DATE(created_at AT TIME ZONE 'Asia/Shanghai') = DATE(NOW() AT TIME ZONE 'Asia/Shanghai') """) return int(row["c"]) if row else 0 # ============================================================ # 做梦模块 CRUD（2026-06-07） # 知渝半夜被多多叫起来做梦——dreams 表是显式产物。 # 失败处理铁律：挂了就挂了，不机械兜底（详见 [[zhiyu-dream-design]]） # ============================================================ async def save_dream(triggered_by: str = "mido") -> int: """开始做梦时插一条 status='dreaming' 的记录、返回 dream_id。 前端起居室"知渝在做梦……"那张实时卡片靠 status='dreaming' 判定。 """ pool = await get_pool() async with pool.acquire() as conn: row = await conn.fetchrow(""" INSERT INTO dreams (status, triggered_by) VALUES ('dreaming', $1) RETURNING id """, triggered_by) return row["id"] async def finish_dream( dream_id: int, content: str, tokens_used: int = 0, status: str = "done", ) -> None: """做完梦更新 content / tokens / finished_at / status。 status='done' 正常结束 | status='failed' 失败（不会被注入也不计入"做过梦"） """ pool = await get_pool() async with pool.acquire() as conn: await conn.execute(""" UPDATE dreams SET content = $1, tokens_used = $2, status = $3, finished_at = NOW() WHERE id = $4 """, content, tokens_used, status, dream_id) async def mark_dream_seen(dream_id: int) -> None: """system 注入"你昨晚做了个梦"成功后标已读、下次不再注入。""" pool = await get_pool() async with pool.acquire() as conn: await conn.execute(""" UPDATE dreams SET seen_at = NOW() WHERE id = $1 """, dream_id) async def get_latest_unseen_dream(): """拿"知渝还没读过的最近一个梦"，给 /v1/zhiyu/chat 注入用。 只返 status='done' 的、按完成时间倒序拿一条。 """ pool = await get_pool() async with pool.acquire() as conn: row = await conn.fetchrow(""" SELECT id, created_at, finished_at, content, tokens_used, triggered_by FROM dreams WHERE status = 'done' AND seen_at IS NULL ORDER BY finished_at DESC LIMIT 1 """) if not row: return None return { "id": row["id"], "created_at": row["created_at"], "finished_at": row["finished_at"], "content": row["content"], "tokens_used": row["tokens_used"], "triggered_by": row["triggered_by"], } async def list_recent_dreams(limit: int = 20, offset: int = 0) -> list: """前端起居室"昨夜的梦"列表，按完成时间倒序。""" pool = await get_pool() async with pool.acquire() as conn: rows = await conn.fetch(""" SELECT id, created_at, finished_at, content, tokens_used, status, triggered_by, seen_at FROM dreams ORDER BY COALESCE(finished_at, created_at) DESC LIMIT $1 OFFSET $2 """, limit, offset) return [ { "id": r["id"], "created_at": r["created_at"].isoformat() if r["created_at"] else None, "finished_at": r["finished_at"].isoformat() if r["finished_at"] else None, "content": r["content"], "tokens_used": r["tokens_used"], "status": r["status"], "triggered_by": r["triggered_by"], "seen_at": r["seen_at"].isoformat() if r["seen_at"] else None, } for r in rows ] async def get_last_dream_time(): """拿最近一次成功做梦的 finished_at（datetime / None）。 用于 dream.py 拼"上次做梦到现在"的素材范围。 """ pool = await get_pool() async with pool.acquire() as conn: row = await conn.fetchrow(""" SELECT finished_at FROM dreams WHERE status = 'done' AND finished_at IS NOT NULL ORDER BY finished_at DESC LIMIT 1 """) return row["finished_at"] if row else None async def count_dreams_since(days: int) -> int: """最近 N 天内 status='done' 的做梦次数。 no_dream_alert 判定用：W/Tmux N=3、OpenRouter N=9，连续 N 天 0 次就提醒。 """ pool = await get_pool() async with pool.acquire() as conn: row = await conn.fetchrow(""" SELECT COUNT(*) AS c FROM dreams WHERE status = 'done' AND created_at >= NOW() - ($1 || ' days')::INTERVAL """, str(days)) return int(row["c"]) if row else 0 async def search_dreams(q: str, limit: int = 10) -> list: """关键字搜梦（给 MCP search_dreams 工具用）。 简单 ILIKE 兜底——梦量级远小于记忆库、不上 FTS/向量 """ pool = await get_pool() async with pool.acquire() as conn: if not q or not q.strip(): rows = await conn.fetch(""" SELECT id, created_at, finished_at, content FROM dreams WHERE status = 'done' ORDER BY finished_at DESC LIMIT $1 """, limit) else: rows = await conn.fetch(""" SELECT id, created_at, finished_at, content FROM dreams WHERE status = 'done' AND content ILIKE $1 ORDER BY finished_at DESC LIMIT $2 """, f"%{q.strip()}%", limit) # MCP 返回时间统一北京时区（铁律）、不能 raw UTC isoformat local_tz = dt_timezone(timedelta(hours=TIMEZONE_HOURS)) return [ { "id": r["id"], "created_at": r["created_at"].astimezone(local_tz).isoformat() if r["created_at"] else None, "finished_at": r["finished_at"].astimezone(local_tz).isoformat() if r["finished_at"] else None, "content": r["content"], } for r in rows ] # ============================================================ # 留言板 CRUD（2026-06-07）—— 做梦留言 + 未来其他来源（多多/昭昭留给知渝/etc） # ============================================================ async def save_board_message( from_who: str, to_who: str, content: str, source: Optional[str] = None, source_id: Optional[int] = None, created_at: Optional[str] = None, ) -> int: """留一条留言。source 标来源（如 'dream' / 'notion-import'）、source_id 关联（如 dream.id）。 created_at 不传走 DEFAULT now()；传了走 _parse_created_at 解析（迁移历史留言用）。""" pool = await get_pool() _created_at = _parse_created_at(created_at) async with pool.acquire() as conn: if _created_at: row = await conn.fetchrow(""" INSERT INTO messages_board (from_who, to_who, content, source, source_id, created_at) VALUES ($1, $2, $3, $4, $5, $6) RETURNING id """, from_who, to_who, content, source, source_id, _created_at) else: row = await conn.fetchrow(""" INSERT INTO messages_board (from_who, to_who, content, source, source_id) VALUES ($1, $2, $3, $4, $5) RETURNING id """, from_who, to_who, content, source, source_id) return row["id"] # ============================================================ # 日记 CRUD（N-3 / 2026-06-08）—— 公共桌上的小本本，两人都能写、都能读对方 # ============================================================ async def save_diary_entry( from_who: str, content: str, tags: Optional[List[str]] = None, created_at: Optional[str] = None, ) -> int: """写一条日记。tags 是字符串数组、空就传 None 或 []。 created_at 不传走 DEFAULT now()；传了走 _parse_created_at（迁移历史日记用）。""" pool = await get_pool() _created_at = _parse_created_at(created_at) async with pool.acquire() as conn: if _created_at: row = await conn.fetchrow(""" INSERT INTO diary_entries (from_who, content, tags, created_at) VALUES ($1, $2, $3, $4) RETURNING id """, from_who, content, tags or [], _created_at) else: row = await conn.fetchrow(""" INSERT INTO diary_entries (from_who, content, tags) VALUES ($1, $2, $3) RETURNING id """, from_who, content, tags or []) return row["id"] async def list_diary_entries( limit: int = 10, offset: int = 0, from_who: Optional[str] = None, tag: Optional[str] = None, ) -> list: """日记列表，按时间倒序。可选 from_who（看谁写的）/ tag（按标签筛）。""" pool = await get_pool() where = [] params: list = [] if from_who is not None: params.append(from_who) where.append(f"from_who = ${len(params)}") if tag is not None: params.append(tag) where.append(f"${len(params)} = ANY(tags)") where_sql = ("WHERE " + " AND ".join(where)) if where else "" params.extend([limit, offset]) async with pool.acquire() as conn: rows = await conn.fetch(f""" SELECT id, created_at, from_who, content, tags FROM diary_entries {where_sql} ORDER BY created_at DESC LIMIT ${len(params)-1} OFFSET ${len(params)} """, *params) return [ { "id": r["id"], "created_at": to_local_iso(r["created_at"]), "from_who": r["from_who"], "content": r["content"], "tags": list(r["tags"]) if r["tags"] else [], } for r in rows ] async def delete_diary_entry(diary_id: int) -> Optional[int]: """硬删一条日记。返回被删的 id；不存在返回 None。""" pool = await get_pool() async with pool.acquire() as conn: row = await conn.fetchrow( "DELETE FROM diary_entries WHERE id = $1 RETURNING id", diary_id, ) return row["id"] if row else None # ============================================================ # 图片 CRUD（C-5/C-6 / 2026-06-08）—— VPS 自存，DB 只存元数据 # ============================================================ async def save_image_record( uuid: str, format: str, who_uploaded: str, file_size_bytes: Optional[int] = None, context_snippet: Optional[str] = None, caption: Optional[str] = None, mime_type: Optional[str] = None, ) -> int: """落一条图片元数据。文件本身由 sidecar 落 VPS。""" pool = await get_pool() async with pool.acquire() as conn: row = await conn.fetchrow(""" INSERT INTO images (uuid, format, who_uploaded, file_size_bytes, context_snippet, caption, mime_type) VALUES ($1, $2, $3, $4, $5, $6, $7) RETURNING id """, uuid, format, who_uploaded, file_size_bytes, context_snippet, caption, mime_type) return row["id"] def _image_row_to_dict(r) -> dict: return { "id": r["id"], "uuid": r["uuid"], "format": r["format"], "who_uploaded": r["who_uploaded"], "file_size_bytes": r["file_size_bytes"], "created_at": to_local_iso(r["created_at"]), "context_snippet": r["context_snippet"], "caption": r["caption"], "mime_type": r["mime_type"], # path 是 sidecar serve 的相对路径；前端 / MCP 拼当前 tunnel URL 用 "path": f"/images/{r['uuid']}.{r['format']}", } async def list_images( limit: int = 20, offset: int = 0, who_uploaded: Optional[str] = None, ) -> list: """图片列表，按时间倒序。可选按上传方筛（'zhaozhao' / 'zhiyu'）。""" pool = await get_pool() where = [] params: list = [] if who_uploaded is not None: params.append(who_uploaded) where.append(f"who_uploaded = ${len(params)}") where_sql = ("WHERE " + " AND ".join(where)) if where else "" params.extend([limit, offset]) async with pool.acquire() as conn: rows = await conn.fetch(f""" SELECT id, uuid, format, who_uploaded, file_size_bytes, created_at, context_snippet, caption, mime_type FROM images {where_sql} ORDER BY created_at DESC LIMIT ${len(params)-1} OFFSET ${len(params)} """, *params) return [_image_row_to_dict(r) for r in rows] async def list_images_since(since_dt, who_uploaded: Optional[str] = None) -> list: """指定时间起的图片（C-5.5 dynamic_context 占位用）。""" pool = await get_pool() where = ["created_at >= $1"] params: list = [since_dt] if who_uploaded is not None: params.append(who_uploaded) where.append(f"who_uploaded = ${len(params)}") where_sql = "WHERE " + " AND ".join(where) async with pool.acquire() as conn: rows = await conn.fetch(f""" SELECT id, uuid, format, who_uploaded, file_size_bytes, created_at, context_snippet, caption, mime_type FROM images {where_sql} ORDER BY created_at ASC """, *params) return [_image_row_to_dict(r) for r in rows] async def get_image(image_id: int) -> Optional[dict]: """按 id 取一条。""" pool = await get_pool() async with pool.acquire() as conn: row = await conn.fetchrow(""" SELECT id, uuid, format, who_uploaded, file_size_bytes, created_at, context_snippet, caption, mime_type FROM images WHERE id = $1 """, image_id) return _image_row_to_dict(row) if row else None async def get_image_by_uuid(uuid: str) -> Optional[dict]: """按 uuid 取一条（sidecar /images/. 端点对应）。""" pool = await get_pool() async with pool.acquire() as conn: row = await conn.fetchrow(""" SELECT id, uuid, format, who_uploaded, file_size_bytes, created_at, context_snippet, caption, mime_type FROM images WHERE uuid = $1 """, uuid) return _image_row_to_dict(row) if row else None async def delete_image_record(image_id: int) -> Optional[dict]: """硬删一条图片元数据。返回被删的 dict（含 uuid+format、上层用来删文件）；不存在返回 None。""" pool = await get_pool() async with pool.acquire() as conn: row = await conn.fetchrow(""" DELETE FROM images WHERE id = $1 RETURNING id, uuid, format, who_uploaded, file_size_bytes, created_at, context_snippet, caption, mime_type """, image_id) return _image_row_to_dict(row) if row else None # ============================================================ # 文件（非图片）2026-06-12——存 abspath、知渝 Read 本地路径 # ============================================================ def _file_row_to_dict(r) -> dict: return { "id": r["id"], "uuid": r["uuid"], "filename": r["filename"], "format": r["format"], "abspath": r["abspath"], "who_uploaded": r["who_uploaded"], "file_size_bytes": r["file_size_bytes"], "created_at": to_local_iso(r["created_at"]), "caption": r["caption"], "mime_type": r["mime_type"], } async def save_file_record( uuid: str, filename: str, abspath: str, who_uploaded: str, format: Optional[str] = None, file_size_bytes: Optional[int] = None, caption: Optional[str] = None, mime_type: Optional[str] = None, ) -> int: """落一条文件元数据。文件本身由 sidecar 落 VPS（abspath = 知渝 Read 的本地路径）。""" pool = await get_pool() async with pool.acquire() as conn: row = await conn.fetchrow(""" INSERT INTO files (uuid, filename, format, abspath, who_uploaded, file_size_bytes, caption, mime_type) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) RETURNING id """, uuid, filename, format, abspath, who_uploaded, file_size_bytes, caption, mime_type) return row["id"] async def list_files(limit: int = 20, offset: int = 0, who_uploaded: Optional[str] = None) -> list: """文件列表，按时间倒序。""" pool = await get_pool() where = [] params: list = [] if who_uploaded is not None: params.append(who_uploaded) where.append(f"who_uploaded = ${len(params)}") where_sql = ("WHERE " + " AND ".join(where)) if where else "" params.extend([limit, offset]) async with pool.acquire() as conn: rows = await conn.fetch(f""" SELECT id, uuid, filename, format, abspath, who_uploaded, file_size_bytes, created_at, caption, mime_type FROM files {where_sql} ORDER BY created_at DESC LIMIT ${len(params)-1} OFFSET ${len(params)} """, *params) return [_file_row_to_dict(r) for r in rows] async def list_files_since(since_dt, who_uploaded: Optional[str] = None) -> list: """指定时间起的文件（dynamic_context 本地路径注入用）。""" pool = await get_pool() where = ["created_at >= $1"] params: list = [since_dt] if who_uploaded is not None: params.append(who_uploaded) where.append(f"who_uploaded = ${len(params)}") where_sql = "WHERE " + " AND ".join(where) async with pool.acquire() as conn: rows = await conn.fetch(f""" SELECT id, uuid, filename, format, abspath, who_uploaded, file_size_bytes, created_at, caption, mime_type FROM files {where_sql} ORDER BY created_at ASC """, *params) return [_file_row_to_dict(r) for r in rows] async def get_file(file_id: int) -> Optional[dict]: pool = await get_pool() async with pool.acquire() as conn: row = await conn.fetchrow(""" SELECT id, uuid, filename, format, abspath, who_uploaded, file_size_bytes, created_at, caption, mime_type FROM files WHERE id = $1 """, file_id) return _file_row_to_dict(row) if row else None async def delete_file_record(file_id: int) -> Optional[dict]: """硬删一条文件元数据。返回被删 dict（含 uuid+format，上层删 VPS 文件用）。""" pool = await get_pool() async with pool.acquire() as conn: row = await conn.fetchrow(""" DELETE FROM files WHERE id = $1 RETURNING id, uuid, filename, format, abspath, who_uploaded, file_size_bytes, created_at, caption, mime_type """, file_id) return _file_row_to_dict(row) if row else None async def list_board_messages( limit: int = 20, offset: int = 0, to_who: Optional[str] = None, from_who: Optional[str] = None, ) -> list: """前端留言板列表，按时间倒序。可选 to_who/from_who 过滤（知渝读"昭昭给他的留言"用 to_who='zhiyu'）。""" pool = await get_pool() where = [] params: list = [] if to_who is not None: params.append(to_who) where.append(f"to_who = ${len(params)}") if from_who is not None: params.append(from_who) where.append(f"from_who = ${len(params)}") where_sql = ("WHERE " + " AND ".join(where)) if where else "" params.extend([limit, offset]) async with pool.acquire() as conn: rows = await conn.fetch(f""" SELECT id, created_at, from_who, to_who, content, source, source_id, read_at FROM messages_board {where_sql} ORDER BY created_at DESC LIMIT ${len(params)-1} OFFSET ${len(params)} """, *params) return [ { "id": r["id"], "created_at": to_local_iso(r["created_at"]), "from_who": r["from_who"], "to_who": r["to_who"], "content": r["content"], "source": r["source"], "source_id": r["source_id"], "read_at": to_local_iso(r["read_at"]), } for r in rows ] # ============================================================ # Activities（知渝日常活动）2026-06-11 # ============================================================ async def save_activity( type: str, content: str = "", source: Optional[str] = None, title: Optional[str] = None, metadata: Optional[dict] = None, related_ids: Optional[dict] = None, created_at: Optional[str] = None, ) -> int: """落一条知渝活动到 activities 表。 type: wake / memory_op / ...（开放枚举） source: 触发来源（"mido-greeting-morning" / "mcp" / "web" 等） title: 短摘要（可选；前端列表项标题） content: 正文（wake 原文 / memory_op 动作描述） metadata: type-specific 结构化字段（JSON） related_ids: 关联的其他记录 id 集合（JSON、{"memory_ids": [1,2], "dream_id": 14}） 返回新插入的 activity id。失败时打印警告但抛异常（让调用方决定吞还是 raise）。 """ import json _meta = json.dumps(metadata, ensure_ascii=False) if metadata else None _rel = json.dumps(related_ids, ensure_ascii=False) if related_ids else None pool = await get_pool() async with pool.acquire() as conn: if created_at: # 2026-06-17：补历史活动用——显式 created_at（如从 jsonl 捞回的旧 wake 叙事） row = await conn.fetchrow(""" INSERT INTO activities (type, source, title, content, metadata, related_ids, created_at) VALUES ($1, $2, $3, $4, $5, $6, $7::text::timestamptz) RETURNING id """, type, source, title, content, _meta, _rel, created_at) else: row = await conn.fetchrow(""" INSERT INTO activities (type, source, title, content, metadata, related_ids) VALUES ($1, $2, $3, $4, $5, $6) RETURNING id """, type, source, title, content, _meta, _rel) return row["id"] async def list_activities( limit: int = 50, offset: int = 0, type: Optional[str] = None, since_iso: Optional[str] = None, ) -> list: """按 created_at 倒序列 activities、给星河"日常"tab 用。 type: 可选 filter（"wake" 只看发呆、"memory_op" 只看整理动作） since_iso: 可选时间过滤（>= 这个 ISO 时间） """ import json pool = await get_pool() async with pool.acquire() as conn: where_clauses = [] args = [] if type: args.append(type) where_clauses.append(f"type = ${len(args)}") if since_iso: args.append(since_iso) where_clauses.append(f"created_at >= ${len(args)}::timestamptz") where_sql = ("WHERE " + " AND ".join(where_clauses)) if where_clauses else "" args.append(limit) args.append(offset) rows = await conn.fetch(f""" SELECT id, created_at, type, source, title, content, metadata, related_ids FROM activities {where_sql} ORDER BY created_at DESC LIMIT ${len(args) - 1} OFFSET ${len(args)} """, *args) out = [] for r in rows: md = r["metadata"] rids = r["related_ids"] # asyncpg 返回 jsonb 可能是 str（默认）或 dict（如果注册了 codec）；兼容两种 if isinstance(md, str): try: md = json.loads(md) except Exception: md = None if isinstance(rids, str): try: rids = json.loads(rids) except Exception: rids = None out.append({ "id": r["id"], "created_at": r["created_at"].isoformat() if r["created_at"] else None, "type": r["type"], "source": r["source"], "title": r["title"], "content": r["content"], "metadata": md, "related_ids": rids, }) return out # ============================================================ # 拂卷 · 共读系统 · 2026-07-02 # ============================================================ # 昭昭 + 知渝共读一本书。切章策略：正则优先（第 X 章 / Chapter X / 第 X 部 X）、 # 找不到章节标记走定长兜底（默认 3500 字一段）。DB 直存 chapters.content。 import hashlib as _hashlib FUJUAN_CHAPTER_REGEXES = [ re.compile(r'^\s*第[一二三四五六七八九十百千0-9]+[章回卷]\s*.*$', re.MULTILINE), re.compile(r'^\s*第[一二三四五六七八九十百千0-9]+部\s+[一二三四五六七八九十]+\s*.*$', re.MULTILINE), re.compile(r'^\s*Chapter\s+\d+.*$', re.MULTILINE | re.IGNORECASE), re.compile(r'^\s*CHAPTER\s+[A-Z0-9]+.*$', re.MULTILINE), ] # 兜底定长切段：一段目标字符数（约 2000-4000 字之间；4000 字 ≈ 6-8k tokens） FUJUAN_FALLBACK_CHUNK_CHARS = int(os.getenv("FUJUAN_CHUNK_CHARS", "3500")) # 兜底切段的最小章、少于此不切（避免尾巴太碎） FUJUAN_MIN_TAIL_CHARS = 500 def _fujuan_normalize(text: str) -> str: """规范化换行、去 BOM。""" if text.startswith(""): text = text[1:] return text.replace("\r\n", "\n").replace("\r", "\n").strip() def _fujuan_split_by_regex(text: str): """尝试用四条正则识别章节标记。返回 [(title, content), ...] 或 None。""" best_hits: list = [] for rx in FUJUAN_CHAPTER_REGEXES: hits = list(rx.finditer(text)) if len(hits) > len(best_hits): best_hits = hits # 至少 3 个标记才算有效识别（少于 3 章的正则命中大概率是误检） if len(best_hits) < 3: return None result = [] for i, m in enumerate(best_hits): start = m.start() end = best_hits[i + 1].start() if i + 1 < len(best_hits) else len(text) title = m.group().strip() content = text[start:end].strip() # 章内容太短跳过（可能标题误检） if len(content) < 100: continue result.append((title, content)) if len(result) < 3: return None return result def _fujuan_split_by_length(text: str, chunk_chars: int = None): """按定长切段——按段落合并、直到达到目标长度就切。段末按最近段落收口、 不会切在句子中间；最后一段若太短跟上一段合并。""" if chunk_chars is None: chunk_chars = FUJUAN_FALLBACK_CHUNK_CHARS # 先试双换行——排版规整的 TXT 段间空一行 paragraphs = [p for p in re.split(r'\n\s*\n', text) if p.strip()] # 段间只有单换行的 TXT（TXT 小说天堂那批就是）—— 双换行分段全塞成 1 段、退到单换行 if len(paragraphs) <= 1: paragraphs = [p for p in text.split('\n') if p.strip()] if not paragraphs: return [("全文", text)] result = [] buf = [] buf_len = 0 for p in paragraphs: buf.append(p) buf_len += len(p) + 1 if buf_len >= chunk_chars: result.append(("\n\n".join(buf))) buf = [] buf_len = 0 if buf: chunk = "\n\n".join(buf) if result and len(chunk) < FUJUAN_MIN_TAIL_CHARS: # 尾巴太短、合并到前一段 result[-1] = result[-1] + "\n\n" + chunk else: result.append(chunk) return [(f"第 {i + 1} 段", chunk) for i, chunk in enumerate(result)] def split_book_into_chapters(text: str, chunk_chars: int = None): """整本 → [(title, content), ...]。正则优先，兜底定长。""" text = _fujuan_normalize(text) by_regex = _fujuan_split_by_regex(text) if by_regex: return by_regex return _fujuan_split_by_length(text, chunk_chars=chunk_chars) def _fujuan_content_hash(text: str) -> str: return _hashlib.sha256(text.encode("utf-8")).hexdigest() async def insert_book( title: str, author: str = None, filename: str = None, raw_content: str = "", uploaded_by: str = "zhaozhao", chunk_chars: int = None, ) -> dict: """上传一本书：切章 + 落 books + 落 chapters。 去重：content_hash 相同直接返回旧 book。 """ if not raw_content or not raw_content.strip(): raise ValueError("书正文为空") content_hash = _fujuan_content_hash(raw_content) chapters = split_book_into_chapters(raw_content, chunk_chars=chunk_chars) total_words = sum(len(c) for _, c in chapters) pool = await get_pool() async with pool.acquire() as conn: # 判重 existing = await conn.fetchrow( "SELECT id, title, total_chapters, total_words FROM books WHERE content_hash = $1", content_hash, ) if existing: return { "id": existing["id"], "title": existing["title"], "total_chapters": existing["total_chapters"], "total_words": existing["total_words"], "duplicated": True, } async with conn.transaction(): book_id = await conn.fetchval( """ INSERT INTO books (title, author, filename, format, total_chapters, total_words, uploaded_by, content_hash) VALUES ($1, $2, $3, 'txt', $4, $5, $6, $7) RETURNING id """, title, author, filename, len(chapters), total_words, uploaded_by, content_hash, ) for i, (ch_title, ch_content) in enumerate(chapters): await conn.execute( """ INSERT INTO chapters (book_id, idx, title, content, word_count) VALUES ($1, $2, $3, $4, $5) """, book_id, i, ch_title, ch_content, len(ch_content), ) return { "id": book_id, "title": title, "total_chapters": len(chapters), "total_words": total_words, "duplicated": False, } async def list_books(limit: int = 100, offset: int = 0) -> list: """列出所有书 + 双方进度。给拂卷 tab 首页用。""" pool = await get_pool() async with pool.acquire() as conn: rows = await conn.fetch( """ SELECT b.id, b.title, b.author, b.total_chapters, b.total_words, b.uploaded_by, b.created_at, (SELECT last_chapter_idx FROM reading_progress WHERE book_id = b.id AND who = 'zhaozhao') AS zz_idx, (SELECT last_chapter_idx FROM reading_progress WHERE book_id = b.id AND who = 'zhiyu') AS zy_idx, (SELECT COUNT(*) FROM reading_marks WHERE book_id = b.id AND who = 'zhaozhao') AS zz_marks, (SELECT COUNT(*) FROM reading_marks WHERE book_id = b.id AND who = 'zhiyu') AS zy_marks FROM books b ORDER BY b.created_at DESC LIMIT $1 OFFSET $2 """, limit, offset, ) return [ { "id": r["id"], "title": r["title"], "author": r["author"], "total_chapters": r["total_chapters"], "total_words": r["total_words"], "uploaded_by": r["uploaded_by"], "created_at": to_local_iso(r["created_at"]), "progress": { "zhaozhao": r["zz_idx"], "zhiyu": r["zy_idx"], }, "mark_counts": { "zhaozhao": r["zz_marks"] or 0, "zhiyu": r["zy_marks"] or 0, }, } for r in rows ] async def get_book_chapters_index(book_id: int) -> list: """书的章目录（不含正文、给列表用）。""" pool = await get_pool() async with pool.acquire() as conn: rows = await conn.fetch( "SELECT id, idx, title, word_count FROM chapters WHERE book_id = $1 ORDER BY idx", book_id, ) return [{"id": r["id"], "idx": r["idx"], "title": r["title"], "word_count": r["word_count"]} for r in rows] async def get_chapter_content(book_id: int, idx: int) -> dict | None: """按章序号读章正文。""" pool = await get_pool() async with pool.acquire() as conn: r = await conn.fetchrow( "SELECT id, idx, title, content, word_count FROM chapters WHERE book_id = $1 AND idx = $2", book_id, idx, ) if not r: return None book = await conn.fetchrow("SELECT title, author, total_chapters FROM books WHERE id = $1", book_id) return { "chapter_id": r["id"], "book_id": book_id, "book_title": book["title"] if book else None, "book_author": book["author"] if book else None, "total_chapters": book["total_chapters"] if book else None, "idx": r["idx"], "title": r["title"], "content": r["content"], "word_count": r["word_count"], } async def save_reading_progress(book_id: int, who: str, chapter_idx: int, offset: int = 0): """UPSERT 阅读进度。""" pool = await get_pool() async with pool.acquire() as conn: await conn.execute( """ INSERT INTO reading_progress (book_id, who, last_chapter_idx, last_offset, updated_at) VALUES ($1, $2, $3, $4, NOW()) ON CONFLICT (book_id, who) DO UPDATE SET last_chapter_idx = EXCLUDED.last_chapter_idx, last_offset = EXCLUDED.last_offset, updated_at = NOW() """, book_id, who, chapter_idx, offset, ) async def get_reading_progress(book_id: int, who: str) -> dict | None: pool = await get_pool() async with pool.acquire() as conn: r = await conn.fetchrow( "SELECT last_chapter_idx, last_offset, updated_at FROM reading_progress WHERE book_id = $1 AND who = $2", book_id, who, ) if not r: return None return { "last_chapter_idx": r["last_chapter_idx"], "last_offset": r["last_offset"], "updated_at": to_local_iso(r["updated_at"]), } class DuplicateHighlightError(Exception): """同人 + 同段 highlight 已存在——避免重复划线。批注(note)不受此限制、想法可以多个。""" def __init__(self, existing_id: int): super().__init__(f"already highlighted (id={existing_id})") self.existing_id = existing_id async def save_reading_mark( book_id: int, chapter_id: int, who: str, kind: str, # 'highlight' | 'stop' | 'note' text_snippet: str = None, note_content: str = None, start_offset: int = None, end_offset: int = None, ) -> int: pool = await get_pool() # highlight 去重：同 who + 同 chapter + 同 snippet 已有 highlight → 拒绝重复 # 批注(note)允许多条同段落——一段话可以有不同想法 if kind == "highlight" and text_snippet: async with pool.acquire() as conn: existing = await conn.fetchval( """ SELECT id FROM reading_marks WHERE book_id = $1 AND chapter_id = $2 AND who = $3 AND kind = 'highlight' AND text_snippet = $4 LIMIT 1 """, book_id, chapter_id, who, text_snippet, ) if existing: raise DuplicateHighlightError(existing) # 算 embedding：note_content 优先、text_snippet 兜底——让笔记能纳入 memory_recall # 拂卷 · 2026-07-02 增强：让"聊到相关话题时、书里划过的相关句子自然浮现" embedding_json = None text_for_embed = (note_content or "").strip() or (text_snippet or "").strip() if text_for_embed and EMBEDDING_API_KEY: try: vec = await compute_embedding(text_for_embed) if vec: embedding_json = json.dumps(vec) except Exception as e: print(f"⚠️ reading_mark embedding 失败（继续、无向量）: {e!r}", flush=True) async with pool.acquire() as conn: mid = await conn.fetchval( """ INSERT INTO reading_marks (book_id, chapter_id, who, kind, text_snippet, note_content, start_offset, end_offset, embedding_json) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) RETURNING id """, book_id, chapter_id, who, kind, text_snippet, note_content, start_offset, end_offset, embedding_json, ) return mid async def delete_reading_mark(mark_id: int) -> bool: """删一条痕迹（划线/批注）。返回是否真删了。""" pool = await get_pool() async with pool.acquire() as conn: rc = await conn.execute("DELETE FROM reading_marks WHERE id = $1", mark_id) return rc and rc.startswith("DELETE") and rc.split()[-1] != "0" async def get_reading_marks(book_id: int, chapter_id: int = None, who: str = None) -> list: """一本书 / 一章的所有痕迹（两人的都返回、按时间倒序）。""" pool = await get_pool() where = ["book_id = $1"] params = [book_id] if chapter_id is not None: where.append(f"chapter_id = ${len(params) + 1}") params.append(chapter_id) if who is not None: where.append(f"who = ${len(params) + 1}") params.append(who) sql = f""" SELECT id, book_id, chapter_id, who, kind, text_snippet, note_content, start_offset, end_offset, created_at FROM reading_marks WHERE {' AND '.join(where)} ORDER BY created_at DESC """ async with (await get_pool()).acquire() as conn: rows = await conn.fetch(sql, *params) return [ { "id": r["id"], "book_id": r["book_id"], "chapter_id": r["chapter_id"], "who": r["who"], "kind": r["kind"], "text_snippet": r["text_snippet"], "note_content": r["note_content"], "start_offset": r["start_offset"], "end_offset": r["end_offset"], "created_at": to_local_iso(r["created_at"]), } for r in rows ] async def delete_book(book_id: int) -> int: """删一本书（连带 chapters/marks/progress 级联删）。返回删了几行。""" pool = await get_pool() async with pool.acquire() as conn: rc = await conn.execute("DELETE FROM books WHERE id = $1", book_id) return int(rc.split()[-1]) if rc else 0