Paddleocr Text Recognition

"""
PaddleOCR Text Recognition Library

Simple OCR API wrapper for PaddleOCR text recognition.
"""

import base64
import logging
import math
import os
from pathlib import Path
from typing import Any, Optional
from urllib.parse import unquote, urlparse

import httpx

logger = logging.getLogger(__name__)

# =============================================================================
# Constants
# =============================================================================

DEFAULT_TIMEOUT = 120  # seconds
API_GUIDE_URL = "https://paddleocr.com"
FILE_TYPE_PDF = 0
FILE_TYPE_IMAGE = 1
IMAGE_EXTENSIONS = (".png", ".jpg", ".jpeg", ".bmp", ".tiff", ".tif", ".webp")


# =============================================================================
# Environment
# =============================================================================


def _get_env(key: str) -> str:
    """Get environment variable, defaulting to empty string with whitespace stripped."""
    return os.getenv(key, "").strip()


def _http_timeout_from_env(env_key: str, default_seconds: float) -> float:
    """
    Read HTTP client timeout in seconds from the environment.

    Returns a positive finite float. If the variable is missing, empty,
    unparsable, non-finite, or not greater than zero, logs a warning and uses the
    default_seconds argument value.
    """
    raw = os.getenv(env_key)
    if raw is None:
        return float(default_seconds)
    stripped = raw.strip()
    if not stripped:
        return float(default_seconds)
    try:
        timeout = float(stripped)
    except (ValueError, TypeError):
        logger.warning(
            "Invalid %s value %r; using default %ss",
            env_key,
            raw,
            default_seconds,
        )
        return float(default_seconds)
    if not math.isfinite(timeout) or timeout <= 0:
        logger.warning(
            "%s must be a finite number > 0 (got %r); using default %ss",
            env_key,
            raw,
            default_seconds,
        )
        return float(default_seconds)
    return timeout


def _resolve_api_url(api_url: str, env_var: str) -> str:
    """Require https; allow host-only values by prepending https://."""
    if api_url.startswith("http://"):
        raise ValueError(f"{env_var} must use https://; http:// is not allowed.")
    if not api_url.startswith("https://"):
        return f"https://{api_url}"
    return api_url


def get_config() -> tuple[str, str]:
    """
    Get API URL and token from environment.

    Returns:
        tuple of (api_url, token)

    Raises:
        ValueError: If required env vars are missing, API URL uses http://,
            or URL path doesn't end with /ocr
    """
    api_url = _get_env("PADDLEOCR_OCR_API_URL")
    token = _get_env("PADDLEOCR_ACCESS_TOKEN")

    if not api_url:
        raise ValueError(
            f"PADDLEOCR_OCR_API_URL not configured. Get your API at: {API_GUIDE_URL}"
        )
    if not token:
        raise ValueError(
            f"PADDLEOCR_ACCESS_TOKEN not configured. Get your API at: {API_GUIDE_URL}"
        )

    api_url = _resolve_api_url(api_url, "PADDLEOCR_OCR_API_URL")
    api_path = urlparse(api_url).path.rstrip("/")
    if not api_path.endswith("/ocr"):
        raise ValueError(
            "PADDLEOCR_OCR_API_URL must be a full endpoint ending with /ocr. "
            "Example: https://your-service.paddleocr.com/ocr"
        )

    return api_url, token


# =============================================================================
# File Utilities
# =============================================================================


def _detect_file_type(path_or_url: str) -> int:
    """Detect file type: 0=PDF, 1=Image."""
    path = path_or_url.lower()
    if path.startswith(("http://", "https://")):
        path = unquote(urlparse(path).path)

    if path.endswith(".pdf"):
        return FILE_TYPE_PDF
    elif path.endswith(IMAGE_EXTENSIONS):
        return FILE_TYPE_IMAGE
    else:
        raise ValueError(f"Unsupported file format: {path_or_url}")


