francwa
2df7843d8b
Replaces the monolithic FileManager class + scattered helpers in
alfred/infrastructure/filesystem with five free functions, each
single-responsibility and pathlib-native:
list_dir / create_dir / link_file / move_file / move_dir
The infra layer now raises typed exceptions (FilesystemError base
+ SourceNotFound / DestinationExists / NotADirectory / NotAFile /
PermissionDenied / CrossDevice / FilesystemOSError) instead of
returning {status: ok|error} dicts. No more get_memory() reads
from infra.
Application layer mirrors the same split: five free use cases
(<op>_use_case) wrap each infra op, guard inputs against escaping
the new DirectoryRoots VO (downloads / torrents / movies /
tv_shows), catch infra exceptions, and return frozen DTOs. Roots
are injected — no global state.
Legacy files kept on disk with _OLD suffix for reference during
the follow-up rewiring (FileManager, MediaOrganizer,
create_folder/move helpers; CreateSeedLinks/ListFolder/MoveMedia/
ManageSubtitles use cases, resolve_destination). They are no
longer exported from __init__, which intentionally breaks current
agent tool wrappers and downstream tests — re-wiring is the next
chunk of work on the unfuck branch.
74 lines
2.0 KiB
Python
74 lines
2.0 KiB
Python
"""Low-level filesystem operations — one responsibility per function."""
|
|
|
|
import logging
|
|
import subprocess
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _err(error: str, message: str) -> dict[str, Any]:
|
|
return {"status": "error", "error": error, "message": message}
|
|
|
|
|
|
def create_folder(path: str) -> dict[str, Any]:
|
|
"""
|
|
Create a directory and all missing parents.
|
|
|
|
Args:
|
|
path: Absolute path to the directory to create.
|
|
|
|
Returns:
|
|
Dict with status and path, or error details.
|
|
"""
|
|
try:
|
|
p = Path(path)
|
|
p.mkdir(parents=True, exist_ok=True)
|
|
logger.info(f"Folder ready: {p}")
|
|
return {"status": "ok", "path": str(p)}
|
|
except OSError as e:
|
|
logger.error(f"create_folder failed: {e}")
|
|
return _err("mkdir_failed", str(e))
|
|
|
|
|
|
def move(source: str, destination: str) -> dict[str, Any]:
|
|
"""
|
|
Move a file or folder to a destination path.
|
|
|
|
Uses the system mv command — instant on the same filesystem (ZFS rename).
|
|
|
|
Args:
|
|
source: Absolute path to the source file or folder.
|
|
destination: Absolute path to the destination.
|
|
|
|
Returns:
|
|
Dict with status, source, destination — or error details.
|
|
"""
|
|
src = Path(source)
|
|
dst = Path(destination)
|
|
|
|
if not src.exists():
|
|
return _err("source_not_found", f"Source does not exist: {source}")
|
|
|
|
if dst.exists():
|
|
return _err("destination_exists", f"Destination already exists: {destination}")
|
|
|
|
try:
|
|
result = subprocess.run(
|
|
["mv", str(src), str(dst)],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
if result.returncode != 0:
|
|
logger.error(f"mv failed: {result.stderr}")
|
|
return _err("move_failed", result.stderr.strip())
|
|
|
|
logger.info(f"Moved: {src} -> {dst}")
|
|
return {"status": "ok", "source": str(src), "destination": str(dst)}
|
|
|
|
except OSError as e:
|
|
logger.error(f"move failed: {e}")
|
|
return _err("move_failed", str(e))
|