feat(selfcheck): /api/v1/selfcheck 统一取数端点 + tools/selfcheck_full.py 全量快照（牧尘批「三档」）
This commit is contained in:
xiaoyi
@@ -0,0 +1,285 @@
|
||||
package routes
|
||||
|
||||
// 织忆自检端点（2026-10-26，牧尘批「三档」）
|
||||
//
|
||||
// 背景：织忆相关的定时自检脚本有 ~20 个，取数方式三套并存——
|
||||
// ① 直连生产库 lancedb.connect("/var/lib/memoryweave")（绕过服务、schema 一改就静默腐烂）
|
||||
// ② IPC socket 自定义协议
|
||||
// ③ HTTP /api/v1/{stats,popwatch,metrics/self}
|
||||
// 三套契约对同一个库，schema 变更时分开损坏。典型事故：pattern 自检脚本写死
|
||||
// type='pattern'，库里实际是 pattern_stat ⇒ 连续数月静默 0 输出。
|
||||
//
|
||||
// 本端点是**唯一取数契约**：把「读织忆内部状态」收敛成一份结构化 JSON。
|
||||
// 设计边界（不做什么）：
|
||||
// - 不做判定/告警/投递 —— 阈值、渠道、评审留在外部 cron（观测者不能住在被观测者体内：
|
||||
// zhiyid 一挂自检就跟着哑，内外同时沉默是更坏的故障）。
|
||||
// - 不跑重活（eval 全量、备份、vacuum）—— 那些抢生产资源，留在 systemd-run 独立 scope。
|
||||
//
|
||||
// level 语义：
|
||||
// - cheap（默认，<50ms）：只读内存态/计数，不扫全表。
|
||||
// - full（秒级）：额外扫全表算碎片率/冗余率/近 7 天质量/写量分布（原 quality-watch 口径）。
|
||||
//
|
||||
// 兼容铁律：本端点返回的比率一律给「分子/分母/样本是否足够」，缺数据不返回 0
|
||||
// （0 会被下游读成"很好"）。沿用 selfoptimize 的 metricNA(-1) 语义。
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"math"
|
||||
"net/http"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/xiaoxue/memoryweave/internal/distill"
|
||||
"github.com/xiaoxue/memoryweave/internal/selfoptimize"
|
||||
"github.com/xiaoxue/memoryweave/internal/storage"
|
||||
)
|
||||
|
||||
// SelfcheckSchemaVersion 变更返回结构时递增；消费方据此判兼容。
|
||||
const SelfcheckSchemaVersion = "1"
|
||||
|
||||
// 与 zhiyi-quality-watch.py 同口径（该脚本已改为读本端点，口径只此一处）
|
||||
const (
|
||||
selfcheckFragLen = 20 // <20 字符 = 碎片
|
||||
selfcheckDupMinSample = 5000 // 冗余率最小样本（活行），不足只报样本不判率
|
||||
selfcheckScanLimit = 300000
|
||||
)
|
||||
|
||||
type selfcheckQuality struct {
|
||||
TotalRows int `json:"total_rows"`
|
||||
LiveRows int `json:"live_rows"`
|
||||
ScannedRows int `json:"scanned_rows"`
|
||||
ScanTruncated bool `json:"scan_truncated"`
|
||||
Coverage float64 `json:"coverage"` // 扫描行 / store 真实总量；<1 表示结果不完整，禁止据此下结论
|
||||
FragLen int `json:"frag_len"`
|
||||
|
||||
FragRowsAll int `json:"frag_rows_all"`
|
||||
FragRateAll float64 `json:"frag_rate_all"`
|
||||
FragRowsLive int `json:"frag_rows_live"`
|
||||
FragRateLive float64 `json:"frag_rate_live"`
|
||||
|
||||
DupGroups int `json:"dup_groups"`
|
||||
DupExtraRows int `json:"dup_extra_rows"`
|
||||
DupRate float64 `json:"dup_rate"`
|
||||
DupSampleOK bool `json:"dup_sample_ok"`
|
||||
|
||||
NewRows7d int `json:"new_rows_7d"`
|
||||
NewFragRows7d int `json:"new_frag_rows_7d"`
|
||||
NewFragRate7d *float64 `json:"new_frag_rate_7d"` // nil = 近 7 天无新增（无分母）
|
||||
}
|
||||
|
||||
type selfcheckDayCount struct {
|
||||
Date string `json:"date"`
|
||||
Rows int `json:"rows"`
|
||||
Live int `json:"live"`
|
||||
}
|
||||
|
||||
type selfcheckWriteVolume struct {
|
||||
Days []selfcheckDayCount `json:"days"`
|
||||
Today int `json:"today"`
|
||||
Last7d int `json:"last_7d"`
|
||||
Median7d float64 `json:"median_7d"`
|
||||
P25_7d float64 `json:"p25_7d"`
|
||||
}
|
||||
|
||||
// Selfcheck GET /api/v1/selfcheck?level=cheap|full
|
||||
func (a *API) Selfcheck(w http.ResponseWriter, r *http.Request) {
|
||||
start := time.Now()
|
||||
if r.Method != http.MethodGet {
|
||||
respondError(w, http.StatusMethodNotAllowed, "method not allowed")
|
||||
return
|
||||
}
|
||||
level := strings.ToLower(strings.TrimSpace(r.URL.Query().Get("level")))
|
||||
if level == "" {
|
||||
level = "cheap"
|
||||
}
|
||||
if level != "cheap" && level != "full" {
|
||||
respondError(w, http.StatusBadRequest, "level must be cheap|full")
|
||||
return
|
||||
}
|
||||
|
||||
rep := map[string]interface{}{
|
||||
"schema_version": SelfcheckSchemaVersion,
|
||||
"checked_at": time.Now().Format(time.RFC3339),
|
||||
"level": level,
|
||||
}
|
||||
degraded := []string{}
|
||||
|
||||
// ── 存储总量 ───────────────────────────────
|
||||
if st, err := a.LanceDB.Stats(); err != nil {
|
||||
degraded = append(degraded, "store: "+err.Error())
|
||||
} else {
|
||||
total := numOf(st["total_memories"])
|
||||
tomb := numOf(st["tombstone_count"])
|
||||
live := total - tomb
|
||||
if live < 0 {
|
||||
live = 0
|
||||
}
|
||||
store := map[string]interface{}{
|
||||
"backend": st["backend"],
|
||||
"data_dir": st["data_dir"],
|
||||
"total_memories": total,
|
||||
"total_episodes": numOf(st["total_episodes"]),
|
||||
"tombstone_count": tomb,
|
||||
"live_memories": live,
|
||||
}
|
||||
if total > 0 {
|
||||
store["tombstone_rate"] = round4(tomb / total)
|
||||
} else {
|
||||
store["tombstone_rate"] = -1.0 // N/A：空库无分母
|
||||
}
|
||||
rep["store"] = store
|
||||
}
|
||||
|
||||
// ── 图谱 ───────────────────────────────────
|
||||
if a.GraphStore != nil {
|
||||
n, e, d := a.GraphStore.Stats()
|
||||
rep["graph"] = map[string]interface{}{
|
||||
"node_count": n,
|
||||
"edge_count": e,
|
||||
"density": round4(d),
|
||||
}
|
||||
} else {
|
||||
degraded = append(degraded, "graph: store not initialised")
|
||||
}
|
||||
|
||||
// ── 面板指标（反馈/蒸馏/gap/cascade，含 _valid/_samples 语义）──
|
||||
rep["dashboard"] = selfoptimize.Dash.Metrics()
|
||||
|
||||
// ── 整合链路（内存态，<1ms）────────────────
|
||||
tracker := GetConsolTracker()
|
||||
consolStatus, consolFails, sinceSuccess, lastErr := tracker.HealthStatus()
|
||||
_, consolSkips, lastSkip, _ := tracker.Snapshot()
|
||||
rep["consolidate"] = map[string]interface{}{
|
||||
"status": consolStatus,
|
||||
"consec_failures": consolFails,
|
||||
"skipped": consolSkips,
|
||||
"last_skip": lastSkip,
|
||||
"last_success_age": sinceSuccess.String(),
|
||||
"last_error": lastErr,
|
||||
}
|
||||
|
||||
// ── 缓存 / FTS ─────────────────────────────
|
||||
cache := map[string]interface{}{"fts_index": storage.FtsIndexStats()}
|
||||
if storage.SearchCacheInstance != nil {
|
||||
cache["search_cache"] = storage.SearchCacheInstance.Stats()
|
||||
}
|
||||
rep["cache"] = cache
|
||||
|
||||
// ── 空壳拦截器（t_e11207f8 取证）───────────
|
||||
checks, blocks, loaded, rulesPath := distill.PayloadGuardStats()
|
||||
rep["payload_guard"] = map[string]interface{}{
|
||||
"checks": checks,
|
||||
"blocks": blocks,
|
||||
"loaded": loaded,
|
||||
"rules_path": rulesPath,
|
||||
}
|
||||
|
||||
// ── 全量质量段（level=full）：读卫星快照，不在本进程扫全表 ────────
|
||||
// 设计：全表扫描是重活（4 万行 × 内容），且 daemon 还有 recall 要跑。
|
||||
// 取数由仓库内 tools/selfcheck_full.py 完成（列投影只读，1.7s / 250MB，独立进程），
|
||||
// 写 ~/.hermes/state/zhiyi-quality.json；本端点只读快照并标注新鲜度。
|
||||
// 这样"取数"只有一处实现（跟 schema 同仓），"判定/告警"仍在外部 cron。
|
||||
if level == "full" {
|
||||
snap, err := loadQualitySnapshot()
|
||||
if err != nil {
|
||||
degraded = append(degraded, "quality: "+err.Error())
|
||||
} else {
|
||||
rep["quality"] = snap.Quality
|
||||
rep["write_volume"] = snap.WriteVolume
|
||||
rep["quality_as_of"] = snap.GeneratedAt
|
||||
rep["quality_age_s"] = snap.AgeSeconds()
|
||||
rep["quality_stale"] = snap.AgeSeconds() > qualityStaleAfter
|
||||
}
|
||||
}
|
||||
|
||||
if len(degraded) > 0 {
|
||||
rep["degraded"] = degraded
|
||||
}
|
||||
rep["duration_ms"] = time.Since(start).Milliseconds()
|
||||
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
json.NewEncoder(w).Encode(rep)
|
||||
}
|
||||
|
||||
// qualityStaleAfter：快照超过这个时长未刷新 ⇒ 标记 stale（消费方不得据此下结论）。
|
||||
const qualityStaleAfter = 6 * 3600
|
||||
|
||||
// qualitySnapshot 对应仓库内 tools/selfcheck_full.py 的产出。
|
||||
// 字段名与 python 侧严格一致（schema_version=1）。
|
||||
type qualitySnapshot struct {
|
||||
SchemaVersion int `json:"schema_version"`
|
||||
GeneratedAt string `json:"generated_at"`
|
||||
Source struct {
|
||||
DB string `json:"db"`
|
||||
Table string `json:"table"`
|
||||
Mode string `json:"mode"`
|
||||
} `json:"source"`
|
||||
DurationMS int `json:"duration_ms"`
|
||||
Quality json.RawMessage `json:"quality"`
|
||||
WriteVolume json.RawMessage `json:"write_volume"`
|
||||
fileModTime time.Time
|
||||
}
|
||||
|
||||
func (q *qualitySnapshot) AgeSeconds() int64 {
|
||||
if q.fileModTime.IsZero() {
|
||||
return -1
|
||||
}
|
||||
return int64(time.Since(q.fileModTime).Seconds())
|
||||
}
|
||||
|
||||
// QualitySnapshotPath 允许用 env 覆盖（测试/灰度）。
|
||||
func QualitySnapshotPath() string {
|
||||
if p := strings.TrimSpace(os.Getenv("ZHIYI_QUALITY_SNAPSHOT")); p != "" {
|
||||
return p
|
||||
}
|
||||
home, _ := os.UserHomeDir()
|
||||
return filepath.Join(home, ".hermes", "state", "zhiyi-quality.json")
|
||||
}
|
||||
|
||||
func loadQualitySnapshot() (*qualitySnapshot, error) {
|
||||
p := QualitySnapshotPath()
|
||||
fi, err := os.Stat(p)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("快照不存在（先跑 tools/selfcheck_full.py）：%w", err)
|
||||
}
|
||||
raw, err := os.ReadFile(p)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var snap qualitySnapshot
|
||||
if err := json.Unmarshal(raw, &snap); err != nil {
|
||||
return nil, fmt.Errorf("快照解析失败: %w", err)
|
||||
}
|
||||
snap.fileModTime = fi.ModTime()
|
||||
return &snap, nil
|
||||
}
|
||||
|
||||
// numOf 把 stats 里可能是 float64/int/json.Number 的数值统一成 float64。
|
||||
func numOf(v interface{}) float64 {
|
||||
switch x := v.(type) {
|
||||
case float64:
|
||||
return x
|
||||
case float32:
|
||||
return float64(x)
|
||||
case int:
|
||||
return float64(x)
|
||||
case int64:
|
||||
return float64(x)
|
||||
case json.Number:
|
||||
f, _ := x.Float64()
|
||||
return f
|
||||
default:
|
||||
return 0
|
||||
}
|
||||
}
|
||||
|
||||
func round4(f float64) float64 {
|
||||
if math.IsNaN(f) || math.IsInf(f, 0) {
|
||||
return -1
|
||||
}
|
||||
return math.Round(f*10000) / 10000
|
||||
}
|
||||
@@ -0,0 +1,99 @@
|
||||
package routes
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// 快照契约测试：/api/v1/selfcheck?level=full 依赖 tools/selfcheck_full.py 的产出格式。
|
||||
// 这里钉住 schema_version 与关键字段，防止 python 侧改字段名后 Go 侧静默返回空。
|
||||
func TestLoadQualitySnapshot(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
p := filepath.Join(dir, "zhiyi-quality.json")
|
||||
body := `{
|
||||
"schema_version": 1,
|
||||
"generated_at": "2026-10-26T13:20:00+0800",
|
||||
"source": {"db": "/var/lib/memoryweave", "table": "memories", "mode": "read_only_projection"},
|
||||
"duration_ms": 1680,
|
||||
"quality": {
|
||||
"frag_len": 20, "total_rows": 39957, "live_rows": 39836,
|
||||
"frag_rows_all": 313, "frag_rate_all": 0.0078,
|
||||
"frag_rows_live": 313, "frag_rate_live": 0.0079,
|
||||
"dup_groups": 34, "dup_extra_rows": 103, "dup_rate": 0.0026,
|
||||
"new_rows_7d": 1176, "new_frag_rows_7d": 35, "new_frag_rate_7d": 0.0298
|
||||
},
|
||||
"write_volume": {
|
||||
"days": [{"date": "2026-10-25", "rows": 1473, "live": 1471}],
|
||||
"today": 332, "today_date": "2026-10-26", "last_7d": 7706, "nonzero_days": 60
|
||||
}
|
||||
}`
|
||||
if err := os.WriteFile(p, []byte(body), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Setenv("ZHIYI_QUALITY_SNAPSHOT", p)
|
||||
|
||||
snap, err := loadQualitySnapshot()
|
||||
if err != nil {
|
||||
t.Fatalf("loadQualitySnapshot: %v", err)
|
||||
}
|
||||
if snap.SchemaVersion != 1 {
|
||||
t.Fatalf("schema_version=%d want 1", snap.SchemaVersion)
|
||||
}
|
||||
|
||||
var q struct {
|
||||
TotalRows int `json:"total_rows"`
|
||||
LiveRows int `json:"live_rows"`
|
||||
DupRate float64 `json:"dup_rate"`
|
||||
FragRowsAll int `json:"frag_rows_all"`
|
||||
}
|
||||
if err := json.Unmarshal(snap.Quality, &q); err != nil {
|
||||
t.Fatalf("quality 段解析失败: %v", err)
|
||||
}
|
||||
if q.TotalRows != 39957 || q.LiveRows != 39836 {
|
||||
t.Fatalf("quality 口径不符: %+v", q)
|
||||
}
|
||||
if q.FragRowsAll != 313 || q.DupRate != 0.0026 {
|
||||
t.Fatalf("quality 数值不符: %+v", q)
|
||||
}
|
||||
|
||||
var wv struct {
|
||||
Today int `json:"today"`
|
||||
Days []struct {
|
||||
Date string `json:"date"`
|
||||
Live int `json:"live"`
|
||||
} `json:"days"`
|
||||
}
|
||||
if err := json.Unmarshal(snap.WriteVolume, &wv); err != nil {
|
||||
t.Fatalf("write_volume 段解析失败: %v", err)
|
||||
}
|
||||
if wv.Today != 332 || len(wv.Days) != 1 || wv.Days[0].Live != 1471 {
|
||||
t.Fatalf("write_volume 不符: %+v", wv)
|
||||
}
|
||||
|
||||
// 新鲜度：刚写的文件 age 应 < 10s；人为改旧 → stale 判定为真
|
||||
if age := snap.AgeSeconds(); age < 0 || age > 10 {
|
||||
t.Fatalf("AgeSeconds=%d，本应接近 0", age)
|
||||
}
|
||||
old := time.Now().Add(-8 * time.Hour)
|
||||
if err := os.Chtimes(p, old, old); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
snap2, err := loadQualitySnapshot()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if snap2.AgeSeconds() <= qualityStaleAfter {
|
||||
t.Fatalf("8h 前的快照应判 stale（>%d 秒），得到 %d", qualityStaleAfter, snap2.AgeSeconds())
|
||||
}
|
||||
}
|
||||
|
||||
// 快照缺失 ⇒ 必须报错（由 handler 转成 degraded），不能静默返回零值。
|
||||
func TestLoadQualitySnapshotMissing(t *testing.T) {
|
||||
t.Setenv("ZHIYI_QUALITY_SNAPSHOT", filepath.Join(t.TempDir(), "nope.json"))
|
||||
if _, err := loadQualitySnapshot(); err == nil {
|
||||
t.Fatal("缺失快照应返回 error")
|
||||
}
|
||||
}
|
||||
@@ -613,6 +613,10 @@ func NewServer() http.Handler {
|
||||
// 反馈统一走 /api/v1/feedback/{useful,not-useful,deprecate,correct}（feedback.go，写拆分后新字段）。
|
||||
mux.HandleFunc("/api/v1/bootstrap", api.Bootstrap)
|
||||
mux.HandleFunc("/api/v1/stats", api.Stats)
|
||||
// 2026-10-26（牧尘批「三档」）：自检统一取数端点。
|
||||
// 唯一契约 —— 自检脚本一律读这里，禁止再 lancedb.connect() 直连生产库。
|
||||
// ?level=cheap（默认，内存态计数）| full（+读 tools/selfcheck_full.py 的全量快照：碎片/冗余/近7天质量/写量）
|
||||
mux.HandleFunc("/api/v1/selfcheck", api.Selfcheck)
|
||||
mux.HandleFunc("/api/v1/batch-commit", api.BatchCommit)
|
||||
// 主症A（t_fa6c9104）：显式写入直落 memories 表（嵌入 + 立刻可召回）。
|
||||
// 纯新增路由：commit/recall/蒸馏路径零改动；旧服务端 404 时插件自动回落 commit。
|
||||
|
||||
@@ -0,0 +1,200 @@
|
||||
#!/usr/bin/env python3
|
||||
"""织忆自检·全量取数器（唯一允许直连生产库的脚本）
|
||||
|
||||
为什么存在
|
||||
----------
|
||||
织忆相关的定时自检脚本曾有 ~20 个，其中两个直接 `lancedb.connect("/var/lib/memoryweave")`
|
||||
绕过服务读全表。那种写法的病根不是"直连"，而是**取数逻辑散落多处、与 schema 一起腐烂**：
|
||||
pattern 自检写死 `type='pattern'`（库里其实是 `pattern_stat`）⇒ 连续数月静默 0 输出。
|
||||
|
||||
本文件把"全量取数"收成一处、放进仓库：改 schema 时同一个 commit 就能改它，且它有输出可复核。
|
||||
判定/告警/投递**不在这里**（留在外部 cron）—— 观测者不能住在被观测者体内。
|
||||
|
||||
它产出 `~/.hermes/state/zhiyi-quality.json`，服务端 `GET /api/v1/selfcheck?level=full` 读这份快照。
|
||||
|
||||
口径（必须与历史 zhiyi-quality-watch.py 一致，改了就改告警语义）
|
||||
------------------------------------------------------------
|
||||
- 碎片：`len(content.strip()) < FRAG_LEN(20)`，按**字符数**（不是字节，中文按 1 算）
|
||||
- 冗余：live 行内 content 完全相同；冗余行数 = Σ(组内行数-1)
|
||||
- 近 7 天：created_at 在 7 天内（live）
|
||||
- 写量：按 created_at 的**本地日期**分桶，同时给 rows（含已删）与 live（仅存活）
|
||||
|
||||
性能
|
||||
----
|
||||
只投影 3 列（content/created_at/is_deleted），不读 1024 维向量：
|
||||
4 万行 0.06s（旧实现 `to_arrow()` 全列要数秒且吃 1G+ 内存）。
|
||||
|
||||
用法
|
||||
----
|
||||
python3 tools/selfcheck_full.py [--out PATH] [--json] [--days 60]
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import datetime as dt
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
from collections import Counter
|
||||
|
||||
DB_PATH = os.environ.get("ZHIYI_DB", "/var/lib/memoryweave")
|
||||
FRAG_LEN = 20
|
||||
OUT_DEFAULT = os.path.expanduser("~/.hermes/state/zhiyi-quality.json")
|
||||
SCHEMA = 1
|
||||
|
||||
|
||||
def _reexec_with_lancedb() -> None:
|
||||
"""cron 默认解释器没有 lancedb 时，切到 lance-compact venv 再跑（一次性）。"""
|
||||
try:
|
||||
import lancedb # noqa: F401
|
||||
return
|
||||
except ImportError:
|
||||
pass
|
||||
venv_py = os.path.expanduser("~/.hermes/venvs/lance-compact/bin/python")
|
||||
if os.path.exists(venv_py) and os.path.realpath(sys.prefix) != os.path.realpath(
|
||||
os.path.dirname(os.path.dirname(venv_py))
|
||||
):
|
||||
os.execv(venv_py, [venv_py] + sys.argv)
|
||||
sys.stderr.write("[selfcheck_full] lancedb 不可用（需 lance-compact venv）\n")
|
||||
sys.exit(2)
|
||||
|
||||
|
||||
def _as_bool(v) -> bool:
|
||||
if v is None:
|
||||
return False
|
||||
if isinstance(v, bool):
|
||||
return v
|
||||
return str(v).strip().lower() in ("true", "1")
|
||||
|
||||
|
||||
def collect(days: int = 60) -> dict:
|
||||
import lancedb
|
||||
|
||||
t0 = time.time()
|
||||
db = lancedb.connect(DB_PATH)
|
||||
table = db.open_table("memories")
|
||||
tbl = table.to_lance().to_table(columns=["content", "created_at", "is_deleted"])
|
||||
|
||||
contents = tbl.column("content").to_pylist()
|
||||
created = tbl.column("created_at").to_pylist()
|
||||
deleted = tbl.column("is_deleted").to_pylist()
|
||||
|
||||
total = len(contents)
|
||||
frag_all = 0
|
||||
live = 0
|
||||
frag_live = 0
|
||||
dup_counter: Counter[str] = Counter()
|
||||
daily: dict[str, dict[str, int]] = {}
|
||||
now = dt.datetime.now()
|
||||
week_ago = now - dt.timedelta(days=7)
|
||||
new_7d = 0
|
||||
new_frag_7d = 0
|
||||
|
||||
for i in range(total):
|
||||
raw = contents[i]
|
||||
text = str(raw).strip() if raw is not None else ""
|
||||
n = len(text)
|
||||
is_frag = n < FRAG_LEN
|
||||
if is_frag:
|
||||
frag_all += 1
|
||||
|
||||
is_del = _as_bool(deleted[i] if i < len(deleted) else False)
|
||||
|
||||
# created_at：存量行有空串/畸形 → 归 unknown，不炸整批
|
||||
day = "unknown"
|
||||
ts = None
|
||||
if i < len(created):
|
||||
raw_ts = created[i]
|
||||
if raw_ts:
|
||||
try:
|
||||
if isinstance(raw_ts, dt.datetime):
|
||||
ts = raw_ts
|
||||
else:
|
||||
ts = dt.datetime.fromisoformat(str(raw_ts).replace("Z", "+00:00"))
|
||||
if ts.tzinfo is not None:
|
||||
ts = ts.astimezone().replace(tzinfo=None)
|
||||
day = ts.strftime("%Y-%m-%d")
|
||||
except Exception:
|
||||
ts = None
|
||||
|
||||
slot = daily.setdefault(day, {"rows": 0, "live": 0})
|
||||
slot["rows"] += 1
|
||||
|
||||
if is_del:
|
||||
continue
|
||||
live += 1
|
||||
slot["live"] += 1
|
||||
dup_counter[text] += 1
|
||||
if is_frag:
|
||||
frag_live += 1
|
||||
if ts is not None and ts >= week_ago:
|
||||
new_7d += 1
|
||||
if is_frag:
|
||||
new_frag_7d += 1
|
||||
|
||||
dup_groups = sum(1 for c in dup_counter.values() if c > 1)
|
||||
dup_extra = sum(c - 1 for c in dup_counter.values() if c > 1)
|
||||
|
||||
# 只保留最近 days 天（含 unknown），按日期升序
|
||||
keep = sorted(daily.keys())[-days:]
|
||||
day_rows = [{"date": d, "rows": daily[d]["rows"], "live": daily[d]["live"]} for d in keep]
|
||||
|
||||
live_days = [x["live"] for x in day_rows if x["date"] != "unknown"]
|
||||
today = now.strftime("%Y-%m-%d")
|
||||
today_live = next((x["live"] for x in day_rows if x["date"] == today), 0)
|
||||
|
||||
return {
|
||||
"schema_version": SCHEMA,
|
||||
"generated_at": now.strftime("%Y-%m-%dT%H:%M:%S%z"),
|
||||
"source": {"db": DB_PATH, "table": "memories", "mode": "read_only_projection"},
|
||||
"duration_ms": int((time.time() - t0) * 1000),
|
||||
"quality": {
|
||||
"frag_len": FRAG_LEN,
|
||||
"total_rows": total,
|
||||
"live_rows": live,
|
||||
"frag_rows_all": frag_all,
|
||||
"frag_rate_all": round(frag_all / total, 4) if total else None,
|
||||
"frag_rows_live": frag_live,
|
||||
"frag_rate_live": round(frag_live / live, 4) if live else None,
|
||||
"dup_groups": dup_groups,
|
||||
"dup_extra_rows": dup_extra,
|
||||
"dup_rate": round(dup_extra / live, 4) if live else None,
|
||||
"new_rows_7d": new_7d,
|
||||
"new_frag_rows_7d": new_frag_7d,
|
||||
"new_frag_rate_7d": round(new_frag_7d / new_7d, 4) if new_7d else None,
|
||||
},
|
||||
"write_volume": {
|
||||
"days": day_rows,
|
||||
"today": today_live,
|
||||
"today_date": today,
|
||||
# 近 7 个完整日（不含今天/unknown）
|
||||
"last_7d": sum(x["live"] for x in day_rows if x["date"] not in (today, "unknown")) if day_rows else 0,
|
||||
"nonzero_days": sum(1 for x in live_days if x > 0),
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser(description="织忆自检全量取数（只读）")
|
||||
ap.add_argument("--out", default=OUT_DEFAULT)
|
||||
ap.add_argument("--json", action="store_true", help="打印到 stdout")
|
||||
ap.add_argument("--days", type=int, default=60)
|
||||
a = ap.parse_args()
|
||||
|
||||
_reexec_with_lancedb()
|
||||
snap = collect(a.days)
|
||||
|
||||
if a.out:
|
||||
os.makedirs(os.path.dirname(a.out), exist_ok=True)
|
||||
tmp = a.out + ".tmp"
|
||||
with open(tmp, "w", encoding="utf-8") as f:
|
||||
json.dump(snap, f, ensure_ascii=False)
|
||||
os.replace(tmp, a.out) # 原子替换：读方永不会看到半个文件
|
||||
if a.json:
|
||||
print(json.dumps(snap, ensure_ascii=False, indent=1))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user