src/kittykat_agent/core/utils.py

import asyncio
import base64
from pathlib import Path
import re
from urllib.parse import urlparse
import shortuuid
from kittykat_agent.core.model import PaginationMeta
from typing import Coroutine, Iterable, List, Optional, TypeVar, Generic
from pydantic import BaseModel
import json
import copy
from typing import Any, Dict
from logger import logger

T = TypeVar("T")


class BaseApiResponse(BaseModel, Generic[T]):
    status_code: int
    message: str
    data: Optional[T] = None


def build_pagination_meta(skip: int, limit: int, total: int) -> PaginationMeta:
    has_next = skip + limit < total
    has_previous = skip > 0

    return PaginationMeta(
        total=total,
        skip=skip,
        limit=limit,
        has_next=has_next,
        has_previous=has_previous,
        next_skip=(skip + limit) if has_next else None,
        previous_skip=(max(skip - limit, 0) if has_previous else None)
    )


def get_filename_from_url(url: str) -> str:
    """
    Extracts the base filename (without extension) from a URL.

    Example:
        https://cdn.example.com/images/my_image_123.png -> "my_image_123"
    """
    return Path(urlparse(url).path).stem


def get_extension_from_url(url: str) -> str:
    """
    Extracts the file extension (without the dot) from a URL.

    Example:
        https://cdn.example.com/images/my_image_123.png -> "png"
    """
    return Path(urlparse(url).path).suffix.lstrip(".")


def get_image_type_from_base64(b64_string: str) -> str:
    """
    Detects the image type (png, jpeg, gif, webp, etc.) from a base64 string.
    Falls back to 'png' if type cannot be determined.
    """

    match = re.match(r"^data:image/([a-zA-Z0-9+]+);base64,", b64_string)
    if match:
        return match.group(1).lower()

    # Remove prefix if present
    if "," in b64_string:
        b64_string = b64_string.split(",", 1)[1]

    try:
        # decode first bytes only
        decoded = base64.b64decode(b64_string[:100], validate=True)
    except Exception:
        return "png"  # fallback if invalid base64

    if decoded.startswith(b"\x89PNG\r\n\x1a\n"):
        return "png"
    elif decoded.startswith(b"\xff\xd8\xff"):
        return "jpeg"
    elif decoded.startswith(b"GIF87a") or decoded.startswith(b"GIF89a"):
        return "gif"
    elif decoded.startswith(b"RIFF") and decoded[8:12] == b"WEBP":
        return "webp"

    # ✅ Fallback
    return "png"


def slugify(text: str) -> str:
    # Lowercase and strip
    text = text.strip().lower()
    # Replace all non-alphanumeric characters (including spaces) with underscores
    text = re.sub(r"[^\w]", "_", text)
    # Collapse multiple consecutive underscores to a single underscore
    text = re.sub(r"_+", "_", text)
    # Optional: Remove leading/trailing underscores
    text = text.strip("_")
    return text


def generate_team_name(name: Optional[str] = None, email: Optional[str] = None) -> str:
    """
    Generate a clean team name from user name or email.

    Args:
        name: User's full name (preferred)
        email: User's email (fallback)

    Returns:
        Formatted team name like "John Doe's workspace"

    Example:
        generate_team_name("John Doe") -> "John Doe's workspace"
        generate_team_name(None, "john.doe@example.com") -> "John's workspace"
    """
    if name and name.strip():
        source = name.strip()
    elif email:
        # Extract username part from email
        source = email.split("@")[0]
    else:
        source = "User"

    # Remove extra whitespace
    cleaned = re.sub(r"\s+", " ", source)
    # Remove special characters, keep only alphanumeric and spaces
    cleaned = re.sub(r"[^A-Za-z0-9\s]", "", cleaned)
    # Title case and create team name
    team_name = f"{cleaned.strip().title()}'s workspace"

    return team_name


