934 lines
33 KiB
Python
934 lines
33 KiB
Python
"""Scryfall API helpers for the MTG reference connector."""
|
|
|
|
import base64
|
|
import json
|
|
import re
|
|
from urllib import error, parse, request
|
|
|
|
from odoo import _, models
|
|
from odoo.exceptions import UserError
|
|
|
|
from .constants import (
|
|
DEFAULT_SCRYFALL_API_BASE_URL,
|
|
DEFAULT_SCRYFALL_IMPORT_LANGUAGES,
|
|
DEFAULT_SCRYFALL_TIMEOUT_SECONDS,
|
|
DEFAULT_SCRYFALL_USER_AGENT,
|
|
)
|
|
SCRYFALL_ODOO_LANGUAGE_ALIASES = {
|
|
"de": ("de_DE",),
|
|
"en": ("en_US",),
|
|
"es": ("es_ES", "es_419"),
|
|
"fr": ("fr_FR",),
|
|
"it": ("it_IT",),
|
|
"ja": ("ja_JP",),
|
|
"ko": ("ko_KR",),
|
|
"pt": ("pt_PT", "pt_BR"),
|
|
"ru": ("ru_RU",),
|
|
"zhs": ("zh_CN",),
|
|
"zht": ("zh_TW",),
|
|
}
|
|
SCRYFALL_LANGUAGE_CODES = {
|
|
"de",
|
|
"en",
|
|
"es",
|
|
"fr",
|
|
"it",
|
|
"ja",
|
|
"ko",
|
|
"ph",
|
|
"pt",
|
|
"ru",
|
|
"zhs",
|
|
"zht",
|
|
}
|
|
SCRYFALL_LOOKUP_MODES = {"exact", "fuzzy", "url"}
|
|
_SCRYFALL_CARD_UUID_PATTERN = re.compile(
|
|
r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
|
|
def normalize_language_codes(raw_value, default=None):
|
|
"""Normalize Scryfall language codes into a stable tuple.
|
|
|
|
Args:
|
|
raw_value: Iterable or comma-separated language codes.
|
|
default: Fallback language codes when the input is empty.
|
|
|
|
Returns:
|
|
tuple[str, ...]: Deduplicated lower-case Scryfall language codes.
|
|
"""
|
|
if isinstance(raw_value, str):
|
|
chunks = re.split(r"[\s,;]+", raw_value)
|
|
else:
|
|
chunks = list(raw_value or ())
|
|
|
|
values = []
|
|
for chunk in chunks:
|
|
code = str(chunk or "").strip().lower()
|
|
if code and code not in values:
|
|
values.append(code)
|
|
|
|
fallback = list(default or DEFAULT_SCRYFALL_IMPORT_LANGUAGES)
|
|
return tuple(values or fallback)
|
|
|
|
|
|
class MvdTcgMtgScryfallApi(models.AbstractModel):
|
|
"""Provide Scryfall lookups for the MTG connector."""
|
|
|
|
_name = "mvd.tcg.mtg.scryfall.api"
|
|
_description = "MTG Scryfall API"
|
|
_SCRYFALL_ALLOWED_API_HOSTS = frozenset({"api.scryfall.com"})
|
|
_SCRYFALL_ALLOWED_IMAGE_HOSTS = frozenset(
|
|
{"api.scryfall.com", "cards.scryfall.io"}
|
|
)
|
|
_SCRYFALL_ALLOWED_IMAGE_HOST_SUFFIXES = (".scryfall.io",)
|
|
|
|
def _mtg_scryfall_check_manager_access(self):
|
|
"""Require manager-level access for Scryfall connector actions."""
|
|
if self.env.is_superuser() or any(
|
|
self.env.user.has_group(xmlid)
|
|
for xmlid in (
|
|
"mvd_tcg_base.mvd_tcg_base_group_manager",
|
|
"base.group_system",
|
|
)
|
|
):
|
|
return
|
|
raise UserError(_("Only TCG managers can run Scryfall connector actions."))
|
|
|
|
@classmethod
|
|
def _validate_allowed_url(
|
|
cls,
|
|
url,
|
|
*,
|
|
label,
|
|
allowed_hosts,
|
|
allowed_host_suffixes=(),
|
|
):
|
|
"""Validate one configured or upstream URL against a strict allowlist.
|
|
|
|
Args:
|
|
url: Absolute URL that should be validated.
|
|
label: Human label used in error messages.
|
|
allowed_hosts: Exact hostnames allowed for this URL class.
|
|
allowed_host_suffixes: Optional allowed hostname suffixes.
|
|
|
|
Returns:
|
|
str: Normalized absolute URL.
|
|
|
|
Raises:
|
|
UserError: If the URL uses a disallowed scheme or host.
|
|
"""
|
|
normalized_url = (url or "").strip()
|
|
parsed_url = parse.urlparse(normalized_url)
|
|
hostname = (parsed_url.hostname or "").lower()
|
|
if parsed_url.scheme != "https" or not hostname:
|
|
raise UserError(
|
|
_("%(label)s must use HTTPS and a trusted host.")
|
|
% {"label": label}
|
|
)
|
|
if hostname in allowed_hosts:
|
|
return normalized_url
|
|
if any(hostname.endswith(suffix) for suffix in allowed_host_suffixes):
|
|
return normalized_url
|
|
raise UserError(
|
|
_("%(label)s must use a trusted Scryfall host.")
|
|
% {"label": label}
|
|
)
|
|
|
|
@classmethod
|
|
def _validate_api_url(cls, url):
|
|
"""Validate one Scryfall API URL.
|
|
|
|
Args:
|
|
url: Absolute API URL.
|
|
|
|
Returns:
|
|
str: Validated absolute API URL.
|
|
"""
|
|
return cls._validate_allowed_url(
|
|
url,
|
|
label=_("Scryfall API URL"),
|
|
allowed_hosts=cls._SCRYFALL_ALLOWED_API_HOSTS,
|
|
)
|
|
|
|
@classmethod
|
|
def _validate_image_url(cls, url):
|
|
"""Validate one Scryfall-hosted image URL.
|
|
|
|
Args:
|
|
url: Absolute image URL.
|
|
|
|
Returns:
|
|
str: Validated absolute image URL.
|
|
"""
|
|
return cls._validate_allowed_url(
|
|
url,
|
|
label=_("Scryfall image URL"),
|
|
allowed_hosts=cls._SCRYFALL_ALLOWED_IMAGE_HOSTS,
|
|
allowed_host_suffixes=cls._SCRYFALL_ALLOWED_IMAGE_HOST_SUFFIXES,
|
|
)
|
|
|
|
def _get_config_parameter(self, name, default=False):
|
|
"""Return one connector configuration parameter.
|
|
|
|
Args:
|
|
name: Technical config parameter key.
|
|
default: Fallback value when the parameter is unset.
|
|
|
|
Returns:
|
|
str | bool: Stored config value or the provided fallback.
|
|
"""
|
|
return self.env["ir.config_parameter"].sudo().get_param(name, default)
|
|
|
|
def get_api_base_url(self):
|
|
"""Return the configured Scryfall API base URL.
|
|
|
|
Returns:
|
|
str: Absolute base URL for API requests.
|
|
"""
|
|
return self._validate_api_url(
|
|
(
|
|
self._get_config_parameter(
|
|
"mvd_tcg_mtg_scryfall.api_base_url",
|
|
DEFAULT_SCRYFALL_API_BASE_URL,
|
|
)
|
|
or DEFAULT_SCRYFALL_API_BASE_URL
|
|
).strip()
|
|
)
|
|
|
|
def get_timeout_seconds(self):
|
|
"""Return the configured Scryfall request timeout.
|
|
|
|
Returns:
|
|
int: Timeout in seconds.
|
|
"""
|
|
raw_value = self._get_config_parameter(
|
|
"mvd_tcg_mtg_scryfall.timeout_seconds",
|
|
DEFAULT_SCRYFALL_TIMEOUT_SECONDS,
|
|
)
|
|
try:
|
|
return max(1, int(raw_value))
|
|
except (TypeError, ValueError):
|
|
return DEFAULT_SCRYFALL_TIMEOUT_SECONDS
|
|
|
|
def get_user_agent(self):
|
|
"""Return the configured Scryfall HTTP user agent.
|
|
|
|
Returns:
|
|
str: User agent string for API and image requests.
|
|
"""
|
|
return (
|
|
self._get_config_parameter(
|
|
"mvd_tcg_mtg_scryfall.user_agent",
|
|
DEFAULT_SCRYFALL_USER_AGENT,
|
|
)
|
|
or DEFAULT_SCRYFALL_USER_AGENT
|
|
).strip()
|
|
|
|
def get_default_max_cards_per_set(self):
|
|
"""Return the default per-set card limit for controlled imports.
|
|
|
|
Returns:
|
|
int: Configured limit or ``0`` for no limit.
|
|
"""
|
|
raw_value = self._get_config_parameter(
|
|
"mvd_tcg_mtg_scryfall.import_max_cards_per_set",
|
|
0,
|
|
)
|
|
try:
|
|
return max(0, int(raw_value))
|
|
except (TypeError, ValueError):
|
|
return 0
|
|
|
|
def get_default_include_tokens(self):
|
|
"""Return whether token cards should be included by default.
|
|
|
|
Returns:
|
|
bool: ``True`` when controlled set imports should include tokens.
|
|
"""
|
|
raw_value = self._get_config_parameter(
|
|
"mvd_tcg_mtg_scryfall.import_include_tokens",
|
|
False,
|
|
)
|
|
return str(raw_value).strip().lower() in {"1", "true", "yes", "on"}
|
|
|
|
def _build_api_url(self, path, params=None):
|
|
"""Build one absolute Scryfall API URL.
|
|
|
|
Args:
|
|
path: Relative Scryfall API path.
|
|
params: Optional query parameters.
|
|
|
|
Returns:
|
|
str: Absolute API URL.
|
|
"""
|
|
url = f"{self.get_api_base_url().rstrip('/')}/{path.lstrip('/')}"
|
|
if params:
|
|
url = f"{url}?{parse.urlencode(params, doseq=True)}"
|
|
return url
|
|
|
|
def _request_json(self, url):
|
|
"""Perform one JSON request against the Scryfall API.
|
|
|
|
Args:
|
|
url: Absolute API URL.
|
|
|
|
Returns:
|
|
dict: Parsed JSON payload.
|
|
|
|
Raises:
|
|
UserError: If Scryfall rejects the request or returns invalid JSON.
|
|
"""
|
|
validated_url = self._validate_api_url(url)
|
|
http_request = request.Request(
|
|
validated_url,
|
|
headers={
|
|
"User-Agent": self.get_user_agent(),
|
|
"Accept": "application/json;q=0.9,*/*;q=0.8",
|
|
},
|
|
)
|
|
try:
|
|
with request.urlopen(
|
|
http_request,
|
|
timeout=self.get_timeout_seconds(),
|
|
) as response:
|
|
return json.loads(response.read().decode("utf-8"))
|
|
except error.HTTPError as exc:
|
|
details = exc.read().decode("utf-8", errors="replace")
|
|
try:
|
|
payload = json.loads(details)
|
|
except json.JSONDecodeError:
|
|
payload = {}
|
|
message = payload.get("details") or payload.get("error") or details or str(exc)
|
|
raise UserError(_("Scryfall lookup failed: %s") % message) from exc
|
|
except (error.URLError, OSError, json.JSONDecodeError) as exc:
|
|
raise UserError(_("Scryfall lookup failed: %s") % exc) from exc
|
|
|
|
def _request_json_path(self, path, params=None):
|
|
"""Perform one JSON request against a relative Scryfall API path.
|
|
|
|
Args:
|
|
path: Relative Scryfall API path.
|
|
params: Optional query parameters.
|
|
|
|
Returns:
|
|
dict: Parsed JSON payload.
|
|
"""
|
|
return self._request_json(self._build_api_url(path, params=params))
|
|
|
|
def _normalize_lookup_mode(self, lookup_mode):
|
|
"""Return a validated lookup mode for card imports.
|
|
|
|
Args:
|
|
lookup_mode: Raw lookup mode from a wizard or batch entry.
|
|
|
|
Returns:
|
|
str: One of ``exact``, ``fuzzy`` or ``url``.
|
|
|
|
Raises:
|
|
UserError: If the lookup mode is unknown.
|
|
"""
|
|
normalized_mode = (lookup_mode or "").strip().lower()
|
|
if normalized_mode not in SCRYFALL_LOOKUP_MODES:
|
|
raise UserError(_("Unsupported Scryfall lookup mode: %s") % (lookup_mode or ""))
|
|
return normalized_mode
|
|
|
|
def map_translation_language_codes(self, scryfall_language_code):
|
|
"""Map one Scryfall language code to active Odoo languages.
|
|
|
|
Args:
|
|
scryfall_language_code: Raw Scryfall language code such as ``de``.
|
|
|
|
Returns:
|
|
list[str]: Active Odoo language codes for the given Scryfall code.
|
|
"""
|
|
normalized_scryfall_code = (scryfall_language_code or "").strip().lower()
|
|
if not normalized_scryfall_code:
|
|
return []
|
|
|
|
installed_codes = {
|
|
code.lower().replace("-", "_"): code
|
|
for code in self.env["res.lang"].search([("active", "=", True)]).mapped("code")
|
|
}
|
|
resolved_codes = []
|
|
for alias in SCRYFALL_ODOO_LANGUAGE_ALIASES.get(normalized_scryfall_code, ()):
|
|
normalized_alias = alias.lower().replace("-", "_")
|
|
installed_code = installed_codes.get(normalized_alias)
|
|
if installed_code and installed_code not in resolved_codes:
|
|
resolved_codes.append(installed_code)
|
|
for normalized_code, installed_code in installed_codes.items():
|
|
if normalized_code.split("_", 1)[0] == normalized_scryfall_code:
|
|
if installed_code not in resolved_codes:
|
|
resolved_codes.append(installed_code)
|
|
return resolved_codes
|
|
|
|
def map_scryfall_language_code(self, odoo_language_code):
|
|
"""Map one Odoo language code to a Scryfall language code.
|
|
|
|
Args:
|
|
odoo_language_code: Odoo language code such as ``de_DE``.
|
|
|
|
Returns:
|
|
str | bool: Matching Scryfall language code or ``False``.
|
|
"""
|
|
normalized_odoo_code = (
|
|
(odoo_language_code or "").strip().lower().replace("-", "_")
|
|
)
|
|
if not normalized_odoo_code:
|
|
return False
|
|
|
|
for scryfall_code, aliases in SCRYFALL_ODOO_LANGUAGE_ALIASES.items():
|
|
normalized_aliases = {
|
|
alias.lower().replace("-", "_") for alias in aliases
|
|
}
|
|
if normalized_odoo_code in normalized_aliases:
|
|
return scryfall_code
|
|
|
|
base_language = normalized_odoo_code.split("_", 1)[0]
|
|
return base_language if base_language in SCRYFALL_LANGUAGE_CODES else False
|
|
|
|
def get_import_language_codes(self):
|
|
"""Return active Scryfall language codes that should be imported.
|
|
|
|
Returns:
|
|
tuple[str, ...]: Normalized Scryfall language codes.
|
|
"""
|
|
configured_languages = self._get_config_parameter(
|
|
"mvd_tcg_mtg_scryfall.import_language_codes",
|
|
False,
|
|
)
|
|
return normalize_language_codes(
|
|
configured_languages or DEFAULT_SCRYFALL_IMPORT_LANGUAGES
|
|
)
|
|
|
|
def _parse_card_url_or_id(self, identifier):
|
|
"""Parse one Scryfall URL or UUID into a reusable lookup descriptor.
|
|
|
|
Args:
|
|
identifier: Human card URL, API URL or raw Scryfall UUID.
|
|
|
|
Returns:
|
|
dict[str, object]: Parsed descriptor containing request path.
|
|
|
|
Raises:
|
|
UserError: If the identifier is not a supported Scryfall reference.
|
|
"""
|
|
normalized_identifier = (identifier or "").strip()
|
|
if not normalized_identifier:
|
|
raise UserError(_("Enter a Scryfall card URL or card id."))
|
|
|
|
if _SCRYFALL_CARD_UUID_PATTERN.match(normalized_identifier):
|
|
return {
|
|
"lookup_mode": "url",
|
|
"query": normalized_identifier,
|
|
"path": f"/cards/{normalized_identifier}",
|
|
"params": None,
|
|
}
|
|
|
|
parsed_url = parse.urlparse(normalized_identifier)
|
|
hostname = (parsed_url.hostname or "").lower()
|
|
path_segments = [segment for segment in parsed_url.path.split("/") if segment]
|
|
if hostname not in {"scryfall.com", "www.scryfall.com", "api.scryfall.com"}:
|
|
raise UserError(_("Enter a Scryfall card URL or card id."))
|
|
if not path_segments:
|
|
raise UserError(_("Enter a Scryfall card URL or card id."))
|
|
|
|
if path_segments[0] == "cards" and len(path_segments) >= 2:
|
|
return {
|
|
"lookup_mode": "url",
|
|
"query": normalized_identifier,
|
|
"path": f"/cards/{parse.quote(path_segments[1])}",
|
|
"params": None,
|
|
}
|
|
|
|
if path_segments[0] == "card" and len(path_segments) >= 3:
|
|
set_code = parse.quote(path_segments[1])
|
|
collector_number = parse.quote(path_segments[2], safe="")
|
|
path = f"/cards/{set_code}/{collector_number}"
|
|
language_code = (path_segments[3] or "").lower() if len(path_segments) >= 4 else ""
|
|
if language_code in SCRYFALL_LANGUAGE_CODES:
|
|
path = f"{path}/{parse.quote(path_segments[3])}"
|
|
return {
|
|
"lookup_mode": "url",
|
|
"query": normalized_identifier,
|
|
"path": path,
|
|
"params": None,
|
|
}
|
|
|
|
raise UserError(_("Enter a Scryfall card URL or card id."))
|
|
|
|
def parse_lookup_descriptor(self, query, lookup_mode):
|
|
"""Parse one card lookup into a reusable request descriptor.
|
|
|
|
Args:
|
|
query: Raw card lookup query from the user.
|
|
lookup_mode: Lookup mode such as ``exact``, ``fuzzy`` or ``url``.
|
|
|
|
Returns:
|
|
dict[str, object]: Parsed descriptor with request path and parameters.
|
|
|
|
Raises:
|
|
UserError: If the query is empty or malformed for the chosen mode.
|
|
"""
|
|
normalized_query = (query or "").strip()
|
|
if not normalized_query:
|
|
raise UserError(_("Enter a card name or Scryfall card URL first."))
|
|
|
|
normalized_mode = self._normalize_lookup_mode(lookup_mode)
|
|
if normalized_mode == "url":
|
|
return self._parse_card_url_or_id(normalized_query)
|
|
|
|
return {
|
|
"lookup_mode": normalized_mode,
|
|
"query": normalized_query,
|
|
"path": "/cards/named",
|
|
"params": {normalized_mode: normalized_query},
|
|
}
|
|
|
|
def load_image_base64(self, image_url):
|
|
"""Download one card image and return it in Odoo's base64 format.
|
|
|
|
Args:
|
|
image_url: Absolute image URL from Scryfall.
|
|
|
|
Returns:
|
|
str | bool: Base64-encoded image data or ``False`` on download issues.
|
|
"""
|
|
if not image_url:
|
|
return False
|
|
validated_image_url = self._validate_image_url(image_url)
|
|
|
|
http_request = request.Request(
|
|
validated_image_url,
|
|
headers={
|
|
"User-Agent": self.get_user_agent(),
|
|
"Accept": "image/avif,image/webp,image/*,*/*;q=0.8",
|
|
},
|
|
)
|
|
try:
|
|
with request.urlopen(
|
|
http_request,
|
|
timeout=self.get_timeout_seconds(),
|
|
) as response:
|
|
return base64.b64encode(response.read()).decode("ascii")
|
|
except (error.URLError, OSError):
|
|
return False
|
|
|
|
def lookup_card_payload(self, query, lookup_mode):
|
|
"""Resolve one MTG card payload from a parsed lookup descriptor.
|
|
|
|
Args:
|
|
query: Raw lookup query.
|
|
lookup_mode: Lookup mode such as ``exact``, ``fuzzy`` or ``url``.
|
|
|
|
Returns:
|
|
dict: Raw Scryfall card payload.
|
|
"""
|
|
self._mtg_scryfall_check_manager_access()
|
|
descriptor = self.parse_lookup_descriptor(query, lookup_mode)
|
|
return self._request_json_path(
|
|
descriptor["path"],
|
|
params=descriptor.get("params"),
|
|
)
|
|
|
|
def lookup_card_payloads(self, query, lookup_mode, languages=None):
|
|
"""Resolve one card lookup and expand it to localized print payloads.
|
|
|
|
Args:
|
|
query: Raw lookup query.
|
|
lookup_mode: Lookup mode such as ``exact``, ``fuzzy`` or ``url``.
|
|
languages: Optional import language scope.
|
|
|
|
Returns:
|
|
list[dict]: Deduplicated localized payloads for one print group.
|
|
"""
|
|
payload = self.lookup_card_payload(query, lookup_mode)
|
|
return self.get_localized_print_payloads(payload, languages=languages)
|
|
|
|
def lookup_exact(self, card_name):
|
|
"""Look up one card by exact name on Scryfall."""
|
|
return self.lookup_card_payload(card_name, "exact")
|
|
|
|
def lookup_fuzzy(self, card_name):
|
|
"""Look up one card by fuzzy name on Scryfall."""
|
|
return self.lookup_card_payload(card_name, "fuzzy")
|
|
|
|
def lookup_by_url_or_id(self, identifier):
|
|
"""Look up one card from a Scryfall URL or card UUID."""
|
|
return self.lookup_card_payload(identifier, "url")
|
|
|
|
def iter_search_card_payloads(self, query, *, unique="cards", limit=None, include_multilingual=False, order=None, direction=None):
|
|
"""Yield card payloads from the paginated Scryfall search API.
|
|
|
|
Args:
|
|
query: Raw Scryfall search query.
|
|
unique: Requested Scryfall uniqueness mode.
|
|
limit: Optional maximum number of returned payloads.
|
|
include_multilingual: Whether Scryfall should include multilingual prints.
|
|
order: Optional Scryfall ordering key.
|
|
direction: Optional Scryfall sort direction.
|
|
|
|
Yields:
|
|
dict: Card payloads from the result stream.
|
|
"""
|
|
self._mtg_scryfall_check_manager_access()
|
|
query_params = {
|
|
"q": (query or "").strip(),
|
|
"unique": unique,
|
|
}
|
|
if include_multilingual:
|
|
query_params["include_multilingual"] = "true"
|
|
if order:
|
|
query_params["order"] = order
|
|
if direction:
|
|
query_params["dir"] = direction
|
|
next_url = self._build_api_url("/cards/search", query_params)
|
|
remaining_limit = int(limit) if limit and int(limit) > 0 else None
|
|
|
|
while next_url:
|
|
page_payload = self._request_json(next_url)
|
|
page_card_payloads = [
|
|
card_payload
|
|
for card_payload in page_payload.get("data", [])
|
|
if card_payload.get("object") == "card" and card_payload.get("id")
|
|
]
|
|
if remaining_limit is not None:
|
|
page_card_payloads = page_card_payloads[:remaining_limit]
|
|
|
|
for card_payload in page_card_payloads:
|
|
yield card_payload
|
|
|
|
if remaining_limit is not None:
|
|
remaining_limit -= len(page_card_payloads)
|
|
if remaining_limit <= 0:
|
|
break
|
|
if not page_payload.get("has_more"):
|
|
break
|
|
next_url = page_payload.get("next_page") or ""
|
|
|
|
def search_cards(self, query, *, unique="cards", limit=None, include_multilingual=False):
|
|
"""Search card payloads on Scryfall with optional pagination.
|
|
|
|
Args:
|
|
query: Raw Scryfall search query.
|
|
unique: Requested Scryfall uniqueness mode.
|
|
limit: Optional maximum number of returned payloads.
|
|
include_multilingual: Whether Scryfall should include multilingual prints.
|
|
|
|
Returns:
|
|
list[dict]: Card payloads returned by the Scryfall search API.
|
|
"""
|
|
return list(
|
|
self.iter_search_card_payloads(
|
|
query,
|
|
unique=unique,
|
|
limit=limit,
|
|
include_multilingual=include_multilingual,
|
|
)
|
|
)
|
|
|
|
def fetch_search_results_by_url(self, search_url, *, limit=None):
|
|
"""Fetch paginated Scryfall search results from an absolute URL.
|
|
|
|
Args:
|
|
search_url: Absolute search URL such as ``prints_search_uri``.
|
|
limit: Optional maximum number of returned payloads.
|
|
|
|
Returns:
|
|
list[dict]: Card payloads from the paginated result stream.
|
|
"""
|
|
self._mtg_scryfall_check_manager_access()
|
|
if search_url:
|
|
parsed_url = parse.urlsplit(search_url)
|
|
query_params = parse.parse_qsl(parsed_url.query, keep_blank_values=True)
|
|
query_params = [
|
|
(key, value)
|
|
for key, value in query_params
|
|
if key != "include_multilingual"
|
|
]
|
|
query_params.append(("include_multilingual", "true"))
|
|
search_url = parse.urlunsplit(
|
|
(
|
|
parsed_url.scheme,
|
|
parsed_url.netloc,
|
|
parsed_url.path,
|
|
parse.urlencode(query_params),
|
|
parsed_url.fragment,
|
|
)
|
|
)
|
|
|
|
next_url = (search_url or "").strip()
|
|
remaining_limit = int(limit) if limit and int(limit) > 0 else None
|
|
payloads = []
|
|
while next_url:
|
|
page_payload = self._request_json(next_url)
|
|
page_card_payloads = [
|
|
card_payload
|
|
for card_payload in page_payload.get("data", [])
|
|
if card_payload.get("object") == "card" and card_payload.get("id")
|
|
]
|
|
if remaining_limit is not None:
|
|
page_card_payloads = page_card_payloads[:remaining_limit]
|
|
payloads.extend(page_card_payloads)
|
|
if remaining_limit is not None:
|
|
remaining_limit -= len(page_card_payloads)
|
|
if remaining_limit <= 0:
|
|
break
|
|
if not page_payload.get("has_more"):
|
|
break
|
|
next_url = page_payload.get("next_page") or ""
|
|
return payloads
|
|
|
|
def get_localized_print_payloads(self, payload, *, languages=None):
|
|
"""Return localized payloads for one MTG print group.
|
|
|
|
Args:
|
|
payload: Seed Scryfall card payload.
|
|
languages: Optional iterable of Scryfall language codes.
|
|
|
|
Returns:
|
|
list[dict]: Deduplicated payloads for the same set and collector number.
|
|
"""
|
|
if payload.get("object") != "card" or not payload.get("id"):
|
|
return []
|
|
|
|
import_language_codes = normalize_language_codes(
|
|
languages,
|
|
default=self.get_import_language_codes(),
|
|
)
|
|
wanted_languages = set(import_language_codes)
|
|
same_print_payloads = [payload]
|
|
search_url = payload.get("prints_search_uri")
|
|
if search_url:
|
|
same_print_payloads.extend(
|
|
candidate_payload
|
|
for candidate_payload in self.fetch_search_results_by_url(search_url)
|
|
if (
|
|
candidate_payload.get("set") == payload.get("set")
|
|
and candidate_payload.get("collector_number")
|
|
== payload.get("collector_number")
|
|
)
|
|
)
|
|
|
|
deduplicated_payloads = {}
|
|
for candidate_payload in same_print_payloads:
|
|
candidate_id = candidate_payload.get("id")
|
|
candidate_language = (candidate_payload.get("lang") or "").strip().lower()
|
|
if not candidate_id:
|
|
continue
|
|
if wanted_languages and candidate_language and candidate_language not in wanted_languages:
|
|
continue
|
|
deduplicated_payloads[candidate_id] = candidate_payload
|
|
|
|
language_sort_order = {
|
|
code: index
|
|
for index, code in enumerate(import_language_codes)
|
|
}
|
|
return sorted(
|
|
deduplicated_payloads.values(),
|
|
key=lambda candidate_payload: (
|
|
language_sort_order.get(
|
|
(candidate_payload.get("lang") or "").strip().lower(),
|
|
999,
|
|
),
|
|
candidate_payload.get("lang") or "",
|
|
candidate_payload.get("id") or "",
|
|
),
|
|
)
|
|
|
|
def expand_payloads_with_localized_prints(self, payloads, *, languages=None):
|
|
"""Expand seed payloads to localized print payloads.
|
|
|
|
Args:
|
|
payloads: Seed Scryfall card payloads.
|
|
languages: Optional iterable of Scryfall language codes.
|
|
|
|
Returns:
|
|
list[dict]: Deduplicated localized payloads.
|
|
"""
|
|
expanded_payloads = {}
|
|
for payload in payloads:
|
|
for localized_payload in self.get_localized_print_payloads(
|
|
payload,
|
|
languages=languages,
|
|
):
|
|
expanded_payloads[localized_payload["id"]] = localized_payload
|
|
return list(expanded_payloads.values())
|
|
|
|
def resolve_lookup_payload(self, query, lookup_mode):
|
|
"""Resolve one Scryfall lookup input to a seed payload."""
|
|
return self.lookup_card_payload(query, lookup_mode)
|
|
|
|
def resolve_lookup_payloads(self, query, lookup_mode, languages=None):
|
|
"""Resolve one lookup query and expand it to localized payloads."""
|
|
return self.lookup_card_payloads(query, lookup_mode, languages=languages)
|
|
|
|
def parse_batch_queries(self, raw_batch_input, default_lookup_mode="url"):
|
|
"""Parse a multiline batch input into normalized lookup entries.
|
|
|
|
Each non-empty line is treated as one lookup item. Lines may optionally
|
|
start with ``url:``, ``exact:`` or ``fuzzy:`` to override the default
|
|
lookup mode for that line.
|
|
|
|
Args:
|
|
raw_batch_input: Multiline raw text from the import wizard.
|
|
default_lookup_mode: Fallback lookup mode for plain lines.
|
|
|
|
Returns:
|
|
list[dict[str, str]]: Cleaned batch entries with ``query`` and
|
|
``lookup_mode`` keys.
|
|
"""
|
|
normalized_default_mode = self._normalize_lookup_mode(default_lookup_mode)
|
|
parsed_queries = []
|
|
for raw_line in (raw_batch_input or "").splitlines():
|
|
normalized_line = raw_line.strip()
|
|
if not normalized_line or normalized_line.startswith("#"):
|
|
continue
|
|
|
|
lookup_mode = normalized_default_mode
|
|
query = normalized_line
|
|
if ":" in normalized_line:
|
|
prefix, suffix = normalized_line.split(":", 1)
|
|
normalized_prefix = prefix.strip().lower()
|
|
if normalized_prefix in SCRYFALL_LOOKUP_MODES and suffix.strip():
|
|
lookup_mode = normalized_prefix
|
|
query = suffix.strip()
|
|
|
|
parsed_queries.append(
|
|
{
|
|
"lookup_mode": lookup_mode,
|
|
"query": query,
|
|
}
|
|
)
|
|
return parsed_queries
|
|
|
|
def parse_set_codes(self, raw_text):
|
|
"""Parse one or many MTG set codes from free-text input.
|
|
|
|
Args:
|
|
raw_text: Multiline or comma-separated set-code input.
|
|
|
|
Returns:
|
|
list[str]: Deduplicated lower-case set codes.
|
|
"""
|
|
set_codes = []
|
|
for chunk in re.split(r"[\s,;]+", raw_text or ""):
|
|
set_code = (chunk or "").strip().lower()
|
|
if set_code and set_code not in set_codes:
|
|
set_codes.append(set_code)
|
|
return set_codes
|
|
|
|
def fetch_set_seed_payloads(self, set_code, *, card_limit=0, include_tokens=False):
|
|
"""Fetch seed payloads for one MTG set code.
|
|
|
|
Args:
|
|
set_code: Scryfall set code such as ``tdm``.
|
|
card_limit: Optional maximum number of returned cards.
|
|
include_tokens: Whether token cards should be included.
|
|
|
|
Returns:
|
|
list[dict]: Seed payloads for the requested set import.
|
|
"""
|
|
normalized_set_code = (set_code or "").strip().lower()
|
|
if not normalized_set_code:
|
|
raise UserError(_("Enter a set code first."))
|
|
|
|
query_terms = [f"e:{normalized_set_code}", "game:paper"]
|
|
if not include_tokens:
|
|
query_terms.append("-is:token")
|
|
|
|
seed_payloads = list(
|
|
self.iter_search_card_payloads(
|
|
" ".join(query_terms),
|
|
unique="prints",
|
|
limit=card_limit or None,
|
|
order="set",
|
|
direction="asc",
|
|
)
|
|
)
|
|
if not seed_payloads:
|
|
raise UserError(
|
|
_("Scryfall returned no cards for the set code %s.")
|
|
% normalized_set_code.upper()
|
|
)
|
|
return seed_payloads
|
|
|
|
def get_set_print_group_payloads(
|
|
self,
|
|
set_code,
|
|
*,
|
|
languages=None,
|
|
limit=None,
|
|
include_tokens=False,
|
|
):
|
|
"""Return localized payload groups for one MTG set import.
|
|
|
|
Args:
|
|
set_code: Raw MTG set code.
|
|
languages: Optional iterable of wanted Scryfall language codes.
|
|
limit: Optional maximum number of imported print groups.
|
|
include_tokens: Whether token cards should be included.
|
|
|
|
Returns:
|
|
list[list[dict]]: Localized payload groups keyed by print reference.
|
|
"""
|
|
import_language_codes = normalize_language_codes(
|
|
languages,
|
|
default=self.get_import_language_codes(),
|
|
)
|
|
wanted_languages = set(import_language_codes)
|
|
grouped_payloads = {}
|
|
card_model = self.env["mvd.tcg.card"]
|
|
normalized_set_code = (set_code or "").strip().lower()
|
|
if not normalized_set_code:
|
|
raise UserError(_("Enter a set code first."))
|
|
|
|
query_terms = [f"e:{normalized_set_code}", "game:paper"]
|
|
if not include_tokens:
|
|
query_terms.append("-is:token")
|
|
|
|
for payload in self.iter_search_card_payloads(
|
|
" ".join(query_terms),
|
|
unique="prints",
|
|
include_multilingual=True,
|
|
order="set",
|
|
direction="asc",
|
|
):
|
|
external_ref = card_model._mtg_scryfall_build_external_ref(payload)
|
|
if not external_ref or external_ref == ":":
|
|
continue
|
|
payload_language = (payload.get("lang") or "").strip().lower()
|
|
if wanted_languages and payload_language and payload_language not in wanted_languages:
|
|
continue
|
|
grouped_payloads.setdefault(external_ref, {})[payload.get("id")] = payload
|
|
|
|
sorted_group_payloads = []
|
|
language_sort_order = {
|
|
code: index
|
|
for index, code in enumerate(import_language_codes)
|
|
}
|
|
for external_ref in sorted(grouped_payloads):
|
|
localized_payloads = sorted(
|
|
grouped_payloads[external_ref].values(),
|
|
key=lambda candidate_payload: (
|
|
language_sort_order.get(
|
|
(candidate_payload.get("lang") or "").strip().lower(),
|
|
999,
|
|
),
|
|
candidate_payload.get("lang") or "",
|
|
candidate_payload.get("id") or "",
|
|
),
|
|
)
|
|
if localized_payloads:
|
|
sorted_group_payloads.append(localized_payloads)
|
|
if limit and len(sorted_group_payloads) >= int(limit):
|
|
break
|
|
|
|
if not sorted_group_payloads:
|
|
raise UserError(
|
|
_("Scryfall returned no cards for the set code %s.")
|
|
% normalized_set_code.upper()
|
|
)
|
|
return sorted_group_payloads
|