353 lines
8.9 KiB
Bash
Executable File
353 lines
8.9 KiB
Bash
Executable File
#!/bin/bash
|
|
# llm-wiki 缓存脚本
|
|
|
|
set -euo pipefail
|
|
|
|
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
|
|
# shellcheck disable=SC1091
|
|
source "$SCRIPT_DIR/shared-config.sh"
|
|
|
|
usage() {
|
|
cat <<'EOF'
|
|
用法：
|
|
bash scripts/cache.sh check <file>
|
|
bash scripts/cache.sh update <file> <source_page>
|
|
bash scripts/cache.sh invalidate <file>
|
|
EOF
|
|
}
|
|
|
|
require_file() {
|
|
local file_path="$1"
|
|
|
|
[ -n "$file_path" ] || {
|
|
usage
|
|
exit 1
|
|
}
|
|
|
|
[ -f "$file_path" ] || {
|
|
echo "文件不存在：$file_path" >&2
|
|
exit 1
|
|
}
|
|
}
|
|
|
|
find_wiki_root() {
|
|
local file_path="$1"
|
|
local dir parent
|
|
|
|
dir="$(cd "$(dirname "$file_path")" && pwd)"
|
|
|
|
while true; do
|
|
if [ -f "$dir/.wiki-cache.json" ] || [ -f "$dir/.wiki-schema.md" ]; then
|
|
printf '%s\n' "$dir"
|
|
return 0
|
|
fi
|
|
|
|
parent="$(dirname "$dir")"
|
|
[ "$parent" = "$dir" ] && return 1
|
|
dir="$parent"
|
|
done
|
|
}
|
|
|
|
cache_file_path() {
|
|
printf '%s/.wiki-cache.json\n' "$1"
|
|
}
|
|
|
|
ensure_cache_file() {
|
|
local cache_file="$1"
|
|
|
|
if [ ! -f "$cache_file" ]; then
|
|
cat > "$cache_file" <<'EOF'
|
|
{
|
|
"version": 1,
|
|
"entries": {}
|
|
}
|
|
EOF
|
|
fi
|
|
}
|
|
|
|
relative_path() {
|
|
require_python_cmd
|
|
|
|
"$PYTHON_CMD" - "$1" "$2" <<'PY'
|
|
import os
|
|
import sys
|
|
|
|
print(os.path.relpath(os.path.realpath(sys.argv[2]), os.path.realpath(sys.argv[1])))
|
|
PY
|
|
}
|
|
|
|
normalized_source_page() {
|
|
local wiki_root="$1"
|
|
local source_page="$2"
|
|
|
|
if [ -z "$source_page" ]; then
|
|
printf '%s\n' ""
|
|
return 0
|
|
fi
|
|
|
|
case "$source_page" in
|
|
/*)
|
|
require_python_cmd
|
|
|
|
"$PYTHON_CMD" - "$wiki_root" "$source_page" <<'PY'
|
|
import os
|
|
import sys
|
|
|
|
wiki_root = os.path.realpath(sys.argv[1])
|
|
source_page = os.path.realpath(sys.argv[2])
|
|
|
|
try:
|
|
common = os.path.commonpath([wiki_root, source_page])
|
|
except ValueError:
|
|
common = ""
|
|
|
|
if common == wiki_root:
|
|
print(os.path.relpath(source_page, wiki_root))
|
|
else:
|
|
print(sys.argv[2])
|
|
PY
|
|
;;
|
|
*)
|
|
printf '%s\n' "$source_page"
|
|
;;
|
|
esac
|
|
}
|
|
|
|
file_hash() {
|
|
require_python_cmd
|
|
|
|
"$PYTHON_CMD" - "$1" "$2" <<'PY'
|
|
import hashlib
|
|
import pathlib
|
|
import sys
|
|
|
|
relative_path = sys.argv[1].encode("utf-8")
|
|
file_path = pathlib.Path(sys.argv[2])
|
|
content = file_path.read_bytes()
|
|
|
|
digest = hashlib.sha256(relative_path + b"\0" + content).hexdigest()
|
|
print(f"sha256:{digest}")
|
|
PY
|
|
}
|
|
|
|
cache_check() {
|
|
local file_path="$1"
|
|
local wiki_root cache_file relative_path_value current_hash result
|
|
|
|
require_file "$file_path"
|
|
wiki_root="$(find_wiki_root "$file_path")" || {
|
|
echo "未找到知识库根目录：$file_path" >&2
|
|
exit 1
|
|
}
|
|
cache_file="$(cache_file_path "$wiki_root")"
|
|
|
|
if [ ! -f "$cache_file" ]; then
|
|
printf 'MISS\n'
|
|
return 0
|
|
fi
|
|
|
|
require_python_cmd
|
|
|
|
relative_path_value="$(relative_path "$wiki_root" "$file_path")"
|
|
current_hash="$(file_hash "$relative_path_value" "$file_path")"
|
|
|
|
result="$(
|
|
"$PYTHON_CMD" - "$cache_file" "$wiki_root" "$relative_path_value" "$current_hash" <<'PY'
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import sys
|
|
|
|
cache_file, wiki_root, relative_path, current_hash = sys.argv[1:5]
|
|
|
|
with open(cache_file, "r", encoding="utf-8") as fh:
|
|
data = json.load(fh)
|
|
|
|
entry = data.get("entries", {}).get(relative_path)
|
|
|
|
# 无 cache entry → 尝试自愈（exact filename stem match + source_path 验证）
|
|
if not entry:
|
|
raw_stem = pathlib.Path(relative_path).stem
|
|
sources_dir = os.path.join(wiki_root, "wiki", "sources")
|
|
if os.path.isdir(sources_dir):
|
|
for f in os.listdir(sources_dir):
|
|
if pathlib.Path(f).stem == raw_stem and f.endswith(".md"):
|
|
source_page = os.path.join("wiki", "sources", f)
|
|
source_abs = os.path.join(wiki_root, source_page)
|
|
# 验证 source 页面的 source_path frontmatter 是否指向当前 raw 文件
|
|
source_path_match = False
|
|
try:
|
|
with open(source_abs, "r", encoding="utf-8") as sf:
|
|
in_frontmatter = False
|
|
for line in sf:
|
|
stripped = line.strip()
|
|
if stripped == "---":
|
|
if in_frontmatter:
|
|
break # end of frontmatter
|
|
in_frontmatter = True
|
|
continue
|
|
if in_frontmatter and stripped.startswith("source_path:"):
|
|
fm_value = stripped.split(":", 1)[1].strip()
|
|
# 匹配相对路径的末尾部分
|
|
if relative_path.endswith(fm_value) or fm_value.endswith(relative_path) or fm_value == relative_path:
|
|
source_path_match = True
|
|
break
|
|
except (OSError, UnicodeDecodeError):
|
|
pass
|
|
if not source_path_match:
|
|
# stem 匹配但 source_path 不一致 → 不信任，需要验证
|
|
print("MISS:repaired_needs_verify")
|
|
raise SystemExit(0)
|
|
# stem + source_path 都匹配 → 安全自愈
|
|
timestamp = __import__("datetime").datetime.now(__import__("datetime").timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
entries = data.setdefault("entries", {})
|
|
entries[relative_path] = {
|
|
"hash": current_hash,
|
|
"ingested_at": timestamp,
|
|
"source_page": source_page,
|
|
}
|
|
tmp_file = cache_file + ".tmp"
|
|
with open(tmp_file, "w", encoding="utf-8") as fh2:
|
|
json.dump(data, fh2, ensure_ascii=False, indent=2)
|
|
fh2.write("\n")
|
|
os.replace(tmp_file, cache_file)
|
|
print("HIT(repaired)")
|
|
raise SystemExit(0)
|
|
print("MISS:no_entry")
|
|
raise SystemExit(0)
|
|
|
|
if entry.get("hash") != current_hash:
|
|
print("MISS:hash_changed")
|
|
raise SystemExit(0)
|
|
|
|
source_page = entry.get("source_page")
|
|
if not source_page:
|
|
print("MISS:no_entry")
|
|
raise SystemExit(0)
|
|
|
|
source_path = source_page
|
|
if not os.path.isabs(source_path):
|
|
source_path = os.path.join(wiki_root, source_path)
|
|
|
|
if not os.path.isfile(source_path):
|
|
print("MISS:no_source")
|
|
else:
|
|
print("HIT")
|
|
PY
|
|
)"
|
|
|
|
printf '%s\n' "$result"
|
|
}
|
|
|
|
cache_update() {
|
|
local file_path="$1"
|
|
local source_page="$2"
|
|
local wiki_root cache_file relative_path_value current_hash normalized_source timestamp
|
|
|
|
require_file "$file_path"
|
|
wiki_root="$(find_wiki_root "$file_path")" || {
|
|
echo "未找到知识库根目录：$file_path" >&2
|
|
exit 1
|
|
}
|
|
cache_file="$(cache_file_path "$wiki_root")"
|
|
ensure_cache_file "$cache_file"
|
|
|
|
require_python_cmd
|
|
|
|
relative_path_value="$(relative_path "$wiki_root" "$file_path")"
|
|
current_hash="$(file_hash "$relative_path_value" "$file_path")"
|
|
normalized_source="$(normalized_source_page "$wiki_root" "$source_page")"
|
|
timestamp="$(date -u +"%Y-%m-%dT%H:%M:%SZ")"
|
|
|
|
"$PYTHON_CMD" - "$cache_file" "$relative_path_value" "$current_hash" "$timestamp" "$normalized_source" <<'PY'
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
cache_file, relative_path, file_hash_value, timestamp, source_page = sys.argv[1:6]
|
|
|
|
with open(cache_file, "r", encoding="utf-8") as fh:
|
|
data = json.load(fh)
|
|
|
|
entries = data.setdefault("entries", {})
|
|
entries[relative_path] = {
|
|
"hash": file_hash_value,
|
|
"ingested_at": timestamp,
|
|
"source_page": source_page,
|
|
}
|
|
|
|
tmp_file = cache_file + ".tmp"
|
|
with open(tmp_file, "w", encoding="utf-8") as fh:
|
|
json.dump(data, fh, ensure_ascii=False, indent=2)
|
|
fh.write("\n")
|
|
os.replace(tmp_file, cache_file)
|
|
PY
|
|
|
|
printf 'UPDATED\n'
|
|
}
|
|
|
|
cache_invalidate() {
|
|
local file_path="$1"
|
|
local wiki_root cache_file relative_path_value
|
|
|
|
# 不调用 require_file：文件可能已被删除（级联删除场景）
|
|
# 直接通过路径查找缓存条目
|
|
wiki_root="$(find_wiki_root "$file_path")" || {
|
|
echo "未找到知识库根目录：$file_path" >&2
|
|
exit 1
|
|
}
|
|
cache_file="$(cache_file_path "$wiki_root")"
|
|
|
|
if [ ! -f "$cache_file" ]; then
|
|
printf 'INVALIDATED\n'
|
|
return 0
|
|
fi
|
|
|
|
require_python_cmd
|
|
|
|
relative_path_value="$(relative_path "$wiki_root" "$file_path")"
|
|
|
|
"$PYTHON_CMD" - "$cache_file" "$relative_path_value" <<'PY'
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
cache_file, relative_path = sys.argv[1:3]
|
|
|
|
with open(cache_file, "r", encoding="utf-8") as fh:
|
|
data = json.load(fh)
|
|
|
|
data.setdefault("entries", {}).pop(relative_path, None)
|
|
|
|
tmp_file = cache_file + ".tmp"
|
|
with open(tmp_file, "w", encoding="utf-8") as fh:
|
|
json.dump(data, fh, ensure_ascii=False, indent=2)
|
|
fh.write("\n")
|
|
os.replace(tmp_file, cache_file)
|
|
PY
|
|
|
|
printf 'INVALIDATED\n'
|
|
}
|
|
|
|
command_name="${1:-}"
|
|
|
|
case "$command_name" in
|
|
check)
|
|
[ "$#" -eq 2 ] || { usage; exit 1; }
|
|
cache_check "$2"
|
|
;;
|
|
update)
|
|
[ "$#" -eq 3 ] || { usage; exit 1; }
|
|
cache_update "$2" "$3"
|
|
;;
|
|
invalidate)
|
|
[ "$#" -eq 2 ] || { usage; exit 1; }
|
|
cache_invalidate "$2"
|
|
;;
|
|
*)
|
|
usage
|
|
exit 1
|
|
;;
|
|
esac
|