flowdeck/app/services/og_fetcher.py

"""FlowDeck — Bookmark cards (v5.5.0): Open Graph metadata via httpx.

Fetches a URL server-side, extracts OG/Twitter meta tags (title, description,
image, site name, favicon) and returns a safe, compact payload used to render
Notion-style bookmark cards. Robust to missing tags, non-HTML bodies and
slow/unreachable hosts.
"""
from __future__ import annotations

import logging
import re
from urllib.parse import urljoin, urlparse

logger = logging.getLogger(__name__)

_META_RE = re.compile(
    r'<meta\b[^>]*?(?:content=[\'"]([^\'"]*)[\'"]|property=[\'"]([^\'"]*)[\'"]|name=[\'"]([^\'"]*)[\'"]|content=[\'"]([^\'"]*)[\'"])[^>]*?>',
    re.I,
)

# Property/name keys we look for, in priority order, mapped to our payload keys.
_OG_TITLE = ("og:title", "twitter:title", "title", "og:site_name")
_OG_DESC = ("og:description", "twitter:description", "description")
_OG_IMG = ("og:image", "twitter:image", "twitter:image:src")
_OG_SITE = ("og:site_name", "twitter:site", "application-name")


def _extract_og(html: str) -> dict:
    text = html[:400_000]  # only scan the beginning — that's where <head> lives
    props: dict[str, str] = {}
    for m in _META_RE.finditer(text):
        m.groups()
        content = ""
        prop = ""
        for s in re.findall(r"(?:content|property|name)=[\"']([^\"']*)[\"']", m.group(0)):
            if prop == "":
                prop = s
            else:
                content = s
                break
        if prop:
            props[prop.lower()] = content
    return props


def _pick(props: dict, keys: tuple) -> str:
    for k in keys:
        v = props.get(k)
        if v:
            return v
    return ""


def _title_of(props: dict, raw_title: str) -> str:
    t = _pick(props, _OG_TITLE)
    if t:
        return t
    m = re.search(r"<title[^>]*>(.*?)</title>", raw_title[:200_000], re.I | re.S)
    return (m.group(1).strip() if m else "") or urlparse(props.get("_url", "")).netloc


def _site_name(url: str) -> str:
    host = urlparse(url).netloc.replace("www.", "")
    return host.split(".")[0].capitalize() if host else ""


async def fetch_og_metadata(url: str, timeout: float = 6.0) -> dict:
    """Fetch ``url`` and return {url, title, description, image, site_name,
    favicon}. Empty strings are omitted. Never raises for network errors."""
    src = url.strip()
    if not src.startswith(("http://", "https://")):
        src = "https://" + src
    base = {"url": src, "title": "", "description": "", "image": "", "site_name": "", "favicon": ""}
    try:
        import httpx
        headers = {
            "User-Agent": "FlowDeck/5.10 bookmark-fetcher (+https://flowdeck.dracodev.net)",
            "Accept": "text/html,application/xhtml+xml",
        }
        async with httpx.AsyncClient(follow_redirects=True, timeout=timeout) as client:
            resp = await client.get(src, headers=headers)
            resp.raise_for_status()
    except Exception as exc:  # noqa: BLE001 - network/parse failures are non-fatal
        logger.debug("og fetch failed for %s: %s", src, exc)
        base["title"] = urlparse(src).netloc or src
        base["site_name"] = _site_name(src)
        return base

    ctype = (resp.headers.get("content-type") or "").lower()
    if "text/html" not in ctype and "xhtml" not in ctype:
        base["title"] = urlparse(src).netloc or src
        base["site_name"] = _site_name(src)
        return base

    import html as htmlmod

    body = resp.text
    props = _extract_og(body)
    props["_url"] = src
    title = htmlmod.unescape(_title_of(props, body))
    desc = htmlmod.unescape(_pick(props, _OG_DESC))
    img = _pick(props, _OG_IMG)
    site = htmlmod.unescape(_pick(props, _OG_SITE)) or _site_name(src)

    def abs_url(u: str) -> str:
        if not u:
            return ""
        return urljoin(src, htmlmod.unescape(u))

    favicon = ""
    fm = re.search(r"<link[^>]+rel=[\"'](?:shortcut )?icon[\"'][^>]+href=[\"']([^\"']+)[\"']", body, re.I)
    if not fm:
        fm = re.search(r"<link[^>]+href=[\"']([^\"']+)[\"'][^>]+rel=[\"'](?:shortcut )?icon[\"']", body, re.I)
    if fm:
        favicon = abs_url(fm.group(1))

    return {
        "url": src,
        "title": title.strip()[:200] or urlparse(src).netloc or src,
        "description": desc.strip()[:400],
        "image": abs_url(img),
        "site_name": site.strip()[:100],
        "favicon": favicon,
    }