kuhyx/testsAndMisc
1
0
Fork 0
mirror of https://github.com/kuhyx/testsAndMisc.git synced 2026-11-10 20:07:04 +01:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

Split wsg_grabber/catalog.py under the 250-line cap

Browse Source 
catalog.py was 251 lines. Its five defensive accessors for the loosely-typed
/wsg/ JSON -- _get, _as_list, _as_int, _as_md5 and the _MD5_SHAPE / int64
range constants they depend on -- form a cohesive unit distinct from the
parsing logic above them, so they move to a sibling _coerce.py and become
module-public there.

No caller or test referenced the private helpers (downloader.py uses only the
six public parse functions), so no re-export shim was needed and the public
API is byte-for-byte unchanged.

catalog.py 251 -> 175, new _coerce.py 87. Both modules stay at 100% branch
coverage; 277 wsg_grabber tests pass.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011w2Pj4Ek3ZNMQVYcFHgXDY
This commit is contained in:
Krzysztof Rudnicki 2026-08-16 15:36:02 +02:00
parent d6b376effb
commit 77c6d9f815
3 changed files with 146 additions and 97 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
docs/superpowers/evidence/wsg-grabber-catalog-split-2026-08-16.json Normal file
View File

@ -0,0 +1,38 @@
{
"intent": "Bring python_pkg/wsg_grabber/catalog.py under the 250-line file cap by extracting its defensive JSON coercion helpers into a sibling module, without changing any public behaviour.",
"scope": [
"python_pkg/wsg_grabber/catalog.py",
"python_pkg/wsg_grabber/_coerce.py (new)",
"Non-goal: no behaviour change, no test changes, no public API change"
],
"changes": [
"Extracted _get/_as_list/_as_int/_as_md5 plus the _MD5_SHAPE and _INT64_MIN/_INT64_MAX constants into python_pkg/wsg_grabber/_coerce.py as module-public get/as_list/as_int/as_md5.",
"catalog.py now imports those four helpers from _coerce; the six public functions (media_url, parse_thread_list, parse_archive, parse_thread, deleted_md5s, new_files) keep identical signatures and behaviour.",
"catalog.py 251 -> 175 lines; _coerce.py is 87 lines. Both under the 250-line cap."
],
"verification": [
{
"command": "python3 -m pytest python_pkg/wsg_grabber/tests/ -q --cov=python_pkg.wsg_grabber --cov-report=term-missing",
"result": "pass",
"evidence": "277 passed in 15.02s. catalog.py 61 stmts / 30 branches 100%; new _coerce.py 20 stmts / 6 branches 100%. Every wsg_grabber module at 100%, so no coverage was lost by the move and the new module is measured rather than silently skipped."
},
{
"command": "pre-commit run --files python_pkg/wsg_grabber/catalog.py python_pkg/wsg_grabber/_coerce.py",
"result": "pass",
"evidence": "All hooks Passed, including ruff, ruff format, mypy, pylint, bandit, and 'pytest with coverage enforcement'. No noqa or suppression was added."
},
{
"command": "grep -rn 'catalog' --include='*.py' python_pkg/",
"result": "pass",
"evidence": "Only importer is downloader.py, which calls the public functions parse_thread_list/parse_archive/parse_thread/new_files/deleted_md5s/media_url. No caller and no test referenced the extracted private helpers, so no re-export shim was required."
}
],
"risks": [
"A formatter stripped the new _coerce import once mid-edit (it ran while call sites still used the old _as_* names); the import was restored and re-verified by grep before the gate was run.",
"Low residual risk: the change is a pure move of private helpers with no signature change, covered by 277 existing tests at 100% branch coverage."
],
"rollback": [
"git revert the commit; catalog.py returns to its 251-line single-file form and _coerce.py is deleted.",
"After rollback, run python3 -m pytest python_pkg/wsg_grabber/tests/ -q to confirm 277 tests still pass."
]
}

87
python_pkg/wsg_grabber/_coerce.py Normal file
View File

