214 lines
7.9 KiB
Python
214 lines
7.9 KiB
Python
"""nxm:// handling and the Nexus Mods API client."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Callable
|
|
|
|
from .. import __version__
|
|
from .util import atomic_write_json, read_json
|
|
|
|
GAME_DOMAIN = "monsterhunterwilds"
|
|
API = "https://api.nexusmods.com"
|
|
_SECRET_ATTRS = {"application": "mhwilds-mod-manager", "kind": "nexus-api-key"}
|
|
|
|
|
|
class NexusError(Exception):
|
|
pass
|
|
|
|
|
|
@dataclass
|
|
class NxmLink:
|
|
game: str
|
|
mod_id: int
|
|
file_id: int
|
|
key: str = ""
|
|
expires: str = ""
|
|
|
|
@property
|
|
def is_premium_flow(self) -> bool:
|
|
return not self.key
|
|
|
|
|
|
def parse_nxm(url: str) -> NxmLink:
|
|
"""nxm://monsterhunterwilds/mods/<mod>/files/<file>?key=…&expires=… (key/expires pass through)."""
|
|
u = urllib.parse.urlparse(url.strip())
|
|
if u.scheme != "nxm":
|
|
raise NexusError(f"not an nxm:// link: {url}")
|
|
m = re.fullmatch(r"/mods/(\d+)/files/(\d+)", u.path)
|
|
if not u.netloc or not m:
|
|
raise NexusError(f"malformed nxm link: {url}")
|
|
q = urllib.parse.parse_qs(u.query)
|
|
return NxmLink(u.netloc.lower(), int(m.group(1)), int(m.group(2)),
|
|
q.get("key", [""])[0], q.get("expires", [""])[0])
|
|
|
|
|
|
# -- API key storage -----------------------------------------------------------------
|
|
|
|
def _secret_mod():
|
|
try:
|
|
import gi
|
|
gi.require_version("Secret", "1")
|
|
from gi.repository import Secret
|
|
return Secret
|
|
except (ImportError, ValueError):
|
|
return None
|
|
|
|
|
|
class KeyStore:
|
|
"""libsecret when importable, else a 0600 JSON file."""
|
|
|
|
def __init__(self, fallback_file: Path):
|
|
self.file = fallback_file
|
|
|
|
def _schema(self, Secret):
|
|
return Secret.Schema.new("ca.schantli.MHWildsModManager", Secret.SchemaFlags.NONE,
|
|
{"application": Secret.SchemaAttributeType.STRING,
|
|
"kind": Secret.SchemaAttributeType.STRING})
|
|
|
|
def get(self) -> str:
|
|
Secret = _secret_mod()
|
|
if Secret:
|
|
try:
|
|
v = Secret.password_lookup_sync(self._schema(Secret), _SECRET_ATTRS, None)
|
|
if v:
|
|
return v
|
|
except Exception:
|
|
pass
|
|
return str(read_json(self.file, {}).get("nexus_api_key", ""))
|
|
|
|
def set(self, key: str) -> None:
|
|
Secret = _secret_mod()
|
|
if Secret:
|
|
try:
|
|
if Secret.password_store_sync(self._schema(Secret), _SECRET_ATTRS, Secret.COLLECTION_DEFAULT,
|
|
"Nexus Mods API key", key, None):
|
|
if self.file.exists():
|
|
self.file.unlink()
|
|
return
|
|
except Exception:
|
|
pass
|
|
atomic_write_json(self.file, {"nexus_api_key": key}, mode=0o600)
|
|
|
|
|
|
# -- client ----------------------------------------------------------------------------
|
|
|
|
@dataclass
|
|
class RateLimits:
|
|
daily_remaining: str = ""
|
|
daily_limit: str = ""
|
|
hourly_remaining: str = ""
|
|
hourly_limit: str = ""
|
|
|
|
def summary(self) -> str:
|
|
if not (self.daily_remaining or self.hourly_remaining):
|
|
return ""
|
|
return f"API quota: {self.hourly_remaining}/{self.hourly_limit} hourly, {self.daily_remaining}/{self.daily_limit} daily"
|
|
|
|
|
|
Opener = Callable[[urllib.request.Request], "object"]
|
|
|
|
|
|
class NexusClient:
|
|
def __init__(self, api_key: str, opener: Opener | None = None):
|
|
self.api_key = api_key
|
|
self._open = opener or (lambda req: urllib.request.urlopen(req, timeout=60))
|
|
self.limits = RateLimits()
|
|
|
|
def _headers(self) -> dict[str, str]:
|
|
return {"apikey": self.api_key, "Application-Name": "MHWildsModManager",
|
|
"Application-Version": __version__, "User-Agent": f"mhwilds-mod-manager/{__version__}",
|
|
"Accept": "application/json"}
|
|
|
|
def _get_json(self, path: str, params: dict[str, str] | None = None):
|
|
if not self.api_key:
|
|
raise NexusError("No Nexus API key set (Preferences → Nexus Mods).")
|
|
url = f"{API}{path}" + (f"?{urllib.parse.urlencode(params)}" if params else "")
|
|
try:
|
|
resp = self._open(urllib.request.Request(url, headers=self._headers()))
|
|
self._note_limits(resp.headers)
|
|
return json.loads(resp.read())
|
|
except urllib.error.HTTPError as e:
|
|
self._note_limits(e.headers)
|
|
msg = {401: "Invalid Nexus API key.", 403: "Nexus refused the download (premium required or link expired).",
|
|
404: "Not found on Nexus.", 429: "Nexus rate limit reached; try again later."}.get(e.code, f"Nexus HTTP {e.code}")
|
|
raise NexusError(msg) from e
|
|
except (urllib.error.URLError, OSError) as e:
|
|
raise NexusError(f"Network error: {e}") from e
|
|
|
|
def _note_limits(self, headers) -> None:
|
|
if headers is None:
|
|
return
|
|
g = headers.get
|
|
self.limits = RateLimits(g("x-rl-daily-remaining", ""), g("x-rl-daily-limit", ""),
|
|
g("x-rl-hourly-remaining", ""), g("x-rl-hourly-limit", ""))
|
|
|
|
def download_links(self, link: NxmLink) -> list[dict]:
|
|
params = {}
|
|
if link.key: # what lets non-premium accounts download
|
|
params = {"key": link.key, "expires": link.expires}
|
|
return self._get_json(f"/v1/games/{link.game}/mods/{link.mod_id}/files/{link.file_id}/download_link.json",
|
|
params)
|
|
|
|
def mod_info(self, link: NxmLink) -> dict:
|
|
return self._get_json(f"/v1/games/{link.game}/mods/{link.mod_id}.json")
|
|
|
|
|
|
def prefill(link: NxmLink, info: dict) -> dict:
|
|
return {"name": info.get("name", ""), "version": info.get("version", ""), "author": info.get("author", ""),
|
|
"summary": info.get("summary", ""), "picture_url": info.get("picture_url", ""),
|
|
"mod_id": link.mod_id, "file_id": link.file_id,
|
|
"url": f"https://www.nexusmods.com/{link.game}/mods/{link.mod_id}"}
|
|
|
|
|
|
def download(url: str, dest_dir: Path, progress: Callable[[int, int], None] | None = None,
|
|
opener: Opener | None = None) -> Path:
|
|
"""Stream a CDN URL to dest_dir; the file name comes from the URL path."""
|
|
url = urllib.parse.quote(url, safe=":/&?=%")
|
|
u = urllib.parse.urlparse(url)
|
|
if u.scheme != "https":
|
|
raise NexusError("refusing non-HTTPS download")
|
|
name = os.path.basename(urllib.parse.unquote(u.path)) or "download.bin"
|
|
name = re.sub(r"[^\w.\- ]", "_", name).lstrip(".") or "download.bin"
|
|
dest_dir.mkdir(parents=True, exist_ok=True)
|
|
dest = dest_dir / name
|
|
part = dest.with_name(dest.name + ".part")
|
|
req = urllib.request.Request(url, headers={"User-Agent": f"mhwilds-mod-manager/{__version__}"})
|
|
try:
|
|
resp = (opener or (lambda r: urllib.request.urlopen(r, timeout=60)))(req)
|
|
total = int(resp.headers.get("Content-Length") or 0)
|
|
done = 0
|
|
with open(part, "wb") as fh:
|
|
while chunk := resp.read(1 << 20):
|
|
fh.write(chunk)
|
|
done += len(chunk)
|
|
if progress:
|
|
progress(done, total)
|
|
os.replace(part, dest)
|
|
except (urllib.error.URLError, OSError) as e:
|
|
part.unlink(missing_ok=True)
|
|
raise NexusError(f"Download failed: {e}") from e
|
|
return dest
|
|
|
|
|
|
def fetch_thumbnail(url: str, dest: Path, opener: Opener | None = None, limit: int = 8 * 1024 * 1024) -> bool:
|
|
"""Best-effort mod picture download; never raises."""
|
|
if not url.startswith("https://"):
|
|
return False
|
|
try:
|
|
req = urllib.request.Request(url, headers={"User-Agent": f"mhwilds-mod-manager/{__version__}"})
|
|
resp = (opener or (lambda r: urllib.request.urlopen(r, timeout=30)))(req)
|
|
data = resp.read(limit + 1)
|
|
if len(data) > limit or not data:
|
|
return False
|
|
dest.write_bytes(data)
|
|
return True
|
|
except (urllib.error.URLError, OSError):
|
|
return False
|