roux/scripts/convert-to-cooklang

#!/usr/bin/env python3
"""Convert unstructured recipe content to Cooklang format using an
Anthropic-compatible API (Messages API).

Accepts text input (--text or --stdin) or file input (--file for PDF/images).
Reads API configuration from a YAML config file (--config).

Outputs the Cooklang text on stdout.
Exits 0 on success, non-zero with error details on stderr on failure.

Usage:
    scripts/convert-to-cooklang --config roux-config.yaml --text "Recipe text..."
    scripts/convert-to-cooklang --config roux-config.yaml --stdin < recipe.txt
    scripts/convert-to-cooklang --config roux-config.yaml --file recipe.pdf
"""

import argparse
import base64
import mimetypes
import os
import re
import sys

import yaml
from anthropic import Anthropic


# -- Config -------------------------------------------------------------------

def load_config(path: str) -> dict:
    with open(path) as f:
        cfg = yaml.safe_load(f)
    anthropic_cfg = cfg.get("anthropic", {})
    if not anthropic_cfg.get("api_key"):
        print("Error: anthropic.api_key is required in config", file=sys.stderr)
        sys.exit(1)
    return anthropic_cfg


# -- Prompt loading -----------------------------------------------------------

def load_prompts() -> tuple[str, str, str]:
    """Load system prompt, converter instructions, and user template.

    Tries locations in order:
    1. ../prompts/ relative to the script (development from project root)
    2. /usr/local/share/roux/prompts/ (production Docker deployment)
    """
    script_dir = os.path.dirname(os.path.abspath(__file__))
    candidates = [
        os.path.join(script_dir, "..", "prompts"),
        "/usr/local/share/roux/prompts",
    ]

    prompts_dir = None
    for candidate in candidates:
        candidate = os.path.abspath(candidate)
        if os.path.isdir(candidate):
            prompts_dir = candidate
            break

    if prompts_dir is None:
        print(
            "Error: prompts directory not found. "
            "Searched: " + ", ".join(candidates),
            file=sys.stderr,
        )
        sys.exit(1)

    def read(name: str) -> str:
        with open(os.path.join(prompts_dir, name)) as f:
            return f.read()

    return read("system-prompt.md"), read("cooklang-converter.md"), read("user-prompt-template.md")


# -- Content preparation ------------------------------------------------------

def build_text_content(text: str) -> list[dict]:
    return [{"type": "text", "text": text}]


def build_file_content(filepath: str) -> list[dict]:
    """Prepare a file (PDF or image) for the Anthropic Messages API.

    Returns a list with one document/image content block.
    """
    SUPPORTED_DOCUMENT_TYPES = {"application/pdf"}
    SUPPORTED_IMAGE_TYPES = {"image/jpeg", "image/png", "image/gif", "image/webp"}

    mime_type, _ = mimetypes.guess_type(filepath)
    if mime_type is None:
        mime_type = "application/octet-stream"

    if mime_type not in SUPPORTED_DOCUMENT_TYPES | SUPPORTED_IMAGE_TYPES:
        print(
            f"Error: unsupported file type '{mime_type}'. "
            f"Supported types: PDF, JPEG, PNG, GIF, WebP",
            file=sys.stderr,
        )
        sys.exit(1)

    MAX_FILE_SIZE = 20 * 1024 * 1024  # 20MB
    file_size = os.path.getsize(filepath)
    if file_size > MAX_FILE_SIZE:
        print(
            f"Error: file too large ({file_size} bytes). Maximum file size is 20MB.",
            file=sys.stderr,
        )
        sys.exit(1)

    with open(filepath, "rb") as f:
        data = base64.b64encode(f.read()).decode("utf-8")

    is_pdf = mime_type == "application/pdf"
    block_type = "document" if is_pdf else "image"

    return [
        {
            "type": block_type,
            "source": {
                "type": "base64",
                "media_type": mime_type,
                "data": data,
            },
        }
    ]


# -- API call -----------------------------------------------------------------

def cooklang_from_anthropic(
    api_key: str,
    base_url: str | None,
    model: str | None,
    system_prompt: str,
    user_message_content: list[dict],
) -> str:
    """Send the recipe content to the Anthropic API and return Cooklang text."""
    client = Anthropic(api_key=api_key, base_url=base_url) if base_url else Anthropic(api_key=api_key)
    resolved_model = model or "claude-sonnet-4-6"

    response = client.messages.create(
        model=resolved_model,
        max_tokens=4096,
        system=[{"type": "text", "text": system_prompt}],
        messages=[
            {
                "role": "user",
                "content": user_message_content,
            }
        ],
    )

    # Warn if response was truncated
    if response.stop_reason == "max_tokens":
        print(
            "Warning: response may be truncated (reached max_tokens limit). "
            "Consider splitting the recipe into smaller parts.",
            file=sys.stderr,
        )

    # Extract text from response content blocks
    parts: list[str] = []
    for block in response.content:
        if block.type == "text":
            parts.append(block.text)

    cooklang = "\n".join(parts).strip()

    # Strip markdown code fences if present (the prompts may instruct the
    # model to wrap output in ```cooklang blocks)
    cooklang = re.sub(r"^```\w*\n?", "", cooklang)
    cooklang = re.sub(r"\n```\s*$", "", cooklang)
    cooklang = cooklang.strip()

    # Basic validation: must start with front matter marker
    if not cooklang.startswith("---"):
        print(
            "Error: AI response does not appear to be valid Cooklang "
            "(missing front matter). Response preview: "
            + cooklang[:200],
            file=sys.stderr,
        )
        sys.exit(1)

    return cooklang


# -- Main ---------------------------------------------------------------------

def main() -> None:
    parser = argparse.ArgumentParser(
        description="Convert unstructured recipe content to Cooklang using an Anthropic-compatible API."
    )
    parser.add_argument(
        "--config",
        required=True,
        help="Path to YAML config file (must contain anthropic.api_key)",
    )
    input_group = parser.add_mutually_exclusive_group(required=True)
    input_group.add_argument("--text", help="Inline recipe text")
    input_group.add_argument("--stdin", action="store_true", help="Read recipe text from stdin")
    input_group.add_argument("--file", help="Path to recipe PDF or image file")
    args = parser.parse_args()

    # Load config
    config = load_config(args.config)
    api_key = config["api_key"]
    base_url = config.get("base_url")
    model = config.get("model")

    # Load prompts
    system_prompt, converter_instructions, user_template = load_prompts()

    # Prepare user message content
    if args.file:
        # File input (PDF or image): send file content + converter instructions
        file_content = build_file_content(args.file)
        user_message_content = (
            [{"type": "text", "text": converter_instructions}]
            + file_content
        )
    elif args.stdin:
        raw_text = sys.stdin.read()
        user_message_content = build_text_content(
            user_template.replace("{{FULL_RECIPE_TEXT}}", raw_text)
            .replace("{{TITLE}}", "Untitled")
            .replace("{{URL}}", "")
        )
    else:
        # --text
        user_message_content = build_text_content(
            user_template.replace("{{FULL_RECIPE_TEXT}}", args.text)
            .replace("{{TITLE}}", "Untitled")
            .replace("{{URL}}", "")
        )

    # Call API
    try:
        cooklang = cooklang_from_anthropic(
            api_key=api_key,
            base_url=base_url,
            model=model,
            system_prompt=system_prompt,
            user_message_content=user_message_content,
        )
    except Exception as e:
        print(f"Error calling Anthropic API: {e}", file=sys.stderr)
        sys.exit(1)

    # Output Cooklang text
    print(cooklang)


if __name__ == "__main__":
    main()