@ -0,0 +1,87 @@
"""Defensive accessors for the loosely-typed /wsg/ JSON payloads.
The API omits file fields entirely on posts without an attachment and is fed by
an anonymous imageboard, so nothing here trusts the shape of what it is handed:
every accessor answers "not usable" rather than raising.
"""
from __future__ import annotations
import re
from typing import Any
# base64 of a 16-byte digest is always 22 characters plus "==". Matching the
# exact shape -- not merely the length -- is what keeps a NUL byte out of a
# filename: one such value poisons the index permanently, because the crash
# happens before the row can ever be marked gone.
_MD5_SHAPE = re.compile(r"[A-Za-z0-9+/]{22}==")
# sqlite stores signed 64-bit integers; anything wider raises OverflowError deep
# inside executemany, which used to take the worker thread down with it.
_INT64_MIN = -(2**63)
_INT64_MAX = 2**63 - 1
def get(container: object, key: str) -> object:
"""Read *key* from *container* when it is a mapping.
Args:
container: Anything; only dicts yield a value.
key: Key to read.
Returns:
object: The value, or None.
"""
if isinstance(container, dict):
return container.get(key)
return None
def as_list(value: object) -> list[Any]:
"""Return *value* when it is a list, else an empty list.
Args:
value: Candidate.
Returns:
list[Any]: Safe list to iterate.
"""
return value if isinstance(value, list) else []
def as_int(value: object) -> int | None:
"""Coerce *value* to int when it is a real integer.
Booleans are rejected: ``True`` is an int in Python but never a valid post
number or timestamp. Values outside sqlite's signed 64-bit range are
rejected too -- they raise OverflowError on insert, which is not in the
worker's recoverable set and would kill it silently.
Args:
value: Candidate.
Returns:
int | None: The integer, or None.
"""
if isinstance(value, bool) or not isinstance(value, int):
return None
return value if _INT64_MIN <= value <= _INT64_MAX else None
def as_md5(value: object) -> str | None:
"""Validate the API's base64 md5 field.
The charset check is not decoration: the tail of this value ends up in a
filename via ``store.local_name``, and it arrives from an anonymous
imageboard. Length alone would let through a NUL byte, which raises deep
inside the filesystem call instead of being rejected here.
Args:
value: Candidate.
Returns:
str | None: The 24-character digest, or None when malformed.
"""
if not isinstance(value, str):
return None
return value if _MD5_SHAPE.fullmatch(value) else None

118
python_pkg/wsg_grabber/catalog.py
View File

