utube transcription

Dockerfile

@@ -20,6 +20,8 @@ RUN pip install --no-cache-dir --upgrade pip && \
     pip install --no-cache-dir -r requirements.txt && \
     python -m spacy download en_core_web_sm
 
+RUN pip install --no-cache-dir "youtube-transcript-api>=1.2.4"
+
 COPY --chown=appuser:appuser . .
 
 RUN mkdir -p /app/logs && \

app/api/server.py

@@ -11,6 +11,7 @@ from app.config import get_settings
 from app.core.database import pool_manager
 from app.core.logger import get_logger
 from app.api.v1.router import api_v1_router
+from app.api.verify import router as verify_router
 
 _logger = get_logger(__name__)
 _settings = get_settings()
@@ -51,6 +52,7 @@ def create_application() -> FastAPI:
             {"name": "Convert", "description": "Single-file and single-URL conversion"},
             {"name": "Batch", "description": "Bulk conversion of files and URLs"},
             {"name": "System", "description": "Health, info, and supported formats"},
+            {"name": "Verify", "description": "Phone number and identity verification"},
         ],
         lifespan=lifespan,
     )
@@ -64,6 +66,7 @@ def create_application() -> FastAPI:
     )
 
     app.include_router(api_v1_router, prefix="/api/v1")
+    app.include_router(verify_router, prefix="/verify")
 
     @app.get("/", include_in_schema=False)
     async def root():

app/api/verify.py

@@ -0,0 +1,19 @@
+from __future__ import annotations
+
+from fastapi import APIRouter, HTTPException, Query
+
+from app.services.verify_service import VerifyService
+
+router = APIRouter(tags=["Verify"])
+_service = VerifyService()
+
+
+@router.get("/phone")
+async def verify_phone(
+    number: str = Query(..., description="Phone number (E.164 format like +14155552671, or local with country_code)"),
+    country_code: str | None = Query(None, description="ISO 3166-1 alpha-2 country code (e.g. US, IN, GB)"),
+):
+    result = _service.verify_phone(number, country_code)
+    if not result.valid:
+        raise HTTPException(status_code=400, detail=result.error)
+    return result.dict()

app/services/converter_service.py

@@ -3,11 +3,13 @@ from __future__ import annotations
 import hashlib
 import io
 import mimetypes
+import re
 import time
 from pathlib import Path
 from typing import Optional
-from urllib.parse import urlparse
+from urllib.parse import urlparse, parse_qs
 
+import httpx
 from markitdown import MarkItDown
 
 from app.config import get_settings
@@ -19,6 +21,84 @@ from app.services.ocr_service import ocr_image, ocr_pdf
 _logger = get_logger(__name__)
 _settings = get_settings()
 
+_YOUTUBE_URL_PATTERN = re.compile(
+    r"(?:https?://)?"
+    r"(?:www\.|m\.)?"
+    r"(?:youtube\.com/(?:watch\?v=|embed/|v/|shorts/|live/)|youtu\.be/)"
+    r"([\w-]{11})"
+)
+
+_YOUTUBE_DOMAINS = {"youtube.com", "www.youtube.com", "m.youtube.com", "youtu.be", "www.youtu.be"}
+
+
+def _is_youtube_url(url: str) -> bool:
+    return _YOUTUBE_URL_PATTERN.match(url) is not None
+
+
+def _extract_youtube_video_id(url: str) -> str | None:
+    match = _YOUTUBE_URL_PATTERN.match(url)
+    if match:
+        return match.group(1)
+    parsed = urlparse(url)
+    if parsed.hostname in _YOUTUBE_DOMAINS:
+        qs = parse_qs(parsed.query)
+        return qs.get("v", [None])[0]
+    return None
+
+
+def _fetch_youtube_oembed(video_id: str) -> dict | None:
+    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:
+    try:
+        from youtube_transcript_api import YouTubeTranscriptApi
+        api = YouTubeTranscriptApi()
+        transcript = api.fetch(video_id, languages=["en"])
+        return " ".join(part.text for part in transcript)
+    except Exception as exc:
+        _logger.warning("YouTube transcript failed for %s: %s", video_id, exc)
+        return None
+
+
+def _convert_youtube(url: str) -> ConversionResult | None:
+    video_id = _extract_youtube_video_id(url)
+    if not video_id:
+        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)
@@ -98,6 +178,24 @@ class ConverterService:
             )
 
         start = time.perf_counter()
