68 lines
1.6 KiB
Python
68 lines
1.6 KiB
Python
"""
|
||
GLM 文件 ID 缓存（基于磁盘的简单 KV，sha256 → file_id，3天有效期）
|
||
"""
|
||
|
||
import hashlib
|
||
import json
|
||
import threading
|
||
import time
|
||
from pathlib import Path
|
||
|
||
_CACHE_FILE = Path(__file__).parent.parent / "uploads" / ".glm_file_cache.json"
|
||
_lock = threading.Lock()
|
||
_TTL = 3 * 24 * 3600 # 3天
|
||
|
||
|
||
def _load() -> dict:
|
||
try:
|
||
if _CACHE_FILE.exists():
|
||
return json.loads(_CACHE_FILE.read_text(encoding="utf-8"))
|
||
except Exception:
|
||
pass
|
||
return {}
|
||
|
||
|
||
def _save(data: dict) -> None:
|
||
try:
|
||
_CACHE_FILE.parent.mkdir(parents=True, exist_ok=True)
|
||
_CACHE_FILE.write_text(
|
||
json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8"
|
||
)
|
||
except Exception as e:
|
||
print(f"[file_cache] 写入失败：{e}")
|
||
|
||
|
||
def sha256_of_file(file_path: Path) -> str:
|
||
h = hashlib.sha256()
|
||
with open(file_path, "rb") as f:
|
||
for chunk in iter(lambda: f.read(65536), b""):
|
||
h.update(chunk)
|
||
return h.hexdigest()
|
||
|
||
|
||
def get(file_hash: str) -> dict | None:
|
||
with _lock:
|
||
data = _load()
|
||
entry = data.get(file_hash)
|
||
if not entry:
|
||
return None
|
||
if entry.get("expires_at", 0) <= time.time():
|
||
data.pop(file_hash, None)
|
||
_save(data)
|
||
return None
|
||
return entry
|
||
|
||
|
||
def set(file_hash: str, file_id: str) -> None:
|
||
with _lock:
|
||
data = _load()
|
||
data[file_hash] = {"file_id": file_id, "expires_at": time.time() + _TTL}
|
||
_save(data)
|
||
|
||
|
||
def delete(file_hash: str) -> None:
|
||
with _lock:
|
||
data = _load()
|
||
data.pop(file_hash, None)
|
||
_save(data)
|