from __future__ import annotations import os from pathlib import Path from app.config import settings SHARED_DIR_MODE = 0o2775 def _managed_directory_chain(path: Path) -> list[Path]: """Return upload-root-relative directories that should share writable perms.""" resolved_path = path.resolve(strict=False) upload_root = Path(settings.upload_dir).resolve(strict=False) if resolved_path != upload_root and upload_root not in resolved_path.parents: return [path] chain: list[Path] = [upload_root] current = upload_root try: relative_parts = resolved_path.relative_to(upload_root).parts except ValueError: return [path] for part in relative_parts: current = current / part chain.append(current) return chain def _normalize_directory_mode(path: Path, *, mode: int = SHARED_DIR_MODE) -> None: try: current_mode = path.stat().st_mode & 0o7777 except OSError: return desired_mode = mode if current_mode == desired_mode: return try: os.chmod(path, desired_mode) except OSError: # Best-effort only: callers still get the path, but existing root-owned # trees can be repaired when the process has sufficient permissions. return def ensure_group_writable_dir(path: str | Path, *, mode: int = SHARED_DIR_MODE) -> Path: """Create a directory and normalize upload-tree permissions for shared workers.""" dir_path = Path(path) for candidate in _managed_directory_chain(dir_path): candidate.mkdir(parents=True, exist_ok=True) _normalize_directory_mode(candidate, mode=mode) return dir_path def resolve_public_asset_url(url: str | None) -> Path | None: """Resolve a public static asset URL like /renders/... to a local disk path.""" if not url: return None normalized = url.replace("\\", "/") if normalized.startswith("/renders/"): candidate = Path(settings.upload_dir) / "renders" / normalized[len("/renders/"):] elif normalized.startswith("/thumbnails/"): candidate = Path(settings.upload_dir) / "thumbnails" / normalized[len("/thumbnails/"):] else: return None return candidate def resolve_result_path(result_path: str | None) -> Path | None: """Resolve stored result_path variants to a local disk path. Supports canonical /app/uploads/... paths, legacy /shared/... paths, public URLs, and bare storage keys such as renders//file.png. """ if not result_path: return None normalized = result_path.replace("\\", "/") for marker in ("/uploads/", "/shared/"): if marker in normalized: relative = normalized.split(marker, 1)[1].lstrip("/") return Path(settings.upload_dir) / relative public_candidate = resolve_public_asset_url(normalized) if public_candidate is not None: return public_candidate stripped = normalized.lstrip("/") if stripped.startswith(("renders/", "thumbnails/", "exports/", "usd/", "step_files/")): return Path(settings.upload_dir) / stripped if Path(normalized).is_absolute(): return Path(normalized) return None def result_path_to_storage_key(result_path: str | None) -> str | None: """Normalize stored paths to a canonical relative storage key when possible.""" if not result_path: return None normalized = result_path.replace("\\", "/") disk_path = resolve_result_path(result_path) if disk_path is not None: try: return disk_path.relative_to(Path(settings.upload_dir)).as_posix() except ValueError: pass public_candidate = normalized.lstrip("/") if public_candidate.startswith(("renders/", "thumbnails/", "exports/", "usd/", "step_files/")): return public_candidate return normalized def result_path_to_public_url( result_path: str | None, *, require_exists: bool = False, ) -> str | None: """Convert internal result paths to a servable public URL. Returns only /renders/... or /thumbnails/... URLs. Non-public internal paths like step_files/renders stay hidden from API/UI callers. """ if not result_path: return None disk_path = resolve_result_path(result_path) if require_exists: if disk_path is None or not disk_path.is_file(): return None normalized = result_path.replace("\\", "/") for marker in ("/renders/", "/thumbnails/"): if marker in normalized: idx = normalized.index(marker) public_url = normalized[idx:] candidate = resolve_public_asset_url(public_url) if require_exists and (candidate is None or not candidate.is_file()): return None return public_url if disk_path is None: return None try: relative = disk_path.relative_to(Path(settings.upload_dir)) except ValueError: return None relative_str = relative.as_posix() if relative_str.startswith(("renders/", "thumbnails/")): if require_exists and not disk_path.is_file(): return None return f"/{relative_str}" return None def build_order_line_step_render_path( step_path: str | Path, order_line_id: str, filename: str, *, ensure_exists: bool = False, ) -> Path: """Build a unique per-order-line render-worker artifact path beside the STEP file.""" artifact_dir = Path(step_path).parent / "renders" / str(order_line_id) if ensure_exists: ensure_group_writable_dir(artifact_dir) return artifact_dir / filename def build_order_line_export_path( order_line_id: str, filename: str, *, ensure_exists: bool = False, ) -> Path: """Build a unique per-order-line export artifact path under the shared upload root.""" artifact_dir = Path(settings.upload_dir) / "exports" / str(order_line_id) if ensure_exists: ensure_group_writable_dir(artifact_dir) return artifact_dir / filename