app/services/converter_service.py

from __future__ import annotations

import hashlib
import io
import mimetypes
import re
import time
from pathlib import Path
from urllib.parse import urlparse

# import httpx  # DISABLED (OOM mitigation) — only used by YouTube
from markitdown import MarkItDown

from app.core.constants import IMAGE_EXTENSIONS, IMAGE_MIME_PREFIXES
from app.core.logger import get_logger
from app.models.domain import ConversionError, ConversionResult
from app.services.ocr_service import ocr_image, ocr_pdf

_logger = get_logger(__name__)


# def _extract_youtube_video_id(url_or_id: str) -> str:  # DISABLED (OOM mitigation)
#     if len(url_or_id) == 11 and not url_or_id.startswith("http"):
#         return url_or_id
#     pattern = r"(?:v=|\/)([0-9A-Za-z_-]{11}).*"
#     match = re.search(pattern, url_or_id)
#     if match:
#         return match.group(1)
#     raise ValueError(f"Could not extract a valid YouTube ID from: {url_or_id}")
# 
# 
# def _fetch_youtube_oembed(video_id: str) -> dict | None:  # DISABLED (OOM mitigation)
#     oembed_url = f"https://www.youtube.com/oembed?url=https://www.youtube.com/watch?v={video_id}&format=json"
#     try:
#         resp = httpx.get(oembed_url, timeout=10.0)
#         resp.raise_for_status()
#         return resp.json()
#     except Exception as exc:
#         _logger.warning("YouTube oEmbed failed for %s: %s", video_id, exc)
#         return None
# 
# 
# def _fetch_youtube_transcript(video_id: str) -> str | None:  # DISABLED (OOM mitigation)
#     try:
#         import json as _json
#         import os as _os
# 
#         from youtube_transcript_api import YouTubeTranscriptApi
#         from youtube_transcript_api.formatters import TextFormatter
#         from youtube_transcript_api._errors import (
#             TranscriptsDisabled,
#             NoTranscriptFound,
#             VideoUnavailable,
#         )
# 
#         kwargs: dict = {}
#         cookies_raw = _os.environ.get("YOUTUBE_COOKIES", "").strip()
#         if cookies_raw:
#             try:
#                 import requests as _requests
#                 session = _requests.Session()
#                 session.cookies.update(_json.loads(cookies_raw))
#                 kwargs["http_client"] = session
#             except Exception as exc:
#                 _logger.warning("Failed to apply YOUTUBE_COOKIES: %s", exc)
# 
#         ytt_api = YouTubeTranscriptApi(**kwargs)
#         transcript_data = ytt_api.fetch(video_id, languages=["en"])
#         formatter = TextFormatter()
#         return formatter.format_transcript(transcript_data)
#     except TranscriptsDisabled:
#         _logger.warning("Transcripts disabled for video %s", video_id)
#     except NoTranscriptFound:
#         _logger.warning("No transcript found for video %s in language 'en'", video_id)
#     except VideoUnavailable:
#         _logger.warning("Video %s is unavailable, deleted, or private", video_id)
#     except ValueError as ve:
#         _logger.warning("Invalid video ID %s: %s", video_id, ve)
#     except Exception as exc:
#         _logger.warning("YouTube transcript failed for %s: %s", video_id, exc)
#     return None
# 
# 
# def _convert_youtube(url: str) -> ConversionResult | None:  # DISABLED (OOM mitigation)
#     try:
#         video_id = _extract_youtube_video_id(url)
#     except ValueError:
#         return None
# 
#     oembed = _fetch_youtube_oembed(video_id)
#     transcript = _fetch_youtube_transcript(video_id)
# 
#     lines = ["# YouTube\n"]
#     title = (oembed or {}).get("title", "")
#     if title:
#         lines.append(f"\n## {title}\n")
# 
#     author = (oembed or {}).get("author_name", "")
#     if author:
#         lines.append(f"\n- **Channel:** {author}\n")
# 
#     desc = (oembed or {}).get("description", "")
#     if desc:
#         lines.append(f"\n### Description\n{desc}\n")
# 
#     if transcript:
#         lines.append(f"\n### Transcript\n{transcript}\n")
#     else:
#         if not title and not desc:
#             return None
#         lines.append("\n> No transcript available.\n")
# 
#     markdown = "".join(lines)
#     return _build_result(url, markdown, 0, "text/html", 0.0)


def _is_image(ext: str, mime: str) -> bool:
    return ext.lower() in IMAGE_EXTENSIONS or any(mime.startswith(p) for p in IMAGE_MIME_PREFIXES)


