174 lines
8.4 KiB
Python
174 lines
8.4 KiB
Python
"""Stage SQLite/JSON and publish them with a single Neo4j data transaction.
|
|
|
|
There is no distributed transaction between a filesystem and Neo4j. Cooperating
|
|
readers therefore use the publication barrier. Ordinary failures restore local
|
|
files; ambiguous commit/crash leaves a durable marker and readers fail closed.
|
|
Re-running sync creates a complete fresh generation and clears that marker.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import shutil
|
|
import sqlite3
|
|
import subprocess
|
|
import tempfile
|
|
import uuid
|
|
from contextlib import closing
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from tools.index_state import (PENDING_FILE, VERSION_FILE, durable_json, fsync_directory,
|
|
publication_lock, sqlite_version)
|
|
from tools.rag.scan_zpwiki import scan_pages
|
|
from tools.rag.build_chunks import build_chunks
|
|
from tools.rag.build_sqlite_index import build_database
|
|
from tools.rag.embedding_utils import embedding_model_name
|
|
from tools.publication import PUBLICATION_POLICY_VERSION
|
|
from tools.graphrag.graphrag import build_graph_payload
|
|
from tools.graphrag.graph_db import create_neo4j_driver, get_neo4j_settings
|
|
from tools.graphrag.build_graphrag import (create_constraints, graph_version,
|
|
write_graph_transaction, graph_counts)
|
|
|
|
FILES = ('documents.json', 'chunks.json', 'zp_index.sqlite', VERSION_FILE)
|
|
|
|
|
|
def digest(path: Path) -> str:
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
|
|
def source_files(wiki_root: Path) -> dict[str, str]:
|
|
if not (wiki_root / 'pages').is_dir():
|
|
raise RuntimeError(f'Chýba wiki priečinok pages: {wiki_root}')
|
|
return {str(p.relative_to(wiki_root)): digest(p)
|
|
for p in sorted((wiki_root / 'pages').rglob('README.md'))}
|
|
|
|
|
|
def git_revision(wiki_root: Path) -> str | None:
|
|
if not (wiki_root / '.git').exists():
|
|
return None
|
|
result = subprocess.run(['git', '-C', str(wiki_root), 'rev-parse', 'HEAD'],
|
|
capture_output=True, text=True, timeout=15, check=True)
|
|
return result.stdout.strip()
|
|
|
|
|
|
def prepare_snapshot(wiki_root: Path, stage: Path) -> tuple[dict, dict, dict]:
|
|
revision = git_revision(wiki_root); sources = source_files(wiki_root)
|
|
frozen_wiki = stage / 'wiki'
|
|
for relative in sources:
|
|
target = frozen_wiki / relative; target.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copyfile(wiki_root / relative, target)
|
|
if source_files(frozen_wiki) != sources:
|
|
raise RuntimeError('Wiki sa počas vytvárania snapshotu zmenila.')
|
|
documents = scan_pages(wiki_root=frozen_wiki, output=stage / 'documents.json')
|
|
chunks = build_chunks(wiki_root=frozen_wiki, output=stage / 'chunks.json')
|
|
counts = build_database(documents_file=stage / 'documents.json', chunks_file=stage / 'chunks.json',
|
|
db_file=stage / 'zp_index.sqlite', cache_file=stage.parent / 'embedding_cache.sqlite')
|
|
version = {'version': uuid.uuid4().hex, 'git_revision': revision,
|
|
'created_at_utc': datetime.now(timezone.utc).isoformat(),
|
|
'source_sha256': hashlib.sha256(json.dumps(sources, sort_keys=True).encode()).hexdigest(),
|
|
'documents_sha256': digest(stage / 'documents.json'), 'chunks_sha256': digest(stage / 'chunks.json'),
|
|
'embedding_model': embedding_model_name(), 'publication_policy': PUBLICATION_POLICY_VERSION}
|
|
if chunks:
|
|
from tools.rag.embedding_utils import model_profile
|
|
version['embedding_fingerprint'] = model_profile(embedding_model_name())['fingerprint']
|
|
with closing(sqlite3.connect(stage / 'zp_index.sqlite')) as conn, conn:
|
|
conn.execute('CREATE TABLE index_state (id INTEGER PRIMARY KEY CHECK(id=1), metadata_json TEXT NOT NULL)')
|
|
conn.execute('INSERT INTO index_state VALUES (1, ?)', (json.dumps(version, sort_keys=True),))
|
|
durable_json(stage / VERSION_FILE, version)
|
|
if source_files(wiki_root) != sources or git_revision(wiki_root) != revision:
|
|
raise RuntimeError('Wiki sa počas indexovania zmenila; pripravené indexy sa neaktivujú.')
|
|
return version, counts, build_graph_payload(documents, chunks)
|
|
|
|
|
|
def replace_file(source: Path, target: Path) -> None:
|
|
"""Separate operation so tests can inject a publication I/O failure."""
|
|
os.replace(source, target)
|
|
|
|
|
|
def publish_snapshot(data_dir: Path, stage: Path, version: dict, payload: dict, session) -> dict:
|
|
with publication_lock(data_dir, exclusive=True):
|
|
had_pending = (data_dir / PENDING_FILE).exists()
|
|
old_graph_version = graph_version(session)
|
|
originals = []
|
|
backup = stage / 'previous'; backup.mkdir()
|
|
for name in FILES:
|
|
current = data_dir / name
|
|
if current.exists():
|
|
shutil.copy2(current, backup / name)
|
|
with (backup / name).open('rb') as handle:
|
|
os.fsync(handle.fileno())
|
|
originals.append(name)
|
|
fsync_directory(backup)
|
|
durable_json(data_dir / PENDING_FILE, {'target_version': version['version'],
|
|
'staging_directory': stage.name, 'original_files': originals})
|
|
commit_attempted = False
|
|
try:
|
|
with session.begin_transaction() as tx:
|
|
write_graph_transaction(tx, payload, version=version)
|
|
node_counts, relation_counts = graph_counts(tx)
|
|
graph_summary = {'nodes': sum(n for _, n in node_counts),
|
|
'relationships': sum(n for _, n in relation_counts)}
|
|
for name in FILES:
|
|
# Flush staged contents before the durable directory switch.
|
|
with (stage / name).open('rb') as handle:
|
|
os.fsync(handle.fileno())
|
|
replace_file(stage / name, data_dir / name)
|
|
fsync_directory(data_dir)
|
|
commit_attempted = True
|
|
tx.commit()
|
|
except BaseException:
|
|
if commit_attempted:
|
|
# A lost COMMIT response is not proof of rollback. Never restore
|
|
# old SQLite against a possibly committed new graph.
|
|
try:
|
|
current_graph_version = graph_version(session)
|
|
except Exception:
|
|
raise # Keep marker + backups; reads stay unavailable.
|
|
if current_graph_version == version and sqlite_version(data_dir / 'zp_index.sqlite') == version:
|
|
(data_dir / PENDING_FILE).unlink(); fsync_directory(data_dir)
|
|
return graph_summary
|
|
if current_graph_version != old_graph_version:
|
|
raise # Unknown external writer; fail closed.
|
|
for name in FILES:
|
|
if name in originals:
|
|
os.replace(backup / name, data_dir / name)
|
|
else:
|
|
(data_dir / name).unlink(missing_ok=True)
|
|
fsync_directory(data_dir)
|
|
if not had_pending:
|
|
(data_dir / PENDING_FILE).unlink(); fsync_directory(data_dir)
|
|
raise
|
|
(data_dir / PENDING_FILE).unlink()
|
|
fsync_directory(data_dir)
|
|
return graph_summary
|
|
|
|
|
|
def synchronize_indexes(wiki_root: Path, data_dir: Path, *, progress=None) -> dict:
|
|
data_dir.mkdir(parents=True, exist_ok=True)
|
|
stage = Path(tempfile.mkdtemp(prefix='.sync-stage-', dir=data_dir))
|
|
try:
|
|
version, counts, payload = prepare_snapshot(wiki_root, stage)
|
|
if progress:
|
|
progress('publishing_indexes')
|
|
settings = get_neo4j_settings()
|
|
with create_neo4j_driver(settings) as driver, driver.session(database=settings.database) as session:
|
|
driver.verify_connectivity()
|
|
create_constraints(session)
|
|
graph_summary = publish_snapshot(data_dir, stage, version, payload, session)
|
|
return {'counts': counts, 'documents_scanned': counts['documents'], 'chunks_created': counts['chunks'],
|
|
'database_path': str(data_dir / 'zp_index.sqlite'), 'index_version': version,
|
|
'graph_counts': graph_summary}
|
|
finally:
|
|
# Keep recovery material when a crash/ambiguous commit is unresolved.
|
|
pending = data_dir / PENDING_FILE
|
|
keep_stage = False
|
|
if pending.exists():
|
|
try:
|
|
keep_stage = json.loads(pending.read_text()).get('staging_directory') == stage.name
|
|
except (ValueError, OSError):
|
|
keep_stage = True
|
|
if not keep_stage:
|
|
shutil.rmtree(stage)
|