openai_llm.py

"""
openai_llm.py — LLM book matcher via OpenAI-compatible API.

Works with any OpenAI-compatible endpoint: vLLM, Ollama, LiteLLM, OpenAI, etc.

.env:
    LLM_PROVIDER=openai_llm      # Must be "openai_llm" to activate
    LLM_BASE_URL=http://localhost:8000/v1
    LLM_API_KEY=your_key
    LLM_MODEL=your_model_name
"""

import logging
import re

import requests

log = logging.getLogger(__name__)


def pick_best_match(
    title: str,
    author: str,
    results: list,
    base_url: str = "",
    api_key: str = "",
    model: str = "",
    max_tokens: int = 256,
    timeout: int = 120,
    language: str = "english",
) -> tuple[int | None, str]:
  """
  Ask an OpenAI-compatible LLM which result best matches the book.
  Returns (index, reason). index is None on any failure.

  Reasons:
    "matched"              — LLM picked a result
    "none"                 — LLM explicitly said NONE
    "no_results"           — empty results list passed in
    "not_configured"       — missing base_url or model
    "connection_error"     — could not reach the LLM endpoint
    "http_XXX"             — LLM returned an HTTP error
    "invalid_response"     — could not parse LLM output
    "empty_response"       — LLM returned empty content
  """
  if not results:
    return None, "no_results"
  if not base_url or not model:
    return None, "not_configured"

  # Build candidate list
  lines = []
  for i, r in enumerate(results):
    parts = [f"[{i}] {r.get('title', '?')}"]
    if r.get("author"):
      parts.append(f"by {r['author']}")
    if r.get("narrator"):
      parts.append(f"(narrated by {r['narrator']})")
    if r.get("series"):
      seq = r.get("series_sequence", "")
      parts.append(f"[{r['series']}{' #' + seq if seq else ''}]")
    lines.append(" ".join(parts))

  candidates = "\n".join(lines)

  prompt = (
      f'Match: "{title}" by {author}\n\n'
      f"{candidates}\n\n"
      f"Pick the best match in {language} only. Ignore results in other languages.\n"
      f"Reply with ONLY the index number or NONE. Nothing else.\n"
      f"/no_think"
  )

  system = (
      f"You match audiobooks. Only pick {language} language results. "
      "Output ONLY a single number or the word NONE. "
      "No reasoning. No explanation. No other text."
  )

  # Call the API
  url = f"{base_url.rstrip('/')}/chat/completions"
  headers = {"Content-Type": "application/json"}
  if api_key:
    headers["Authorization"] = f"Bearer {api_key}"

  try:
    r = requests.post(
        url,
        headers=headers,
        json={
            "model": model,
            "messages": [
                {"role": "system", "content": system},
                {"role": "user", "content": prompt},
            ],
            "max_tokens": max_tokens,
            "temperature": 0,
        },
        timeout=timeout,
    )
  except requests.ConnectionError:
    return None, f"connection_error: cannot reach {base_url}"
  except requests.Timeout:
    return None, f"connection_error: timeout reaching {base_url}"
  except requests.RequestException as e:
    return None, f"connection_error: {e}"

  if r.status_code != 200:
    body = r.text[:200] if r.text else ""
    return None, f"http_{r.status_code}: {body}"

  try:
    data = r.json()
    msg = data["choices"][0]["message"]
    text = msg.get("content") or ""

    # Qwen3 and similar models put thinking in reasoning_content,
    # and sometimes content is null/empty with the answer only
    # at the tail of reasoning_content
    if not text.strip() and msg.get("reasoning_content"):
      # Try to extract answer from end of reasoning text
      reasoning = msg["reasoning_content"]
      # Look for a final line that's just a number or NONE
      lines = reasoning.strip().splitlines()
      for line in reversed(lines):
        cleaned_line = line.strip()
        if re.match(r"^\d+$", cleaned_line) or cleaned_line.upper() == "NONE":
          text = cleaned_line
          break
      if not text.strip():
        return None, "empty_response: content empty, reasoning had no clear answer"

  except (KeyError, IndexError, ValueError) as e:
    return None, f"invalid_response: could not extract content: {e}"

  if not text or not text.strip():
    return None, "empty_response"

  # Strip <think>...</think> tags from reasoning models
  text = re.sub(r"<think>.*?</think>", "", text, flags=re.DOTALL).strip()

  if not text:
    return None, "empty_response: only thinking tags, no answer"

  # Parse
  upper = text.upper()
  if "NONE" in upper:
    return None, "none"

  match = re.search(r"\d+", text)
  if match:
    idx = int(match.group())
    if 0 <= idx < len(results):
      return idx, "matched"
    return None, f"invalid_response: index {idx} out of range 0-{len(results)-1}"

  return None, f"invalid_response: unparseable '{text[:50]}'"


