93 lines
2.6 KiB
Python
93 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
from pathlib import Path
|
|
|
|
from dockervault.models import ComposeProject, DEFAULT_COMPOSE_FILENAMES, normalize_path
|
|
|
|
SERVICE_LINE_RE = re.compile(r"^\s{2}([A-Za-z0-9_.-]+):\s*$")
|
|
|
|
|
|
def find_compose_files(base_path: Path) -> list[Path]:
|
|
"""Find likely Docker Compose files under base_path."""
|
|
matches: list[Path] = []
|
|
|
|
for path in base_path.rglob("*"):
|
|
if path.is_file() and path.name in DEFAULT_COMPOSE_FILENAMES:
|
|
matches.append(path)
|
|
|
|
return sorted(matches)
|
|
|
|
|
|
def parse_services_from_compose(compose_path: Path) -> list[str]:
|
|
"""
|
|
Light parser for service names.
|
|
|
|
Keeps dependencies minimal for v0.
|
|
It looks for the `services:` block and collects entries indented by two spaces.
|
|
"""
|
|
try:
|
|
lines = compose_path.read_text(encoding="utf-8").splitlines()
|
|
except UnicodeDecodeError:
|
|
lines = compose_path.read_text(encoding="utf-8", errors="ignore").splitlines()
|
|
|
|
in_services = False
|
|
services: list[str] = []
|
|
|
|
for line in lines:
|
|
stripped = line.strip()
|
|
|
|
if not stripped or stripped.startswith("#"):
|
|
continue
|
|
|
|
if not in_services:
|
|
if stripped == "services:":
|
|
in_services = True
|
|
continue
|
|
|
|
# Leaving top-level services block
|
|
if not line.startswith(" ") and not line.startswith("\t"):
|
|
break
|
|
|
|
match = SERVICE_LINE_RE.match(line)
|
|
if match:
|
|
services.append(match.group(1))
|
|
|
|
return sorted(set(services))
|
|
|
|
|
|
def group_projects_by_folder(compose_files: list[Path]) -> list[ComposeProject]:
|
|
grouped: dict[Path, list[Path]] = {}
|
|
|
|
for compose_file in compose_files:
|
|
grouped.setdefault(compose_file.parent, []).append(compose_file)
|
|
|
|
projects: list[ComposeProject] = []
|
|
|
|
for folder, files in sorted(grouped.items()):
|
|
service_names: set[str] = set()
|
|
|
|
for compose_file in files:
|
|
service_names.update(parse_services_from_compose(compose_file))
|
|
|
|
projects.append(
|
|
ComposeProject(
|
|
name=folder.name,
|
|
root_path=normalize_path(folder),
|
|
compose_files=[file.name for file in sorted(files)],
|
|
services=sorted(service_names),
|
|
)
|
|
)
|
|
|
|
return projects
|
|
|
|
|
|
def scan_projects(base_path: Path) -> list[ComposeProject]:
|
|
if not base_path.exists():
|
|
raise FileNotFoundError(f"Path does not exist: {base_path}")
|
|
|
|
if not base_path.is_dir():
|
|
raise NotADirectoryError(f"Path is not a directory: {base_path}")
|
|
|
|
compose_files = find_compose_files(base_path)
|
|
return group_projects_by_folder(compose_files)
|