+
+        if _is_youtube_url(url):
+            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:

app/services/verify_service.py

@@ -0,0 +1,80 @@
+from __future__ import annotations
+
+import phonenumbers
+from phonenumbers import carrier, geocoder, PhoneNumberType
+
+
+class PhoneVerificationResult:
+    def __init__(
+        self,
+        valid: bool,
+        international_format: str | None = None,
+        national_format: str | None = None,
+        country_code: int | None = None,
+        location: str | None = None,
+        carrier_name: str | None = None,
+        line_type: str | None = None,
+        error: str | None = None,
+    ) -> None:
+        self.valid = valid
+        self.international_format = international_format
+        self.national_format = national_format
+        self.country_code = country_code
+        self.location = location
+        self.carrier = carrier_name
+        self.line_type = line_type
+        self.error = error
+
+    def dict(self) -> dict:
+        return {
+            "valid": self.valid,
+            "international_format": self.international_format,
+            "national_format": self.national_format,
+            "country_code": self.country_code,
+            "location": self.location,
+            "carrier": self.carrier,
+            "line_type": self.line_type,
+            "error": self.error,
+        }
+
+
+_TYPE_NAMES = {
+    PhoneNumberType.FIXED_LINE: "fixed_line",
+    PhoneNumberType.MOBILE: "mobile",
+    PhoneNumberType.FIXED_LINE_OR_MOBILE: "fixed_line_or_mobile",
+    PhoneNumberType.TOLL_FREE: "toll_free",
+    PhoneNumberType.PREMIUM_RATE: "premium_rate",
+    PhoneNumberType.SHARED_COST: "shared_cost",
+    PhoneNumberType.VOIP: "voip",
+    PhoneNumberType.PERSONAL_NUMBER: "personal_number",
+    PhoneNumberType.PAGER: "pager",
+    PhoneNumberType.UAN: "uan",
+    PhoneNumberType.VOICEMAIL: "voicemail",
+    PhoneNumberType.UNKNOWN: "unknown",
+}
+
+
+class VerifyService:
+    def verify_phone(self, number: str, country_code: str | None = None) -> PhoneVerificationResult:
+        try:
+            parsed = phonenumbers.parse(number, country_code)
+        except phonenumbers.NumberParseException as exc:
+            return PhoneVerificationResult(valid=False, error=f"Parsing error: {exc}")
+
+        if not phonenumbers.is_valid_number(parsed):
+            return PhoneVerificationResult(valid=False, error="Invalid phone number")
+
+        number_type = phonenumbers.number_type(parsed)
+        line_type = _TYPE_NAMES.get(number_type, "unknown")
+        location = geocoder.description_for_number(parsed, "en") or None
+        carrier_name = carrier.name_for_number(parsed, "en") or None
+
+        return PhoneVerificationResult(
+            valid=True,
+            international_format=phonenumbers.format_number(parsed, phonenumbers.PhoneNumberFormat.INTERNATIONAL),
+            national_format=phonenumbers.format_number(parsed, phonenumbers.PhoneNumberFormat.NATIONAL),
+            country_code=parsed.country_code,
+            location=location,
+            carrier_name=carrier_name,
+            line_type=line_type,
+        )

pyproject.toml

@@ -18,6 +18,7 @@ dependencies = [
     "pydantic-settings>=2.0.0",
     "python-multipart>=0.0.9",
     "httpx>=0.27",
+    "youtube-transcript-api>=1.2.4",
     "pyfiglet>=1.0.0",
     "rich>=13.7",
     "numpy>=1.26.0",
@@ -27,6 +28,7 @@ dependencies = [
     "pypdfium2>=4.30.0",
     "pandas>=2.0.0",
     "spacy>=3.7.0",
+    "phonenumbers>=8.13.0",
 ]
 
 [project.optional-dependencies]

requirements.txt

@@ -12,6 +12,7 @@ pillow>=10.0.0
 pypdfium2>=4.30.0
 pandas>=2.0.0
 spacy>=3.7.0
+phonenumbers>=8.13.0
 
 # Async database drivers
 aiomysql>=0.3.2