def _load_file_as_base64(file_path: str) -> str:
    """Load local file and encode as base64."""
    path = Path(file_path)
    if not path.exists():
        raise FileNotFoundError(f"File not found: {file_path}")
    if path.stat().st_size == 0:
        raise ValueError(f"File is empty (0 bytes): {file_path}")
    return base64.b64encode(path.read_bytes()).decode("utf-8")


# =============================================================================
# API Request
# =============================================================================


def _make_api_request(
    api_url: str, token: str, params: dict[str, Any]
) -> dict[str, Any]:
    """
    Make PaddleOCR API request.

    Args:
        api_url: API endpoint URL
        token: Access token
        params: Request parameters

    Returns:
        API response dict

    Raises:
        RuntimeError: On API errors
    """
    headers = {
        "Authorization": f"token {token}",
        "Content-Type": "application/json",
        "Client-Platform": "official-skill",
    }

    timeout = _http_timeout_from_env("PADDLEOCR_OCR_TIMEOUT", float(DEFAULT_TIMEOUT))

    try:
        with httpx.Client(timeout=timeout) as client:
            try:
                resp = client.post(api_url, json=params, headers=headers)
            except TypeError as e:
                raise RuntimeError(
                    "Request parameters cannot be JSON-encoded; use only JSON-serializable "
                    f"option values ({e})"
                ) from e
    except httpx.TimeoutException:
        raise RuntimeError(f"API request timed out after {timeout}s")
    except httpx.RequestError as e:
        raise RuntimeError(f"API request failed: {e}")

    if resp.status_code != 200:
        error_detail = ""
        try:
            error_body = resp.json()
            if isinstance(error_body, dict):
                error_detail = str(error_body.get("errorMsg", "")).strip()
        except Exception:
            pass

        if not error_detail:
            error_detail = (resp.text[:200] or "No response body").strip()

        if resp.status_code == 403:
            raise RuntimeError(f"Authentication failed (403): {error_detail}")
        elif resp.status_code == 429:
            raise RuntimeError(f"API rate limit exceeded (429): {error_detail}")
        elif resp.status_code >= 500:
            raise RuntimeError(
                f"API service error ({resp.status_code}): {error_detail}"
            )
        else:
            raise RuntimeError(f"API error ({resp.status_code}): {error_detail}")

    try:
        result = resp.json()
    except Exception:
        raise RuntimeError(f"Invalid JSON response: {resp.text[:200]}")

    if not isinstance(result, dict):
        raise RuntimeError(
            f"Unexpected JSON shape (expected object): {resp.text[:200]}"
        )

    if result.get("errorCode", 0) != 0:
        msg = result.get("errorMsg", "Unknown error")
        raise RuntimeError(f"API error: {msg}")

    return result


# =============================================================================
# Main API
# =============================================================================


