"""Media discovery from HTML, Open Graph, JSON-LD and microdata."""
from __future__ import annotations
import re
from dataclasses import dataclass
from pathlib import PurePosixPath
from typing import Any
from urllib.parse import urldefrag, urljoin, urlsplit
from bs4 import BeautifulSoup, Tag
from app.adapters.base import MediaCandidate, VideoPageLink
from app.services.caption_parser import (
CaptionResult,
SelectorAdapter,
iter_jsonld_objects,
normalize_text,
)
from app.utils.files import ALLOWED_MEDIA_EXTENSIONS
_MEDIA_MIME_BY_EXT = {
".mp4": "video/mp4",
".m4v": "video/mp4",
".webm": "video/webm",
".mkv": "video/x-matroska",
".mov": "video/quicktime",
".avi": "video/x-msvideo",
".flv": "video/x-flv",
".mpg": "video/mpeg",
".mpeg": "video/mpeg",
".ts": "video/mp2t",
".m2ts": "video/mp2t",
".ogv": "video/ogg",
".m3u8": "application/vnd.apple.mpegurl",
".mpd": "application/dash+xml",
}
_STREAM_EXTENSIONS = {".m3u8", ".mpd"}
_ALL_MEDIA_EXTENSIONS = ALLOWED_MEDIA_EXTENSIONS | _STREAM_EXTENSIONS
_DURATION_ISO_RE = re.compile(
r"^P(?:(?P\d+)D)?(?:T(?:(?P\d+)H)?(?:(?P\d+)M)?"
r"(?:(?P[\d.]+)S)?)?$",
re.IGNORECASE,
)
_VIDEO_OBJECT_TYPES = {"videoobject", "clip", "movie"}
_OG_VIDEO_PROPS = (
"og:video",
"og:video:url",
"og:video:secure_url",
"twitter:player:stream",
)
_OG_IMAGE_PROPS = ("og:image", "og:image:secure_url", "twitter:image")
@dataclass(slots=True)
class _RawVideo:
media_url: str | None
title: str
caption: str | None
thumbnail_url: str | None
duration_seconds: float | None
media_type: str
file_size: int | None
source: str
identity_url: str | None = None
direct_file: bool = False
def parse_duration(value: Any) -> float | None:
if value is None:
return None
if isinstance(value, (int, float)):
return float(value) if value > 0 else None
text = str(value).strip()
if not text:
return None
if text.isdigit():
result = float(text)
return result if result > 0 else None
match = _DURATION_ISO_RE.fullmatch(text)
if not match:
return None
days = int(match.group("days") or 0)
hours = int(match.group("hours") or 0)
minutes = int(match.group("minutes") or 0)
seconds = float(match.group("seconds") or 0)
total = days * 86400 + hours * 3600 + minutes * 60 + seconds
return total if total > 0 else None
def _clean_url(base_url: str, raw: Any) -> str | None:
if not isinstance(raw, str):
return None
value = raw.strip()
if not value or value.startswith(("javascript:", "data:", "blob:", "file:", "mailto:")):
return None
absolute = urldefrag(urljoin(base_url, value))[0]
if urlsplit(absolute).scheme.lower() not in {"http", "https"}:
return None
return absolute
def _extension(url: str) -> str:
return PurePosixPath(urlsplit(url).path).suffix.lower()
def _is_direct_file(url: str) -> bool:
return _extension(url) in _ALL_MEDIA_EXTENSIONS
def _media_type(url: str | None, hint: str | None = None) -> str:
if hint:
cleaned = hint.split(";", 1)[0].strip().lower()
if cleaned.startswith("video/") or cleaned in {
"application/vnd.apple.mpegurl",
"application/x-mpegurl",
"application/dash+xml",
}:
return cleaned
if url:
extension = _extension(url)
if extension in _MEDIA_MIME_BY_EXT:
return _MEDIA_MIME_BY_EXT[extension]
return "video/unknown"
def _title_of(tag: Tag) -> str:
for parent in tag.parents:
if not isinstance(parent, Tag):
continue
if parent.name not in {"article", "li", "section", "div", "figure"}:
continue
heading = parent.find(["h1", "h2", "h3", "h4", "h5"])
if isinstance(heading, Tag):
title = normalize_text(heading.get_text(" ", strip=True))
if title:
return title
itemprop_name = parent.find(attrs={"itemprop": "name"})
if isinstance(itemprop_name, Tag):
title = normalize_text(
str(itemprop_name.get("content") or itemprop_name.get_text(" ", strip=True))
)
if title:
return title
return ""
def _card_description(tag: Tag) -> str:
for parent in tag.parents:
if not isinstance(parent, Tag):
continue
if parent.name not in {"article", "li", "section", "div", "figure"}:
continue
for candidate in parent.select(
"[itemprop='description'], .description, .caption, .excerpt, p"
):
if isinstance(candidate, Tag):
text = normalize_text(candidate.get_text(" ", strip=True))
if len(text) >= 20:
return text
return ""
def _meta_contents(
soup: BeautifulSoup, properties: tuple[str, ...], base_url: str
) -> list[str]:
values: list[str] = []
for prop in properties:
for attrs in ({"property": prop}, {"name": prop}):
for tag in soup.find_all("meta", attrs=attrs):
content = _clean_url(base_url, tag.get("content"))
if content:
values.append(content)
return values
class MediaAdapter:
"""Extract raw media candidates without trusting or downloading them."""
def __init__(self, caption_parser: SelectorAdapter | None = None) -> None:
self._caption_parser = caption_parser
def extract(
self,
soup: BeautifulSoup,
url: str,
caption: CaptionResult,
adapter: SelectorAdapter | None = None,
video_page_selectors: tuple[str, ...] = (),
) -> tuple[list[MediaCandidate], list[VideoPageLink]]:
raws: list[_RawVideo] = []
page_thumbnail = self._page_thumbnail(soup, url)
# Rich JSON-LD VideoObject entries usually provide the most specific data.
for obj in iter_jsonld_objects(soup):
raw = self._from_jsonld(obj, url, page_thumbnail)
if raw is not None:
raws.append(raw)
# Standard Open Graph / Twitter video metadata.
og_type = ""
for meta in soup.find_all("meta"):
prop = str(meta.get("property") or meta.get("name") or "").lower()
if prop == "og:video:type":
og_type = str(meta.get("content") or "")
for media_url in _meta_contents(soup, _OG_VIDEO_PROPS, url):
raws.append(
_RawVideo(
media_url=media_url,
title=caption.title,
caption=caption.caption,
thumbnail_url=page_thumbnail,
duration_seconds=None,
media_type=_media_type(media_url, og_type),
file_size=None,
source="open_graph",
direct_file=_is_direct_file(media_url),
)
)
# Actual