"""Utility functions for My iMessage MCP server."""
from pathlib import Path
from typing import Any, TypeVar
from .constants import DEFAULT_PAGINATION_LIMIT, DEFAULT_PAGINATION_OFFSET
T = TypeVar("T")
def ensure_file_uri(file_path: str) -> str:
"""Convert file path to proper file URI.
Args:
file_path: Path to the file (absolute, relative, or already a URI)
Returns:
Properly formatted file:// URI
"""
if file_path.startswith("file://"):
return file_path
path = Path(file_path)
if not path.is_absolute():
path = Path.cwd() / path
return f"file://{path.absolute()}"
def apply_pagination(
items: list[T],
offset: int = DEFAULT_PAGINATION_OFFSET,
limit: int = DEFAULT_PAGINATION_LIMIT,
add_offset_field: bool = True,
) -> tuple[list[T | dict[str, Any]], dict[str, Any]]:
"""Apply pagination to a list of items.
Args:
items: The full list of items to paginate
offset: Number of items to skip
limit: Maximum number of items to return
add_offset_field: Whether to add an 'offset' field to each item
Returns:
Tuple of (paginated_items, metadata_dict)
"""
total_items = len(items)
start_idx = min(offset, total_items)
end_idx = min(start_idx + limit, total_items)
paginated = items[start_idx:end_idx]
result_items: list[T | dict[str, Any]]
if add_offset_field:
processed_items: list[dict[str, Any]] = []
for i, item in enumerate(paginated):
if isinstance(item, dict):
processed_item = item.copy()
else:
processed_item = {"item": item}
processed_item["offset"] = start_idx + i
processed_items.append(processed_item)
result_items = processed_items # type: ignore[assignment]
else:
result_items = list(paginated)
has_more = end_idx < total_items
metadata = {
"totalItems": total_items,
"offset": offset,
"limit": limit,
"hasMore": has_more,
"nextOffset": end_idx if has_more else None,
}
return result_items, metadata
