API Reference

def cleanup_temp_file(file_id: str) -> None:
    """Remove temp file and its registry entry.

    Args:
        file_id: Unique identifier for the file.
    """
    try:
        if not TEMP_FILES_REGISTRY.exists():
            return

        with open(TEMP_FILES_REGISTRY, 'r') as f:
            registry = json.load(f)

        if file_id in registry:
            path = registry[file_id]['path']
            try:
                os.unlink(path)
            except:
                pass

            del registry[file_id]

            with open(TEMP_FILES_REGISTRY, 'w') as f:
                json.dump(registry, f)
    except:
        pass

def create_response_with_files(processed: dict[str, Any]) -> dict[str, Any]:
    """Create JSON response with file downloads.

    Args:
        processed: Processed result from process_result().

    Returns:
        Response dictionary with file IDs and metadata.
    """
    response = {"success": True, "result_type": processed['type']}

    if processed['type'] == 'download':
        file_id = str(uuid.uuid4())
        register_temp_file(file_id, processed['path'], processed['filename'])
        response['file_id'] = file_id
        response['filename'] = processed['filename']
    elif processed['type'] == 'downloads':
        files = []
        for f in processed['files']:
            file_id = str(uuid.uuid4())
            register_temp_file(file_id, f['path'], f['filename'])
            files.append({
                'file_id': file_id,
                'filename': f['filename']
            })
        response['files'] = files
    else:
        response['result'] = processed['data']

    return response

def get_temp_file(file_id: str) -> dict[str, str] | None:
    """Get temp file info from registry.

    Args:
        file_id: Unique identifier for the file.

    Returns:
        Dictionary with 'path' and 'filename' keys, or None if not found.
    """
    try:
        if not TEMP_FILES_REGISTRY.exists():
            return None

        with open(TEMP_FILES_REGISTRY, 'r') as f:
            registry = json.load(f)

        return registry.get(file_id)
    except:
        return None

async def handle_form_submission(request: Request, func: Callable, params: dict[str, ParamInfo]) -> JSONResponse:
    """Handle form submission for any function.

    Args:
        request: FastAPI request object.
        func: Function to call with validated parameters.
        params: Parameter metadata from analyze().

    Returns:
        JSON response with result or error.
    """
    try:
        form_data = await request.form()
        data = {}

        for name, value in form_data.items():
            if hasattr(value, 'filename'):
                suffix = os.path.splitext(value.filename)[1]
                data[name] = await save_uploaded_file(value, suffix)
            else:
                data[name] = value

        validated = validate_params(data, params)
        result = func(**validated)
        processed = process_result(result)
        response = create_response_with_files(processed)

        return JSONResponse(response)
    except Exception as e:
        return JSONResponse({"success": False, "error": str(e)}, status_code=400)

def register_temp_file(file_id: str, path: str, filename: str) -> None:
    """Register a temp file for download.

    Args:
        file_id: Unique identifier for the file.
        path: File system path to the temporary file.
        filename: Original filename for download.
    """
    try:
        if TEMP_FILES_REGISTRY.exists():
            with open(TEMP_FILES_REGISTRY, 'r') as f:
                registry = json.load(f)
        else:
            registry = {}

        registry[file_id] = {'path': path, 'filename': filename}

        with open(TEMP_FILES_REGISTRY, 'w') as f:
            json.dump(registry, f)
    except:
        pass