def ocr(
    file_path: Optional[str] = None,
    file_url: Optional[str] = None,
    file_type: Optional[int] = None,
    **options: Any,
) -> dict[str, Any]:
    """
    Perform OCR on image or PDF.

    Args:
        file_path: Local file path (mutually exclusive with file_url)
        file_url: URL to file (mutually exclusive with file_path)
        file_type: Optional file type override (0=PDF, 1=Image)
        **options: Additional API options (passed directly to API)

    Returns:
        {
            "ok": True,
            "text": "extracted text...",
            "result": { raw API result },
            "error": None
        }
        or on error:
        {
            "ok": False,
            "text": "",
            "result": None,
            "error": {"code": "...", "message": "..."}
        }
    """
    if file_path is not None and not isinstance(file_path, str):
        return _error("INPUT_ERROR", "file_path must be a string or None")
    if file_url is not None and not isinstance(file_url, str):
        return _error("INPUT_ERROR", "file_url must be a string or None")

    fp = file_path.strip() if file_path else ""
    fu = file_url.strip() if file_url else ""
    if fp and fu:
        return _error(
            "INPUT_ERROR",
            "Provide only one of file_path or file_url, not both",
        )
    if not fp and not fu:
        return _error("INPUT_ERROR", "file_path or file_url required")
    if file_type is not None and file_type not in (FILE_TYPE_PDF, FILE_TYPE_IMAGE):
        return _error("INPUT_ERROR", "file_type must be 0 (PDF) or 1 (Image)")

    try:
        api_url, token = get_config()
    except ValueError as e:
        return _error("CONFIG_ERROR", str(e))

    # Build request params
    try:
        resolved_file_type: Optional[int] = None
        if fu:
            params = {"file": fu}
            if file_type is not None:
                resolved_file_type = file_type
            else:
                try:
                    resolved_file_type = _detect_file_type(fu)
                except ValueError:
                    resolved_file_type = None
        else:
            resolved_file_type = (
                file_type if file_type is not None else _detect_file_type(fp)
            )
            params = {"file": _load_file_as_base64(fp)}

        params["visualize"] = (
            False  # reduce response payload; callers can override via options
        )
        params.update(options)
        if resolved_file_type is not None:
            params["fileType"] = resolved_file_type
        else:
            params.pop("fileType", None)

    except (ValueError, OSError, MemoryError) as e:
        return _error("INPUT_ERROR", str(e))

    try:
        result = _make_api_request(api_url, token, params)
    except RuntimeError as e:
        return _error("API_ERROR", str(e))

    try:
        text = _extract_text(result)
    except ValueError as e:
        return _error("API_ERROR", str(e))

    return {
        "ok": True,
        "text": text,
        "result": result,
        "error": None,
    }


def _extract_text(result: dict[str, Any]) -> str:
    """Extract text from OCR result."""
    if not isinstance(result, dict):
        raise ValueError("Invalid API response: top-level response must be an object")

    raw_result = result.get("result")
    if not isinstance(raw_result, dict):
        raise ValueError("Invalid API response: missing 'result' object")

    pages = raw_result.get("ocrResults")
    if not isinstance(pages, list):
        raise ValueError("Invalid API response: result.ocrResults must be an array")

    all_text = []
    for i, item in enumerate(pages):
        if not isinstance(item, dict):
            raise ValueError(
                f"Invalid API response: result.ocrResults[{i}] must be an object"
            )

        pruned = item.get("prunedResult")
        if not isinstance(pruned, dict):
            raise ValueError(
                f"Invalid API response: result.ocrResults[{i}].prunedResult must be an object"
            )

        texts = pruned.get("rec_texts", [])
        if not isinstance(texts, list):
            raise ValueError(
                f"Invalid API response: result.ocrResults[{i}].prunedResult.rec_texts "
                "must be an array"
            )
        line_parts: list[str] = []
        for j, t in enumerate(texts):
            if not isinstance(t, str):
                raise ValueError(
                    f"Invalid API response: result.ocrResults[{i}].prunedResult."
                    f"rec_texts[{j}] must be a string"
                )
            line_parts.append(t)
        if line_parts:
            all_text.append("\n".join(line_parts))
    return "\n\n".join(all_text)


def _error(code: str, message: str) -> dict[str, Any]:
    """Create error response."""
    return {
        "ok": False,
        "text": "",
        "result": None,
        "error": {"code": code, "message": message},
    }

"""
PaddleOCR Text Recognition Caller

Simple CLI wrapper for the PaddleOCR text recognition library.

Usage:
    uv run scripts/ocr_caller.py --file-url "URL"
    uv run scripts/ocr_caller.py --file-path "image.png" --pretty
"""

# /// script
# requires-python = ">=3.9"
# dependencies = [
#   "httpx>=0.24.0",
# ]
# ///

import argparse
import io
import json
import sys
import tempfile
import uuid
from datetime import datetime
from pathlib import Path
from typing import Optional

