fastapi_app/lib/utils/server_utils.py

"""
Server utility functions for PDF-TEI-Editor.

This module provides framework-agnostic server utilities with dependency injection.
Includes FastAPI-specific helpers for request handling.
"""

import re
import socket
from datetime import datetime
from pathlib import Path
from typing import Optional


class ApiError(RuntimeError):
    """
    Custom exception class for API-specific errors.

    Attributes:
        message -- explanation of the error
        status_code -- the HTTP (or other) status code associated with the error
    """
    def __init__(self, message, status_code=400):
        super().__init__(message)
        self.status_code = status_code


# Pure utility functions (no dependencies)


def has_internet() -> bool:
    """
    Test internet connectivity by attempting to connect to Google's DNS server.
    This is faster than HTTP requests as it only checks connectivity without data transfer.

    Returns:
        True if internet connection is available, False otherwise
    """
    try:
        socket.setdefaulttimeout(3)
        socket.socket(socket.AF_INET, socket.SOCK_STREAM).connect(("8.8.8.8", 53))
        return True
    except socket.error:
        return False


def make_timestamp() -> str:
    """
    Create a timestamp string in format "YYYY-MM-DD HH:MM:SS".

    Returns:
        Formatted timestamp string
    """
    now = datetime.now()
    return now.strftime("%Y-%m-%d %H:%M:%S")


def make_version_timestamp() -> str:
    """
    Create a timestamp formatted for version filenames (safe for filesystem).

    Format: YYYY-MM-DD_HH-MM-SS

    Returns:
        Filesystem-safe timestamp string
    """
    return make_timestamp().replace(" ", "_").replace(":", "-")


# Version timestamp format constants
VERSION_TIMESTAMP_REGEX = re.compile(r'^\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2}-')


def strip_version_timestamp_prefix(filename: str) -> str:
    """
    Remove version timestamp prefix from filename if present.

    Args:
        filename: Filename that may have timestamp prefix

    Returns:
        Filename without timestamp prefix
    """
    return VERSION_TIMESTAMP_REGEX.sub('', filename)


def safe_file_path(file_path: str) -> str:
    """
    Sanitize file path for safe filesystem operations.

    Removes non-alphabetic leading characters, strips "/data" prefix,
    and replaces filesystem-incompatible characters with underscores.

    Args:
        file_path: File path to sanitize

    Returns:
        Sanitized file path

    Raises:
        ApiError: If file path is invalid or empty
    """
    if not file_path:
        raise ApiError("Invalid file path: path is empty", status_code=400)

    # Remove leading non-alphabetic characters
    while len(file_path) > 0 and not file_path[0].isalpha():
        file_path = file_path[1:]

    if not file_path.startswith("data/"):
        raise ApiError(f"Invalid file path: {file_path}", status_code=400)

    # Remove the 'data/' prefix
    cleaned_path = file_path.removeprefix('data/')

    # Split path into components, sanitize each component, then rejoin
    path_parts = cleaned_path.split('/')
    sanitized_parts = []

    for part in path_parts:
        if part:  # Skip empty parts
            # Replace characters incompatible with Windows/POSIX filesystems
            sanitized_part = ''
            for char in part:
                if char in '%<>:"|?*\\' or ord(char) < 32:
                    sanitized_part += '_'
                else:
                    sanitized_part += char
            sanitized_parts.append(sanitized_part)

    return '/'.join(sanitized_parts)


# Functions with dependency injection


def get_data_file_path(document_id: str, data_root: Path, file_type: str, version: str = None) -> Path:
    """
    Get the full path to a data file.

    Note: This is a legacy function. New code should use hash-based storage
    via hash_utils.get_storage_path().

    Args:
        document_id: Document identifier
        data_root: Root directory for data files
        file_type: Type of file (for extension determination)
        version: Optional version identifier

    Returns:
        Full path to the data file
    """
    # For now, simple path construction
    # Will be replaced with hash-based lookup in Phase 2
    safe_path = safe_file_path(f"data/{document_id}")
    return data_root / safe_path


def get_version_path(data_root: Path, file_hash: str, version: str) -> Path:
    """
    Get the path for a versioned file.

    Args:
        data_root: Root directory for data files
        file_hash: Hash of the file
        version: Version timestamp

    Returns:
        Path to the version file
    """
    return data_root / "versions" / file_hash / f"{version}-{file_hash}.tei.xml"


def resolve_document_identifier(doc_id: str, db_dir: Path, logger=None) -> str:
    """
    Resolve a document identifier that can be either a file path or a hash.

    This is a placeholder for Phase 2 when SQLite lookup will be implemented.

    Args:
        doc_id: Document identifier (path or hash)
        db_dir: Database directory
        logger: Optional logger

    Returns:
        Resolved file path

    Raises:
        ApiError: If identifier cannot be resolved
    """
    if not doc_id:
        raise ApiError("Document identifier is empty", status_code=400)

    # Direct path handling (test mode only)
    if doc_id.startswith('/data/'):
        import os
        if not os.environ.get('TEST_IN_PROGRESS'):
            raise ApiError("Direct file paths are only allowed in test mode", status_code=400)
        return doc_id

    # Hash lookup - placeholder for Phase 2 SQLite implementation
    # For now, treat as error
    raise ApiError(f"Hash lookup not yet implemented in Phase 1", status_code=501)


# FastAPI-specific helpers


try:
    from fastapi import Request

    def get_session_id_from_request(request: Request) -> Optional[str]:
        """
        Extract session ID from FastAPI request.

        Checks in order:
        1. Headers (X-Session-Id) - Primary method for per-tab sessions
        2. Query parameters (sessionId) - For EventSource/SSE connections
        3. Cookies (sessionId) - Fallback (note: shared across tabs)

        Args:
            request: FastAPI Request object

        Returns:
            Session ID if found, None otherwise
        """
        # 1. Try headers (primary method - supports per-tab sessions)
        session_id = request.headers.get('X-Session-Id')
        if session_id:
            return session_id

        # 2. Try query parameters (for EventSource/SSE)
        session_id = request.query_params.get('sessionId')
        if session_id:
            return session_id

        # 3. Fall back to cookies (shared across tabs)
        return request.cookies.get('sessionId')

except ImportError:
    # FastAPI not available, skip FastAPI-specific functions
    pass