def run(
    func_or_list: Callable[..., Any] | list[Callable[..., Any]], 
    host: str = "0.0.0.0", 
    port: int = 8000, 
    template_dir: str | Path | None = None
) -> None:
    """Generate and run a web UI for one or more Python functions.

    Single function mode: Creates a form at root (/) for the function.
    Multiple functions mode: Creates an index page with links to individual function forms.

    Args:
        func_or_list: A single function or list of functions to wrap.
        host: Server host address (default: "0.0.0.0").
        port: Server port (default: 8000).
        template_dir: Optional custom template directory.

    Raises:
        FileNotFoundError: If template directory doesn't exist.
        TypeError: If function parameters use unsupported types.
    """

    funcs = func_or_list if isinstance(func_or_list, list) else [func_or_list]

    app = FastAPI()

    if template_dir is None:
        template_dir = Path(__file__).parent / "templates"
    else:
        template_dir = Path(template_dir)

    if not template_dir.exists():
        raise FileNotFoundError(
            f"Template directory '{template_dir}' not found."
        )

    templates = Jinja2Templates(directory=str(template_dir))
    app.mount("/static", StaticFiles(directory=template_dir / "static"), name="static")

    # Download endpoint for streaming files
    @app.get("/download/{file_id}")
    async def download_file(file_id: str):
        file_info = get_temp_file(file_id)

        if not file_info:
            return JSONResponse({"error": "File not found"}, status_code=404)

        path = file_info['path']
        filename = file_info['filename']

        if not os.path.exists(path):
            cleanup_temp_file(file_id)
            return JSONResponse({"error": "File expired"}, status_code=404)

        response = FastAPIFileResponse(
            path=path,
            filename=filename,
            media_type='application/octet-stream'
        )

        # Cleanup after sending
        async def cleanup():
            cleanup_temp_file(file_id)

        response.background = cleanup

        return response

    # Single function mode
    if len(funcs) == 1:
        func = funcs[0]
        params = analyze(func)
        func_name = func.__name__.replace('_', ' ').title()
        description = inspect.getdoc(func)

        @app.get("/")
        async def form(request: Request):
            fields = build_form_fields(params)
            return templates.TemplateResponse(
                "form.html",
                {
                    "request": request, 
                    "title": func_name, 
                    "description": description,
                    "fields": fields, 
                    "submit_url": "/submit"
                }
            )

        @app.post("/submit")
        async def submit(request: Request):
            return await handle_form_submission(request, func, params)

    # Multiple functions mode
    else:
        @app.get("/")
        async def index(request: Request):
            tools = [{
                "name": f.__name__.replace('_', ' ').title(),
                "path": f"/{f.__name__}"
            } for f in funcs]
            return templates.TemplateResponse(
                "index.html",
                {"request": request, "tools": tools}
            )

        for func in funcs:
            params = analyze(func)
            func_name = func.__name__.replace('_', ' ').title()
            description = inspect.getdoc(func)
            route = f"/{func.__name__}"
            submit_route = f"{route}/submit"

            def make_form_handler(title: str, prms: dict[str, ParamInfo], desc: str | None, submit_path: str):
                async def form_view(request: Request):
                    flds = build_form_fields(prms)
                    return templates.TemplateResponse(
                        "form.html",
                        {
                            "request": request, 
                            "title": title, 
                            "description": desc,
                            "fields": flds, 
                            "submit_url": submit_path
                        }
                    )
                return form_view

            def make_submit_handler(fn: Callable, prms: dict[str, ParamInfo]):
                async def submit_view(request: Request):
                    return await handle_form_submission(request, fn, prms)
                return submit_view

            app.get(route)(make_form_handler(func_name, params, description, submit_route))
            app.post(submit_route)(make_submit_handler(func, params))

    config = uvicorn.Config(
        app, 
        host=host, 
        port=port, 
        reload=False,
        limit_concurrency=100,
        limit_max_requests=1000,
        timeout_keep_alive=30,
        h11_max_incomplete_event_size=16 * 1024 * 1024
    )
    server = uvicorn.Server(config)
    asyncio.run(server.serve())

async def save_uploaded_file(uploaded_file: Any, suffix: str) -> str:
    """Save an uploaded file to a temporary location.

    Args:
        uploaded_file: The uploaded file object from FastAPI.
        suffix: File extension to use for the temp file.

    Returns:
        Path to the saved temporary file.
    """
    with tempfile.NamedTemporaryFile(delete=False, suffix=suffix, buffering=FILE_BUFFER_SIZE) as tmp:
        while chunk := await uploaded_file.read(CHUNK_SIZE):
            tmp.write(chunk)
        return tmp.name

