#!/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()