194 lines
8.6 KiB
Python
194 lines
8.6 KiB
Python
from __future__ import annotations
|
|
|
|
from .shared import (
|
|
Any,
|
|
BACKUP_RESTORE_POLICY,
|
|
CASHFLOW_KIND_SQL,
|
|
DEFAULT_LEDGER_BACKUP_DIR,
|
|
KST,
|
|
Path,
|
|
current_kst_month,
|
|
datetime,
|
|
db,
|
|
fail,
|
|
hashlib,
|
|
init_db,
|
|
ok,
|
|
resolve_db_path,
|
|
row_to_dict,
|
|
sqlite3,
|
|
timedelta,
|
|
)
|
|
from .shared import _current_kst_datetime, _month_range
|
|
|
|
def health_check(db_path: str | None = None) -> dict[str, Any]:
|
|
path = init_db(db_path)
|
|
with db(path) as conn:
|
|
schema_version = conn.execute("SELECT value FROM schema_meta WHERE key='schema_version'").fetchone()[0]
|
|
tx_count = conn.execute("SELECT COUNT(*) FROM transactions WHERE status='active'").fetchone()[0]
|
|
return ok(server="jarca-ledger-mcp", db_path=path, schema_version=schema_version, active_transactions=tx_count)
|
|
|
|
|
|
def _sha256_file(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as f:
|
|
for chunk in iter(lambda: f.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def _file_metadata(path: Path, include_hash: bool = True) -> dict[str, Any]:
|
|
stat = path.stat()
|
|
metadata: dict[str, Any] = {
|
|
"path": str(path.resolve()),
|
|
"size_bytes": stat.st_size,
|
|
"mtime": datetime.fromtimestamp(stat.st_mtime).replace(microsecond=0).isoformat(),
|
|
}
|
|
if include_hash:
|
|
metadata["sha256"] = _sha256_file(path)
|
|
return metadata
|
|
|
|
|
|
def _readonly_sqlite_connection(path: Path) -> sqlite3.Connection:
|
|
conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True)
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
|
|
def _safe_table_counts(conn: sqlite3.Connection) -> dict[str, int]:
|
|
counts: dict[str, int] = {}
|
|
for table in ("accounts", "transactions", "budgets", "import_batches", "category_rules", "ledger_categories", "manual_assets", "asset_snapshots", "review_decisions", "reconciliation_sessions"):
|
|
exists = conn.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (table,)).fetchone()
|
|
if exists:
|
|
counts[table] = int(conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0])
|
|
return counts
|
|
|
|
|
|
def _safe_monthly_aggregates(conn: sqlite3.Connection) -> dict[str, Any]:
|
|
current_month = current_kst_month()
|
|
start, next_month = _month_range(current_month)
|
|
rows = conn.execute(
|
|
f"""SELECT {CASHFLOW_KIND_SQL} AS cashflow_kind, SUM(t.amount) AS total, COUNT(*) AS count
|
|
FROM transactions t
|
|
LEFT JOIN ledger_categories lc ON lc.active=1 AND t.category=lc.category AND t.subcategory=lc.subcategory
|
|
WHERE t.status='active' AND t.transaction_date>=? AND t.transaction_date<?
|
|
GROUP BY {CASHFLOW_KIND_SQL}""",
|
|
(start, next_month),
|
|
).fetchall()
|
|
return {
|
|
"month": current_month,
|
|
"by_cashflow_kind": {
|
|
str(row["cashflow_kind"]): {"amount": int(row["total"] or 0), "count": int(row["count"] or 0)}
|
|
for row in rows
|
|
},
|
|
}
|
|
|
|
|
|
def _ledger_db_status_for_path(path: Path, include_hash: bool = True, include_integrity: bool = True) -> dict[str, Any]:
|
|
status: dict[str, Any] = {"db_path": str(path.resolve()), "exists": path.exists()}
|
|
if not path.exists():
|
|
return status
|
|
status.update(_file_metadata(path, include_hash=include_hash))
|
|
try:
|
|
with _readonly_sqlite_connection(path) as conn:
|
|
if include_integrity:
|
|
status["integrity_check"] = str(conn.execute("PRAGMA integrity_check").fetchone()[0])
|
|
status["table_counts"] = _safe_table_counts(conn)
|
|
status["aggregate_totals"] = _safe_monthly_aggregates(conn)
|
|
except Exception as e:
|
|
status["integrity_check"] = "error"
|
|
status["error"] = str(e)
|
|
return status
|
|
|
|
|
|
def get_ledger_db_status(db_path: str | None = None, include_hash: bool = True, include_integrity: bool = True) -> dict[str, Any]:
|
|
"""Return non-sensitive metadata evidence for the selected ledger DB."""
|
|
path = resolve_db_path(db_path)
|
|
if not path.exists():
|
|
return fail("not_found", f"ledger DB not found: {path}", db_path=str(path), exists=False)
|
|
return ok(**_ledger_db_status_for_path(path, include_hash=include_hash, include_integrity=include_integrity))
|
|
|
|
|
|
def _resolve_backup_dir(backup_dir: str | None = None) -> Path:
|
|
return Path(backup_dir).expanduser().resolve() if backup_dir else DEFAULT_LEDGER_BACKUP_DIR.resolve()
|
|
|
|
|
|
def _backup_file_metadata(path: Path, include_hash: bool = False) -> dict[str, Any]:
|
|
metadata = _file_metadata(path, include_hash=include_hash)
|
|
metadata["name"] = path.name
|
|
metadata["restore_policy"] = BACKUP_RESTORE_POLICY
|
|
return metadata
|
|
|
|
|
|
def _apply_backup_retention(backup_dir: Path, retention_days: int = 30, retention_count: int = 20) -> list[str]:
|
|
backups = sorted(backup_dir.glob("*.sqlite3"), key=lambda p: p.stat().st_mtime, reverse=True)
|
|
keep_count = max(1, int(retention_count))
|
|
# Retention comparison needs sub-second precision for retention_days=0:
|
|
# the timestamp used for backup names is second-truncated, but filesystem mtimes
|
|
# can identify files created just before retention runs as older than "now".
|
|
cutoff = datetime.now(KST) - timedelta(days=max(0, int(retention_days)))
|
|
pruned: list[str] = []
|
|
for idx, path in enumerate(backups):
|
|
mtime = datetime.fromtimestamp(path.stat().st_mtime, tz=KST)
|
|
should_keep = idx < keep_count or mtime >= cutoff
|
|
if should_keep:
|
|
continue
|
|
path.unlink()
|
|
pruned.append(str(path.resolve()))
|
|
return pruned
|
|
|
|
|
|
def create_ledger_backup(db_path: str | None = None, backup_dir: str | None = None, retention_days: int = 30, retention_count: int = 20) -> dict[str, Any]:
|
|
"""Create a timestamped SQLite backup and return metadata-only evidence."""
|
|
source_path = resolve_db_path(db_path)
|
|
if not source_path.exists():
|
|
return fail("not_found", f"ledger DB not found: {source_path}", db_path=str(source_path), exists=False)
|
|
backup_root = _resolve_backup_dir(backup_dir)
|
|
backup_root.mkdir(parents=True, exist_ok=True)
|
|
timestamp = _current_kst_datetime().strftime("%Y%m%d%H%M%S%f")
|
|
backup_path = backup_root / f"ledger_{timestamp}.sqlite3"
|
|
try:
|
|
# Use SQLite's online backup API so WAL-mode databases are copied consistently.
|
|
with sqlite3.connect(f"file:{source_path}?mode=ro", uri=True) as source_conn:
|
|
with sqlite3.connect(backup_path) as backup_conn:
|
|
source_conn.backup(backup_conn)
|
|
except Exception:
|
|
if backup_path.exists():
|
|
backup_path.unlink()
|
|
raise
|
|
source_status = _ledger_db_status_for_path(source_path, include_hash=True, include_integrity=True)
|
|
backup_status = _ledger_db_status_for_path(backup_path, include_hash=True, include_integrity=True)
|
|
pruned_paths = _apply_backup_retention(backup_root, retention_days=retention_days, retention_count=retention_count)
|
|
return ok(
|
|
source_db=source_status,
|
|
backup=backup_status,
|
|
backup_dir=str(backup_root),
|
|
retention={"retention_days": int(retention_days), "retention_count": int(retention_count), "pruned_paths": pruned_paths},
|
|
restore_policy=BACKUP_RESTORE_POLICY,
|
|
)
|
|
|
|
|
|
def list_ledger_backups(backup_dir: str | None = None, include_hash: bool = False, limit: int = 50) -> dict[str, Any]:
|
|
backup_root = _resolve_backup_dir(backup_dir)
|
|
if not backup_root.exists():
|
|
return ok(backup_dir=str(backup_root), backups=[], count=0, restore_policy=BACKUP_RESTORE_POLICY)
|
|
max_items = max(1, min(int(limit), 500))
|
|
files = sorted(backup_root.glob("*.sqlite3"), key=lambda p: p.stat().st_mtime, reverse=True)[:max_items]
|
|
backups = [_backup_file_metadata(path, include_hash=include_hash) for path in files]
|
|
return ok(backup_dir=str(backup_root), backups=backups, count=len(backups), restore_policy=BACKUP_RESTORE_POLICY)
|
|
|
|
|
|
def get_ledger_backup_status(file_path: str, include_hash: bool = True, include_integrity: bool = True) -> dict[str, Any]:
|
|
path = Path(file_path).expanduser().resolve()
|
|
if not path.exists():
|
|
return fail("not_found", f"backup file not found: {path}", path=str(path), exists=False)
|
|
return ok(backup=_ledger_db_status_for_path(path, include_hash=include_hash, include_integrity=include_integrity), restore_policy=BACKUP_RESTORE_POLICY)
|
|
|
|
from .accounts_write_services import create_ledger_account as create_ledger_account
|
|
|
|
def list_ledger_accounts(active_only: bool = True, db_path: str | None = None) -> dict[str, Any]:
|
|
sql = "SELECT * FROM accounts" + (" WHERE active=1" if active_only else "") + " ORDER BY name"
|
|
with db(db_path) as conn:
|
|
rows = conn.execute(sql).fetchall()
|
|
return ok(accounts=[row_to_dict(r) for r in rows], count=len(rows))
|