def analyze(func: Callable[..., Any]) -> dict[str, ParamInfo]:
    """Analyze a function's signature and extract parameter metadata.

    Args:
        func: The function to analyze.

    Returns:
        Mapping of parameter names to ParamInfo objects.

    Raises:
        TypeError: If parameter type is not supported.
        TypeError: If list has no type parameter.
        TypeError: If list item type is not supported.
        TypeError: If list of Literal is used (conceptually confusing).
        TypeError: If list default is not a list.
        TypeError: If list default items have wrong type.
        ValueError: If default value doesn't match Literal options.
        ValueError: If Literal options are invalid.
        ValueError: If Union has multiple non-None types.
        ValueError: If default value type doesn't match parameter type.
    """

    result = {}

    for name, p in inspect.signature(func).parameters.items():
        default = None if p.default == inspect.Parameter.empty else p.default
        t = p.annotation
        f = None
        list_f = None  # Field info for the list itself
        dynamic_func = None
        is_optional = False
        optional_default_enabled = None  # None = auto, True = enabled, False = disabled
        is_list = False

        # 1. Extract base type from Annotated (OUTER level)
        # This could be constraints for the list itself
        if get_origin(t) is Annotated:
            args = get_args(t)
            t = args[0]
            if len(args) > 1:
                # Store this temporarily - we'll decide if it's for list or item later
                list_f = args[1]

        # 2. Check for Union types (including | None syntax) BEFORE list detection
        if get_origin(t) is types.UnionType or str(get_origin(t)) == 'typing.Union':
            union_args = get_args(t)

            # First pass: detect markers and check for None
            has_none = type(None) in union_args

            for arg in union_args:
                if get_origin(arg) is Annotated:
                    annotated_args = get_args(arg)
                    # Check if this is Annotated[None, Marker]
                    if annotated_args[0] is type(None) and len(annotated_args) > 1:
                        for marker in annotated_args[1:]:
                            if isinstance(marker, _OptionalEnabledMarker):
                                optional_default_enabled = True
                                is_optional = True
                            elif isinstance(marker, _OptionalDisabledMarker):
                                optional_default_enabled = False
                                is_optional = True

            # Second pass: extract the actual type (not None, not markers)
            if has_none or is_optional:
                is_optional = True
                non_none_types = []

                for arg in union_args:
                    # Skip plain None
                    if arg is type(None):
                        continue

                    # Skip Annotated[None, Marker] (the markers)
                    if get_origin(arg) is Annotated:
                        annotated_args = get_args(arg)
                        if annotated_args[0] is type(None):
                            continue

                    # This is the actual type
                    non_none_types.append(arg)

                if len(non_none_types) == 0:
                    raise TypeError(f"'{name}': Cannot have only None type")
                elif len(non_none_types) > 1:
                    raise TypeError(f"'{name}': Union with multiple non-None types not supported")

                # Extract the actual type
                t = non_none_types[0]

                # Check again if this is Annotated (for Field constraints)
                if get_origin(t) is Annotated:
                    args = get_args(t)
                    t = args[0]
                    if len(args) > 1 and list_f is None:
                        list_f = args[1]

        # 3. Detect list type
        if get_origin(t) is list:
            is_list = True
            list_args = get_args(t)

            if not list_args:
                raise TypeError(f"'{name}': list must have type parameter (e.g., list[int])")

            # Extract item type
            t = list_args[0]

            # Check if item type is Literal (before extracting Annotated)
            if get_origin(t) is Literal:
                raise TypeError(f"'{name}': list of Literal not supported")

            # 4. Extract Annotated from ITEM type
            if get_origin(t) is Annotated:
                args = get_args(t)
                t = args[0]

                # Check again for Literal after extracting Annotated
                if get_origin(t) is Literal:
                    raise TypeError(f"'{name}': list of Literal not supported")

                if len(args) > 1:
                    f = args[1]  # Field constraints for EACH ITEM
        elif t is list:
            # Handle bare 'list' without type parameter
            raise TypeError(f"'{name}': list must have type parameter (e.g., list[int])")

        # If not a list, then list_f is actually the field_info for the item
        if not is_list and list_f is not None:
            f = list_f
            list_f = None

        # 5. Handle Literal types (dropdowns)
        if get_origin(t) is Literal:
            opts = get_args(t)

            # Check if opts contains a single callable (dynamic Literal)
            if len(opts) == 1 and callable(opts[0]):
                dynamic_func = opts[0]
                result_value = dynamic_func()

                # Convert result to tuple properly
                if isinstance(result_value, (list, tuple)):
                    opts = tuple(result_value)
                else:
                    opts = (result_value,)

            # Validate options
            if opts:
                types_set = {type(o) for o in opts}
                if len(types_set) > 1:
                    raise TypeError(f"'{name}': mixed types in Literal")

                # For lists, we can't validate default against Literal here (it's a list)
                if not is_list and default is not None and default not in opts:
                    raise ValueError(f"'{name}': default '{default}' not in options {opts}")

                f = Literal[opts] if len(opts) > 0 else t
                t = types_set.pop() if types_set else type(None)
            else:
                t = type(None)

        # 6. Validate base type
        if t not in VALID:
            raise TypeError(f"'{name}': {t} not supported")

        # 7. Validate default value
        if default is not None:
            if is_list:
                # Must be a list
                if not isinstance(default, list):
                    raise TypeError(f"'{name}': default must be a list")

                # Validate list-level constraints BEFORE converting empty list to None
                if list_f and hasattr(list_f, 'metadata'):
                    TypeAdapter(Annotated[list[t], list_f]).validate_python(default)

                # Validate each item
                for item in default:
                    # Check type
                    if not isinstance(item, t):
                        raise TypeError(f"'{name}': list item type mismatch in default")

                    # Validate against Field constraints (for items)
                    if f and hasattr(f, 'metadata'):
                        TypeAdapter(Annotated[t, f]).validate_python(item)

                # Convert empty list to None AFTER validation
                if len(default) == 0:
                    default = None
            else:
                # Non-list validation (existing logic)
                if not is_optional and get_origin(f) is not Literal:
                    if not isinstance(default, t):
                        raise TypeError(f"'{name}': default value type mismatch")

                # Validate default value against field constraints
                if f and hasattr(f, 'metadata'):
                    TypeAdapter(Annotated[t, f]).validate_python(default)

        # 8. Determine optional_enabled state
        # Priority: explicit marker > default value presence > False
        if optional_default_enabled is not None:
            # Explicit marker takes priority
            final_optional_enabled = optional_default_enabled
        elif default is not None:
            # Has default value, start enabled
            final_optional_enabled = True
        else:
            # No default, start disabled
            final_optional_enabled = False

        result[name] = ParamInfo(t, default, f, dynamic_func, is_optional, final_optional_enabled, is_list, list_f)

    return result

