feat(posts): extract + record external file-host links (Phase 3)
Capture off-platform links (mega/gdrive/mediafire/dropbox/pixeldrain) embedded in post bodies so they're never silently dropped, and surface them in the post view. The download worker (Phase 4) walks these rows. - link_extract.py: pure extractor — <a href> + bare URLs, unwraps Patreon redirect shims, PRESERVES the full url incl. #fragment (mega's key), dedups. Reusable by every platform (runs off Post.description). - external_link model + migration 0049: post_id/artist_id/host/url/label/status /attempts/last_error/attachment_id/timing; CHECK whitelists (full enum incl. worker statuses up front) + (post_id,url) unique. - importer._sync_external_links: insert-missing on both import paths (_apply_sidecar + upsert_post_record) so a re-import never resets a link's status; runs for all platforms. - post_feed_service.get_post: returns external_links (detail-only). - PostCard: renders the links (host chip + label + status) once expanded. - tests: extractor (5 hosts, fragment, shim unwrap, dedup), importer (record + no-dup on reimport), serializer. Refs FC #830. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,105 @@
|
||||
"""Extract off-platform file-host links from a post body.
|
||||
|
||||
Pure (no I/O) so it's unit-testable and reusable by EVERY in-house downloader's
|
||||
import path — every platform stores its body in `Post.description`, so running
|
||||
this there covers them all. Finds links to the supported external file hosts in
|
||||
a post's HTML body (both `<a href="...">` anchors and bare URLs in text),
|
||||
unwraps a Patreon outbound-redirect wrapper, and preserves the FULL url
|
||||
including the `#fragment` (mega.nz puts the decryption key there) and the query
|
||||
string — without those a mega download is impossible.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from html import unescape
|
||||
from urllib.parse import parse_qs, urlsplit
|
||||
|
||||
# Supported file-host enum values (kept in sync with the external_link CHECK).
|
||||
SUPPORTED_HOSTS = ("mega", "gdrive", "mediafire", "dropbox", "pixeldrain")
|
||||
|
||||
# Bare-domain suffix → canonical host label. Matched against the URL netloc
|
||||
# (exact or as a dotted suffix, so www./dl. subdomains resolve too).
|
||||
_HOST_MAP = {
|
||||
"mega.nz": "mega",
|
||||
"mega.co.nz": "mega",
|
||||
"drive.google.com": "gdrive",
|
||||
"mediafire.com": "mediafire",
|
||||
"dropbox.com": "dropbox",
|
||||
"pixeldrain.com": "pixeldrain",
|
||||
}
|
||||
|
||||
# `<a href="...">label</a>` — DOTALL so a label spanning tags/newlines is caught.
|
||||
_HREF_RE = re.compile(
|
||||
r"""<a\b[^>]*?\bhref=["']([^"']+)["'][^>]*>(.*?)</a>""",
|
||||
re.IGNORECASE | re.DOTALL,
|
||||
)
|
||||
_TAG_RE = re.compile(r"<[^>]+>")
|
||||
# Bare http(s) URL in text. Stops at whitespace, quotes, angle brackets, and
|
||||
# closing brackets — but KEEPS `#`, `&`, `?`, `=` so fragments/queries survive.
|
||||
_URL_RE = re.compile(r"""https?://[^\s"'<>)\]}]+""", re.IGNORECASE)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ExtractedLink:
|
||||
host: str # one of SUPPORTED_HOSTS
|
||||
url: str # full url incl. query + #fragment
|
||||
label: str | None # visible anchor text, when present
|
||||
|
||||
|
||||
def _netloc(url: str) -> str:
|
||||
# Lowercase host without credentials or port.
|
||||
return urlsplit(url).netloc.lower().split("@")[-1].split(":")[0]
|
||||
|
||||
|
||||
def host_for(url: str) -> str | None:
|
||||
"""Canonical host label for a url, or None if it's not a supported host."""
|
||||
netloc = _netloc(url)
|
||||
for suffix, host in _HOST_MAP.items():
|
||||
if netloc == suffix or netloc.endswith("." + suffix):
|
||||
return host
|
||||
return None
|
||||
|
||||
|
||||
def _unwrap(url: str) -> str:
|
||||
"""Unwrap a Patreon outbound-redirect wrapper to its inner target, so a
|
||||
wrapped mega/gdrive link resolves to the real host. Patreon has used both
|
||||
`www.patreon.com/...?url=<encoded>` and the `l.patreon.com` shim; check the
|
||||
common target-param names. Non-Patreon urls pass through untouched."""
|
||||
netloc = _netloc(url)
|
||||
if not (netloc == "patreon.com" or netloc.endswith(".patreon.com")):
|
||||
return url
|
||||
qs = parse_qs(urlsplit(url).query)
|
||||
for key in ("url", "u", "ext_url", "redirect", "target"):
|
||||
vals = qs.get(key)
|
||||
if vals and vals[0]:
|
||||
return unescape(vals[0]).strip()
|
||||
return url
|
||||
|
||||
|
||||
def extract_external_links(html: str | None) -> list[ExtractedLink]:
|
||||
"""All supported-host links in `html`, de-duplicated by url (first wins, so
|
||||
an anchor's label is kept over a later bare sighting of the same url)."""
|
||||
if not html:
|
||||
return []
|
||||
found: dict[str, ExtractedLink] = {}
|
||||
|
||||
# 1) Anchors first — they carry a human label ("Mega - Streamable").
|
||||
for raw_href, inner in _HREF_RE.findall(html):
|
||||
url = _unwrap(unescape(raw_href).strip())
|
||||
host = host_for(url)
|
||||
if host is None:
|
||||
continue
|
||||
label = unescape(_TAG_RE.sub("", inner)).strip() or None
|
||||
found.setdefault(url, ExtractedLink(host=host, url=url, label=label))
|
||||
|
||||
# 2) Bare URLs pasted in text (no label). Trailing prose punctuation is
|
||||
# trimmed; the href values already captured above de-dup away here.
|
||||
for raw in _URL_RE.findall(html):
|
||||
url = _unwrap(unescape(raw).strip().rstrip(".,;"))
|
||||
host = host_for(url)
|
||||
if host is None:
|
||||
continue
|
||||
found.setdefault(url, ExtractedLink(host=host, url=url, label=None))
|
||||
|
||||
return list(found.values())
|
||||
Reference in New Issue
Block a user