bruno
4c4e415975
CI / lint (push) Successful in 58s
CI / security (push) Successful in 40s
CI / test (push) Successful in 1m15s
CI / build (push) Successful in 37s
CI / e2e (push) Successful in 10m15s
Desktop Build / build-windows (push) Canceled after 0s
Desktop Build / build-linux (push) Canceled after 0s
87 lines
2.9 KiB
Python
87 lines
2.9 KiB
Python
"""File reading services shared by REST routes and the AI tool layer."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
from backend.services.errors import ServiceError
|
|
from backend.services.paths import resolve_safe_path
|
|
from backend.services.vaults import get_vault_root
|
|
|
|
logger = logging.getLogger("obsigate.services.files")
|
|
|
|
|
|
def read_raw_file(vault_name: str, path: str) -> dict[str, Any]:
|
|
"""Return the raw text content of a vault file (no redaction)."""
|
|
root = get_vault_root(vault_name)
|
|
file_path = resolve_safe_path(root, path)
|
|
|
|
if not file_path.exists() or not file_path.is_file():
|
|
raise ServiceError(
|
|
f"File not found: {path}",
|
|
code="not_found",
|
|
status=404,
|
|
details={"vault": vault_name, "path": path},
|
|
)
|
|
|
|
try:
|
|
raw = file_path.read_text(encoding="utf-8", errors="replace")
|
|
except PermissionError as e:
|
|
logger.error(f"Permission denied reading raw file {path}: {e}")
|
|
raise ServiceError(f"Permission denied: cannot read file {path}", code="permission_denied", status=403) from e
|
|
except UnicodeDecodeError:
|
|
try:
|
|
raw = file_path.read_bytes().decode("utf-8", errors="replace")
|
|
except Exception as e:
|
|
logger.error(f"Error reading binary raw file {path}: {e}")
|
|
raise ServiceError(f"Cannot read file: {e!s}", code="read_error", status=500) from e
|
|
except Exception as e:
|
|
logger.error(f"Unexpected error reading raw file {path}: {e}")
|
|
raise ServiceError(f"Error reading file: {e!s}", code="read_error", status=500) from e
|
|
|
|
return {"vault": vault_name, "path": path, "raw": raw}
|
|
|
|
|
|
def read_file_text(
|
|
vault_name: str,
|
|
path: str,
|
|
*,
|
|
redact: bool = True,
|
|
max_bytes: int | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Return a vault file's text content, optionally redacted and size-capped.
|
|
|
|
Raises:
|
|
ServiceError: ``not_found`` (404), ``file_too_large`` (413) or a read
|
|
error (500).
|
|
"""
|
|
root = get_vault_root(vault_name)
|
|
target = resolve_safe_path(root, path)
|
|
|
|
if not target.exists() or not target.is_file():
|
|
raise ServiceError(
|
|
f"File not found: {path}",
|
|
code="not_found",
|
|
status=404,
|
|
details={"vault": vault_name, "path": path},
|
|
)
|
|
|
|
size = target.stat().st_size
|
|
if max_bytes is not None and size > max_bytes:
|
|
raise ServiceError(
|
|
f"File too large ({size} bytes > {max_bytes})",
|
|
code="file_too_large",
|
|
status=413,
|
|
details={"vault": vault_name, "path": path, "size": size},
|
|
)
|
|
|
|
content = target.read_text(encoding="utf-8", errors="replace")
|
|
|
|
if redact:
|
|
from backend.secret_redactor import redact_file_content
|
|
|
|
content = redact_file_content(content, path)
|
|
|
|
return {"vault": vault_name, "path": path, "size": size, "content": content}
|