def build_form_fields(params_info: dict) -> list[dict[str, Any]]:
    """Build form field specifications from parameter metadata.

    Re-executes dynamic functions to get fresh options.

    This function takes the analyzed parameter information from analyze() and
    converts it into a list of field specifications that can be used by the
    template engine to generate HTML form inputs.

    Process:
        1. Iterate through each parameter's ParamInfo
        2. Determine the appropriate HTML input type (text, number, select, etc.)
        3. Extract constraints and convert them to HTML attributes
        4. Handle special cases (optional fields, dynamic literals, files, etc.)
        5. Serialize defaults to JSON-safe format
        6. Return list of field dictionaries ready for template rendering

    Args:
        params_info: Mapping of parameter names to ParamInfo objects.
            Keys are parameter names (str), values are ParamInfo objects with 
            type, default, field_info, etc.

    Returns:
        List of field dictionaries for template rendering. Each dictionary contains:

        - name (str): Parameter name
        - type (str): HTML input type ('text', 'number', 'select', etc.)
        - default (Any): Default value for the field (JSON-serialized)
        - required (bool): Whether field is required (lists are ALWAYS required)
        - is_optional (bool): Whether field has optional toggle
        - optional_enabled (bool): Whether optional field starts enabled
        - is_list (bool): Whether this is a list field
        - list_min_length (int): For list fields, minimum number of items
        - list_max_length (int): For list fields, maximum number of items
        - options (tuple): For select fields, the dropdown options
        - min/max (int/float): For number fields, numeric constraints
        - minlength/maxlength (int): For text fields, length constraints
        - pattern (str): Regex pattern for validation
        - accept (str): For file fields, accepted file extensions
        - step (str): For number fields, '1' for int, 'any' for float

    Field Type Detection:
        - Literal types → 'select' (dropdown)
        - bool → 'checkbox'
        - date → 'date' (date picker)
        - time → 'time' (time picker)
        - int/float → 'number' (with constraints)
        - str with file pattern → 'file' (file upload)
        - str with color pattern → 'color' (color picker)
        - str with email pattern → 'email' (email input)
        - str (default) → 'text' (text input)
    """
    fields = []

    for name, info in params_info.items():
        # Serialize default value to JSON-safe format
        serialized_default = serialize_for_json(info.default)

        field = {
            'name': name, 
            'default': serialized_default,
            # REGLA SIMPLE: Las listas SIEMPRE son required=True
            # Si el campo está habilitado (no opcional O toggle ON), debe tener valor
            # Las listas nunca pueden estar vacías, si se quiere vacío usar None
            'required': True if info.is_list else not info.is_optional,
            'is_optional': info.is_optional,
            'optional_enabled': info.optional_enabled,
            'is_list': info.is_list
        }

        # Si es una lista, extraer los constraints de lista (min_length, max_length)
        if info.is_list and info.list_field_info and hasattr(info.list_field_info, 'metadata'):
            for c in info.list_field_info.metadata:
                cn = type(c).__name__
                if cn == 'MinLen':
                    field['list_min_length'] = c.min_length
                if cn == 'MaxLen':
                    field['list_max_length'] = c.max_length

        # Dropdown select
        if get_origin(info.field_info) is Literal:
            field['type'] = 'select'

            # Re-execute dynamic function if present
            if info.dynamic_func is not None:
                result_value = info.dynamic_func()

                # Convert result to tuple properly
                if isinstance(result_value, (list, tuple)):
                    fresh_options = tuple(result_value)
                else:
                    fresh_options = (result_value,)

                field['options'] = fresh_options
                info.field_info = Literal[fresh_options]
            else:
                field['options'] = get_args(info.field_info)

        # Checkbox
        elif info.type is bool:
            field['type'] = 'checkbox'
            field['required'] = False

        # Date picker
        elif info.type is date:
            field['type'] = 'date'
            # Already serialized above, no need to do it again

        # Time picker
        elif info.type is time:
            field['type'] = 'time'
            # Already serialized above, no need to do it again

        # Number input
        elif info.type in (int, float):
            field['type'] = 'number'
            field['step'] = '1' if info.type is int else 'any'

            # Extract numeric constraints from Pydantic Field
            if info.field_info and hasattr(info.field_info, 'metadata'):
                for c in info.field_info.metadata:
                    cn = type(c).__name__
                    if cn == 'Ge': field['min'] = c.ge
                    elif cn == 'Le': field['max'] = c.le
                    elif cn == 'Gt': field['min'] = c.gt + (1 if info.type is int else 0.01)
                    elif cn == 'Lt': field['max'] = c.lt - (1 if info.type is int else 0.01)

        # Text/email/color/file input
        else:
            field['type'] = 'text'

            if info.field_info and hasattr(info.field_info, 'metadata'):
                for c in info.field_info.metadata:
                    cn = type(c).__name__

                    # Check for pattern constraints
                    if hasattr(c, 'pattern') and c.pattern:
                        pattern = c.pattern

                        # File input detection
                        if pattern.startswith(r'^.+\.(') and pattern.endswith(r')$'):
                            field['type'] = 'file'
                            exts = pattern[6:-2].split('|')
                            field['accept'] = '.' + ',.'.join(exts)
                        # Special input types (color, email)
                        elif pattern in PATTERN_TO_HTML_TYPE:
                            field['type'] = PATTERN_TO_HTML_TYPE[pattern]

                        field['pattern'] = pattern

                    # String length constraints
                    if cn == 'MinLen': 
                        field['minlength'] = c.min_length
                    if cn == 'MaxLen':
                        field['maxlength'] = c.max_length

        fields.append(field)

    return fields