@ -7,26 +7,15 @@ wrong-typed field as "not a video" rather than trusting the shape.
from __future__ import annotations
import re
from typing import TYPE_CHECKING, Any
from typing import TYPE_CHECKING
from python_pkg.wsg_grabber._coerce import as_int, as_list, as_md5, get
from python_pkg.wsg_grabber.constants import MEDIA_HOST, VIDEO_EXTENSIONS
from python_pkg.wsg_grabber.models import RemoteFile, ThreadRef
if TYPE_CHECKING:
from collections.abc import Iterable, Sequence
# base64 of a 16-byte digest is always 22 characters plus "==". Matching the
# exact shape -- not merely the length -- is what keeps a NUL byte out of a
# filename: one such value poisons the index permanently, because the crash
# happens before the row can ever be marked gone.
_MD5_SHAPE = re.compile(r"[A-Za-z0-9+/]{22}==")
# sqlite stores signed 64-bit integers; anything wider raises OverflowError deep
# inside executemany, which used to take the worker thread down with it.
_INT64_MIN = -(2**63)
_INT64_MAX = 2**63 - 1
def media_url(tim: int, ext: str) -> str:
"""Return the CDN address of an attachment.
@ -51,15 +40,15 @@ def parse_thread_list(payload: object) -> list[ThreadRef]:
list[ThreadRef]: Every advertised thread, in page order.
"""
refs: list[ThreadRef] = []
for page in _as_list(payload):
for entry in _as_list(_get(page, "threads")):
thread_no = _as_int(_get(entry, "no"))
for page in as_list(payload):
for entry in as_list(get(page, "threads")):
thread_no = as_int(get(entry, "no"))
if thread_no is None:
continue
refs.append(
ThreadRef(
thread_no=thread_no,
api_last_modified=_as_int(_get(entry, "last_modified")) or 0,
api_last_modified=as_int(get(entry, "last_modified")) or 0,
),
)
return refs
@ -78,8 +67,8 @@ def parse_archive(payload: object) -> list[ThreadRef]:
list[ThreadRef]: Archived threads.
"""
refs: list[ThreadRef] = []
for entry in _as_list(payload):
thread_no = _as_int(entry)
for entry in as_list(payload):
thread_no = as_int(entry)
if thread_no is not None:
refs.append(ThreadRef(thread_no=thread_no, api_last_modified=0))
return refs
@ -99,7 +88,7 @@ def parse_thread(thread_no: int, payload: object) -> list[RemoteFile]:
list[RemoteFile]: Attachments with a video extension.
"""
files: list[RemoteFile] = []
for post in _as_list(_get(payload, "posts")):
for post in as_list(get(payload, "posts")):
parsed = _parse_post(thread_no, post)
if parsed is not None:
files.append(parsed)
@ -116,10 +105,10 @@ def deleted_md5s(payload: object) -> set[str]:
set[str]: Identities that should be written off rather than retried.
"""
gone: set[str] = set()
for post in _as_list(_get(payload, "posts")):
if not _as_int(_get(post, "filedeleted")):
for post in as_list(get(payload, "posts")):
if not as_int(get(post, "filedeleted")):
continue
digest = _as_md5(_get(post, "md5"))
digest = as_md5(get(post, "md5"))
if digest is not None:
gone.add(digest)
return gone
@ -162,90 +151,25 @@ def _parse_post(thread_no: int, post: object) -> RemoteFile | None:
Returns:
RemoteFile | None: None when the post has no video attachment.
"""
if _as_int(_get(post, "filedeleted")):
if as_int(get(post, "filedeleted")):
return None
tim = _as_int(_get(post, "tim"))
ext = _get(post, "ext")
digest = _as_md5(_get(post, "md5"))
post_no = _as_int(_get(post, "no"))
tim = as_int(get(post, "tim"))
ext = get(post, "ext")
digest = as_md5(get(post, "md5"))
post_no = as_int(get(post, "no"))
if tim is None or post_no is None or digest is None:
return None
if not isinstance(ext, str) or ext not in VIDEO_EXTENSIONS:
return None
name = _get(post, "filename")
name = get(post, "filename")
return RemoteFile(
md5=digest,
tim=tim,
ext=ext,
orig_name=name if isinstance(name, str) else str(tim),
fsize=_as_int(_get(post, "fsize")) or 0,
width=_as_int(_get(post, "w")) or 0,
height=_as_int(_get(post, "h")) or 0,
fsize=as_int(get(post, "fsize")) or 0,
width=as_int(get(post, "w")) or 0,
height=as_int(get(post, "h")) or 0,
thread_no=thread_no,
post_no=post_no,
)
def _get(container: object, key: str) -> object:
"""Read *key* from *container* when it is a mapping.
Args:
container: Anything; only dicts yield a value.
key: Key to read.
Returns:
object: The value, or None.
"""
if isinstance(container, dict):
return container.get(key)
return None
def _as_list(value: object) -> list[Any]:
"""Return *value* when it is a list, else an empty list.
Args:
value: Candidate.
Returns:
list[Any]: Safe list to iterate.
"""
return value if isinstance(value, list) else []
def _as_int(value: object) -> int | None:
"""Coerce *value* to int when it is a real integer.
Booleans are rejected: ``True`` is an int in Python but never a valid post
number or timestamp. Values outside sqlite's signed 64-bit range are
rejected too -- they raise OverflowError on insert, which is not in the
worker's recoverable set and would kill it silently.
Args:
value: Candidate.
Returns:
int | None: The integer, or None.
"""
if isinstance(value, bool) or not isinstance(value, int):
return None
return value if _INT64_MIN <= value <= _INT64_MAX else None
def _as_md5(value: object) -> str | None:
"""Validate the API's base64 md5 field.
The charset check is not decoration: the tail of this value ends up in a
filename via ``store.local_name``, and it arrives from an anonymous
imageboard. Length alone would let through a NUL byte, which raises deep
inside the filesystem call instead of being rejected here.
Args:
value: Candidate.
Returns:
str | None: The 24-character digest, or None when malformed.
"""
if not isinstance(value, str):
return None
return value if _MD5_SHAPE.fullmatch(value) else None