86 lines
2.6 KiB
Python
86 lines
2.6 KiB
Python
"""URL to local path conversion for Azure Blob Storage URLs."""
|
|
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
from urllib.parse import unquote
|
|
|
|
from ..config import Config, URLMapping
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class URLConverter:
|
|
"""Converts Azure Blob Storage URLs to local file paths."""
|
|
|
|
def __init__(self, config: Config):
|
|
"""Initialize converter with configuration.
|
|
|
|
Args:
|
|
config: Application configuration with URL mappings
|
|
"""
|
|
self.config = config
|
|
|
|
def convert_url_to_path(self, url: str) -> Optional[Path]:
|
|
"""Convert Azure Blob Storage URL to local file path.
|
|
|
|
Args:
|
|
url: Azure Blob Storage URL
|
|
|
|
Returns:
|
|
Local file path if mapping found, None otherwise
|
|
|
|
Example:
|
|
>>> converter.convert_url_to_path(
|
|
... "https://devagravitystg.file.core.windows.net/devagravitysync/aN5PysnXIuRECzcRbvHkjL7g0/file.png"
|
|
... )
|
|
Path("Z:/aN5PysnXIuRECzcRbvHkjL7g0/file.png")
|
|
"""
|
|
if not url:
|
|
return None
|
|
|
|
# URL decode (handles special characters like spaces)
|
|
url = unquote(url)
|
|
|
|
# Find matching URL mapping
|
|
for mapping in self.config.url_mappings:
|
|
if url.startswith(mapping.url_prefix):
|
|
# Extract relative path after prefix
|
|
relative_path = url[len(mapping.url_prefix):]
|
|
|
|
# Combine with local path
|
|
local_path = Path(mapping.local_path) / relative_path
|
|
|
|
# Normalize path (resolve .. and .) but don't follow symlinks yet
|
|
try:
|
|
# On Windows, normalize separators
|
|
local_path = Path(str(local_path).replace("/", "\\"))
|
|
except (OSError, RuntimeError) as e:
|
|
logger.warning(f"Failed to normalize path {local_path}: {e}")
|
|
return None
|
|
|
|
logger.debug(f"Converted URL to path: {url} -> {local_path}")
|
|
return local_path
|
|
|
|
logger.debug(f"No mapping found for URL: {url}")
|
|
return None
|
|
|
|
def is_azure_url(self, text: str) -> bool:
|
|
"""Check if text is an Azure Blob Storage URL.
|
|
|
|
Args:
|
|
text: Text to check
|
|
|
|
Returns:
|
|
True if text matches configured URL prefixes
|
|
"""
|
|
if not text:
|
|
return False
|
|
|
|
text = text.strip()
|
|
|
|
for mapping in self.config.url_mappings:
|
|
if text.startswith(mapping.url_prefix):
|
|
return True
|
|
|
|
return False
|