# ============================================================
# Raw LLM call helper
# ============================================================

def _call_llm(
    system: str,
    prompt: str,
    base_url: str,
    api_key: str,
    model: str,
    max_tokens: int,
    timeout: int,
) -> tuple[str | None, str]:
  """
  Make a raw LLM call. Returns (text, reason).
  text is None on failure.
  """
  url = f"{base_url.rstrip('/')}/chat/completions"
  headers = {"Content-Type": "application/json"}
  if api_key:
    headers["Authorization"] = f"Bearer {api_key}"

  try:
    r = requests.post(
        url,
        headers=headers,
        json={
            "model": model,
            "messages": [
                {"role": "system", "content": system},
                {"role": "user", "content": prompt},
            ],
            "max_tokens": max_tokens,
            "temperature": 0,
        },
        timeout=timeout,
    )
  except requests.ConnectionError:
    return None, f"connection_error: cannot reach {base_url}"
  except requests.Timeout:
    return None, f"connection_error: timeout reaching {base_url}"
  except requests.RequestException as e:
    return None, f"connection_error: {e}"

  if r.status_code != 200:
    body = r.text[:200] if r.text else ""
    return None, f"http_{r.status_code}: {body}"

  try:
    data = r.json()
    msg = data["choices"][0]["message"]
    text = msg.get("content") or ""

    if not text.strip() and msg.get("reasoning_content"):
      reasoning = msg["reasoning_content"]
      lines = reasoning.strip().splitlines()
      for line in reversed(lines):
        cleaned = line.strip()
        if cleaned and not cleaned.startswith("*"):
          text = cleaned
          break

    if not text or not text.strip():
      return None, "empty_response"

    text = re.sub(r"<think>.*?</think>", "", text, flags=re.DOTALL).strip()
    if not text:
      return None, "empty_response: only thinking tags"

    return text, "ok"

  except (KeyError, IndexError, ValueError) as e:
    return None, f"invalid_response: {e}"


# ============================================================
# Identify book from partial/chapter title
# ============================================================

def identify_book(
    title: str,
    author: str,
    base_url: str = "",
    api_key: str = "",
    model: str = "",
    max_tokens: int = 256,
    timeout: int = 120,
    language: str = "english",
) -> tuple[str | None, str | None, str]:
  """
  Ask the LLM to identify what book a partial/chapter title belongs to.
  Returns (book_title, book_author, reason).

  Examples:
      "04 - The Ring Goes East" by Rob Inglis
        -> ("The Two Towers", "J.R.R. Tolkien", "identified")

      "Appendix" by Rob Inglis
        -> ("The Return of the King", "J.R.R. Tolkien", "identified")

      "Dune 05 - The Faces of a Martyr" by Brian Herbert
        -> ("The Machine Crusade", "Brian Herbert", "identified")
  """
  if not base_url or not model:
    return None, None, "not_configured"

  prompt = (
      f'What {language} audiobook is this from?\n'
      f'Title: "{title}"\n'
      f'Author/Narrator: {author}\n\n'
      f'This might be a chapter, part, or section of a larger audiobook.\n'
      f'Reply with ONLY the full book title and author in this exact format:\n'
      f'TITLE: <book title>\n'
      f'AUTHOR: <author name>\n'
      f'If you cannot identify it, reply UNKNOWN.\n'
      f'/no_think'
  )

  system = (
      "You identify audiobooks from partial titles, chapter names, or section names. "
      "Reply in the exact format requested. No explanation."
  )

  text, reason = _call_llm(system, prompt, base_url,
                           api_key, model, max_tokens, timeout)
  if text is None:
    return None, None, reason

  upper = text.upper()
  if "UNKNOWN" in upper:
    return None, None, "unknown"

  # Parse TITLE: ... AUTHOR: ...
  title_match = re.search(r'TITLE:\s*(.+)', text, re.IGNORECASE)
  author_match = re.search(r'AUTHOR:\s*(.+)', text, re.IGNORECASE)

  if title_match:
    book_title = title_match.group(1).strip().strip('"\'')
    book_author = author_match.group(1).strip().strip(
        '"\'') if author_match else author
    return book_title, book_author, "identified"

  return None, None, f"invalid_response: could not parse '{text[:80]}'"