From ef3a294c3fa6e4fc8f9adb1b7158661ff80c9d6e Mon Sep 17 00:00:00 2001 From: nichinichisou Date: Wed, 14 Oct 2026 16:02:12 +0800 Subject: [PATCH] build: check the core and source layers in pyright's strict mode pyproject.toml lists core, source and schemas as strict paths; the rest of the package stays in basic mode. The modules of these layers are fully annotated: - addressables: catalog locations are typed as `Location` records; `relative_path` takes and returns strings. - catalog and jp: the JP source and client are typed, and `jp.open_catalog` documents what it opens. - config and configfile: setting values are strings or lists of strings (`configfile.Value`); the prompt takes typed callables. - cache: the settings are a `Settings` typed dictionary and buckets store `(value, size)` pairs. - jsonio: `is_object` and `is_array` narrow decoded JSON values. - master: `download` takes a typed fetch function. Leaving an ApkSet's `with` block returns None: the archives it closes never suppress an error. Co-Authored-By: Claude Opus 5.5 --- pyproject.toml | 2 + src/nnnotes/core/cache.py | 74 +++++++++++++++--------- src/nnnotes/core/jsonio.py | 47 ++++++++++----- src/nnnotes/schemas/__init__.py | 3 +- src/nnnotes/source/addressables.py | 51 +++++++++-------- src/nnnotes/source/apkset.py | 36 ++++++------ src/nnnotes/source/catalog.py | 68 +++++++++++++--------- src/nnnotes/source/config.py | 43 +++++++------- src/nnnotes/source/configfile.py | 66 +++++++++++++--------- src/nnnotes/source/gameapi.py | 18 +++--- src/nnnotes/source/jp.py | 91 +++++++++++++++++------------- src/nnnotes/source/languages.py | 5 +- src/nnnotes/source/master.py | 51 ++++++++++------- 13 files changed, 332 insertions(+), 223 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index a577ec0..57a1288 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -74,6 +74,8 @@ include = ["src"] pythonVersion = "3.11" pythonPlatform = "Linux" typeCheckingMode = "basic" +# the configuration and source layers, and the schemas, are checked in strict mode +strict = ["src/nnnotes/core", "src/nnnotes/source", "src/nnnotes/schemas"] [tool.importlinter] root_package = "nnnotes" diff --git a/src/nnnotes/core/cache.py b/src/nnnotes/core/cache.py index f74d054..e1ee210 100644 --- a/src/nnnotes/core/cache.py +++ b/src/nnnotes/core/cache.py @@ -28,12 +28,23 @@ from collections import OrderedDict from importlib import metadata from pathlib import Path -from typing import Any +from typing import Any, Protocol, TypedDict, cast DISK_MAGIC = b"NNC1" DEFAULT_MEMORY_MB = 128 -_settings = { + +class Settings(TypedDict): + """The cache settings of a process (configure).""" + + enabled: bool + memory: int # byte budget of each bucket + directory: Path | None # disk layer of the byte buckets + closures: int # loaded bundle closures kept per thread + closure_bytes: int # their bundle bytes + + +_settings: Settings = { "enabled": True, "memory": DEFAULT_MEMORY_MB << 20, "directory": None, @@ -48,10 +59,10 @@ def configure( *, enabled: bool | None = None, memory_mb: int | None = None, - directory=None, + directory: str | os.PathLike[str] | None = None, closures: int | None = None, closure_mb: int | None = None, -) -> dict: +) -> Settings: """Set the caches of this process: `enabled`; `memory_mb` (byte budget of each bucket); `directory` (disk layer of the byte buckets; "" or None for none); `closures` / `closure_mb` (loaded bundle closures kept by unity.load_closure, per thread: count and bundle bytes). Clears the buckets. Returns the settings.""" @@ -71,9 +82,9 @@ def configure( return settings() -def settings() -> dict: +def settings() -> Settings: """The current settings (a copy).""" - return dict(_settings) + return _settings.copy() def enabled() -> bool: @@ -90,7 +101,7 @@ def bucket(name: str, *, salt: str = "", disk: bool = False, max_item: int | Non return b -def stats() -> dict: +def stats() -> dict[str, dict[str, int]]: """Per bucket: items, bytes, hits, misses, disk hits, disk writes, disk write errors.""" return {n: b.stats() for n, b in sorted(_buckets.items())} @@ -101,17 +112,22 @@ def clear() -> None: # ---------------------------------------------------------------- keys -def _feed(h, part) -> None: +class _Hash(Protocol): + def update(self, data: bytes | memoryview, /) -> None: ... + + +def _feed(h: _Hash, part: object) -> None: if isinstance(part, (bytes, bytearray, memoryview)): - mv = memoryview(part).cast("B") + mv = memoryview(cast("bytes", part)).cast("B") # any buffer, read as bytes h.update(b"b" + struct.pack(" None: raise TypeError(f"cache key part of type {type(part).__name__}") -def key(*parts) -> str: +def key(*parts: object) -> str: """sha256 hex of `parts` (bytes, str, int, float, bool, None and tuples / lists of them), unambiguous.""" h = hashlib.sha256() _feed(h, parts) @@ -130,14 +146,14 @@ def key(*parts) -> str: _salts: dict[str, str] = {} -def source_salt(path, *dists: str) -> str: +def source_salt(path: str | os.PathLike[str], *dists: str) -> str: """A salt for keys of values a module produces: sha256 of the module's source file and the installed versions of the distributions `dists` (a missing one counts as "-").""" p = Path(path) k = f"{p}|{'|'.join(dists)}" s = _salts.get(k) if s is None: - versions = [] + versions: list[str] = [] for d in dists: try: versions.append(f"{d}={metadata.version(d)}") @@ -147,7 +163,7 @@ def source_salt(path, *dists: str) -> str: return s -def file_id(path) -> tuple: +def file_id(path: str | os.PathLike[str]) -> tuple[str, int, int]: """(path, size, mtime) of a file: identifies an executable or input file in a key.""" p = Path(path) st = p.stat() @@ -156,14 +172,16 @@ def file_id(path) -> tuple: # ---------------------------------------------------------------- buckets class Bucket: + """A memo cache of values by content key (see the module documentation).""" + def __init__(self, name: str, salt: str, disk: bool, max_item: int | None): self.name, self.salt, self.disk, self._max_item = name, salt, disk, max_item - self._items: OrderedDict[str, tuple] = OrderedDict() + self._items: OrderedDict[str, tuple[Any, int]] = OrderedDict() # key -> (value, byte size) self._bytes = 0 self._lock = threading.Lock() self.hits = self.misses = self.disk_hits = self.disk_writes = self.disk_errors = 0 - def key(self, *parts) -> str: + def key(self, *parts: object) -> str: return key(self.name, self.salt, parts) def _limit(self) -> int: @@ -180,8 +198,9 @@ def get(self, k: str) -> Any: self._items.move_to_end(k) self.hits += 1 return hit[0] - if self.disk and _settings["directory"] is not None: - data = _disk_read(self._path(k)) + directory = _settings["directory"] + if self.disk and directory is not None: + data = _disk_read(self._path(directory, k)) if data is not None: with self._lock: self.disk_hits += 1 @@ -191,21 +210,22 @@ def get(self, k: str) -> Any: self.misses += 1 return None - def put(self, k: str, value, size: int | None = None) -> None: + def put(self, k: str, value: Any, size: int | None = None) -> None: """Store `value` (bytes for a disk bucket) under `k`; `size`: its byte size (default len(value)).""" if not _settings["enabled"]: return n = len(value) if size is None else int(size) self._remember(k, value, n) - if self.disk and _settings["directory"] is not None and n <= max(self._limit(), 64 << 20): - ok = _disk_write(self._path(k), value) + directory = _settings["directory"] + if self.disk and directory is not None and n <= max(self._limit(), 64 << 20): + ok = _disk_write(self._path(directory, k), value) with self._lock: if ok: self.disk_writes += 1 else: self.disk_errors += 1 - def _remember(self, k: str, value, n: int) -> None: + def _remember(self, k: str, value: Any, n: int) -> None: if n > self._limit(): return with self._lock: @@ -219,15 +239,15 @@ def _remember(self, k: str, value, n: int) -> None: _, (_, m) = self._items.popitem(last=False) self._bytes -= m - def _path(self, k: str) -> Path: - return _settings["directory"] / self.name / k[:2] / k + def _path(self, directory: Path, k: str) -> Path: + return directory / self.name / k[:2] / k def clear(self) -> None: with self._lock: self._items.clear() self._bytes = 0 - def stats(self) -> dict: + def stats(self) -> dict[str, int]: with self._lock: return { "items": len(self._items), @@ -298,7 +318,7 @@ def unpack(data: bytes) -> tuple[bytes, list[bytes]]: pos = 8 meta = data[pos : pos + n] pos += n - blobs = [] + blobs: list[bytes] = [] for _ in range(count): (m,) = struct.unpack_from(" TypeGuard[dict[str, Any]]: + """Whether a decoded JSON value is an object.""" + return isinstance(v, dict) + + +def is_array(v: object) -> TypeGuard[list[Any]]: + """Whether a decoded JSON value is an array.""" + return isinstance(v, list) + + class _NaN(ValueError): - def __init__(self): + def __init__(self) -> None: super().__init__() self.path: list[str] = [] -def _finite(v, found: list): +def _finite(v: object, found: list[bool]) -> object: """`v` with every non-finite float replaced by a placeholder string; containers are copied only where something changed, so the caller's document is never modified.""" if isinstance(v, float): @@ -38,8 +51,9 @@ def _finite(v, found: list): found.append(True) return _POS if v > 0 else _NEG if isinstance(v, dict): - out = None - for k, x in v.items(): + d = cast("dict[object, object]", v) + out: dict[object, object] | None = None + for k, x in d.items(): if k == _POS or k == _NEG: raise ValueError(f"key {k!r} collides with the non-finite placeholder") try: @@ -49,28 +63,29 @@ def _finite(v, found: list): raise if y is not x: if out is None: - out = dict(v) + out = dict(d) out[k] = y - return v if out is None else out + return d if out is None else out if isinstance(v, (list, tuple)): - out = None - for i, x in enumerate(v): + items = cast("Sequence[object]", v) + copied: list[object] | None = None + for i, x in enumerate(items): try: y = _finite(x, found) except _NaN as e: e.path.append(f"[{i}]") raise if y is not x: - if out is None: - out = list(v) - out[i] = y - return v if out is None else out + if copied is None: + copied = list(items) + copied[i] = y + return items if copied is None else copied if isinstance(v, str) and (v == _POS or v == _NEG): raise ValueError(f"string {v!r} collides with the non-finite placeholder") return v -def dumps(obj, **kw) -> str: +def dumps(obj: object, **kw: Any) -> str: """json.dumps with non-finite floats as 1e999 / -1e999; raises ValueError on NaN (with its path). Keyword arguments are json.dumps' (indent, ensure_ascii, separators, default, ...).""" try: @@ -80,7 +95,7 @@ def dumps(obj, **kw) -> str: else: if _MARK not in s: return s # (a placeholder string in the data: the walk rejects it) - found: list = [] + found: list[bool] = [] try: clean = _finite(obj, found) except _NaN as e: @@ -92,7 +107,9 @@ def dumps(obj, **kw) -> str: return s -def write_json(path, obj, *, indent: int | None = 1, ensure_ascii: bool = False, **kw) -> Path: +def write_json( + path: str | os.PathLike[str], obj: object, *, indent: int | None = 1, ensure_ascii: bool = False, **kw: Any +) -> Path: """Write `obj` as JSON to `path` (UTF-8, LF); see `dumps`. Returns the path.""" path = Path(path) path.write_text(dumps(obj, indent=indent, ensure_ascii=ensure_ascii, **kw), encoding="utf-8", newline="\n") diff --git a/src/nnnotes/schemas/__init__.py b/src/nnnotes/schemas/__init__.py index ef742d6..c43a960 100644 --- a/src/nnnotes/schemas/__init__.py +++ b/src/nnnotes/schemas/__init__.py @@ -3,6 +3,7 @@ import json from importlib import resources +from typing import Any __all__ = ["load", "names"] @@ -15,7 +16,7 @@ def names() -> list[str]: return sorted(p.name[: -len(SUFFIX)] for p in files.iterdir() if p.name.endswith(SUFFIX)) -def load(name: str) -> dict: +def load(name: str) -> dict[str, Any]: """The schema `name` (`music-data` for music-data.schema.json).""" path = resources.files("nnnotes.schemas") / f"{name}{SUFFIX}" if not path.is_file(): diff --git a/src/nnnotes/source/addressables.py b/src/nnnotes/source/addressables.py index 8d5d756..bac959b 100644 --- a/src/nnnotes/source/addressables.py +++ b/src/nnnotes/source/addressables.py @@ -19,13 +19,19 @@ import hashlib import io import struct +from collections.abc import Sequence from dataclasses import dataclass, field +from typing import Any, TypeAlias from urllib.parse import quote, unquote from Crypto.Cipher import AES __all__ = ["BundleKey", "decrypt", "parse", "parse_header", "parse_keys", "parse_locations"] +# a location of a catalog: {offset, primary_key, internal_id, dependencies} (parse), with the fields parse_locations +# decodes +Location: TypeAlias = "dict[str, Any]" + UNITYFS = b"UnityFS\0" ENCRYPTED_BYTES = 16384 # only the first 16 KiB of a bundle are encrypted CATALOG_MAGIC = 0x0DE38942 @@ -57,7 +63,7 @@ class BundleKey: key: bytes = field(repr=False) seed: bytes = field(repr=False) - def __post_init__(self): + def __post_init__(self) -> None: if len(self.key) != 16: raise ValueError("bundle key must be 16 bytes") @@ -67,11 +73,11 @@ def decrypt(data: bytes, filename: str, key: BundleKey) -> bytes: if data.startswith(UNITYFS): return data nonce = hashlib.sha256(key.seed + filename.encode("utf-8")).digest()[:8] - aes = AES.new(key.key, AES.MODE_CTR, nonce=nonce, initial_value=0) + aes = AES.new(key.key, AES.MODE_CTR, nonce=nonce, initial_value=0) # pyright: ignore[reportUnknownMemberType] (pycryptodome stubs: untyped overloads) return aes.decrypt(data[:ENCRYPTED_BYTES]) + data[ENCRYPTED_BYTES:] -def relative_path(value): +def relative_path(value: str) -> str: """A JP remote asset path (below the CDN base), checked and quoted; ValueError when it is not one.""" decoded = unquote(value) if ( @@ -100,23 +106,23 @@ def remote_path(internal_id: str) -> str | None: return slash + path -def parse(data: bytes) -> list[dict]: +def parse(data: bytes) -> list[Location]: """Every location of a binary catalog: {offset, primary_key, internal_id, dependencies (location offsets)}.""" data = catalog_bytes(data) - def u32(offset): + def u32(offset: int) -> int: return struct.unpack_from(" Sequence[int]: if offset == NONE: return [] return struct.unpack_from(f"<{u32(offset - 4) // 4}I", data, offset) - def text(offset): + def text(offset: int) -> str: if offset == NONE: return "" if offset & 0x40000000: # a path of parts, stored last part first - parts = [] + parts: list[str] = [] while offset != NONE: part, offset = struct.unpack_from(" int: return struct.unpack_from(" str | None: return self.plain(offset) if not sep: raise ValueError(f"catalog string at {offset & OFFSET_MASK}: a part list where a plain string belongs") - parts, link, seen = [], offset, set() + parts: list[str] = [] + link, seen = offset, set[int]() while link != NONE: pos = link & OFFSET_MASK if pos in seen: @@ -221,7 +228,7 @@ def type_name(self, offset: int) -> tuple[str | None, str | None] | None: t = self._types[offset] = (self.string(assembly, "."), self.string(cls, ".")) return t - def object_init(self, offset: int) -> dict | None: + def object_init(self, offset: int) -> dict[str, Any] | None: """ObjectInitializationData.Serializer.Data {id, type, data}: id and data plain strings, type a TypeSerializer.Data.""" if offset == NONE: @@ -242,24 +249,24 @@ def key(self, offset: int) -> tuple[str | None, object]: return cls, struct.unpack_from(" dict | None: + def extra(self, offset: int) -> dict[str, Any] | None: """A location's extra data (ObjectTypeData {type, object}): {"type": class name} plus, for AssetBundleRequestOptions, the decoded options; None for null.""" if offset == NONE: return None typ, obj = struct.unpack_from(" dict: + def request_options(self, offset: int) -> dict[str, Any]: """AssetBundleRequestOptionsSerializationAdapter.SerializedData {u32 hash, u32 bundleName, u32 crc, u32 bundleSize, u32 common}: hash a Hash128 (16 raw bytes, hex in stored order), bundleName read with '_', common a SerializedData.Common {i16 timeout, u8 redirectLimit, u8 retryCount, i32 flags}.""" hash_id, name_id, crc, size, common = struct.unpack_from("<5I", self.data, offset) - out = { + out: dict[str, Any] = { "hash": None if hash_id == NONE else self.data[hash_id : hash_id + 16].hex(), "bundleName": self.string(name_id, "_"), "crc": crc, @@ -275,7 +282,7 @@ def request_options(self, offset: int) -> dict: return out -def parse_header(data: bytes) -> dict: +def parse_header(data: bytes) -> dict[str, Any]: """The header of a binary catalog: magic, version, keysOffset, locatorId, instanceProvider, sceneProvider ({id, assembly, type, data}), initObjects (the same, the providers the catalog initializes), buildResultHash.""" data = catalog_bytes(data) @@ -297,13 +304,13 @@ def parse_header(data: bytes) -> dict: } -def parse_keys(data: bytes) -> list[dict]: +def parse_keys(data: bytes) -> list[dict[str, Any]]: """The key table of a binary catalog in stored order (ContentCatalogData.ResourceLocator.KeyData {u32 key object, u32 location set}): {"key": value, "type": the key's type name, "locations": location offsets}.""" data = catalog_bytes(data) header = parse_header(data) buf = _Buffer(data) - out = [] + out: list[dict[str, Any]] = [] table = header["keysOffset"] n = buf.u32(table - 4) if n % 8: @@ -315,7 +322,7 @@ def parse_keys(data: bytes) -> list[dict]: return out -def parse_locations(data: bytes) -> list[dict]: +def parse_locations(data: bytes) -> list[Location]: """Every location of a binary catalog, fully decoded, by offset: what parse() gives (offset, primary_key, internal_id, dependencies) plus provider, dependency_hash, resource_type (class names) and extra_data. @@ -331,7 +338,7 @@ def parse_locations(data: bytes) -> list[dict]: table = header["keysOffset"] for pos in range(table, table + buf.u32(table - 4), 8): offsets.update(buf.array(buf.u32(pos + 4))) - out = [] + out: list[Location] = [] for pos in sorted(offsets): if pos + LOCATION.size > len(data): raise ValueError(f"catalog location at {pos}: past the end of the catalog") diff --git a/src/nnnotes/source/apkset.py b/src/nnnotes/source/apkset.py index f5e5c8f..d6ad2a5 100644 --- a/src/nnnotes/source/apkset.py +++ b/src/nnnotes/source/apkset.py @@ -7,6 +7,7 @@ from __future__ import annotations +import os import shutil import struct import tempfile @@ -15,6 +16,7 @@ from contextlib import ExitStack from functools import lru_cache from pathlib import Path +from typing import IO, Any from ..core.cache import file_id @@ -23,7 +25,7 @@ @lru_cache(maxsize=2) -def _expanded_apks(identity): +def _expanded_apks(identity: tuple[str, int, int]) -> tuple[tempfile.TemporaryDirectory[str], list[Path]]: """An owned temporary directory and its APK paths; readers retain it across cache eviction.""" with zipfile.ZipFile(identity[0]) as outer: names = [n for n in outer.namelist() if n.lower().endswith(".apk")] @@ -32,7 +34,7 @@ def _expanded_apks(identity): raise ValueError("APK archive must contain exactly one base.apk") directory = tempfile.TemporaryDirectory(prefix="nnnotes-apks-") try: - paths = [] + paths: list[Path] = [] for i, name in enumerate(bases + sorted(n for n in names if n not in bases)): path = Path(directory.name) / f"{i}.apk" with outer.open(name) as stream, path.open("wb") as target: @@ -45,13 +47,15 @@ def _expanded_apks(identity): class ApkSet: - def __init__(self, source): + """The members of an APK set (see the module documentation), read as one archive inside a `with` block.""" + + def __init__(self, source: str | os.PathLike[str] | IO[bytes]): self.source = source self._stack = ExitStack() - self._members = {} - self._expanded = None + self._members: dict[str, tuple[zipfile.ZipFile, zipfile.ZipInfo]] = {} + self._expanded: tuple[tempfile.TemporaryDirectory[str], list[Path]] | None = None - def __enter__(self): + def __enter__(self) -> ApkSet: try: self._open() return self @@ -59,13 +63,13 @@ def __enter__(self): self.__exit__(None, None, None) raise - def _add(self, source): + def _add(self, source: str | os.PathLike[str] | IO[bytes]) -> None: archive = self._stack.enter_context(zipfile.ZipFile(source)) for info in archive.infolist(): self._members.setdefault(info.filename, (archive, info)) - def _open(self): - if hasattr(self.source, "read"): + def _open(self) -> None: + if not isinstance(self.source, (str, os.PathLike)): self._add(self.source) return path = Path(self.source) @@ -85,20 +89,20 @@ def _open(self): for apk in paths: self._add(apk) - def namelist(self): + def namelist(self) -> list[str]: return list(self._members) - def infolist(self): + def infolist(self) -> list[zipfile.ZipInfo]: return [info for _, info in self._members.values()] - def read(self, name): + def read(self, name: str | zipfile.ZipInfo) -> bytes: name = name.filename if isinstance(name, zipfile.ZipInfo) else name archive, info = self._members[name] return archive.read(info) - def __exit__(self, *args): + def __exit__(self, *args: Any) -> None: try: - return self._stack.__exit__(*args) + self._stack.__exit__(*args) finally: self._members.clear() self._expanded = None @@ -109,13 +113,13 @@ def _manifest_attribute(data: bytes, attribute: str) -> tuple[str | None, int | """An attribute of the element of a binary AndroidManifest.xml: (its string, None when it has no string value; the Res_value data type; the Res_value data). None when absent or unreadable.""" try: - pos, strings = 8, [] + pos, strings = 8, list[str]() while pos + 8 <= len(data): typ, hsize, size = struct.unpack_from(" str: rel = remote_path(internal_id) if rel is None: rel = internal_id[len(LOCAL_PREFIX) :] if internal_id.startswith(LOCAL_PREFIX) else internal_id - head, sep, tail = rel.partition("/Android/") + _, sep, tail = rel.partition("/Android/") return tail if sep else rel.rsplit("/", 1)[-1] @@ -96,8 +108,8 @@ def __init__( cdn: Lazy[str] = None, bundle_key: Lazy[BundleKey] = None, apk: Path | None = None, - source=None, - client=None, + source: Source | None = None, + client: Client | None = None, apk_catalog: bytes | None = None, resource_version: str | None = None, ): @@ -108,15 +120,15 @@ def __init__( self.source, self.client = source, client self.resource_version = resource_version self._sources = {"remote": catalog_bytes} - self._locations: list[dict] | None = None - self._parsed: tuple | None = None + self._locations: list[Location] | None = None + self._parsed: tuple[list[Location], dict[int, Location], dict[str, list[Location]]] | None = None if apk_catalog is not None: self._sources["apk"] = apk_catalog elif self.apk is not None: with ApkSet(self.apk) as z: self._sources["apk"] = z.read(APK_CATALOG) - def check_apk(self, archive) -> None: + def check_apk(self, archive: ApkSet) -> None: """Before extracting a JP local file, verify the APK against this catalog's snapshot.""" if self.source is not None and "apk" in self._sources: try: @@ -132,7 +144,7 @@ def local_cache_dir(self) -> Path: return self.cache_dir / "apk" / hashlib.sha256(self._sources["apk"]).hexdigest() return self.cache_dir - def _parse(self) -> tuple: + def _parse(self) -> tuple[list[Location], dict[int, Location], dict[str, list[Location]]]: """(entries, by offset, by primary key), parsed the first time a lookup needs them (a command that only wants the catalog files, sources(), does not pay for it).""" if self._parsed is None: @@ -142,22 +154,22 @@ def _parse(self) -> tuple: e["offset"] += APK_OFFSET_BASE e["dependencies"] = [d + APK_OFFSET_BASE for d in e["dependencies"]] entries.append(e) - by_key: dict[str, list[dict]] = {} + by_key: dict[str, list[Location]] = {} for e in entries: by_key.setdefault(e["primary_key"], []).append(e) self._parsed = (entries, {e["offset"]: e for e in entries}, by_key) return self._parsed @property - def entries(self) -> list[dict]: + def entries(self) -> list[Location]: return self._parse()[0] @property - def _by_off(self) -> dict[int, dict]: + def _by_off(self) -> dict[int, Location]: return self._parse()[1] @property - def _by_key(self) -> dict[str, list[dict]]: + def _by_key(self) -> dict[str, list[Location]]: return self._parse()[2] # --- construction ------------------------------------------------------ @@ -230,10 +242,10 @@ def keys(self, prefix: str = "") -> list[str]: def has(self, key: str) -> bool: return key in self._by_key - def entries_for(self, key: str) -> list[dict]: + def entries_for(self, key: str) -> list[Location]: return list(self._by_key.get(key, [])) - def entry(self, key: str) -> dict: + def entry(self, key: str) -> Location: """The location of `key`: its asset location (internal id under Assets/ or Packages/) when it has one, else its first location. KeyError when the catalog lacks the key.""" ents = self._by_key.get(key) @@ -244,13 +256,13 @@ def entry(self, key: str) -> dict: return e return ents[0] - def dependencies(self, key: str) -> list[dict]: + def dependencies(self, key: str) -> list[Location]: """The locations the location of `key` (entry) depends on, in its dependency order (those the catalogs hold).""" return [self._by_off[d] for d in self.entry(key)["dependencies"] if d in self._by_off] @staticmethod - def as_bundle(e: dict) -> Bundle | None: + def as_bundle(e: Location) -> Bundle | None: """The bundle file a location names (a remote or APK-local .bundle), else None.""" iid = e["internal_id"] if iid.endswith(".bundle") and remote_path(iid) is not None: @@ -263,12 +275,12 @@ def sources(self) -> dict[str, bytes]: """The catalog files this catalog was made of: {"remote": bytes, "apk": bytes (with an APK)}.""" return dict(self._sources) - def locations(self) -> list[dict]: + def locations(self) -> list[Location]: """Every location of the remote catalog and (with an APK) of the APK catalog, fully decoded (addressables.parse_locations), with `catalog` ("remote" / "apk") and `kind` (location_kind). Offsets and dependencies are those of `entries` (APK ones in their own namespace).""" if self._locations is None: - out = [] + out: list[Location] = [] for name, base in (("remote", 0), ("apk", APK_OFFSET_BASE)): if name not in self._sources: continue @@ -292,11 +304,11 @@ def bundles(self) -> list[Bundle]: out.setdefault(b.name, b) return [out[n] for n in sorted(out)] - def raw_files(self) -> list[dict]: + def raw_files(self) -> list[Location]: """Every raw file the catalogs name (a CDN or APK file that is not a bundle, e.g. CRI ACB/AWB/USM data), one entry per file name (file_name), sorted by name, preferring the APK catalog's as bundles() does. Remote ones are what fetch_raw takes.""" - out: dict[str, dict] = {} + out: dict[str, Location] = {} for e in sorted(self.entries, key=lambda e: (e["offset"] < APK_OFFSET_BASE, e["offset"])): if location_kind(e["internal_id"]) == "raw": out.setdefault(file_name(e["internal_id"]), e) @@ -329,7 +341,7 @@ def cached(self, b: Bundle) -> Path | None: dst = (self.cache_dir if b.remote else self.local_cache_dir()) / "bundles" / b.name return dst if dst.is_file() and dst.stat().st_size > 0 else None - def cached_raw(self, e: dict) -> Path | None: + def cached_raw(self, e: Location) -> Path | None: """The file fetch_raw(e) returns when it is in the cache already, else None.""" rel = remote_path(e["internal_id"]) dst = self.cache_dir / "raw" / rel.lstrip("/") if rel is not None else None @@ -368,14 +380,14 @@ def _url(self, internal_id: str) -> str: raise ValueError(f"{internal_id}: not a remote location") return cdn.rstrip("/") + path - def _download(self, url): + def _download(self, url: str) -> bytes: if self.source is not None: if self.client is None: raise ConfigError("JP downloads require a configured JP client") return self.client.get(url, source=self.source) return download(url) - def fetch_key(self, key: str, skip=frozenset()) -> list[Path]: + def fetch_key(self, key: str, skip: Collection[str] = frozenset()) -> list[Path]: """Every bundle of the key's closure but those named in `skip`; APK-local ones only when an APK is set.""" return [self.fetch(b) for b in self.resolve(key) if (b.remote or self.apk) and b.name not in skip] @@ -393,7 +405,7 @@ def bundle_of(self, key: str) -> Bundle: return b raise KeyError(f"{key}: no bundle among its dependencies") - def fetch_raw(self, e: dict) -> Path: + def fetch_raw(self, e: Location) -> Path: """A non-bundle CDN asset (e.g. CRI ACB/AWB data) as stored -- no decryption.""" iid = e["internal_id"] rel = remote_path(iid) @@ -440,7 +452,7 @@ def download(url: str, timeout: float = 120) -> bytes: parts = urllib.parse.urlsplit(url) if parts.scheme not in ("http", "https") or _proxied(parts): return _urlopen(url, timeout) - conns = _KEPT.__dict__.setdefault("conns", {}) + conns: dict[tuple[str, str], http.client.HTTPConnection] = _KEPT.__dict__.setdefault("conns", {}) key = (parts.scheme, parts.netloc) path = (parts.path or "/") + (f"?{parts.query}" if parts.query else "") for _ in range(2): @@ -468,7 +480,7 @@ def download(url: str, timeout: float = 120) -> bytes: return _urlopen(url, timeout) -def _proxied(parts) -> bool: +def _proxied(parts: urllib.parse.SplitResult) -> bool: proxies = urllib.request.getproxies() return parts.scheme in proxies and not urllib.request.proxy_bypass(parts.hostname or "") diff --git a/src/nnnotes/source/config.py b/src/nnnotes/source/config.py index 8ccddd7..4da6c34 100644 --- a/src/nnnotes/source/config.py +++ b/src/nnnotes/source/config.py @@ -20,8 +20,9 @@ import re import shutil import tomllib -from collections.abc import Mapping +from collections.abc import Callable, Mapping from pathlib import Path +from typing import Any, cast __all__ = ["Config", "ConfigError", "config_files", "user_file"] @@ -35,7 +36,7 @@ class ConfigError(Exception): """A setting is missing or malformed (the message names the setting, never its value).""" -def check_catalog_version(value: str) -> str: +def check_catalog_version(value: object) -> str: """One safe international catalog filename component, including the legacy main selector.""" if not isinstance(value, str) or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,63}", value) or ".." in value: raise ValueError("must be a single catalog version without path separators") @@ -110,14 +111,14 @@ class Config: def __init__( self, - data: dict | None = None, + data: dict[str, Any] | None = None, base: Path | None = None, source: str | None = None, environ: Mapping[str, str] | None = None, - overrides: dict | None = None, - flags: dict | None = None, + overrides: Mapping[tuple[str, str], object] | None = None, + flags: Mapping[tuple[str, str], str] | None = None, ): - self._data = data or {} + self._data: dict[str, Any] = data or {} self._base = base self.source = source # the TOML file read, if any (a path, shown in messages) env = os.environ if environ is None else environ @@ -135,8 +136,8 @@ def load( path: str | Path | None = None, *, environ: Mapping[str, str] | None = None, - overrides: dict | None = None, - flags: dict | None = None, + overrides: Mapping[tuple[str, str], object] | None = None, + flags: Mapping[tuple[str, str], str] | None = None, ) -> Config: env = os.environ if environ is None else environ file = find_file(path, env) @@ -149,13 +150,13 @@ def load( return cls(data, base=file.resolve().parent, source=str(file), environ=env, overrides=overrides, flags=flags) # ---------------------------------------------------------------- lookup - def _table(self, section: str) -> dict: - t = self._data + def _table(self, section: str) -> dict[str, Any]: + t: object = self._data for part in section.split("."): - t = t.get(part) if isinstance(t, dict) else None - return t if isinstance(t, dict) else {} + t = cast("dict[str, object]", t).get(part) if isinstance(t, dict) else None + return cast("dict[str, Any]", t) if isinstance(t, dict) else {} - def _raw(self, section: str, key: str): + def _raw(self, section: str, key: str) -> tuple[object, str | None]: """(value, origin) with origin 'flag', 'env' or 'file'; (None, None) when unset.""" if (section, key) in self._over: return self._over[(section, key)], "flag" @@ -180,12 +181,12 @@ def flag(self, section: str, key: str) -> str | None: def file_keys(self) -> list[tuple[str, str]]: """(section, key) of every value of the TOML file, a nested table as a dotted section ('' at the top).""" - out = [] + out: list[tuple[str, str]] = [] - def walk(table: dict, section: str) -> None: + def walk(table: dict[str, Any], section: str) -> None: for k, v in table.items(): if isinstance(v, dict): - walk(v, f"{section}.{k}" if section else k) + walk(cast("dict[str, Any]", v), f"{section}.{k}" if section else k) else: out.append((section, k)) @@ -222,14 +223,14 @@ def require(self, section: str, key: str) -> str: def get_list(self, section: str, key: str) -> list[str]: """A list of strings (TOML array; comma-separated in the environment or on the command line).""" - v, origin = self._raw(section, key) + v, _ = self._raw(section, key) if v is None: return [] if isinstance(v, str): return [s.strip() for s in v.split(",") if s.strip()] - if not isinstance(v, list) or not all(isinstance(s, str) for s in v): + if not isinstance(v, list) or not all(isinstance(s, str) for s in cast("list[object]", v)): raise self._bad(section, key, "must be a list of strings") - return list(v) + return list(cast("list[str]", v)) def path(self, section: str, key: str) -> Path | None: if section == "paths" and key in ("apk", "catalog") and self.origin(section, key) != "flag": @@ -241,7 +242,7 @@ def path(self, section: str, key: str) -> Path | None: return None if not isinstance(v, (str, os.PathLike)): raise self._bad(section, key, "must be a path string") - p = Path(v).expanduser() + p = Path(cast("str | os.PathLike[str]", v)).expanduser() if origin == "file" and not p.is_absolute() and self._base is not None: p = self._base / p return p @@ -334,7 +335,7 @@ def usable_cpus() -> int: """The CPUs this process may run on, which sizes the default worker counts: its affinity mask (taskset, a container's cpuset), not the machine's CPU count. os.process_cpu_count (Python 3.13), else the affinity mask where the platform reports one, else os.cpu_count; at least 1.""" - count = getattr(os, "process_cpu_count", None) + count: Callable[[], int | None] | None = getattr(os, "process_cpu_count", None) if count is not None: n = count() elif hasattr(os, "sched_getaffinity"): diff --git a/src/nnnotes/source/configfile.py b/src/nnnotes/source/configfile.py index 69ebaae..7577587 100644 --- a/src/nnnotes/source/configfile.py +++ b/src/nnnotes/source/configfile.py @@ -13,8 +13,10 @@ import os import re import tomllib +from collections.abc import Callable from dataclasses import dataclass from pathlib import Path +from typing import Any, TypeAlias, cast from urllib.parse import urlsplit from . import languages @@ -39,6 +41,10 @@ _REGION_NAME = re.compile(r"[A-Za-z0-9_-]+") +# the value of a setting: a string (a path made absolute), or a list of strings +Value: TypeAlias = "str | list[str]" + + @dataclass(frozen=True) class Setting: section: str # "bundle", REGION, FONTS, ... @@ -49,7 +55,9 @@ class Setting: def _template_settings() -> list[Setting]: - out, section, comment = [], None, [] + out: list[Setting] = [] + section: str | None = None + comment: list[str] = [] for line in template().decode("utf-8").splitlines(): s = line.strip() if not s: @@ -97,9 +105,10 @@ def resolve(name: str) -> tuple[Setting, str]: raise ConfigError(f"unknown setting {name}: `nnnotes config check` lists the settings") -def parse_value(setting: Setting, text: str): +def parse_value(setting: Setting, text: str) -> Value: """The value of a setting from its text form (a list: comma-separated; a path: made absolute), checked.""" text = text.strip() + value: Value if setting.kind == "list": value = [s.strip() for s in text.split(",") if s.strip()] elif setting.kind == "path" and text: @@ -111,10 +120,16 @@ def parse_value(setting: Setting, text: str): return value -def validate(setting: Setting, value) -> None: +def validate(setting: Setting, value: Value) -> None: """Raise ValueError (a message without the value) when `value` is not a valid value of `setting`.""" k = (setting.section, setting.key) - if k in HEX_BYTES: + if k == ("catalog", "language") or k == (REGION, "languages"): + for v in value if isinstance(value, list) else [value]: + if v not in languages.LANGUAGES: + raise ValueError(f"not one of the game's languages ({', '.join(languages.LANGUAGES)})") + elif isinstance(value, list): + pass # the other settings with a list value take any strings + elif k in HEX_BYTES: s = value[2:] if value[:2].lower() == "0x" else value try: b = bytes.fromhex(s) @@ -123,10 +138,6 @@ def validate(setting: Setting, value) -> None: size = HEX_BYTES[k] if size is not None and len(b) != size: raise ValueError(f"must be {size} bytes ({2 * size} hex digits)") - elif k == ("catalog", "language") or k == (REGION, "languages"): - for v in value if isinstance(value, list) else [value]: - if v not in languages.LANGUAGES: - raise ValueError(f"not one of the game's languages ({', '.join(languages.LANGUAGES)})") elif k == (REGION, "cdn"): u = urlsplit(value) if u.scheme not in ("https", "http") or not u.netloc: @@ -144,7 +155,7 @@ def validate(setting: Setting, value) -> None: # ---------------------------------------------------------------- TOML text -def _toml(value) -> str: +def _toml(value: Value) -> str: if isinstance(value, list): return "[" + ", ".join(_toml(v) for v in value) + "]" if "'" not in value and not re.search(r"[\x00-\x1f\x7f]", value): @@ -170,7 +181,7 @@ def _region_block(region: str) -> list[str]: return [f"[servers.{region}]\n"] -def _set_line(text: str, section: str, key: str, value) -> str: +def _set_line(text: str, section: str, key: str, value: Value) -> str: lines = text.splitlines(keepends=True) if lines and not lines[-1].endswith("\n"): lines[-1] += "\n" @@ -200,14 +211,14 @@ def _set_line(text: str, section: str, key: str, value) -> str: return _set_line(text, section, key, value) -def _lookup(data: dict, section: str, key: str): - t = data +def _lookup(data: dict[str, Any], section: str, key: str) -> object: + t: object = data for part in section.split("."): - t = t.get(part) if isinstance(t, dict) else None - return t.get(key) if isinstance(t, dict) else None + t = cast("dict[str, object]", t).get(part) if isinstance(t, dict) else None + return cast("dict[str, object]", t).get(key) if isinstance(t, dict) else None -def edit(text: str, values: list[tuple[str, str, object]], where: str) -> str: +def edit(text: str, values: list[tuple[str, str, Value]], where: str) -> str: """`text` with each (section, key, value) set in place; the result is checked to read back as those values.""" for section, key, value in values: text = _set_line(text, section, key, value) @@ -227,7 +238,7 @@ def edit(text: str, values: list[tuple[str, str, object]], where: str) -> str: return text -def render(values: list[tuple[str, str, object]], regions: list[str]) -> str: +def render(values: list[tuple[str, str, Value]], regions: list[str]) -> str: """The template with one region table per region (the example table renamed, repeated) and the values set.""" text = template().decode("utf-8") if regions: @@ -254,13 +265,16 @@ def write(path: Path, text: str) -> None: # ---------------------------------------------------------------- prompts def prompt( - regions_hint: str = TEMPLATE_REGION, ask=input, ask_secret=getpass.getpass, say=print -) -> tuple[list[tuple[str, str, object]], list[str]]: + regions_hint: str = TEMPLATE_REGION, + ask: Callable[[str], str] = input, + ask_secret: Callable[[str], str] = getpass.getpass, + say: Callable[[str], object] = print, +) -> tuple[list[tuple[str, str, Value]], list[str]]: """Ask for every setting in the terminal (Enter leaves one empty; the keys are read without echo) -> (values, regions).""" - values: list[tuple[str, str, object]] = [] + values: list[tuple[str, str, Value]] = [] - def one(s: Setting, section: str, default: str = ""): + def one(s: Setting, section: str, default: str = "") -> None: name = f"[{section}] {s.key}" hint = " (hidden)" if s.secret else " (comma-separated)" if s.kind == "list" else "" say(f"# {s.description}" if s.description else "#") @@ -280,7 +294,7 @@ def one(s: Setting, section: str, default: str = ""): if s.section in ("bundle", "master"): one(s, s.section) say("# regions to configure: names of your choice, one [servers.] table each") - regions = [] + regions: list[str] = [] while not regions: text = ask(f"regions (comma-separated) [{regions_hint}]: ") or regions_hint regions = [r.strip() for r in text.split(",") if r.strip()] @@ -320,7 +334,7 @@ def _status(cfg: Config, s: Setting, section: str) -> tuple[str, str | None]: elif (s.section, s.key) in DIRS and not p.is_dir(): return "not found", "no such directory" else: - v = cfg.get(section, s.key) + v = cfg.require(section, s.key) # set (its origin is known) validate(s, v) if (section, s.key) == ("catalog", "region") and v not in cfg.regions(): return "invalid", f"names no [servers.{v}] table and no {env_name(f'servers.{v}', 'cdn')}" @@ -331,13 +345,13 @@ def _status(cfg: Config, s: Setting, section: str) -> tuple[str, str | None]: return "ok", None -def check(cfg: Config) -> dict: +def check(cfg: Config) -> dict[str, Any]: """Every setting with its origin (flag, env, file or None) and status (ok, unset, invalid, not found), plus the keys of the file and the NNNOTES_ environment variables that name no setting (unknown). No values.""" - out = [] - known_env = set() + out: list[dict[str, Any]] = [] + known_env: set[str] = set() - def record(s: Setting, section: str, status: str, reason: str | None, origin: str | None): + def record(s: Setting, section: str, status: str, reason: str | None, origin: str | None) -> None: env = env_name(section, s.key) known_env.add(env) out.append( diff --git a/src/nnnotes/source/gameapi.py b/src/nnnotes/source/gameapi.py index 167be44..ffb7630 100644 --- a/src/nnnotes/source/gameapi.py +++ b/src/nnnotes/source/gameapi.py @@ -23,8 +23,10 @@ import time import uuid import zipfile +from collections.abc import Iterable from dataclasses import dataclass, field from pathlib import Path +from typing import Any from urllib.parse import urlsplit import grpc @@ -143,7 +145,7 @@ def parse_version(buf: bytes) -> MasterVersion: def parse_server_list(buf: bytes) -> list[Server]: - out = [] + out: list[Server] = [] for num, v in fields(buf): if num != 1: continue @@ -197,7 +199,7 @@ def call( timeout: float = TIMEOUT, attempts: int = ATTEMPTS, setting: str = "the API root", - response_metadata: dict | None = None, + response_metadata: dict[str, Any] | None = None, ) -> bytes: """One unary call with raw bytes; returns the response message's bytes. `setting` names the root in errors. Raises GameApiError.""" @@ -205,7 +207,9 @@ def call( name = method.rsplit("/", 1)[-1] # e.g. Version channel = grpc.secure_channel(target, grpc.ssl_channel_credentials()) if tls else grpc.insecure_channel(target) try: - stub = channel.unary_unary(method) # no (de)serializers: bytes in, bytes out + stub: grpc.UnaryUnaryMultiCallable[bytes, bytes] = channel.unary_unary( + method + ) # no (de)serializers: bytes in, bytes out attempts = max(1, attempts) for i in range(attempts): last = i + 1 == attempts @@ -229,7 +233,7 @@ def call( raise AssertionError("unreachable") -def failure(name: str, setting: str, code, trailers) -> str: +def failure(name: str, setting: str, code: grpc.StatusCode | None, trailers: Iterable[Any] | None) -> str: """The one-line error of a failed call: status, a hint, and the game's error code when the server sent one.""" status = code.name if code is not None else "no status" hint = STATUS_HINT.get(code) if code is not None else None @@ -327,18 +331,18 @@ def server_list(cfg: Config, *, timeout: float = TIMEOUT) -> list[Server]: def configured_roots(cfg: Config) -> dict[str, set[str]]: """{region: its `cdn` and `api` settings, trailing `/` removed} of the configured regions.""" - out = {} + out: dict[str, set[str]] = {} for r in cfg.regions(): roots = {cfg.get(f"servers.{r}", k) for k in ("cdn", "api")} out[r] = {s.strip().rstrip("/") for s in roots if s} return out -def server_summary(s: Server, configured: dict[str, set[str]], hosts: bool = False) -> dict: +def server_summary(s: Server, configured: dict[str, set[str]], hosts: bool = False) -> dict[str, Any]: """A server list entry for printing: the configured region with one of its roots, root counts, and the roots themselves only when `hosts`.""" region = next((r for r, roots in configured.items() if roots & {*s.cdn, *s.api}), None) - d = { + d: dict[str, Any] = { "name": s.name, "displayName": s.display_name, "areaId": s.area_id, diff --git a/src/nnnotes/source/jp.py b/src/nnnotes/source/jp.py index 26d40a4..7763ec2 100644 --- a/src/nnnotes/source/jp.py +++ b/src/nnnotes/source/jp.py @@ -10,18 +10,21 @@ import hashlib import http.client import json +import os import re import threading import time +from collections.abc import Callable from dataclasses import asdict, dataclass, field from pathlib import Path +from typing import Any, cast from urllib.parse import urlsplit from ..core.cache import write_atomic from . import gameapi -from .addressables import parse_header, relative_path +from .addressables import BundleKey, parse_header, relative_path from .catalog import Catalog -from .config import ConfigError +from .config import Config, ConfigError __all__ = ["Client", "Observation", "Source", "origin"] @@ -30,7 +33,7 @@ FILE_LIMIT = 1024 * 1024 * 1024 -def numeric_version(value): +def numeric_version(value: object) -> tuple[int, ...]: if not isinstance(value, str) or not re.fullmatch(r"[0-9]+(?:\.[0-9]+){1,3}", value): raise ValueError("invalid numeric version") parts = tuple(map(int, value.split("."))) @@ -39,29 +42,31 @@ def numeric_version(value): return parts + (0,) * (4 - len(parts)) -def master_version(value): +def master_version(value: object) -> str: if not isinstance(value, str) or not re.fullmatch(r"[0-9]+(?:\.[0-9]+){1,3}/[0-9a-f]{32}", value): raise ValueError("invalid JP master version") return value -def select_asset(raw, client) -> tuple[str, str] | None: +def select_asset(raw: str | None, client: str) -> tuple[str, str] | None: """Select the largest applicable live minimum; no top-level fallback when live has no match.""" if not raw: return None try: - data = json.loads(raw) + data: object = json.loads(raw) if not isinstance(data, dict): raise ValueError() + data = cast("dict[str, object]", data) live = data.get("live") if live: if not isinstance(live, list): raise ValueError() current = numeric_version(client) - candidates = [] - for entry in live: + candidates: list[tuple[tuple[int, ...], dict[str, object]]] = [] + for entry in cast("list[object]", live): if not isinstance(entry, dict): continue + entry = cast("dict[str, object]", entry) try: minimum = numeric_version(entry.get("minClientVersion")) except ValueError: @@ -80,7 +85,7 @@ def select_asset(raw, client) -> tuple[str, str] | None: raise gameapi.GameApiError("JP Version: invalid asset metadata") from None -def origin(value): +def origin(value: str) -> str: """The origin (`scheme://host[:port]`, without a default port) of an HTTPS URL with no path, query or credentials (HTTP only on a loopback host); a ConfigError for any other value.""" try: @@ -114,7 +119,7 @@ class Source: cdn: str = field(repr=False) platform: str = "Android" - def __post_init__(self): + def __post_init__(self) -> None: if self.provider != "jp" or self.platform != "Android": raise ValueError("unsupported JP asset source") numeric_version(self.version) @@ -124,28 +129,31 @@ def __post_init__(self): raise ValueError("JP source CDN is not a canonical origin") @property - def root(self): + def root(self) -> str: return f"{self.cdn}/asset/{self.version}/Android/{self.hash}" @property - def catalog_url(self): + def catalog_url(self) -> str: return self.root + "/catalog_main.bin" - def url(self, internal_id): + def url(self, internal_id: str) -> str: if not internal_id.startswith(PLACEHOLDER): raise ValueError("JP resource has no remote asset placeholder") return self.root + "/" + relative_path(internal_id[len(PLACEHOLDER) :]) - def to_dict(self): + def to_dict(self) -> dict[str, str]: return asdict(self) @classmethod - def from_dict(cls, data): - if not isinstance(data, dict) or set(data) != {"provider", "version", "hash", "cdn", "platform"}: + def from_dict(cls, data: object) -> Source: + if not isinstance(data, dict): + raise ValueError("invalid JP source metadata") + fields = cast("dict[str, Any]", data) + if set(fields) != {"provider", "version", "hash", "cdn", "platform"}: raise ValueError("invalid JP source metadata") - return cls(**data) + return cls(**fields) - def cache_dir(self, cache): + def cache_dir(self, cache: str | os.PathLike[str]) -> Path: site = hashlib.sha256(self.cdn.encode()).hexdigest()[:16] return Path(cache) / "jp" / site / self.version / self.hash @@ -164,13 +172,13 @@ class Client: asset source and CDN credential; `get()` downloads a file of one source or master version with that credential, observing again when it is older than five minutes.""" - def __init__(self, cfg, region): + def __init__(self, cfg: Config, region: str): self.cfg, self.region = cfg.for_region(region), region - self._observation = None - self._created = 0 + self._observation: Observation | None = None + self._created = 0.0 self._lock = threading.RLock() - def observe(self, *, timeout=gameapi.TIMEOUT): + def observe(self, *, timeout: float = gameapi.TIMEOUT) -> Observation: """Ask the API for the current Observation: master version, asset source, CDN and its credential.""" with self._lock: section = f"servers.{self.region}" @@ -178,7 +186,7 @@ def observe(self, *, timeout=gameapi.TIMEOUT): api = origin(api if "://" in api else "https://" + api) allowed = origin(self.cfg.cdn(self.region)) client = gameapi.client_version(self.cfg, self.region) - headers = {} + headers: dict[str, Any] = {} body = gameapi.call( api, gameapi.VERSION_METHOD, @@ -211,13 +219,15 @@ def observe(self, *, timeout=gameapi.TIMEOUT): self._observation, self._created = result, time.monotonic() return result - def _current(self): + def _current(self) -> Observation: with self._lock: if self._observation is None or time.monotonic() - self._created > 300: return self.observe() return self._observation - def get(self, url, *, source=None, master=None, limit=FILE_LIMIT): + def get( + self, url: str, *, source: Source | None = None, master: str | None = None, limit: int = FILE_LIMIT + ) -> bytes: """Download against one immutable source. Auth refresh never changes the requested snapshot.""" for attempt in range(2): observation = self._current() @@ -244,15 +254,14 @@ def get(self, url, *, source=None, master=None, limit=FILE_LIMIT): response = connection.getresponse() status = response.status if status == 200: - length = response.getheader("Content-Length") - if length is not None and not response.chunked: - if not length.isdecimal(): + header = response.getheader("Content-Length") + length = None + if header is not None and not response.chunked: + if not header.isdecimal(): raise gameapi.GameApiError("JP download has invalid Content-Length") - length = int(length) + length = int(header) if length > limit: raise gameapi.GameApiError("JP download exceeds the size limit") - else: - length = None data = response.read(limit + 1) if len(data) > limit: raise gameapi.GameApiError("JP download exceeds the size limit") @@ -272,11 +281,11 @@ def get(self, url, *, source=None, master=None, limit=FILE_LIMIT): raise AssertionError("unreachable") -def sidecar(path): +def sidecar(path: str | os.PathLike[str]) -> Path: return Path(str(path) + ".source.json") -def read_source(path, data): +def read_source(path: str | os.PathLike[str], data: bytes) -> Source: try: doc = json.loads(sidecar(path).read_bytes()) if doc["catalog_sha256"] != hashlib.sha256(data).hexdigest(): @@ -286,15 +295,21 @@ def read_source(path, data): raise ConfigError("JP catalog needs its matching .source.json sidecar (written by `catalog fetch`)") from None -def write_source(path, data, source): - from ..core.cache import write_atomic - +def write_source(path: str | os.PathLike[str], data: bytes, source: Source) -> None: doc = {"catalog_sha256": hashlib.sha256(data).hexdigest(), "source": source.to_dict()} write_atomic(sidecar(path), (json.dumps(doc, sort_keys=True) + "\n").encode()) -def open_catalog(cfg, region, *, catalog_file=None, bundle_key=None, apk=None): - +def open_catalog( + cfg: Config, + region: str, + *, + catalog_file: str | os.PathLike[str] | None = None, + bundle_key: BundleKey | Callable[[], BundleKey] | None = None, + apk: Path | None = None, +) -> Catalog: + """The catalog of the JP region `region`: `catalog_file` with its .source.json sidecar, else the catalog of the + asset version the API serves now (from the cache, downloaded on first use).""" client = Client(cfg, region) if catalog_file is not None: data = Path(catalog_file).read_bytes() diff --git a/src/nnnotes/source/languages.py b/src/nnnotes/source/languages.py index 50c7a1c..2f91fd3 100644 --- a/src/nnnotes/source/languages.py +++ b/src/nnnotes/source/languages.py @@ -6,6 +6,9 @@ from __future__ import annotations +from collections.abc import Mapping +from typing import Any + from .config import ConfigError # code -> (Fwk.Localization.LanguageMode, text table column) @@ -35,6 +38,6 @@ def column(code: str) -> str: return LANGUAGES[check(code)][1] -def texts(row: dict | None) -> dict: +def texts(row: Mapping[str, Any] | None) -> dict[str, Any]: """{code: text} of a text table row in every language (None for a column the row lacks).""" return {code: (row or {}).get(col) for code, (_, col) in LANGUAGES.items()} diff --git a/src/nnnotes/source/master.py b/src/nnnotes/source/master.py index 3c79c1b..f756f43 100644 --- a/src/nnnotes/source/master.py +++ b/src/nnnotes/source/master.py @@ -23,13 +23,17 @@ import time import urllib.error import urllib.request +from collections.abc import Callable from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field from pathlib import Path +from typing import Any import numpy as np from ..core import jsonio +from ..core.cache import write_atomic +from ..core.jsonio import is_array, is_object __all__ = ["DownloadError", "MasterKey", "decode", "decode_files", "download", "has_row", "table"] @@ -46,7 +50,7 @@ class MasterKey: key: bytes = field(repr=False) iv: bytes = field(repr=False) - def __post_init__(self): + def __post_init__(self) -> None: if len(self.key) != 32 or len(self.iv) != 32: raise ValueError("master key and IV must be 32 bytes each") @@ -131,7 +135,7 @@ def decrypt_blocks(ct: np.ndarray, rk: np.ndarray) -> np.ndarray: s ^= rk[rnd] # AddRoundKey if rnd: # InvMixColumns a0, a1, a2, a3 = s[:, 0, :], s[:, 1, :], s[:, 2, :], s[:, 3, :] - s = np.stack( + s = np.stack( # pyright: ignore[reportUnknownMemberType] (numpy stubs: partly untyped overloads) [ MUL[14][a0] ^ MUL[11][a1] ^ MUL[13][a2] ^ MUL[9][a3], MUL[9][a0] ^ MUL[14][a1] ^ MUL[11][a2] ^ MUL[13][a3], @@ -147,9 +151,10 @@ def decrypt_cbc(data: bytes, key: bytes, iv: bytes, rk: np.ndarray | None = None """Rijndael-256 CBC decryption, padding kept.""" if not data or len(data) % BLOCK: raise ValueError(f"ciphertext length {len(data)} is not a positive multiple of {BLOCK}") - ct = np.frombuffer(data, dtype=np.uint8).reshape(-1, BLOCK) + # pyright: ignore below: numpy stubs give frombuffer and vstack partly untyped overloads + ct = np.frombuffer(data, dtype=np.uint8).reshape(-1, BLOCK) # pyright: ignore[reportUnknownMemberType] pt = decrypt_blocks(ct, round_keys(key) if rk is None else rk) - pt ^= np.vstack([np.frombuffer(iv, dtype=np.uint8)[None, :], ct[:-1]]) + pt ^= np.vstack([np.frombuffer(iv, dtype=np.uint8)[None, :], ct[:-1]]) # pyright: ignore[reportUnknownMemberType] return pt.tobytes() @@ -168,13 +173,13 @@ def table_json(text: bytes) -> str: return jsonio.dumps(json.loads(text.decode("utf-8")), ensure_ascii=False, indent=1) -def decode_files(files: list[Path], out_dir: Path, key: MasterKey, workers: int = 8) -> dict: +def decode_files(files: list[Path], out_dir: Path, key: MasterKey, workers: int = 8) -> dict[str, Any]: """Decode master data files into /.json. Returns {decoded, failed: [{file, error}]}.""" out_dir = Path(out_dir) out_dir.mkdir(parents=True, exist_ok=True) rk = round_keys(key.key) - def one(p: Path): + def one(p: Path) -> dict[str, str] | None: try: text = table_json(decode(p.read_bytes(), key, rk)) except Exception as e: # report and continue with the other tables @@ -187,12 +192,12 @@ def one(p: Path): return {"decoded": len(files) - len(failed), "failed": failed, "out": str(out_dir)} -def table(master_dir: Path, name: str) -> list[dict]: +def table(master_dir: Path, name: str) -> list[dict[str, Any]]: """The rows of a decoded master table: `_allData` of /.json.""" return json.loads((Path(master_dir) / f"{name}.json").read_text(encoding="utf-8"))["_allData"] -def has_row(master_dir: Path, name: str, row_id) -> bool: +def has_row(master_dir: Path, name: str, row_id: object) -> bool: """Whether the decoded master table `name` has a row whose `_id` is `row_id`.""" return any(r.get("_id") == row_id for r in table(master_dir, name)) @@ -231,24 +236,32 @@ def _get(url: str, retries: int = 3, timeout: int = 60) -> bytes: raise AssertionError("unreachable") -def download(cdn: str, version: str, out_dir: Path, workers: int = 16, *, get=None, strict=False) -> dict: +def download( + cdn: str, + version: str, + out_dir: Path, + workers: int = 16, + *, + get: Callable[[str], bytes] | None = None, + strict: bool = False, +) -> dict[str, Any]: """Master data `version` from the CDN into / (the manifest and every listed file, SHA-256 checked; files already present with the right hash are kept).""" out_dir = Path(out_dir) out_dir.mkdir(parents=True, exist_ok=True) base = f"{cdn.rstrip('/')}/master/{version}" - get = get or _get - manifest_raw = get(f"{base}/MasterManifest.json") + fetch = get or _get + manifest_raw = fetch(f"{base}/MasterManifest.json") manifest = json.loads(manifest_raw.decode("utf-8")) - if strict and (not isinstance(manifest, dict) or manifest.get("version") != version): + if strict and (not is_object(manifest) or manifest.get("version") != version): raise DownloadError("master manifest version differs from the requested version") files = manifest.get("files", []) if strict: - if not isinstance(files, list) or not files: + if not is_array(files) or not files: raise DownloadError("JP master manifest has no file list") - seen = set() + seen: set[str] = set() for entry in files: - if not isinstance(entry, dict): + if not is_object(entry): raise DownloadError("invalid JP master manifest entry") name = entry.get("name") if ( @@ -263,11 +276,9 @@ def download(cdn: str, version: str, out_dir: Path, workers: int = 16, *, get=No seen.add(name) else: # Preserve the original partial-download receipt even if a worker raises. - from ..core.cache import write_atomic - write_atomic(out_dir / "MasterManifest.json", manifest_raw) - def one(f: dict): + def one(f: dict[str, Any]) -> dict[str, str] | str: name, sha = f["name"], (f.get("hash") or "").lower() if "/" in name or "\\" in name or name in ("", ".", ".."): return {"file": name, "error": "unexpected file name"} @@ -279,7 +290,7 @@ def one(f: dict): and hashlib.sha256(dst.read_bytes()).hexdigest() == sha ): return "kept" - data = get(f"{base}/{name}") + data = fetch(f"{base}/{name}") if strict and (len(data) != f.get("size") or not sha): return {"file": name, "error": "size or hash missing/different from the manifest"} if sha and hashlib.sha256(data).hexdigest() != sha: @@ -292,8 +303,6 @@ def one(f: dict): with ThreadPoolExecutor(max(1, workers)) as ex: results = list(ex.map(one, files)) if not strict or not any(isinstance(r, dict) for r in results): - from ..core.cache import write_atomic - write_atomic(out_dir / "MasterManifest.json", manifest_raw) return { "version": manifest.get("version", version),