# Fix Windows console encoding
if sys.platform == "win32":
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
    sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace")

# Add scripts dir to path for imports
sys.path.insert(0, str(Path(__file__).parent))

from lib import ocr


def get_default_output_path() -> Path:
    """Build a unique result path under the OS temp directory."""
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
    short_id = uuid.uuid4().hex[:8]
    return (
        Path(tempfile.gettempdir())
        / "paddleocr"
        / "text-recognition"
        / "results"
        / f"result_{timestamp}_{short_id}.json"
    )


def resolve_output_path(output_arg: Optional[str]) -> Path:
    if output_arg:
        return Path(output_arg).expanduser().resolve()
    return get_default_output_path().resolve()


def main() -> None:
    parser = argparse.ArgumentParser(
        description="PaddleOCR Text Recognition - OCR images/PDFs",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  # OCR from URL (result is auto-saved to the system temp directory)
  uv run scripts/ocr_caller.py --file-url "https://example.com/image.png"

  # OCR local file (result is auto-saved to the system temp directory)
  uv run scripts/ocr_caller.py --file-path "./document.pdf" --pretty

  # OCR with explicit file type override
  uv run scripts/ocr_caller.py --file-url "URL" --file-type 1 --pretty

  # Save result to a custom file path
  uv run scripts/ocr_caller.py --file-url "URL" --output "./result.json" --pretty

  # Print JSON to stdout without saving a file
  uv run scripts/ocr_caller.py --file-url "URL" --stdout --pretty

Exit codes:
  0   Success (ok=true in JSON output)
  1   OCR or API error (ok=false in JSON output; see error.code and error.message)
  5   Cannot write result to output file

Configuration:
  Set environment variables: PADDLEOCR_OCR_API_URL, PADDLEOCR_ACCESS_TOKEN
  Optional: PADDLEOCR_OCR_TIMEOUT
        """,
    )

    input_group = parser.add_mutually_exclusive_group(required=True)
    input_group.add_argument("--file-url", help="URL to image or PDF")
    input_group.add_argument("--file-path", help="Local path to image or PDF")

    # Output options
    parser.add_argument(
        "--file-type",
        type=int,
        choices=[0, 1],
        help="Optional file type override (0=PDF, 1=Image)",
    )
    parser.add_argument(
        "--pretty", action="store_true", help="Pretty-print JSON output"
    )
    output_group = parser.add_mutually_exclusive_group()
    output_group.add_argument(
        "--output",
        "-o",
        metavar="FILE",
        help="Save result to JSON file (default: auto-save to system temp directory)",
    )
    output_group.add_argument(
        "--stdout",
        action="store_true",
        help="Print JSON to stdout instead of saving to a file",
    )

    args = parser.parse_args()

    # Unwarping and orientation classification are off to cover common scenarios
    # with faster response times.
    result = ocr(
        file_path=args.file_path,
        file_url=args.file_url,
        file_type=args.file_type,
        useDocUnwarping=False,
        useDocOrientationClassify=False,
    )

    indent = 2 if args.pretty else None
    json_output = json.dumps(result, indent=indent, ensure_ascii=False)

    if args.stdout:
        print(json_output)
    else:
        output_path = resolve_output_path(args.output)

        try:
            output_path.parent.mkdir(parents=True, exist_ok=True)
            output_path.write_text(json_output, encoding="utf-8")
            print(f"Result saved to: {output_path}", file=sys.stderr)
        except (PermissionError, OSError) as e:
            print(f"Error: Cannot write to {output_path}: {e}", file=sys.stderr)
            sys.exit(5)

    sys.exit(0 if result.get("ok") else 1)


if __name__ == "__main__":
    main()

"""
Smoke Test for PaddleOCR Text Recognition

Verifies configuration and API connectivity.

Usage:
    uv run scripts/smoke_test.py
    uv run scripts/smoke_test.py --skip-api-test
    uv run scripts/smoke_test.py --test-url "https://example.com/test.png"
"""

# /// script
# requires-python = ">=3.9"
# dependencies = [
#   "httpx>=0.24.0",
# ]
# ///

import argparse
import sys
from pathlib import Path

# Add scripts dir to path for imports
sys.path.insert(0, str(Path(__file__).parent))


def print_config_guide() -> None:
    """Print friendly configuration guide."""
    from lib import DEFAULT_TIMEOUT

    print(
        f"""
============================================================
HOW TO GET YOUR API CREDENTIALS
============================================================

1. Visit: https://paddleocr.com
2. Sign in to your account
3. Open your model's API call example page
4. Copy the API URL from the example request
5. Copy your access token from the same API setup page

Set environment variables:
  export PADDLEOCR_OCR_API_URL=https://your-api-url.paddleocr.com/ocr
  export PADDLEOCR_ACCESS_TOKEN=your_token_here
  export PADDLEOCR_OCR_TIMEOUT={DEFAULT_TIMEOUT}  # optional

============================================================
"""
    )


def main() -> int:
    parser = argparse.ArgumentParser(
        description="PaddleOCR Text Recognition smoke test"
    )
    parser.add_argument("--test-url", help="Optional: Custom image URL for testing")
    parser.add_argument(
        "--skip-api-test",
        action="store_true",
        help="Skip API connectivity test, only check configuration",
    )
    args = parser.parse_args()

    print("=" * 60)
    print("PaddleOCR Text Recognition - Smoke Test")
    print("=" * 60)

    print("\n[1/3] Checking dependencies...")

    try:
        import httpx

        print(f"  + httpx: {httpx.__version__}")
    except ImportError:
        print("  X httpx not installed")
        print("\nRun this script with uv to auto-resolve dependencies:")
        print("  uv run scripts/smoke_test.py")
        print("\nOr install manually:")
        print("  pip install httpx")
        return 1

    print("\n[2/3] Checking configuration...")

    from lib import get_config

    try:
        api_url, token = get_config()
        print(f"  + PADDLEOCR_OCR_API_URL: {api_url}")
        masked_token = token[:8] + "..." + token[-4:] if len(token) > 12 else "***"
        print(f"  + PADDLEOCR_ACCESS_TOKEN: {masked_token}")
    except ValueError as e:
        print(f"  X {e}")
        print_config_guide()
        return 1

    if args.skip_api_test:
        print("\n[3/3] Skipping API connectivity test (--skip-api-test)")
        print("\n" + "=" * 60)
        print("Configuration Check Complete!")
        print("=" * 60)
        return 0

    print("\n[3/3] Testing API connectivity...")

    test_url = (
        args.test_url
        or "https://paddle-model-ecology.bj.bcebos.com/paddlex/imgs/demo_image/general_ocr_001.png"
    )
    print(f"  Test image: {test_url}")

    from lib import ocr

    result = ocr(file_url=test_url)

    if not result.get("ok"):
        error = result.get("error", {})
        print(f"\n  X API call failed: {error.get('message')}")
        if "Authentication" in error.get("message", ""):
            print("\n  Hint: Check if your token is correct and not expired.")
            print("        Get a new token from your API call example page.")
        return 1

    print("  + API call successful!")

    text = result.get("text", "")
    if text:
        preview = text[:200].replace("\n", " ")
        if len(text) > 200:
            preview += "..."
        print(f"\n  Preview: {preview}")

    print("\n" + "=" * 60)
    print("Smoke Test PASSED")
    print("=" * 60)
    print("\nNext steps:")
    print('  uv run scripts/ocr_caller.py --file-url "URL" --pretty')
    print('  uv run scripts/ocr_caller.py --file-path "image.png" --pretty')
    print(
        "  Results are auto-saved to the system temp directory; the caller prints the saved path."
    )

    return 0


if __name__ == "__main__":
    sys.exit(main())