201 lines
7.1 KiB
Python
201 lines
7.1 KiB
Python
#!/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())
|