Files
flowdeck/app/services/og_fetcher.py
T
bruno 6fe5d2f723 feat(v5.10.0): v5.4.0 + v5.5.0 features — backlinks, page/collection duplication, trash purge, version history, markdown import, embed/bookmark/video/audio blocks, cover & icon, OG metadata
- Migration v7: page_versions table + pages.cover_url/page_icon columns
- Trash service (purge_expired 30-day) + daily scheduler
- Board API: duplicate page, backlinks, versions (snapshot/list/restore), cover/icon upload, markdown/file import
- Collections API: duplicate collection (deep copy properties/views/pages/data sources)
- Export service: render embed/bookmark/video/audio in markdown/HTML/PDF/public
- Public page renderer: cover image + icon
- Editor: slash commands for media blocks, lightbox, version history UI, backlinks panel, cover/icon picker, import modal
- OG metadata endpoint for bookmark cards
- 386 tests passing
2026-09-11 08:56:48 -04:00

124 lines
4.3 KiB
Python

"""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):
groups = 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,
}