实战知识库迁移 03：权限、版本和同步，不要让 AI 读错知识

{
    "user_id": "u_cs_001",
    "departments": ["customer_service"],
    "scopes": ["public", "internal"]
}

import json
from pathlib import Path

CHUNKS_FILE = Path("processed/chunks.jsonl")

def load_chunks() -> list[dict]:
    return [
        json.loads(line)
        for line in CHUNKS_FILE.read_text(encoding="utf-8").splitlines()
        if line.strip()
    ]

def can_access(user: dict, chunk: dict) -> bool:
    permission = chunk.get("permission")

    if permission == "public":
        return True

    if permission == "internal":
        return "internal" in user.get("scopes", [])

    if permission == "department":
        return chunk.get("department") in user.get("departments", [])

    if permission == "restricted":
        return user.get("user_id") in chunk.get("allowed_users", [])

    return False

def is_current(chunk: dict) -> bool:
    is_not_expired = chunk.get("effective_to") in (None, "")
    is_not_deprecated = chunk.get("lifecycle") != "deprecated"
    return is_not_expired and is_not_deprecated

def version_match(chunk: dict, requested_version: str | None = None) -> bool:
    if requested_version:
        return chunk.get("version") == requested_version
    return is_current(chunk)

def filter_chunks(user: dict, chunks: list[dict], requested_version: str | None = None):
    return [
        chunk for chunk in chunks
        if can_access(user, chunk) and version_match(chunk, requested_version)
    ]

def print_visible(user: dict, requested_version: str | None = None):
    chunks = load_chunks()
    visible = filter_chunks(user, chunks, requested_version)
    print(f"user={user['user_id']}, version={requested_version or 'current'}")
    for chunk in visible:
        print("-", chunk["id"], chunk["version"], chunk["permission"], chunk["text"])

if __name__ == "__main__":
    users = [
        {
            "user_id": "u_cs_001",
            "departments": ["customer_service"],
            "scopes": ["public", "internal"],
        },
        {
            "user_id": "u_fin_001",
            "departments": ["finance"],
            "scopes": ["public", "internal"],
        },
        {
            "user_id": "u_manager",
            "departments": ["sales"],
            "scopes": ["public", "internal"],
        },
    ]

    for user in users:
        print_visible(user)
        print()

    print("指定查看旧版本：")
    print_visible(users[0], requested_version="2025.12")

visible_chunks = filter_chunks(user, all_chunks)
results = vector_search(query, visible_chunks)

vector_store.search(
    query=query,
    filters={
        "permission_scope": {"$in": user["scopes"]},
        "lifecycle": "current",
    }
)

import hashlib
import json
from datetime import datetime, timezone
from pathlib import Path

STATE_FILE = Path("tools/sync_state.json")

def file_hash(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()

def load_state() -> dict:
    if not STATE_FILE.exists():
        return {}
    return json.loads(STATE_FILE.read_text(encoding="utf-8"))

def save_state(state: dict):
    STATE_FILE.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8")

def check_file(path: Path, state: dict) -> bool:
    current_hash = file_hash(path)
    old_hash = state.get(str(path), {}).get("last_hash")
    return current_hash != old_hash

def mark_synced(path: Path, state: dict):
    state[str(path)] = {
        "last_hash": file_hash(path),
        "last_synced_at": datetime.now(timezone.utc).isoformat(),
    }

def main():
    state = load_state()
    files = [Path("docs/refund_policy.md"), Path("docs/vip_policy.md")]

    changed_files = []
    for path in files:
        if not path.exists():
            print(f"skip missing: {path}")
            continue
        if check_file(path, state):
            changed_files.append(path)

    if not changed_files:
        print("no changes")
        return

    print("changed files:")
    for path in changed_files:
        print("-", path)
        mark_synced(path, state)

    save_state(state)

if __name__ == "__main__":
    main()