fc3g: ExtensionService.quick_add_source — derive platform+slug, find-or-create Artist+Source

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-23 23:37:39 -04:00
parent 379445c244
commit 46d199450d
1 changed files with 127 additions and 0 deletions
@@ -0,0 +1,127 @@
+"""FC-3g: backend support for the Firefox extension.
+
+`ExtensionService.quick_add_source(url)` derives platform + artist
+slug from a URL using regex patterns mirrored from
+extension/lib/platforms.js, then find-or-creates Artist + Source rows
+and returns a JSON-shaped dict for the API layer.
+"""
+
+from __future__ import annotations
+
+import re
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from ..models import Artist, Source
+from ..utils.slug import slugify
+
+
+class UnknownPlatformError(Exception):
+    """URL didn't match any platform pattern."""
+
+
+class InvalidUrlError(Exception):
+    """URL was empty or missing a scheme."""
+
+
+# Mirrored byte-for-byte from extension/lib/platforms.js
+# PLATFORM_ARTIST_PATTERNS. Keep these two copies in sync by hand —
+# reviewers catch drift.
+_PLATFORM_PATTERNS: list[tuple[str, re.Pattern[str]]] = [
+    ("patreon", re.compile(
+        r"^https?://(?:www\.)?patreon\.com/"
+        r"(?!home$|search\b|messages\b|notifications\b|library\b|settings\b|posts\b|c/)"
+        r"(?P<slug>[^/?#]+)/?$",
+        re.IGNORECASE,
+    )),
+    ("subscribestar", re.compile(
+        r"^https?://(?:www\.)?subscribestar\.(?:com|adult)/"
+        r"(?!feed$|messages$|library$)"
+        r"(?P<slug>[^/?#]+)/?$",
+        re.IGNORECASE,
+    )),
+    ("hentaifoundry", re.compile(
+        r"^https?://(?:www\.)?hentai-foundry\.com/user/(?P<slug>[^/?#]+)",
+        re.IGNORECASE,
+    )),
+    ("deviantart", re.compile(
+        r"^https?://(?:www\.)?deviantart\.com/"
+        r"(?!home$|watch\b|tag\b|browse\b)"
+        r"(?P<slug>[^/?#]+)/?$",
+        re.IGNORECASE,
+    )),
+    ("pixiv", re.compile(
+        r"^https?://(?:www\.)?pixiv\.net/(?:en/)?users/(?P<slug>\d+)",
+        re.IGNORECASE,
+    )),
+]
+
+
+class ExtensionService:
+    def __init__(self, session: AsyncSession) -> None:
+        self.session = session
+
+    async def quick_add_source(self, url: str) -> dict:
+        platform, raw_slug = self._derive(url)
+        artist, created_artist = await self._find_or_create_artist(raw_slug)
+        source, created_source = await self._find_or_create_source(
+            artist_id=artist.id, platform=platform, url=url,
+        )
+        return {
+            "source": {
+                "id": source.id,
+                "artist_id": source.artist_id,
+                "platform": source.platform,
+                "url": source.url,
+                "enabled": source.enabled,
+            },
+            "artist": {
+                "id": artist.id,
+                "name": artist.name,
+                "slug": artist.slug,
+            },
+            "created_source": created_source,
+            "created_artist": created_artist,
+        }
+
+    def _derive(self, url: str) -> tuple[str, str]:
+        if not isinstance(url, str) or not url.strip():
+            raise InvalidUrlError("url is empty")
+        if not url.startswith(("http://", "https://")):
+            raise InvalidUrlError(f"url must include http:// or https:// scheme: {url!r}")
+        for platform, pattern in _PLATFORM_PATTERNS:
+            m = pattern.match(url)
+            if m:
+                return platform, m.group("slug")
+        raise UnknownPlatformError(f"no platform pattern matched {url!r}")
+
+    async def _find_or_create_artist(self, raw_name: str) -> tuple[Artist, bool]:
+        slug = slugify(raw_name)
+        existing = (await self.session.execute(
+            select(Artist).where(Artist.slug == slug)
+        )).scalar_one_or_none()
+        if existing is not None:
+            return existing, False
+        artist = Artist(name=raw_name, slug=slug, is_subscription=True)
+        self.session.add(artist)
+        await self.session.flush()
+        return artist, True
+
+    async def _find_or_create_source(
+        self, *, artist_id: int, platform: str, url: str,
+    ) -> tuple[Source, bool]:
+        existing = (await self.session.execute(
+            select(Source).where(
+                Source.artist_id == artist_id,
+                Source.platform == platform,
+                Source.url == url,
+            )
+        )).scalar_one_or_none()
+        if existing is not None:
+            return existing, False
+        src = Source(artist_id=artist_id, platform=platform, url=url, enabled=True)
+        self.session.add(src)
+        await self.session.flush()
+        await self.session.commit()
+        return src, True