def serialize_for_json(value: Any) -> Any:
    """Serialize a value to be JSON-safe for template rendering.

    Converts date/time objects to ISO format strings.

    Args:
        value: The value to serialize (can be any type).

    Returns:
        JSON-safe serialized value (str for dates/times, or original type).
    """
    if value is None:
        return None

    if isinstance(value, date):
        return value.isoformat()

    if isinstance(value, time):
        return value.isoformat()

    if isinstance(value, list):
        return [serialize_for_json(item) for item in value]

    if isinstance(value, dict):
        return {k: serialize_for_json(v) for k, v in value.items()}

    return value

def process_result(result):
    """
    Convert function result to appropriate display format.

    Handles images (PIL, Matplotlib), single/multiple files, and text.
    Returns a dictionary with 'type' and relevant data.
    Files are saved to temporary files and paths are returned.
    """
    # PIL Image detection
    try:
        from PIL import Image
        if isinstance(result, Image.Image):
            buffer = io.BytesIO()
            result.save(buffer, format='PNG')
            buffer.seek(0)
            img_base64 = base64.b64encode(buffer.read()).decode()
            return {
                'type': 'image',
                'data': f'data:image/png;base64,{img_base64}'
            }
    except ImportError:
        pass

    # Matplotlib Figure detection
    try:
        import matplotlib.pyplot as plt
        from matplotlib.figure import Figure
        if isinstance(result, Figure):
            buffer = io.BytesIO()
            result.savefig(buffer, format='png', bbox_inches='tight')
            buffer.seek(0)
            img_base64 = base64.b64encode(buffer.read()).decode()
            plt.close(result)
            return {
                'type': 'image',
                'data': f'data:image/png;base64,{img_base64}'
            }
    except ImportError:
        pass

    # Single file - save to temp
    if isinstance(result, UserFileResponse):
        with tempfile.NamedTemporaryFile(delete=False, suffix=f"_{result.filename}") as tmp:
            tmp.write(result.data)
            return {
                'type': 'download',
                'path': tmp.name,
                'filename': result.filename
            }

    # Multiple files - save to temp
    if isinstance(result, list) and len(result) > 0 and all(isinstance(f, UserFileResponse) for f in result):
        files = []
        for f in result:
            with tempfile.NamedTemporaryFile(delete=False, suffix=f"_{f.filename}") as tmp:
                tmp.write(f.data)
                files.append({
                    'path': tmp.name,
                    'filename': f.filename
                })
        return {
            'type': 'downloads',
            'files': files
        }

    # Default: convert to string
    return {
        'type': 'text',
        'data': str(result)
    }