diff --git a/py/services/download_manager.py b/py/services/download_manager.py index 570f19c9..29dcca9b 100644 --- a/py/services/download_manager.py +++ b/py/services/download_manager.py @@ -33,6 +33,13 @@ from ..utils.constants import ( VALID_OTHER_CIVITAI_TYPES, ) from ..utils.civitai_utils import normalize_civitai_download_url, rewrite_preview_url +from ..utils.paid_access import ( + is_early_access_deadline_active, + is_gate_active, + is_permanent_paid, + normalize_paid_access, + parse_civitai_timestamp, +) from ..utils.file_utils import calculate_sha256, calculate_autov3 from ..utils.preview_selection import resolve_mature_threshold, select_preview_media from ..utils.utils import calculate_filename_for_model, sanitize_folder_name @@ -1985,40 +1992,32 @@ class DownloadManager: os.makedirs(save_dir, exist_ok=True) # Check if this is a paid or early access model - paid_access = version_info.get("paidAccess") - if isinstance(paid_access, str): - # Some providers (e.g. CivArchive fallback) carry the DTO as JSON text - try: - parsed = json.loads(paid_access) - paid_access = parsed if isinstance(parsed, dict) else None - except (TypeError, ValueError): - paid_access = None - if not isinstance(paid_access, dict): - paid_access = None - # An empty DTO ({"permanent": false, "endsAt": null}) is not a gate - if paid_access and not paid_access.get("permanent") and not paid_access.get("endsAt"): - paid_access = None - if version_info.get("earlyAccessEndsAt") or paid_access: - permanent_paid = bool(paid_access.get("permanent")) if paid_access else False + # CivitAI reports a non-null paidAccess only for an ACTIVE gate, so + # {"permanent": false, "endsAt": null} (a timed gate whose end is not + # recorded yet) still counts as gated here. + paid_access = normalize_paid_access(version_info.get("paidAccess")) + legacy_ea_ends_at = version_info.get("earlyAccessEndsAt") + gate_active = is_gate_active(paid_access) or is_early_access_deadline_active( + legacy_ea_ends_at + ) + if gate_active: + permanent_paid = is_permanent_paid(paid_access) if permanent_paid: early_access_msg = ( "This model requires payment. Please ensure you have " "purchased access and are logged in to Civitai." ) else: - early_access_date = version_info.get("earlyAccessEndsAt") + early_access_date = legacy_ea_ends_at if not early_access_date and paid_access: early_access_date = paid_access.get("endsAt") if not early_access_date: early_access_date = "" # Convert to a readable date if possible try: - from datetime import datetime - - date_obj = datetime.fromisoformat( - early_access_date.replace("Z", "+00:00") - ) - formatted_date = date_obj.strftime("%Y-%m-%d") + formatted_date = parse_civitai_timestamp( + early_access_date + ).strftime("%Y-%m-%d") early_access_msg = ( f"This model requires payment (until {formatted_date}). " ) diff --git a/py/services/model_update_service.py b/py/services/model_update_service.py index 00147828..5040eea1 100644 --- a/py/services/model_update_service.py +++ b/py/services/model_update_service.py @@ -20,6 +20,10 @@ from .settings_manager import get_settings_manager from ..utils.cache_paths import CacheType, resolve_cache_path_with_migration from ..utils.constants import MODEL_WEIGHT_FILE_TYPES from ..utils.civitai_utils import rewrite_preview_url +from ..utils.paid_access import ( + is_early_access_deadline_active, + normalize_paid_access as _normalize_paid_access_payload, +) from ..utils.preview_selection import resolve_mature_threshold, select_preview_media logger = logging.getLogger(__name__) @@ -172,16 +176,10 @@ class ModelUpdateRecord: if version.is_paid and not version.early_access_ends_at: return False - # Phase 2: Precise check with exact end time + # Phase 2: Precise check with exact end time (None when the gate is timed + # but its window end has not been recorded yet -> treated as active below) if version.early_access_ends_at: - try: - ea_date = datetime.fromisoformat( - version.early_access_ends_at.replace("Z", "+00:00") - ) - return ea_date > datetime.now(timezone.utc) - except (ValueError, AttributeError): - # If date parsing fails, treat as active EA (conservative) - return True + return is_early_access_deadline_active(version.early_access_ends_at) # Phase 1: Basic EA flag from bulk API return version.is_early_access @@ -1762,6 +1760,9 @@ class ModelUpdateService: # CivitAI's paidAccess DTO ({"permanent": bool, "endsAt": ISO|null}) # gates versions behind a paid tier while availability stays "Public". + # A non-null DTO from the public API is always an ACTIVE gate: lapsed + # (tombstone) gates come back as null. That includes the timed gate whose + # end is not recorded yet, {"permanent": false, "endsAt": null}. paid_access = self._normalize_paid_access(entry.get("paidAccess")) paid_access_json = json.dumps(paid_access) if paid_access else None is_paid = bool(paid_access.get("permanent")) if paid_access else False @@ -1769,7 +1770,7 @@ class ModelUpdateService: early_access_ends_at = _normalize_string(paid_access.get("endsAt")) # Only timed gates are early access; permanent paid versions are not # (consumers filter them via is_paid), so the stored flag stays accurate. - if not is_early_access and paid_access and paid_access.get("endsAt"): + if not is_early_access and paid_access and not paid_access.get("permanent"): is_early_access = True return ModelVersionRecord( @@ -1797,24 +1798,12 @@ class ModelUpdateService: Accepts a dict, None, or a JSON string (as carried by the by-hash enrichment path) and returns ``{"permanent": bool, "endsAt": str|None}`` or None when the input carries no paid-access signal. + + Delegates to :mod:`py.utils.paid_access` so the update service and the + download gate cannot disagree about what counts as a gate. """ - if value is None: - return None - if isinstance(value, str): - try: - parsed = json.loads(value) - except (TypeError, ValueError): - return None - if not isinstance(parsed, dict): - return None - value = parsed - if not isinstance(value, Mapping): - return None - permanent = bool(value.get("permanent")) - ends_at = _normalize_string(value.get("endsAt")) - if not permanent and ends_at is None: - return None - return {"permanent": permanent, "endsAt": ends_at} + + return _normalize_paid_access_payload(value) @staticmethod def _extract_file_count(files) -> Optional[int]: diff --git a/py/utils/paid_access.py b/py/utils/paid_access.py new file mode 100644 index 00000000..d2d9f15a --- /dev/null +++ b/py/utils/paid_access.py @@ -0,0 +1,158 @@ +"""Shared interpretation of CivitAI's ``paidAccess`` DTO. + +One rule for the whole backend: the update service (badges, update filters, price +tracking) and the download gate both ask these helpers, so a version can never be +"paid" for one consumer and "free" for another. + +Why the rules look like this +---------------------------- +CivitAI's public v1 API only returns a non-null DTO for an *active* gate: a lapsed +gate stays in the database as a tombstone but is filtered out server-side +(``toPublicPaidAccessDto`` in civitai's ``server/services/paid-access.service.ts``). +A timed gate whose end time has not been recorded yet is reported as +``{"permanent": false, "endsAt": null}`` and is still enforced — such a version +reports ``canDownload: false`` on the model page — so it must NOT be discarded. +Dropping it (the previous behaviour) hid real gates, which is the class of bug +reported in issue #1060. +""" + +from __future__ import annotations + +import json +from datetime import datetime, timezone +from typing import Any, Mapping, Optional + +__all__ = [ + "normalize_paid_access", + "is_permanent_paid", + "is_gate_active", + "is_early_access_deadline_active", + "parse_civitai_timestamp", +] + +# A DTO that carries neither key is not a gate signal at all (defensive: CivitAI +# never emits a bare ``{}``, but other metadata sources might). +_PAID_ACCESS_KEYS = ("permanent", "endsAt") + + +def parse_civitai_timestamp(value: Any) -> Optional[datetime]: + """Parse a CivitAI ISO-8601 timestamp into an aware UTC datetime. + + Returns None for anything that is not a parsable string. Naive timestamps are + assumed to be UTC, matching CivitAI's serialization. + """ + + if not isinstance(value, str): + return None + text = value.strip() + if not text: + return None + if text.endswith("Z"): + text = f"{text[:-1]}+00:00" + try: + parsed = datetime.fromisoformat(text) + except ValueError: + return None + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed + + +def normalize_paid_access(value: Any) -> Optional[dict]: + """Normalize a ``paidAccess`` value into ``{"permanent": bool, "endsAt": str|None}``. + + Accepts a mapping, a JSON string (the by-hash enrichment path carries the DTO as + text), or None. Returns None when the value carries no gate signal. + + Note that ``{"permanent": False, "endsAt": None}`` is a *gate*: CivitAI reports + it for a timed gate whose window end is not set yet, and enforces it. + """ + + if value is None: + return None + + if isinstance(value, str): + try: + parsed = json.loads(value) + except (TypeError, ValueError): + return None + if not isinstance(parsed, dict): + return None + value = parsed + + if not isinstance(value, Mapping): + return None + + if not any(key in value for key in _PAID_ACCESS_KEYS): + return None + + ends_at = value.get("endsAt") + normalized_ends_at = ends_at.strip() if isinstance(ends_at, str) and ends_at.strip() else None + return { + "permanent": bool(value.get("permanent")), + "endsAt": normalized_ends_at, + } + + +def is_permanent_paid(paid_access: Optional[Mapping[str, Any]]) -> bool: + """True when the gate never expires (a permanent paid version).""" + + return bool(paid_access and paid_access.get("permanent")) + + +def is_gate_active( + paid_access: Optional[Mapping[str, Any]], + *, + now: Optional[datetime] = None, +) -> bool: + """True when a normalized ``paidAccess`` gate is currently in force. + + Active means permanent, or a timed gate whose end is either still in the future + or not recorded yet (CivitAI enforces the latter too). An unparsable end time is + treated as active rather than free: the download would fail anyway, so the + conservative reading matches what the user will experience. + """ + + if not paid_access: + return False + if paid_access.get("permanent"): + return True + + ends_at = paid_access.get("endsAt") + if not ends_at: + # Timed gate with no recorded end — CivitAI still gates the download. + return True + + parsed = parse_civitai_timestamp(ends_at) + if parsed is None: + return True + + reference = now or datetime.now(timezone.utc) + if reference.tzinfo is None: + reference = reference.replace(tzinfo=timezone.utc) + return parsed > reference + + +def is_early_access_deadline_active( + ends_at: Any, + *, + now: Optional[datetime] = None, +) -> bool: + """True when a legacy ``earlyAccessEndsAt`` deadline is still in the future. + + Kept separate from :func:`is_gate_active` because the legacy field is a bare + timestamp rather than a DTO. A present-but-unparsable value is treated as active, + matching the previous conservative behaviour. + """ + + if not ends_at: + return False + + parsed = parse_civitai_timestamp(ends_at) + if parsed is None: + return True + + reference = now or datetime.now(timezone.utc) + if reference.tzinfo is None: + reference = reference.replace(tzinfo=timezone.utc) + return parsed > reference diff --git a/tests/services/test_model_update_service.py b/tests/services/test_model_update_service.py index 61d48bb2..707d8525 100644 --- a/tests/services/test_model_update_service.py +++ b/tests/services/test_model_update_service.py @@ -826,6 +826,32 @@ def test_extract_single_version_paid_access_permanent(tmp_path): assert version.paid_access is not None +def test_extract_single_version_paid_access_pending_end(tmp_path): + """A timed gate whose window end is not recorded yet + ({"permanent": false, "endsAt": null}) is still an active gate, so it is + early access with no known end date rather than a free version.""" + db_path = tmp_path / "updates.sqlite" + service = ModelUpdateService(str(db_path)) + + entry = { + "id": 42, + "name": "v1 paid", + "availability": "Public", + "paidAccess": {"permanent": False, "endsAt": None}, + "files": [], + "images": [], + } + + version = service._extract_single_version(entry, index=0) + + assert version is not None + assert version.is_early_access is True + assert version.is_paid is False + assert version.early_access_ends_at is None + assert version.paid_access == '{"permanent": false, "endsAt": null}' + assert ModelUpdateRecord._is_early_access_active(version) is True + + def test_normalize_paid_access_accepts_json_string(): """The by-hash enrichment path may hand paidAccess to _normalize_paid_access as a JSON string; both the permanent and timed shapes must normalize.""" @@ -841,14 +867,21 @@ def test_normalize_paid_access_accepts_json_string(): ) assert timed == {"permanent": False, "endsAt": "2026-08-22T18:30:00.000Z"} - empty = ModelUpdateService._normalize_paid_access( + # A timed gate whose window end is not recorded yet. CivitAI only returns a + # non-null DTO for an ACTIVE gate (tombstones come back as null) and enforces + # this shape too - the model page reports canDownload: false for it - so it + # must be kept. Dropping it was the #1060 class of bug. + pending_end = ModelUpdateService._normalize_paid_access( '{"permanent": false, "endsAt": null}' ) - assert empty is None + assert pending_end == {"permanent": False, "endsAt": None} malformed = ModelUpdateService._normalize_paid_access("{not json") assert malformed is None + # No gate keys at all is not a gate signal. + assert ModelUpdateService._normalize_paid_access("{}") is None + def test_has_update_for_base_hide_paid(): """hide_paid also suppresses permanent paid versions in the same-base diff --git a/tests/utils/test_paid_access.py b/tests/utils/test_paid_access.py new file mode 100644 index 00000000..3d63da0e --- /dev/null +++ b/tests/utils/test_paid_access.py @@ -0,0 +1,112 @@ +"""Tests for the shared CivitAI paidAccess interpretation. + +These rules decide both the Paid/Early Access badges and whether a download is +warned about, so the interesting cases are the boundary shapes CivitAI actually +emits. +""" + +from datetime import datetime, timezone + +import pytest + +from py.utils.paid_access import ( + is_early_access_deadline_active, + is_gate_active, + is_permanent_paid, + normalize_paid_access, + parse_civitai_timestamp, +) + +NOW = datetime(2026, 9, 30, 12, 0, 0, tzinfo=timezone.utc) + + +@pytest.mark.parametrize( + "value,expected", + [ + (None, None), + ("", None), + ("{not json", None), + ("[1, 2]", None), + ({}, None), + ({"unknown": 1}, None), + # A non-null DTO from CivitAI's public API is always an active gate, so a + # timed gate with no recorded end must survive normalization. + ({"permanent": False, "endsAt": None}, {"permanent": False, "endsAt": None}), + ({"permanent": True, "endsAt": None}, {"permanent": True, "endsAt": None}), + ( + {"permanent": False, "endsAt": "2026-10-10T13:10:17.404Z"}, + {"permanent": False, "endsAt": "2026-10-10T13:10:17.404Z"}, + ), + ], +) +def test_normalize_paid_access_shapes(value, expected): + assert normalize_paid_access(value) == expected + + +def test_normalize_paid_access_accepts_json_text(): + """The by-hash enrichment path can carry the DTO as a JSON string.""" + + assert normalize_paid_access('{"permanent": true, "endsAt": null}') == { + "permanent": True, + "endsAt": None, + } + + +def test_normalize_paid_access_blank_ends_at_is_none(): + assert normalize_paid_access({"permanent": False, "endsAt": " "}) == { + "permanent": False, + "endsAt": None, + } + + +def test_is_permanent_paid(): + assert is_permanent_paid({"permanent": True, "endsAt": None}) is True + assert is_permanent_paid({"permanent": False, "endsAt": None}) is False + assert is_permanent_paid(None) is False + + +@pytest.mark.parametrize( + "paid_access,expected", + [ + (None, False), + ({"permanent": False, "endsAt": None}, True), + ({"permanent": True, "endsAt": None}, True), + # A permanent gate with a stale endsAt stays active. + ({"permanent": True, "endsAt": "2020-01-01T00:00:00.000Z"}, True), + ({"permanent": False, "endsAt": "2026-10-10T13:10:17.404Z"}, True), + ({"permanent": False, "endsAt": "2026-08-27T16:56:57.438Z"}, False), + ({"permanent": False, "endsAt": "not-a-date"}, True), + ], +) +def test_is_gate_active(paid_access, expected): + assert is_gate_active(paid_access, now=NOW) is expected + + +@pytest.mark.parametrize( + "ends_at,expected", + [ + (None, False), + ("", False), + ("2026-10-10T13:10:17.404Z", True), + ("2026-08-27T16:56:57.438Z", False), + ("garbage", True), + ], +) +def test_is_early_access_deadline_active(ends_at, expected): + assert is_early_access_deadline_active(ends_at, now=NOW) is expected + + +@pytest.mark.parametrize( + "value,expected", + [ + (None, None), + (123, None), + ("", None), + ("garbage", None), + ("2026-10-10T13:10:17.404Z", datetime(2026, 10, 10, 13, 10, 17, 404000, tzinfo=timezone.utc)), + # Naive timestamps are assumed UTC. + ("2026-10-10T13:10:17", datetime(2026, 10, 10, 13, 10, 17, tzinfo=timezone.utc)), + ], +) +def test_parse_civitai_timestamp(value, expected): + assert parse_civitai_timestamp(value) == expected