Import tracks from a saved page, for sites that block GAM (Pixabay)

backend/app/enrichment/page_import.py

@@ -23,7 +23,7 @@
 import tempfile
 from dataclasses import dataclass, fields
 from pathlib import Path
-from typing import Any, Callable, Optional
+from typing import Any, BinaryIO, Callable, Optional
 from urllib.parse import urljoin
 
 from sqlmodel import Session, col, func, select
@@ -32,7 +32,7 @@
 from app.clock import utcnow
 from app.ingest import page_tracks, web
 from app.ingest.embedded_metadata import normalise_partial_date
-from app.ingest.filetypes import SOURCE_URL, TYPE_AUDIO
+from app.ingest.filetypes import AUDIO_EXTENSIONS, SOURCE_UPLOAD, SOURCE_URL, TYPE_AUDIO, extension_of
 from app.ingest.page_tracks import PageExtraction, PageTrack
 from app.models.asset import Asset
 from app.models.job import EnrichmentJob
@@ -107,11 +107,12 @@ def preview(session: Session, user_id: str, url: str, *, use_ai: bool = True) ->
 
 
 def already_imported(session: Session, user_id: str, tracks: list[PageTrack]) -> dict[str, str]:
-    """{audio_url: asset id} for the tracks this library already has."""
+    """{track.key: asset id} for the tracks this library already has."""
     keys: dict[str, str] = {}
     for track in tracks:
-        keys[track.source_url or track.audio_url] = track.audio_url
-        keys[track.audio_url] = track.audio_url
+        for url in (track.source_url, track.audio_url):
+            if url:
+                keys[url] = track.key
     if not keys:
         return {}
     rows = session.exec(
@@ -123,7 +124,7 @@ def already_imported(session: Session, user_id: str, tracks: list[PageTrack]) ->
 
 
 def similar_in_library(session: Session, user_id: str, tracks: list[PageTrack]) -> dict[str, str]:
-    """{audio_url: asset id} for tracks whose title (and artist, when both are known)
+    """{track.key: asset id} for tracks whose title (and artist, when both are known)
     match an audio file already in the library.
 
     A warning, not a refusal. The same piece downloaded from somewhere else is a
@@ -150,7 +151,7 @@ def similar_in_library(session: Session, user_id: str, tracks: list[PageTrack])
         for asset_id, creator in by_title.get(track.title.strip().lower(), []):
             if track.creator and creator and track.creator.strip().lower() != creator.strip().lower():
                 continue
-            found[track.audio_url] = asset_id
+            found[track.key] = asset_id
             break
     return found
 
@@ -309,8 +310,8 @@ def run_track(session: Session, job: EnrichmentJob, progress: Progress) -> Track
     track = decode_track(job.payload)
 
     existing = already_imported(session, job.user_id, [track])
-    if track.audio_url in existing:
-        return TrackImportResult("Already in your library", existing[track.audio_url])
+    if track.key in existing:
+        return TrackImportResult("Already in your library", existing[track.key])
 
     if not capacity.has_room():
         raise PageImportError(capacity.DISK_FULL_MESSAGE)
@@ -424,3 +425,125 @@ def apply_track_metadata(
     except Exception:  # noqa: BLE001 - same
         logger.warning("Could not attach page tags to %s", asset.id, exc_info=True)
     return attached
+
+
+# ─── a saved page and the files downloaded from it ──────────────────────────
+
+
+SAVED_STATUS_IMPORTED = "imported"
+SAVED_STATUS_DUPLICATE = "duplicate"
+SAVED_STATUS_UNMATCHED = "unmatched"
+SAVED_STATUS_REJECTED = "rejected"
+
+
+@dataclass
+class SavedFileResult:
+    filename: str
+    status: str
+    asset_id: Optional[str] = None
+    track_title: Optional[str] = None
+    message: str = ""
+
+
+def import_saved(
+    session: Session,
+    user_id: str,
+    extraction: PageExtraction,
+    files: list[tuple[str, BinaryIO]],
+) -> list[SavedFileResult]:
+    """Catalogue files the person downloaded, with what their saved page said of each.
+
+    For sites that refuse GAM (see `page_tracks.extract_saved`). Nothing is fetched: the
+    person brought the page and the files, and this only matches one to the other and
+    files them, the way an upload is filed. A file that matches no track on the page is
+    still imported, as music without the page's details; the person chose to upload it.
+    The same three duplicate checks as a live import apply, the checksum one here.
+    """
+    matches = page_tracks.match_files(extraction.tracks, [name for name, _ in files])
+    results: list[SavedFileResult] = []
+    with tempfile.TemporaryDirectory(prefix="gam-saved-") as workdir:
+        for index, (name, stream) in enumerate(files):
+            extension = extension_of(name)
+            if extension not in AUDIO_EXTENSIONS:
+                results.append(SavedFileResult(name, SAVED_STATUS_REJECTED, message="Not an audio file"))
+                continue
+            if not capacity.has_room():
+                results.append(
+                    SavedFileResult(name, SAVED_STATUS_REJECTED, message=capacity.DISK_FULL_MESSAGE)
+                )
+                continue
+
+            path = Path(workdir) / f"{index}{extension}"
+            if not _spool(stream, path, web.MAX_TRACK_BYTES):
+                results.append(
+                    SavedFileResult(
+                        name,
+                        SAVED_STATUS_REJECTED,
+                        message=f"Larger than the {web.MAX_TRACK_BYTES // (1024 * 1024)} MB limit",
+                    )
+                )
+                continue
+
+            track = matches.get(name)
+            duplicate = _same_file(session, user_id, path)
+            if duplicate is not None:
+                if track is not None:
+                    apply_track_metadata(session, duplicate, track, fill_blanks_only=True)
+                results.append(
+                    SavedFileResult(
+                        name,
+                        SAVED_STATUS_DUPLICATE,
+                        asset_id=duplicate.id,
+                        track_title=track.title if track else None,
+                        message=f"Already in your library as \u201c{duplicate.name}\u201d",
+                    )
+                )
+                continue
+
+            try:
+                asset = asset_service.ingest_file(
+                    session,
+                    build_storage(),
+                    user_id=user_id,
+                    source_path=path,
+                    filename=name,
+                    name=track.title if track else None,
+                    source=SOURCE_UPLOAD,
+                    # The person brought it from a music library's page: music, so not
+                    # sent to Deepgram, matched or not.
+                    audio_kind=music.AUDIO_KIND_MUSIC,
+                )
+            except asset_service.UnsupportedFile:
+                results.append(SavedFileResult(name, SAVED_STATUS_REJECTED, message="Not an audio file"))
+                continue
+
+            if track is None:
+                results.append(
+                    SavedFileResult(
+                        name,
+                        SAVED_STATUS_UNMATCHED,
+                        asset_id=asset.id,
+                        message="Imported, but not found on the page, so it has no details from it",
+                    )
+                )
+                continue
+            apply_track_metadata(session, asset, track)
+            results.append(
+                SavedFileResult(name, SAVED_STATUS_IMPORTED, asset_id=asset.id, track_title=track.title)
+            )
+    return results
+
+
+def _spool(stream: BinaryIO, target: Path, max_bytes: int) -> bool:
+    """Copy an upload to disk, giving up past `max_bytes`. False when it was too large."""
+    written = 0
+    with open(target, "wb") as handle:
+        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
+            written += len(chunk)
+            if written > max_bytes:
+                break
+            handle.write(chunk)
+    if written > max_bytes:
+        target.unlink(missing_ok=True)
+        return False
+    return True

backend/app/ingest/page_tracks.py

@@ -50,9 +50,14 @@
 
 @dataclass
 class PageTrack:
-    """One track as a page describes it. Every field but the URL and title optional."""
+    """One track as a page describes it. Every field but the title optional.
 
-    audio_url: str
+    `audio_url` is always set on a live page import, which downloads from it. A saved page
+    can lack one (Pixabay's download is a button, not a link): the file then comes from
+    the person, matched to the track by `match_id` or title (`match_files`).
+    """
+
+    audio_url: Optional[str]
     title: str
     creator: Optional[str] = None
     publisher: Optional[str] = None
@@ -70,6 +75,14 @@ class PageTrack:
     instruments: list[str] = field(default_factory=list)
     tags: list[str] = field(default_factory=list)
     bpm: Optional[float] = None
+    # A site's own id for the track, when its downloaded files carry it in their names
+    # (Pixabay's do: `artist-title-604094.mp3`). What matches a saved page to the files.
+    match_id: Optional[str] = None
+
+    @property
+    def key(self) -> str:
+        """What identifies this track within one page, with or without an audio URL."""
+        return self.audio_url or self.source_url or self.title
 
     def to_dict(self) -> dict[str, Any]:
         return asdict(self)
@@ -173,6 +186,171 @@ def page_text(page: str, limit: int = 60_000) -> str:
     return body.strip()[:limit]
 
 
+def extract_saved(page: str) -> PageExtraction:
+    """Tracks from a page the person saved in their own browser. Never touches the network.
+
+    For sites that refuse automated requests (Pixabay answers a server with 403): the
+    person opens the page, saves it, and downloads the files themselves, so the only
+    thing GAM does with the site is read what they handed it. The page's own address is
+    recovered from the saved file, since adapters key on the host and on the path.
+    """
+    page_url = saved_page_url(page) or ""
+    host = (urlsplit(page_url).hostname or "").lower()
+
+    if _is_host(host, "pixabay.com"):
+        tracks = _pixabay(page_url, page)
+        if tracks:
+            return PageExtraction(page_url, METHOD_STRUCTURED, tracks[:MAX_TRACKS], "pixabay")
+
+    try:
+        found = extract_structured(page_url, page, _no_network)
+    except SavedPageNeedsMore as exc:
+        return PageExtraction(page_url, METHOD_STRUCTURED, [], "saved page", note=str(exc))
+    if found is not None:
+        return found
+    links = extract_links(page_url, page)
+    if not links.tracks:
+        links.note = "No tracks were found in the saved page."
+    return links
+
+
+class SavedPageNeedsMore(Exception):
+    """An adapter needed a second file the saved page only points to."""
+
+
+def _no_network(url: str) -> tuple[str, str]:
+    name = urlsplit(url).path.rsplit("/", 1)[-1] or url
+    raise SavedPageNeedsMore(
+        f"This page loads its track list from {name}. Use the page link instead of a saved copy."
+    )
+
+
+def saved_page_url(page: str) -> Optional[str]:
+    """Where a saved page came from: its canonical link, its og:url, or the comment
+    Chrome and Edge write at the top of a saved page."""
+    for pattern in (
+        r"<link\b[^>]*rel=[\"']canonical[\"'][^>]*href=[\"']([^\"']+)[\"']",
+        r"<link\b[^>]*href=[\"']([^\"']+)[\"'][^>]*rel=[\"']canonical[\"']",
+        r"<meta\b[^>]*property=[\"']og:url[\"'][^>]*content=[\"']([^\"']+)[\"']",
+        r"<!--\s*saved from url=\(\d+\)(\S+?)\s*-->",
+    ):
+        match = re.search(pattern, page, re.I)
+        if match and match.group(1).startswith(("http://", "https://")):
+            return html_lib.unescape(match.group(1))
+    return None
+
+
+# ─── matching files to tracks ────────────────────────────────────────────────
+
+
+def _normalise(text: str) -> str:
+    return re.sub(r"[^a-z0-9]+", " ", text.lower()).strip()
+
+
+def match_files(tracks: list[PageTrack], filenames: list[str]) -> dict[str, Optional[PageTrack]]:
+    """Which track each downloaded file is. One file per track, one track per file.
+
+    By the site's id in the filename first (Pixabay names a download
+    `artist-title-604094.mp3`), which cannot be wrong. Then by title contained in the
+    filename, longest title first, so a file named for "Tension Riser" is not taken by
+    the track called "Riser". A file matching nothing maps to None.
+    """
+    result: dict[str, Optional[PageTrack]] = {name: None for name in filenames}
+    taken: set[int] = set()
+    by_id = {t.match_id: i for i, t in enumerate(tracks) if t.match_id}
+
+    for name in filenames:
+        stem = name.rsplit(".", 1)[0]
+        for number in re.findall(r"\d+", stem):
+            index = by_id.get(number)
+            if index is not None and index not in taken:
+                result[name] = tracks[index]
+                taken.add(index)
+                break
+
+    titled = sorted(
+        ((i, _normalise(t.title)) for i, t in enumerate(tracks) if _normalise(t.title)),
+        key=lambda pair: len(pair[1]),
+        reverse=True,
+    )
+    for name in filenames:
+        if result[name] is not None:
+            continue
+        stem = f" {_normalise(name.rsplit('.', 1)[0])} "
+        for index, title in titled:
+            if index not in taken and f" {title} " in stem:
+                result[name] = tracks[index]
+                taken.add(index)
+                break
+    return result
+
+
+# ─── pixabay ─────────────────────────────────────────────────────────────────
+
+# A saved copy of a Pixabay music listing, as the browser rendered it: one row per track.
+# The class names carry build hashes (`audioRow--et3ac`) that change every release, so
+# only their stable prefixes are matched. There is no audio URL in a row, since
+# downloading is a button, so the person's own downloads are matched to these rows.
+_PIXABAY_ROW = re.compile(r"class=\"audioRow--")
+_PIXABAY_TITLE = re.compile(r"<a href=\"(/music/[^\"]+)\"[^>]*class=\"title--[^\"]*\"[^>]*>([^<]*)</a>")
+_PIXABAY_ARTIST = re.compile(r"class=\"name--[^\"]*\"[^>]*>([^<]*)</a>")
+_PIXABAY_DURATION = re.compile(r"class=\"duration--[^\"]*\"[^>]*>([^<]*)<")
+_PIXABAY_GENRE = re.compile(r"class=\"theme--[^\"]*\"[^>]*>([^<]*)</a>")
+_PIXABAY_TAGS = re.compile(r"class=\"tags--[^\"]*\"[^>]*>(.*?)</div>", re.S)
+_PIXABAY_LICENSE = "Pixabay Content License (https://pixabay.com/service/license-summary/)"
+
+
+def _pixabay(page_url: str, page: str) -> list[PageTrack]:
+    starts = [m.start() for m in _PIXABAY_ROW.finditer(page)] + [len(page)]
+    kind, label = _pixabay_page_filter(page_url)
+    tracks: list[PageTrack] = []
+    for start, end in zip(starts, starts[1:]):
+        row = page[start:end]
+        title_match = _PIXABAY_TITLE.search(row)
+        if not title_match:
+            continue
+        link = html_lib.unescape(title_match.group(1))
+        title = _plain(title_match.group(2), 255)
+        if not title:
+            continue
+        artist = _PIXABAY_ARTIST.search(row)
+        duration = _PIXABAY_DURATION.search(row)
+        genre = _PIXABAY_GENRE.search(row)
+        tag_block = _PIXABAY_TAGS.search(row)
+        tags = (
+            [html_lib.unescape(t) for t in re.findall(r">([^<]+)</a>", tag_block.group(1))]
+            if tag_block
+            else []
+        )
+        id_match = re.search(r"-(\d+)/?$", link)
+        track = PageTrack(
+            audio_url=None,
+            title=title,
+            creator=_plain(artist.group(1), MAX_TEXT) if artist else None,
+            publisher="Pixabay",
+            duration_seconds=_clock_seconds(duration.group(1)) if duration else None,
+            license=_PIXABAY_LICENSE,
+            source_url=urljoin("https://pixabay.com/", link),
+            genres=music.clean_terms([html_lib.unescape(genre.group(1))]) if genre else [],
+            tags=music.clean_terms(tags),
+            match_id=id_match.group(1) if id_match else None,
+        )
+        if kind == "mood":
+            track.moods = music.clean_terms([*track.moods, label])
+        elif kind == "genre":
+            track.genres = music.clean_terms([*track.genres, label])
+        tracks.append(track)
+    return tracks
+
+
+def _pixabay_page_filter(page_url: str) -> tuple[Optional[str], str]:
+    """`/music/search/mood/mysterious/` says every track listed is mysterious."""
+    parts = [p for p in urlsplit(page_url).path.split("/") if p]
+    if len(parts) >= 4 and parts[:2] == ["music", "search"] and parts[2] in ("mood", "genre"):
+        return parts[2], unquote(parts[3]).replace("-", " ").title()
+    return None, ""
+
+
 # ─── incompetech ─────────────────────────────────────────────────────────────
 
 # Kevin MacLeod's catalogue. The page renders nothing itself: its script fetches