thejayman77
d472b63cbf
A stored og:image isn't proof it renders: signed/hotlink-protected URLs (e.g. the Guardian's i.guim.co.uk) 401 on a direct browser load, so they counted toward coverage yet always fell back. Now fetch_og_image confirms the image truly returns 200 + image/* (requested no-referrer, same SSRF-safe redirect handling) before storing it. Add prune_broken_images() to clear already-stored URLs that no longer load, so coverage is honest and those cards show the placeholder cleanly. The browser onerror→placeholder remains the final safety net. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
338 lines
12 KiB
Python
338 lines
12 KiB
Python
"""Bounded hero-image enrichment.
|
|
|
|
The grid stays purely typographic; the daily brief's items are the one place we
|
|
make an exception and fetch a real image — because the hero is the single
|
|
intentional visual doorway. We fetch ONLY the article page's <head> metadata
|
|
(og:image / twitter:image), store ONLY the resulting image URL, and never touch
|
|
the body. This is opt-in, brief-only, once per build.
|
|
|
|
Security (this is the one place we fetch user-/source-supplied pages):
|
|
- http(s) only; short timeout; byte cap; redirects followed manually and capped;
|
|
- every hop's host is DNS-resolved and rejected if ANY resolved address is
|
|
private / loopback / link-local / multicast / reserved / unspecified (SSRF).
|
|
Failures are cached by the caller (image_checked_at) so an article is never
|
|
retried forever.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ipaddress
|
|
import re
|
|
import socket
|
|
import sqlite3
|
|
import urllib.error
|
|
import urllib.request
|
|
from urllib.parse import urljoin, urlsplit
|
|
|
|
from .text import canonicalize_url
|
|
|
|
USER_AGENT = "goodNews/0.1 (+local constructive news prototype)"
|
|
TIMEOUT = 6
|
|
MAX_BYTES = 300_000
|
|
MAX_REDIRECTS = 3
|
|
|
|
_META_RE = re.compile(rb"<meta\b[^>]*>", re.IGNORECASE)
|
|
_HEAD_END_RE = re.compile(rb"</head>", re.IGNORECASE)
|
|
|
|
# Substrings that mark a generic placeholder/default share image rather than the
|
|
# article's own picture (e.g. NPR's facebook-default). We'd rather show no image
|
|
# (typographic hero) than a generic logo card. NOTE: do NOT add "branded_news" —
|
|
# that's BBC's normal CDN path for real article photos, so rejecting it threw away
|
|
# every BBC hero image and fell back to the tiny RSS thumbnail.
|
|
_GENERIC_IMAGE_MARKERS = (
|
|
"facebook-default",
|
|
"default-wide",
|
|
"default-fb",
|
|
"og-default",
|
|
"default-og",
|
|
"twitter-default",
|
|
"default-image",
|
|
"/placeholder",
|
|
"share-default",
|
|
"social-default",
|
|
# tracking pixels / spacers / data-URIs — never a real share image
|
|
"data:image",
|
|
"/pixel",
|
|
"1x1",
|
|
"spacer",
|
|
"/blank.",
|
|
"transparent.",
|
|
)
|
|
|
|
|
|
def _is_generic_image(url: str) -> bool:
|
|
lowered = url.lower()
|
|
return any(marker in lowered for marker in _GENERIC_IMAGE_MARKERS)
|
|
|
|
|
|
def _prefer_unbranded(url: str) -> str:
|
|
"""Swap BBC's logo-branded image variant for its clean one.
|
|
|
|
BBC's og:image is served from the "branded_news" CDN path with a "BBC NEWS"
|
|
logo baked into the picture (it shows as "…EWS" once the hero crops it). The
|
|
identical photo is served under "cpsprodpb" with no logo, so prefer that — a
|
|
clean hero at the same full resolution.
|
|
"""
|
|
if "ichef.bbci.co.uk" in url and "/branded_news/" in url:
|
|
return url.replace("/branded_news/", "/cpsprodpb/")
|
|
return url
|
|
|
|
|
|
def _attr(tag: bytes, name: bytes) -> bytes | None:
|
|
m = re.search(name + rb"""\s*=\s*["']([^"']*)["']""", tag, re.IGNORECASE)
|
|
return m.group(1) if m else None
|
|
|
|
|
|
def og_image_from_html(html: bytes) -> str | None:
|
|
"""Extract og:image / twitter:image from a page's <head> bytes."""
|
|
head = html.split(b"</head>", 1)[0] if _HEAD_END_RE.search(html) else html
|
|
for tag in _META_RE.findall(head):
|
|
key = _attr(tag, b"property") or _attr(tag, b"name")
|
|
if key and key.lower() in (b"og:image", b"og:image:url", b"twitter:image"):
|
|
content = _attr(tag, b"content")
|
|
if not content:
|
|
continue
|
|
image = canonicalize_url(content.decode("utf-8", "replace"))
|
|
# Skip generic placeholders; keep scanning for a real one.
|
|
if image and not _is_generic_image(image):
|
|
return _prefer_unbranded(image)
|
|
return None
|
|
|
|
|
|
def _host_is_public(host: str | None) -> bool:
|
|
"""True only if the host resolves and ALL its addresses are public."""
|
|
if not host:
|
|
return False
|
|
try:
|
|
infos = socket.getaddrinfo(host, None)
|
|
except (socket.gaierror, UnicodeError, OSError):
|
|
return False
|
|
addrs = {info[4][0] for info in infos}
|
|
if not addrs:
|
|
return False
|
|
for addr in addrs:
|
|
try:
|
|
ip = ipaddress.ip_address(addr.split("%")[0]) # strip scope id
|
|
except ValueError:
|
|
return False
|
|
if (
|
|
ip.is_private or ip.is_loopback or ip.is_link_local
|
|
or ip.is_multicast or ip.is_reserved or ip.is_unspecified
|
|
):
|
|
return False
|
|
return True
|
|
|
|
|
|
class _NoRedirect(urllib.request.HTTPRedirectHandler):
|
|
# Don't auto-follow — we re-validate each hop's host ourselves.
|
|
def redirect_request(self, *args, **kwargs):
|
|
return None
|
|
|
|
|
|
def fetch_og_image(url: str | None) -> str | None:
|
|
"""Fetch a page's head metadata and return its og:image URL, or None.
|
|
|
|
Best-effort and safe: returns None on any error, bad scheme, redirect loop,
|
|
or a host that resolves to a non-public address.
|
|
"""
|
|
opener = urllib.request.build_opener(_NoRedirect)
|
|
for _ in range(MAX_REDIRECTS + 1):
|
|
if not url:
|
|
return None
|
|
parts = urlsplit(url)
|
|
if parts.scheme not in ("http", "https") or not _host_is_public(parts.hostname):
|
|
return None
|
|
request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT, "Accept": "text/html"})
|
|
try:
|
|
response = opener.open(request, timeout=TIMEOUT)
|
|
except (urllib.error.URLError, OSError, ValueError):
|
|
return None
|
|
status = getattr(response, "status", 200) or 200
|
|
if status in (301, 302, 303, 307, 308):
|
|
location = response.headers.get("Location")
|
|
response.close()
|
|
if not location:
|
|
return None
|
|
url = urljoin(url, location)
|
|
continue
|
|
ctype = response.headers.get("Content-Type", "")
|
|
if "html" not in ctype.lower():
|
|
response.close()
|
|
return None
|
|
try:
|
|
body = response.read(MAX_BYTES)
|
|
finally:
|
|
response.close()
|
|
image = og_image_from_html(body)
|
|
# A stored URL is not proof it renders — confirm it actually loads.
|
|
return image if (image and _image_loads(image)) else None
|
|
return None # too many redirects
|
|
|
|
|
|
def _image_loads(url: str) -> bool:
|
|
"""Confirm an image URL truly returns an image (HTTP 200 + image/* type).
|
|
|
|
Many publishers serve a signed or hotlink-protected og:image that 401/403s
|
|
on a direct request (e.g. the Guardian's i.guim.co.uk), so storing the URL
|
|
would overstate coverage and the card would never render it. We request as
|
|
the browser does — no referrer — with the same per-hop host safety as the
|
|
page fetch. Returns False on any error.
|
|
"""
|
|
opener = urllib.request.build_opener(_NoRedirect)
|
|
for _ in range(MAX_REDIRECTS + 1):
|
|
if not url:
|
|
return False
|
|
parts = urlsplit(url)
|
|
if parts.scheme not in ("http", "https") or not _host_is_public(parts.hostname):
|
|
return False
|
|
request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT, "Accept": "image/*,*/*"})
|
|
try:
|
|
response = opener.open(request, timeout=TIMEOUT)
|
|
except (urllib.error.URLError, OSError, ValueError):
|
|
return False
|
|
try:
|
|
status = getattr(response, "status", 200) or 200
|
|
if status in (301, 302, 303, 307, 308):
|
|
location = response.headers.get("Location")
|
|
if not location:
|
|
return False
|
|
url = urljoin(url, location)
|
|
continue
|
|
ctype = (response.headers.get("Content-Type") or "").lower()
|
|
return status == 200 and ctype.startswith("image/")
|
|
finally:
|
|
response.close()
|
|
return False
|
|
|
|
|
|
def prune_broken_images(conn: sqlite3.Connection, check=_image_loads, limit: int = 3000) -> int:
|
|
"""Clear stored image URLs that no longer load (signed/expired/hotlink-
|
|
protected), so coverage is honest and those cards fall back to the calm
|
|
placeholder cleanly instead of attempting a doomed fetch. Returns count cleared.
|
|
"""
|
|
rows = conn.execute(
|
|
"SELECT id, image_url FROM articles WHERE image_url IS NOT NULL AND image_url != '' "
|
|
"ORDER BY id DESC LIMIT ?",
|
|
(limit,),
|
|
).fetchall()
|
|
cleared = 0
|
|
for row in rows:
|
|
if not check(row["image_url"]):
|
|
conn.execute(
|
|
"UPDATE articles SET image_url = NULL, image_checked_at = CURRENT_TIMESTAMP WHERE id = ?",
|
|
(row["id"],),
|
|
)
|
|
cleared += 1
|
|
conn.commit()
|
|
return cleared
|
|
|
|
|
|
def enrich_brief_images(
|
|
conn: sqlite3.Connection, brief_date: str, fetch=fetch_og_image, limit: int = 7, retry_days: int = 2
|
|
) -> int:
|
|
"""Fetch a hero-quality image for brief items that lack one.
|
|
|
|
Any of the brief's items can become the hero (via the client's fallback or a
|
|
replace), so this covers the whole brief (limit defaults to the brief size, 7),
|
|
not just the top few. Items already carrying an image are left alone; items
|
|
still without one are retried after `retry_days` so a transient fetch failure
|
|
or a weaker earlier extractor doesn't mark an article imageless forever.
|
|
Returns how many images were newly found.
|
|
"""
|
|
# Fetch even when a feed image exists, because feed thumbnails are often tiny
|
|
# and the hero is shown large — a page's og:image is the better hero visual.
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT a.id, a.canonical_url
|
|
FROM daily_briefs b
|
|
JOIN daily_brief_items bi ON bi.brief_id = b.id
|
|
JOIN articles a ON a.id = bi.article_id
|
|
WHERE b.brief_date = ?
|
|
AND (
|
|
a.image_checked_at IS NULL
|
|
OR ((a.image_url IS NULL OR a.image_url = '')
|
|
AND a.image_checked_at < datetime('now', ?))
|
|
)
|
|
ORDER BY bi.rank
|
|
LIMIT ?
|
|
""",
|
|
(brief_date, f"-{retry_days} days", limit),
|
|
).fetchall()
|
|
|
|
found = 0
|
|
for row in rows:
|
|
try:
|
|
image = fetch(row["canonical_url"])
|
|
except Exception:
|
|
image = None
|
|
conn.execute(
|
|
"UPDATE articles SET image_url = COALESCE(?, image_url), image_checked_at = CURRENT_TIMESTAMP "
|
|
"WHERE id = ?",
|
|
(image, row["id"]),
|
|
)
|
|
if image:
|
|
found += 1
|
|
conn.commit()
|
|
return found
|
|
|
|
|
|
def enrich_article_image(
|
|
conn: sqlite3.Connection, article_id: int, fetch=fetch_og_image, retry_days: int = 7
|
|
) -> bool:
|
|
"""Attention-triggered: fetch an og:image for ONE article that lacks one.
|
|
|
|
Called when an article earns a summary (i.e. it's actually being read), so we
|
|
only spend a fetch on articles a reader has reached. Leaves an existing image
|
|
alone; retries a still-imageless article only after `retry_days`. Returns True
|
|
if a new image was stored. Best-effort — never raises.
|
|
"""
|
|
row = conn.execute(
|
|
"""
|
|
SELECT id, canonical_url FROM articles
|
|
WHERE id = ?
|
|
AND (image_url IS NULL OR image_url = '')
|
|
AND (image_checked_at IS NULL OR image_checked_at < datetime('now', ?))
|
|
""",
|
|
(article_id, f"-{retry_days} days"),
|
|
).fetchone()
|
|
if not row:
|
|
return False # has an image already, or checked too recently
|
|
try:
|
|
image = fetch(row["canonical_url"])
|
|
except Exception:
|
|
image = None
|
|
conn.execute(
|
|
"UPDATE articles SET image_url = COALESCE(?, image_url), image_checked_at = CURRENT_TIMESTAMP "
|
|
"WHERE id = ?",
|
|
(image, article_id),
|
|
)
|
|
conn.commit()
|
|
return bool(image)
|
|
|
|
|
|
def enrich_summarized_images(
|
|
conn: sqlite3.Connection, fetch=fetch_og_image, limit: int = 50, retry_days: int = 7
|
|
) -> int:
|
|
"""Slow backfill: give already-summarized, accepted articles an image if they
|
|
lack one. Run in modest batches so we never blast publishers. Returns count
|
|
of newly-found images.
|
|
"""
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT a.id FROM articles a
|
|
JOIN article_summaries m ON m.article_id = a.id
|
|
JOIN article_scores s ON s.article_id = a.id
|
|
WHERE s.accepted = 1 AND a.duplicate_of IS NULL
|
|
AND (a.image_url IS NULL OR a.image_url = '')
|
|
AND (a.image_checked_at IS NULL OR a.image_checked_at < datetime('now', ?))
|
|
ORDER BY a.id DESC
|
|
LIMIT ?
|
|
""",
|
|
(f"-{retry_days} days", limit),
|
|
).fetchall()
|
|
found = 0
|
|
for row in rows:
|
|
if enrich_article_image(conn, row["id"], fetch=fetch, retry_days=retry_days):
|
|
found += 1
|
|
return found
|