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, )