def _build_result(
    source: str,
    markdown: str,
    file_size: int,
    mime_type: str,
    elapsed: float,
) -> ConversionResult:
    lines = markdown.splitlines()
    words = markdown.split()
    content_hash = hashlib.sha256(markdown.encode()).hexdigest()
    return ConversionResult(
        source=source,
        markdown=markdown,
        char_count=len(markdown),
        word_count=len(words),
        line_count=len(lines),
        duration_ms=elapsed,
        file_size_bytes=file_size,
        mime_type=mime_type,
        content_hash=content_hash,
    )


class ConverterService:
    def __init__(self, enable_plugins: bool = False) -> None:
        kwargs: dict = {"enable_plugins": enable_plugins}
        self._engine = MarkItDown(**kwargs)

    def convert_file(self, path: str | Path) -> ConversionResult | ConversionError:
        path = Path(path).resolve()
        start = time.perf_counter()

        if not path.exists():
            return ConversionError(
                source=str(path),
                error_type="FileNotFoundError",
                message=f"File does not exist: {path}",
                duration_ms=0.0,
            )

        file_size = path.stat().st_size
        mime_type, _ = mimetypes.guess_type(str(path))
        mime_type = mime_type or "application/octet-stream"

        try:
            if _is_image(path.suffix, mime_type):
                markdown = ocr_image(str(path))
            else:
                markdown = self._engine.convert(str(path)).text_content
                if not markdown.strip() and path.suffix.lower() == ".pdf":
                    _logger.info("No text from PDF, falling back to OCR")
                    markdown = ocr_pdf(str(path))
            elapsed = (time.perf_counter() - start) * 1000
            return _build_result(str(path), markdown, file_size, mime_type, elapsed)
        except Exception as exc:
            elapsed = (time.perf_counter() - start) * 1000
            return ConversionError(
                source=str(path),
                error_type=type(exc).__name__,
                message=str(exc),
                duration_ms=elapsed,
            )

    def convert_url(self, url: str) -> ConversionResult | ConversionError:
        parsed = urlparse(url)
        if parsed.scheme not in {"http", "https"}:
            return ConversionError(
                source=url,
                error_type="ValueError",
                message=f"Unsupported URL scheme: {parsed.scheme!r}",
                duration_ms=0.0,
            )

        start = time.perf_counter()

        # try:  # DISABLED (OOM mitigation) — YouTube transcription
        #     _extract_youtube_video_id(url)
        # except ValueError:
        #     pass
        # else:
        #     result = _convert_youtube(url)
        #     if result is not None:
        #         result = ConversionResult(
        #             source=result.source,
        #             markdown=result.markdown,
        #             char_count=result.char_count,
        #             word_count=result.word_count,
        #             line_count=result.line_count,
        #             duration_ms=(time.perf_counter() - start) * 1000,
        #             file_size_bytes=result.file_size_bytes,
        #             mime_type=result.mime_type,
        #             content_hash=result.content_hash,
        #         )
        #         return result
        #     _logger.warning("YouTube conversion returned None, falling back to markitdown: %s", url)

        try:
            url_ext = Path(urlparse(url).path).suffix.lower()
            if url_ext in IMAGE_EXTENSIONS:
                markdown = ocr_image(url)
                mime_type = mimetypes.guess_type(url)[0] or "image/jpeg"
            else:
                result = self._engine.convert(url)
                markdown = result.text_content
                mime_type = "text/html"
            elapsed = (time.perf_counter() - start) * 1000
            return _build_result(url, markdown, 0, mime_type, elapsed)
        except Exception as exc:
            elapsed = (time.perf_counter() - start) * 1000
            return ConversionError(
                source=url,
                error_type=type(exc).__name__,
                message=str(exc),
                duration_ms=elapsed,
            )

    def convert_stream(self, data: bytes, filename: str) -> ConversionResult | ConversionError:
        start = time.perf_counter()
        mime_type, _ = mimetypes.guess_type(filename)
        mime_type = mime_type or "application/octet-stream"
        ext = Path(filename).suffix.lower()

        try:
            if _is_image(ext, mime_type):
                markdown = ocr_image(data)
            else:
                result = self._engine.convert_stream(io.BytesIO(data), file_extension=ext)
                markdown = result.text_content
                if not markdown.strip() and ext == ".pdf":
                    _logger.info("No text from PDF stream, falling back to OCR")
                    markdown = ocr_pdf(data)
            elapsed = (time.perf_counter() - start) * 1000
            return _build_result(filename, markdown, len(data), mime_type, elapsed)
        except Exception as exc:
            elapsed = (time.perf_counter() - start) * 1000
            return ConversionError(
                source=filename,
                error_type=type(exc).__name__,
                message=str(exc),
                duration_ms=elapsed,
            )