def build_custom_filename(
    brand_name: Optional[str],
    campaign_title: Optional[str],
    output_format: str
) -> str:
    brand_slug = slugify(brand_name) if brand_name else "brand"
    campaign_slug = slugify(campaign_title) if campaign_title else None
    uid = shortuuid.uuid()[:8]
    ext = output_format.lower()

    if campaign_slug:
        return f"{brand_slug}-{campaign_slug}-{uid}.{ext}"
    return f"{brand_slug}-{uid}.{ext}"


def build_custom_filename_without_ext(
    brand_name: Optional[str],
    campaign_title: Optional[str],
) -> str:
    brand_slug = slugify(brand_name) if brand_name else "brand"
    campaign_slug = slugify(campaign_title) if campaign_title else None

    if campaign_slug:
        return f"{brand_slug}-{campaign_slug}"
    return f"{brand_slug}"


def is_platform_api_success(status_code: int) -> bool:
    return 200 <= status_code < 300


def truncate_strings(obj: Any, max_len: int = 500) -> Any:
    """
    Recursively truncate strings longer than max_len in any nested structure.
    """
    if isinstance(obj, str):
        return obj[:max_len] + "..." if len(obj) > max_len else obj

    elif isinstance(obj, list):
        return [truncate_strings(item, max_len) for item in obj]

    elif isinstance(obj, dict):
        return {key: truncate_strings(value, max_len) for key, value in obj.items()}

    elif isinstance(obj, tuple):
        return tuple(truncate_strings(item, max_len) for item in obj)

    elif isinstance(obj, set):
        return {truncate_strings(item, max_len) for item in obj}

    return obj


def safe_log_dict(
    data: Dict[str, Any],
    truncate_keys: Iterable[str],
) -> str:
    """
    Returns a JSON-safe version of a dict for logging,
    truncating values for specified keys.

    Args:
        data (dict): The dictionary to sanitize
        truncate_keys (Iterable[str]): Keys whose values should be truncated

    Returns:
        str: JSON string with sensitive fields truncated
    """
    safe_data = copy.deepcopy(data)
    for key in set(truncate_keys):
        if key in safe_data:
            val = safe_data[key]
            if isinstance(val, str):
                safe_data[key] = f"[base64 omitted, len={len(val)}]"
            elif isinstance(val, list):
                safe_data[key] = f"[list of {len(val)} base64 items omitted]"
            else:
                safe_data[key] = "[omitted]"
    truncated_obj = truncate_strings(safe_data)

    return json.dumps(truncated_obj, indent=2)


def fire_and_forget(coro: Coroutine) -> None:
    """
    Fire-and-forget utility for async operations that should not block.

    This is particularly useful for logging operations where we don't want
    to wait for the log update to complete before continuing processing.

    Args:
        coro: The coroutine to execute in the background
    """
    def callback(task: asyncio.Task) -> None:
        """Callback to handle task completion and log any errors."""
        try:
            task.result()
        except Exception as e:
            logger.error(
                f"Fire-and-forget task failed: {str(e)}", exc_info=True)

    # Create task and add callback
    task = asyncio.create_task(coro)
    task.add_done_callback(callback)


def format_key(key: str) -> str:
    """Format snake_case key to Title Case with spaces.

    Example: target_audience -> Target Audience
    """
    return " ".join(word.capitalize() for word in key.split("_"))


def format_dict_to_context(data: Dict[str, Any], prefix: str = "") -> List[str]:
    """Convert dictionary to formatted context strings.

    Args:
        data: Dictionary to format
        prefix: Optional prefix for nested keys

    Returns:
        List of formatted strings like "Key: value"
    """
    context_parts = []

    for key, value in data.items():
        if not value:
            continue

        formatted_key = format_key(key)
        if prefix:
            formatted_key = f"{prefix} {formatted_key}"

        if isinstance(value, dict):
            # Format nested dicts as JSON for clean representation
            context_parts.append(f"{formatted_key}: {json.dumps(value)}")
        elif isinstance(value, list):
            if value and isinstance(value[0], str):
                context_parts.append(f"{formatted_key}: {', '.join(value)}")
            else:
                context_parts.append(f"{formatted_key}: {json.dumps(value)}")
        else:
            context_parts.append(f"{formatted_key}: {value}")

    return context_parts