Project Files
reference / exif-sniffer / src / exifsniffer / filesystem_access.py
"""Filesystem tools scoped to a configured base directory (LM Studio filesystem-access style).
Mirrors taderich73/filesystem-access: join base + relative path, reject ``..`` segments in the
relative string, and confine targets with :func:`exifsniffer.paths.resolve_under_root` (resolve +
``relative_to`` guard).
Adds ``extract_metadata`` and ``write_metadata`` (EXIF) helpers that use the same base directory
and ``file_name`` path rules as ``read_file`` / ``write_file``.
"""
from __future__ import annotations
import json
import re
from pathlib import Path
from typing import Any
from exifsniffer.exif_edit import update_image_exif
from exifsniffer.extract import extract_metadata_list, flatten_to_metadata_list
from exifsniffer.paths import resolve_under_root
# Same character class as the reference plugin's zod regex: ^[\w./-]+$
# Match JS ``\w`` (ASCII) plus ``.``, ``/``, ``-`` — same intent as the reference zod regex.
_RELATIVE_NAME_PATTERN = re.compile(r"^[\w./-]+$", re.ASCII)
def validate_relative_name(name: str, *, label: str) -> None:
if not name or not name.strip():
raise ValueError(f"{label} cannot be empty")
if not _RELATIVE_NAME_PATTERN.fullmatch(name):
raise ValueError(
f"{label} may only contain letters, numbers, underscores, hyphens, dots, and slashes"
)
for part in Path(name).parts:
if part == "..":
raise ValueError(f"{label} must not contain parent directory segments")
def configured_base_or_error(base: Path | None) -> Path:
if base is None:
raise ValueError(
"Error: Directory not set. Set environment variable LOCAL_MEDIA_BASE to an absolute "
"path (equivalent to the LM Studio plugin 'Base Directory' / folderName field)."
)
resolved = base.expanduser().resolve()
if not resolved.exists() or not resolved.is_dir():
raise ValueError(f"Error: Directory not set or does not exist ({resolved})")
return resolved
def fs_list_files(base: Path | None) -> list[dict[str, Any]]:
try:
root = configured_base_or_error(base)
except ValueError as exc:
return [{"path": "list_files.error", "value": str(exc)}]
try:
names = sorted(p.name for p in root.iterdir())
except OSError as exc:
return [{"path": "list_files.error", "value": f"Error: cannot list directory: {exc}"}]
if not names:
return [{"path": "list_files.message", "value": "Directory is empty"}]
rows: list[dict[str, Any]] = []
for i, name in enumerate(names):
rows.append({"path": f"list_files.entries[{i}]", "value": name})
return rows
def fs_read_file(base: Path | None, file_name: str) -> list[dict[str, Any]]:
try:
validate_relative_name(file_name, label="file_name")
except ValueError as exc:
return [{"path": "read_file.error", "value": str(exc)}]
try:
root = configured_base_or_error(base)
except ValueError as exc:
return [{"path": "read_file.error", "value": str(exc)}]
try:
full_path = resolve_under_root(root, file_name)
except ValueError as exc:
return [{"path": "read_file.error", "value": f"Error: File path is outside the configured directory: {exc}"}]
if not full_path.is_file():
return [{"path": "read_file.error", "value": "Error: File does not exist"}]
try:
text = full_path.read_text(encoding="utf-8")
except OSError as exc:
return [{"path": "read_file.error", "value": f"Error: cannot read file: {exc}"}]
return [{"path": "read_file.content", "value": text}]
def fs_write_file(base: Path | None, file_name: str, content: str) -> list[dict[str, Any]]:
try:
validate_relative_name(file_name, label="file_name")
except ValueError as exc:
return [{"path": "write_file.error", "value": str(exc)}]
try:
root = configured_base_or_error(base)
except ValueError as exc:
return [{"path": "write_file.error", "value": str(exc)}]
try:
full_path = resolve_under_root(root, file_name)
except ValueError as exc:
return [{"path": "write_file.error", "value": f"Error: File path is outside the configured directory: {exc}"}]
full_path.parent.mkdir(parents=True, exist_ok=True)
try:
full_path.write_text(content, encoding="utf-8")
except OSError as exc:
return [{"path": "write_file.error", "value": f"Error: cannot write file: {exc}"}]
return [{"path": "write_file.message", "value": "File created or updated successfully"}]
def fs_create_directory(base: Path | None, directory_name: str) -> list[dict[str, Any]]:
try:
validate_relative_name(directory_name, label="directory_name")
except ValueError as exc:
return [{"path": "create_directory.error", "value": str(exc)}]
try:
root = configured_base_or_error(base)
except ValueError as exc:
return [{"path": "create_directory.error", "value": str(exc)}]
try:
full_path = resolve_under_root(root, directory_name)
except ValueError as exc:
return [
{
"path": "create_directory.error",
"value": f"Error: Directory path is outside the configured directory: {exc}",
}
]
try:
full_path.mkdir(parents=True, exist_ok=True)
except OSError as exc:
return [{"path": "create_directory.error", "value": f"Error: cannot create directory: {exc}"}]
return [{"path": "create_directory.message", "value": f"Directory '{directory_name}' created successfully"}]
def fs_extract_metadata(
base: Path | None,
source_file_name: str,
output_json_file_name: str | None,
*,
include_piexif: bool = False,
) -> list[dict[str, Any]]:
"""Extract image/video metadata from a file under *base*; optionally write a JSON array under *base*."""
try:
validate_relative_name(source_file_name, label="source_file_name")
except ValueError as exc:
return [{"path": "extract_metadata.error", "value": str(exc)}]
if output_json_file_name is not None and output_json_file_name.strip() != "":
try:
validate_relative_name(output_json_file_name, label="output_json_file_name")
except ValueError as exc:
return [{"path": "extract_metadata.error", "value": str(exc)}]
try:
root = configured_base_or_error(base)
except ValueError as exc:
return [{"path": "extract_metadata.error", "value": str(exc)}]
try:
full_source = resolve_under_root(root, source_file_name)
except ValueError as exc:
return [
{
"path": "extract_metadata.error",
"value": f"Error: File path is outside the configured directory: {exc}",
}
]
if not full_source.is_file():
return [{"path": "extract_metadata.error", "value": "Error: File does not exist"}]
try:
rows = extract_metadata_list(full_source, include_piexif=include_piexif)
except (FileNotFoundError, OSError, RuntimeError) as exc:
return [{"path": "extract_metadata.error", "value": str(exc)}]
out_name = output_json_file_name.strip() if output_json_file_name else ""
if out_name:
try:
full_out = resolve_under_root(root, out_name)
except ValueError as exc:
return [
{
"path": "extract_metadata.error",
"value": f"Error: Output path is outside the configured directory: {exc}",
}
]
try:
full_out.parent.mkdir(parents=True, exist_ok=True)
full_out.write_text(json.dumps(rows, indent=2, ensure_ascii=False), encoding="utf-8")
except OSError as exc:
return [{"path": "extract_metadata.error", "value": f"Error: cannot write JSON file: {exc}"}]
return rows
def fs_write_metadata(
base: Path | None,
file_name: str,
set_tags: dict[str, dict[str, Any]] | None,
remove_tags: dict[str, list[str]] | None,
*,
path_prefix: str = "write_metadata",
) -> list[dict[str, Any]]:
"""Update EXIF on a JPEG/WebP under *base* (same path rules as ``write_file``)."""
err = f"{path_prefix}.error"
try:
validate_relative_name(file_name, label="file_name")
except ValueError as exc:
return [{"path": err, "value": str(exc)}]
try:
root = configured_base_or_error(base)
except ValueError as exc:
return [{"path": err, "value": str(exc)}]
try:
full_path = resolve_under_root(root, file_name)
except ValueError as exc:
return [
{
"path": err,
"value": f"Error: File path is outside the configured directory: {exc}",
}
]
try:
summary = update_image_exif(
full_path,
set_tags=set_tags or {},
remove_tags=remove_tags or {},
)
except (FileNotFoundError, ValueError, RuntimeError, OSError) as exc:
return [{"path": err, "value": str(exc)}]
return flatten_to_metadata_list(summary, prefix=path_prefix)