diff --git a/CHANGELOG.md b/CHANGELOG.md index d8a9229..8e9be16 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,17 @@ first release is 3.0.0 because exit codes and the cookie store changed in ways a ## Unreleased +### Added +- `mam/accept_unsnatched` (default off): when the MAM file-name searches return rows but none marked + `my_snatched`, accept the one row that has the release's file type, one of its authors and the same title (same + numbers, subtitle and production: no sibling volume, part, box set or dramatisation); two or more are ambiguous + and none is used. Seen 2026-10-07 with *Before She Knew Him* (Peter Swanson) and every other recent MAM-pass + miss: both file-name searches returned the release's own torrent, but booktree ran minutes after the download + and MAM had not marked it `my_snatched` yet, so the release was left unmatched in the MAM pass. No extra MAM + request: the rows come from the searches booktree already sends, whose text and cache keys are unchanged. + stdout: `No snatched MAM match; using the only unsnatched one that passes the checks: ...`; JSON log: + `mam_attempt: "unsnatched"`. See CONFIG.md, Unsnatched MAM matches. + ## 3.0.5 - 2026-10-06 ### Fixed diff --git a/CONFIG.md b/CONFIG.md index 04a3090..0e74f23 100644 --- a/CONFIG.md +++ b/CONFIG.md @@ -64,6 +64,7 @@ A copy of default_config.cfg can be found under the /templates folder. It is re | cache/mam_empty_hours | | How long an empty MAM answer (or one without a snatched entry) is reused | 24 | | mam/min_interval_seconds | | Minimum spacing between HTTP requests to MAM (cache hits are free) | 6 | | mam/max_queries_per_run | | Runaway guard: maximum MAM searches in one run (one container invocation); further searches are skipped with a message and those releases stay unmatched until the next run (0 = unlimited). The 6-second spacing is the real safety net; this only stops a loop | 3000 | +| mam/accept_unsnatched | | 1 = when the MAM file-name searches return rows but none marked as snatched by you, accept the one row that passes the file type, author and title checks (no extra MAM request; see [Unsnatched MAM matches](#unsnatched-mam-matches-mamaccept_unsnatched-default-off)) | 0 | | json_log | | Path of a JSON-lines run log, or `true` for `booktree_log_.jsonl` next to the CSV; same as `--json-log` | | | refresh | | List of releases (name or path) to re-process ignoring cached answers and the processed marker; same as `--refresh` | | | pins | | List of `RELEASE=ASIN` strings: use that Audible ASIN for the release and re-process it now; same as `--pin` (see [Correcting a match](#correcting-a-match)) | | @@ -320,6 +321,45 @@ apart, and afterwards once per TTL. its "already processed" marker are ignored while everything else is served from cache. `--no-cache` still does this for the whole run. +### Unsnatched MAM matches (`mam/accept_unsnatched`, default off) + +The MAM pass searches MAM's file-name index for the release's file name (with its authors, then without) and keeps +only torrents MAM marks as snatched by you (`my_snatched`), so the match is the torrent you downloaded. booktree +usually runs minutes after a download completes (a download-complete hook), and MAM sets that mark minutes to hours +later: the search finds the release's own torrent, not yet marked, and the release stays unmatched in the MAM pass +(seen with every MAM-pass miss in October 2026, such as *Before She Knew Him* by Peter Swanson). + +With `"mam": {"accept_unsnatched": 1}`, when neither file-name search returned a snatched row but they did return +rows, booktree looks at those same rows (no extra MAM request; same cache) and accepts one when it is the only row, +by MAM id, that passes all of these checks: + +* the release's file type (`m4b`, `mp3`, ...) is among the row's file types; +* one of the release's authors (id3, or parsed from the release name where the tags are junk) is one of the row's + authors, by the same name comparison used for snatched rows; +* the same title: main titles with a token-sort ratio of 90 or more and the same numbers (also written out, as roman + numerals or as "Book N"); the same production (abridged, dramatised, GraphicAudio, full cast); matching subtitles + when both have one; no subtitle that only the release has; and a subtitle that only MAM has only when it is the + name of that torrent's own series and names no bundle or part; +* the same series position: a "Book N" written in one title must appear in the other title or among the other + side's series parts (MAM's `series_info` for the row, the release's own series tags), otherwise the row is refused. + +So *The Viscount and the Witch* is not *The Witch*, *Cradle: Soulsmith* is not *Cradle: Unsouled*, *Part 1* is not +*Part 2*, *Volume II* is not *Volume I*, *Dune: The Complete Saga* and *Dune: Part One* are not *Dune*, *Mistborn: +Secret History* is not *Mistborn*, *Mother of Learning, Book 2* is not a row *Mother of Learning* that MAM lists as +part 3, while *Leviathan Wakes* (tagged The Expanse #1) is the row *Leviathan Wakes: The Expanse, Book 1* in the +series The Expanse. + +Two or more passing rows are ambiguous and none is used; a snatched row is always used as before. MAM rows carry no +runtime, so these checks are the only protection: in `mam-audible` the Audible search that follows is built from +the MAM match's title and authors, and the Audible runtime there is a preference, not a rejection, so a wrong MAM +match would carry through. Nothing is checked with `flags/verbose` 0 and `flags/ebooks` 0, where booktree does +not rank MAM results at all. stdout says `No snatched MAM match; using the only unsnatched one that passes the checks: by +<authors>`; the JSON log marks such a book with `mam_attempt: "unsnatched"` (and `match.attempt` is +`mam-unsnatched` when the MAM record is the one filed). + +It is off by default because it accepts a MAM torrent not marked as yours, judged by file type, author and title +alone. Turn it on for the config that uses `mam` or `mam-audible` if booktree runs right after downloads complete. + ## JSON run log (`--json-log [PATH]`) In addition to the CSV, one JSON object per line: a `book` record per processed release and a final `run` record. @@ -344,7 +384,7 @@ The CSV columns, file name and stdout are unchanged; the JSON log is the structu ~~~ `match.attempt` says what produced the match: `pinned`, `candidates`, `parsed`, `parsed-authors`, `swapped`, -`title-only`, `legacy` (upstream's own search), `mam`, or `log` (taken from the input log in `log` mode). +`title-only`, `legacy` (upstream's own search), `mam`, `mam-unsnatched` (the MAM match is a torrent not yet marked snatched, `mam/accept_unsnatched`; see also `mam_attempt`), or `log` (taken from the input log in `log` mode). `queries[].cache_key` is the hash printed in `Checking cache: <kind>/<hash>`; a query that failed carries `error`, one skipped by the MAM budget carries `skipped`. `id3.duration_s` is the first file's duration while `expected_duration_min` is the whole release. One `run` record is written per `paths` entry. Give the path as diff --git a/myx_classes.py b/myx_classes.py index 5e2d4da..9c84a66 100644 --- a/myx_classes.py +++ b/myx_classes.py @@ -16,11 +16,140 @@ import myx_hints import myx_names import copy +import html +from thefuzz import fuzz #Module variables authMode="login" verbose=False +#Accepting a MAM torrent not marked my_snatched (Config/mam/accept_unsnatched, MAMBook.pickUnsnatched) +MAM_TITLE_MIN = 90 # token-sort ratio for the main titles, and for the subtitles when both have one +MAM_MAX_TITLE = 200 # untrusted titles (id3, MAM rows) are cut before the title regexes run +_SUBTITLE_SPLIT = re.compile(r"\s*:\s*|\s+[-\u2013\u2014]\s+") +_SERIES_POSITION = re.compile(r"\bbook\s*\d+(?:\.\d+)?\b", re.IGNORECASE) +# a subtitle that only MAM has and that names a bundle or a part, not this book ("Dune: The Complete Saga", +# "Dune: Part One"); checked on the raw subtitle, before noise words such as "complete" are removed +_BUNDLE_WORDS = re.compile(r"\b(?:complete|saga|collection|box\s*set|boxed|omnibus|trilogy|series|books|volumes?|" + r"vol|bundle|part|chapter|episode|season)\b", re.IGNORECASE) +# numbers that tell volumes apart, written out or as roman numerals ("Volume II", "Part One") +_NUMBER_WORDS = {w: str(n) for n, w in enumerate( + "zero one two three four five six seven eight nine ten eleven twelve thirteen fourteen fifteen sixteen seventeen " + "eighteen nineteen twenty".split())} +_NUMBER_WORDS.update({w: str(n) for n, w in enumerate( + "_ i ii iii iv v vi vii viii ix x xi xii xiii xiv xv xvi xvii xviii xix xx".split()) if n}) +_EDITION = re.compile(r"(?<!un)\babridged\b|\bdramati[sz]ed\b|\bdramati[sz]ation\b|\bgraphic\s?audio\b|\bfull\s?cast\b", + re.IGNORECASE) +_NUMBER_WORDS.update({"first": "1", "second": "2", "third": "3", "fourth": "4", "fifth": "5"}) + + +def _titleWords(text): + """Lower-case words for comparing titles: HTML entities decoded (MAM rows carry ' and &), accents and + apostrophes removed ("Ender's" = "Enders", "Shōgun" = "Shogun"), noise words (Unabridged, m4b, ...) and other + punctuation removed.""" + t = html.unescape(str(text or "")).replace("&", " and ") + t = myx_utilities.strip_accents(re.sub(r"['\u2019]", "", t)) + t = myx_names.NOISE_WORD.sub(" ", t) + return " ".join(re.sub(r"[^\w\s]|_", " ", t).lower().split()) + + +def _numbers(words): + """The numbers in a _titleWords string: digits, number words and roman numerals ("i" counts: "Volume I").""" + return sorted(w if w.isdigit() else _NUMBER_WORDS[w] for w in words.split() if w.isdigit() or w in _NUMBER_WORDS) + + +def _clip(title): + """html-unescaped, whitespace collapsed and cut to 2 x MAM_MAX_TITLE: id3 tags and MAM rows are untrusted, + and the title regexes are only linear on bounded, collapsed input.""" + return " ".join(html.unescape(str(title or ""))[:MAM_MAX_TITLE * 4].split())[:MAM_MAX_TITLE * 2] + + +def _edition(title): + """Markers of a different production of the same title: abridged (not unabridged), dramatised, GraphicAudio.""" + return sorted(set(m.lower().replace(" ", "")[:6] for m in _EDITION.findall(title))) + + +def _titleParts(title): + """(main title, subtitle, raw subtitle): the first two as _titleWords with "Book N" series positions dropped and a + "a novel" style subtitle emptied; the raw subtitle only html-unescaped.""" + t = _clip(title) + t = re.sub(r"\s*\((?:un)?abridged\)", "", t, flags=re.IGNORECASE) + parts = _SUBTITLE_SPLIT.split(t, maxsplit=1) + raw = parts[1] if len(parts) > 1 else "" + main = _titleWords(_SERIES_POSITION.sub(" ", parts[0])) + sub = _titleWords(_SERIES_POSITION.sub(" ", raw)) + if myx_names.SUBTITLE_NOISE.match(sub): + sub, raw = "", "" + return main, sub, raw + + +def sameMamTitle(ours, theirs, series=()): + """The title check for an unsnatched MAM row (pickUnsnatched): the same book, not a sibling volume, a box set or another part. + Main titles must agree (token-sort ratio >= MAM_TITLE_MIN) and carry the same numbers, also written out + or in roman numerals ("Part 1" is not "Part 2", "Volume II" is not "Volume I"); when both have a subtitle those + must agree too ("Cradle: Unsouled" is not "Cradle: Soulsmith"); a subtitle only we have is refused ("Thrawn: + Treason" is not "Thrawn"); one only MAM has is accepted only when it is the name of the candidate's own series + (`series`, from MAM's series_info) and names no bundle or part: "Leviathan Wakes: The Expanse, Book 1" in series + The Expanse yes; "Dune: The Complete Saga", "Dune: Part One", "Mistborn: Secret History" no.""" + om, osub, _ = _titleParts(ours) + tm, tsub, traw = _titleParts(theirs) + if not om or not tm or fuzz.token_sort_ratio(om, tm) < MAM_TITLE_MIN: + return False + # "Book 2" is not "Book 3" (the positions are dropped from the words compared below), and an abridged, dramatised + # or GraphicAudio production is not the plain one + position = seriesPosition + if position(ours) != position(theirs) and position(ours) and position(theirs): + return False + if _edition(_clip(ours)) != _edition(_clip(theirs)): + return False + if _numbers(f"{om} {osub}") != _numbers(f"{tm} {tsub}"): + return False + if osub and tsub: + return fuzz.token_sort_ratio(osub, tsub) >= MAM_TITLE_MIN + if osub: + return False + if _BUNDLE_WORDS.search(traw): # also when noise-word removal left nothing ("Dune: Complete") + return False + if not tsub: + return True + names = [_titleWords(_clip(n)) for n in series or ()] + return any(n and fuzz.token_set_ratio(n, tsub) >= MAM_TITLE_MIN for n in names) + + +def _num(part): + """A series part as a comparable string: "01", "1.0" and 1 are all "1"; "" when there is none.""" + try: + return format(float(str(part).strip()), "g") if str(part).strip() else "" + except ValueError: + return str(part).strip().lower() + + +def seriesPosition(title): + """The "Book N" numbers written in a title, as a sorted list of _num strings.""" + return sorted(_num(n) for n in re.findall(r"\d+(?:\.\d+)?", " ".join(_SERIES_POSITION.findall(_clip(title))))) + + +def samePosition(ours, theirs): + """Series position check for an unsnatched MAM row (pickUnsnatched), between two Books: a "Book N" written in one + title must be matched by the other's title or, failing that, by one of its series parts; when the other side has + no position at all, the row is refused ("Mother of Learning, Book 2" is not a row "Mother of Learning" that MAM + files as part 3, nor one with no part).""" + def positions(b): + return set(seriesPosition(b.title)), {_num(s.part) for s in b.series if _num(s.part)} + ot, op = positions(ours) + tt, tp = positions(theirs) + if ot and not tt: + return bool(ot & tp) + if tt and not ot: + return bool(tt & op) + return True # both titles carry positions (sameMamTitle compared them) or neither does + + +def _oneLine(text): + """An untrusted field (id3 tag, MAM row) for a one-line message: whitespace and control characters collapsed.""" + return " ".join("".join(c if c.isprintable() else " " for c in str(text or "")).split())[:MAM_MAX_TITLE] + + #Author and Narrator Classes @dataclass class Contributor: @@ -483,6 +612,7 @@ class MAMBook: parsedName:dict=None refresh:bool=False matchAttempt:str="" + mamAttempt:str="" def getRunTimeLength(self): #add all the duration of the files in the book, and convert into minutes @@ -1105,14 +1235,25 @@ def getMAMBooks(self, cfg, bookFile:BookFile): # Search using book key and authors (using or search in case the metadata is bad) print(f"Searching MAM for\n\tTitleFilename: {title}\n\tauthors:{authors}") - books=myx_mam.getMAMBook(cfg, titleFilename=title, authors=authors, extension=extension, refresh=self.refresh) + # Config/mam/accept_unsnatched: also keep the rows not marked my_snatched (no extra request; pickUnsnatched) + pool = [] if myx_mam.acceptUnsnatched(cfg) else None + keep = {} if pool is None else {"unsnatched": pool} + books=myx_mam.getMAMBook(cfg, titleFilename=title, authors=authors, extension=extension, refresh=self.refresh, **keep) # was the author inaccurate? (Maybe it was LastName, FirstName or accented) # print (f"Trying again because Filename, Author = {len(self.mamMatches)}") if len(books) == 0: #try again, without author this time print(f"Widening MAM search using just\n\tTitleFilename: {title}") - books=myx_mam.getMAMBook(cfg, titleFilename=title, extension=extension, refresh=self.refresh) + books=myx_mam.getMAMBook(cfg, titleFilename=title, extension=extension, refresh=self.refresh, **keep) + + # neither file-name search found a snatched torrent, but they did return rows: accept the one that passes the + # checks, if there is exactly one (pickUnsnatched); the ranking below then treats it like a snatched one + # (not when the ranking below does not run: with verbose off it never does, an upstream quirk) + fromUnsnatched = False + if len(books) == 0 and pool and (verbose or ebooks): + books = self.pickUnsnatched(cfg, rankBook if rankBook is not None else self.ffprobeBook, bookFile, pool) + fromUnsnatched = bool(books) #Find the best match self.mamMatches = books @@ -1156,8 +1297,8 @@ def getMAMBooks(self, cfg, bookFile:BookFile): targetBook = '|'.join([book.title, book.getAuthors(), book.getSeriesParts()]) for abook in books: - #if this book is snatched, include in the match - if abook.snatched: + #if this book is snatched, include in the match (an accepted unsnatched one passed pickUnsnatched) + if abook.snatched or fromUnsnatched: #the author is known, check if this book is this authors book #otherwise, if maybe this title is close enough #print (f"{abook.title} by {abook.authors}...") @@ -1189,12 +1330,53 @@ def getMAMBooks(self, cfg, bookFile:BookFile): self.bestMAMMatch = books[0] + if fromUnsnatched and self.bestMAMMatch is not None: + self.matchAttempt = "mam-unsnatched" + self.mamAttempt = "unsnatched" + #pprint(self.bestMAMMatch) if (books is not None): return self.bestMAMMatch else: return None + def pickUnsnatched(self, cfg, book, bookFile, pool): + """Config/mam/accept_unsnatched: the file-name searches found rows but none marked my_snatched. That is usually + the release's own torrent: the hook runs booktree minutes after a download and MAM sets my_snatched later. + `pool` holds those rows as (key, Book, file types), from both searches. A row is accepted only when it is the + one row (by MAM id) that has the release's file type, one of its authors (exact, as for snatched rows), the same + title (sameMamTitle: no sibling volume, part, bundle or other production) and the same series position + (samePosition: a "Book N" in one title against the other's title or series part); two or more are ambiguous and + none is used. MAM rows carry no runtime. Returns [the Book] or [].""" + if book is None: + return [] + parsed = self.getParsedName(book, cfg) or {} + title = _clip(book.title) + # a title that only repeats the release name is junk, unless the release name is just the title + junk = myx_names.isJunkTitle(title) or (myx_names.isJunkTitle(title, parsed.get("source") or self.name) and + _titleWords(parsed.get("title") or "") != _titleWords(title)) + if junk or not book.authors or myx_names.isJunkAuthors(book.authors): + print("No snatched MAM match; no usable title and author to check the unsnatched ones") + return [] + ext = bookFile.getExtension().lower() + rows = {} + for key, abook, types in pool: + rows.setdefault(key, (abook, types)) + # the silent checks first: isThisMyAuthorsBook prints the row's title (verbose), and a row that is not ours + # need not reach stdout + passed = [abook for abook, types in rows.values() + if ext and ext in types and sameMamTitle(title, abook.title, [x.name for x in abook.series]) + and samePosition(book, abook) and myx_utilities.isThisMyAuthorsBook(book.authors, abook, cfg)] + if len(passed) == 1: + print(f"No snatched MAM match; using the only unsnatched one that passes the checks: " + f"{_oneLine(passed[0].title)} by {_oneLine(passed[0].getAuthors())}") + return passed + if passed: + print(f"No snatched MAM match; {len(passed)} unsnatched ones pass the checks: ambiguous, not using any") + else: + print(f"No snatched MAM match; no unsnatched one passes the title, author and file type check ({len(rows)} row(s))") + return [] + def getHashKey(self): return myx_utilities.getHash(self.name) diff --git a/myx_jsonlog.py b/myx_jsonlog.py index 2682a95..592c5a7 100644 --- a/myx_jsonlog.py +++ b/myx_jsonlog.py @@ -5,7 +5,7 @@ run, release, release_path, files[{path, duration_s, hardlinked}], id3{asin,title,authors,narrators,duration_s (first file)}, parsed_name{title,authors,series,part,asin} | null, hint | null, pinned_asin, matched, metadata_source, match{asin,title,authors,narrators,series,part,runtime_min,match_rate,attempt} | null, - expected_duration_min, runtime_delta_min | null, target_path, hardlinked, mam_count, audible_count, + expected_duration_min, runtime_delta_min | null, target_path, hardlinked, mam_count, mam_attempt ("unsnatched" | null), audible_count, queries[{kind, cache_key, cached, results, error?, skipped?, asin,title,authors,narrators,keywords | text}] Record (type "run", one per Config/paths entry): run, started_utc, finished_utc, metadata, books, matched, unmatched, hardlinked_files, mam_queries (HTTP searches), mam_queries_skipped (budget), audible_queries, errors, csv, exit_code @@ -105,6 +105,7 @@ def record(book, cfg, run_id): "hardlinked": bool(files) and all(f["hardlinked"] for f in files), "already_filed": getattr(book, "alreadyFiled", None) or None, # Config/dedupe_roots: folder that holds the same files "mam_count": len(book.mamMatches or []), + "mam_attempt": getattr(book, "mamAttempt", "") or None, # "unsnatched": Config/mam/accept_unsnatched "audible_count": len(book.audibleMatches or []), "queries": getattr(book, "queries", []) or [], } diff --git a/myx_mam.py b/myx_mam.py index ab547ab..67985a4 100644 --- a/myx_mam.py +++ b/myx_mam.py @@ -1,6 +1,7 @@ import requests import json import os +import re import time from pprint import pprint import myx_classes @@ -183,56 +184,96 @@ def searchMAM(cfg, titleFilename, authors, extension, refresh=False): return None -def getMAMBook(cfg, titleFilename="", authors="", extension="", refresh=False): +def _rowToBook(b): + """One row of a MAM search answer as a Book (snatched set from my_snatched).""" + book=myx_classes.Book() + book.init() + if 'asin' in b: + book.asin=str(b["asin"]) + if 'title' in b: + book.title=str(b["title"]) + if 'author_info'in b: + #format {id:author, id:author} + if len(b["author_info"]): + authors = json.loads(b["author_info"]) + book.authors.extend(myx_classes.contributors(str(author) for author in authors.values())) + if 'narrator_info'in b: + #format {id:narrator, id:narrator} + if ((not b["narrator_info"] is None) and len(b["narrator_info"])): + narrators = json.loads(b["narrator_info"]) + book.narrators.extend(myx_classes.contributors(str(narrator) for narrator in narrators.values())) + if 'series_info'in b: + #format {"35598": ["Kat Dubois", "5"]} + if ((not b["series_info"] is None) and len(b["series_info"])): + series_info = json.loads(b["series_info"]) + for series in series_info.values(): + # a value that is not a list (null, a number, a bare string that list() would split into + # letters) or one with no name is not a series entry: skip it rather than abort the run + if not isinstance(series, (list, tuple)): + continue + s=list(series) + if not s or s[0] in (None, ""): + continue + seriesName = str(s[0]) + seriesName = seriesName.replace("'", "'") + # a series without a part is ["Name"] or ["Name", null]; s[1] used to IndexError + # and abort the run, or become "None" and file as "Series #None - Title" + part = s[1] if len(s) > 1 else "" + book.series.append(myx_classes.Series(seriesName, part)) + if 'lang_code' in b: + book.language=myx_utilities.getLanguage((b["lang_code"])) + if 'my_snatched' in b: + book.snatched=bool((b["my_snatched"])) + + return book + + +def getMAMBook(cfg, titleFilename="", authors="", extension="", refresh=False, unsnatched=None): + """The snatched torrents of a file-name search, as Books. `unsnatched`, when a list, also receives the other rows + as (MAM id, Book, file types) for Config/mam/accept_unsnatched (MAMBook.pickUnsnatched); a malformed one is skipped.""" books=[] mamBook=searchMAM(cfg, titleFilename, authors, extension, refresh=refresh) if (mamBook is not None): for b in mamBook: - #pprint(b) - book=myx_classes.Book() - book.init() - if 'asin' in b: - book.asin=str(b["asin"]) - if 'title' in b: - book.title=str(b["title"]) - if 'author_info'in b: - #format {id:author, id:author} - if len(b["author_info"]): - authors = json.loads(b["author_info"]) - book.authors.extend(myx_classes.contributors(str(author) for author in authors.values())) - if 'narrator_info'in b: - #format {id:narrator, id:narrator} - if ((not b["narrator_info"] is None) and len(b["narrator_info"])): - narrators = json.loads(b["narrator_info"]) - book.narrators.extend(myx_classes.contributors(str(narrator) for narrator in narrators.values())) - if 'series_info'in b: - #format {"35598": ["Kat Dubois", "5"]} - if ((not b["series_info"] is None) and len(b["series_info"])): - series_info = json.loads(b["series_info"]) - for series in series_info.values(): - # a value that is not a list (null, a number, a bare string that list() would split into - # letters) or one with no name is not a series entry: skip it rather than abort the run - if not isinstance(series, (list, tuple)): - continue - s=list(series) - if not s or s[0] in (None, ""): - continue - seriesName = str(s[0]) - seriesName = seriesName.replace("'", "'") - # a series without a part is ["Name"] or ["Name", null]; s[1] used to IndexError - # and abort the run, or become "None" and file as "Series #None - Title" - part = s[1] if len(s) > 1 else "" - book.series.append(myx_classes.Series(seriesName, part)) - if 'lang_code' in b: - book.language=myx_utilities.getLanguage((b["lang_code"])) - if 'my_snatched' in b: - book.snatched=bool((b["my_snatched"])) - + if unsnatched is not None and isinstance(b, dict) and not b.get("my_snatched"): + try: + #key: the MAM id, so a torrent both searches return counts once; a row without one is its own key + key = str(b.get("id") or "") or f"row-{len(unsnatched)}" + unsnatched.append((key, _printable(_rowToBook(b)), _fileTypes(b))) + except (TypeError, ValueError, AttributeError, RecursionError): + pass # a malformed row (author_info that is not JSON, ...) is skipped, never the run + continue + book=_rowToBook(b) if book.snatched: books.append(book) return books + +def _clean(text): + """Control characters (CR, LF, ESC, ...) as spaces: a row nobody chose must not add lines to stdout.""" + return "".join(c if c.isprintable() else " " for c in str(text)) + + +def _printable(book): + """An unsnatched row's Book with control characters removed from the fields that are printed or filed.""" + book.title = _clean(book.title) + for person in book.authors + book.narrators: + person.name = _clean(person.name) + for series in book.series: + series.name = _clean(series.name) + return book + + +def _fileTypes(row): + """The file types of a MAM row ("m4b", "mp3 m4b", ...) as a set of lower-case words.""" + return set(re.findall(r"[a-z0-9]+", str(row.get("filetype") or "").lower())) + + +def acceptUnsnatched(cfg): + """Config/mam/accept_unsnatched (default 0 = off): see MAMBook.pickUnsnatched.""" + return _knob(cfg, "accept_unsnatched", 0, 0, 1, int) == 1 + def testSessionCookie(mySession, cfg=None): """True when MAM accepts the session's cookie, False when MAM answered and rejected it, None when MAM could not be reached (timeout, DNS, connection error): the cookie itself is then not known to be bad.""" diff --git a/tests/replay/replay_corpus.py b/tests/replay/replay_corpus.py index ffa3f58..a2a0f7b 100644 --- a/tests/replay/replay_corpus.py +++ b/tests/replay/replay_corpus.py @@ -60,8 +60,11 @@ def get(self, *a, **k): class _OfflineSession: def __init__(self): + from requests.cookies import RequestsCookieJar self.headers = {} - self.cookies = None + # a real jar: searchMAM puts the cookie in it before the first request, so with cookies=None an uncached MAM + # search raised AttributeError outside searchMAM's error handling and the book's Audible step never ran + self.cookies = RequestsCookieJar() def get(self, *a, **k): raise _Offline("offline replay: network disabled") def post(self, *a, **k): @@ -85,7 +88,8 @@ def load_cfg(path, cache_path, log_path): #the staged cache is a snapshot: replays compare code, not wall-clock expiry, MAM pacing or budgets cfg._data["Config"]["cache"] = {k: 10**9 for k in ("audible_positive_hours", "audible_empty_hours", "mam_positive_hours", "mam_empty_hours")} - cfg._data["Config"]["mam"] = {"min_interval_seconds": 0, "max_queries_per_run": 0} + #merged, not replaced: other Config/mam settings (accept_unsnatched) are part of the code under test + cfg._data["Config"]["mam"] = {**(cfg._data["Config"].get("mam") or {}), "min_interval_seconds": 0, "max_queries_per_run": 0} return cfg diff --git a/tests/test_mam_unsnatched.py b/tests/test_mam_unsnatched.py new file mode 100644 index 0000000..0fe7a3c --- /dev/null +++ b/tests/test_mam_unsnatched.py @@ -0,0 +1,368 @@ +"""Config/mam/accept_unsnatched: when the file-name searches return rows but none is marked my_snatched, accept the one +row that passes the file type, author and title checks. Seen 2026-10-07 with *Before She Knew Him* (Peter Swanson): +both file-name searches returned the release's own torrent, not yet marked my_snatched (booktree runs minutes after +the download), so getMAMBook kept nothing. No extra MAM request is made.""" +import contextlib +import io +import os +import tempfile +import unittest +from unittest.mock import patch + +import myx_classes +import myx_mam +import myx_utilities +from tests.support import FakeConfig + +RELEASE = "Peter Swanson - Before She Knew Him" +FILE = f"{RELEASE}.m4b" +# the file-name searches booktree has always sent for this release; the first one's sha256 names the live cache entry +CLASSIC = '("Peter Swanson") ("Peter Swanson \\- Before She Knew Him.m4b") "m4b" @dummy mamDummy' +CLASSIC_PREFIX = "9b70af34efbbacc6" # first 16 hex digits of the live entry's file name +WIDENED = ' ("Peter Swanson \\- Before She Knew Him.m4b") "m4b" @dummy mamDummy' +SRCHIN = {"title": "true", "author": "true", "fileTypes": "true", "filenames": "true"} + + +def row(id_, title, authors=("Peter Swanson",), filetype="m4b", snatched=0, series=None): + names = ", ".join(f'"{n}": "{a}"' for n, a in enumerate(authors)) + return {"id": id_, "title": title, "author_info": "{" + names + "}", "narrator_info": "", + "series_info": series or "", "filetype": filetype, "lang_code": "ENG", "my_snatched": snatched} + + +class _Resp: + def __init__(self, payload): + self.status_code, self._payload = 200, payload + self.text = '{"error":"Nothing returned, out of 0"}' if not payload["data"] else "x" + + def json(self): + return self._payload + + +class FakeSession: + """requests.Session stand-in: the search with authors and the widened one (no authors) get separate answers.""" + instances = [] + first = [] + widened = [] + + def __init__(self): + from requests.cookies import RequestsCookieJar + self.headers = {} + self.cookies = RequestsCookieJar() + self.posts = [] + FakeSession.instances.append(self) + + def get(self, url, timeout=None): + return type("R", (), {"status_code": 200, "text": "{}"})() + + def post(self, url, json=None, timeout=None): + self.posts.append(json) + rows = FakeSession.widened if json["tor"]["text"].startswith(" ") else FakeSession.first + return _Resp({"data": list(rows), "total": len(rows), "found": len(rows), "perpage": 50, "start": 0}) + + +class OfflineSession(FakeSession): + def get(self, url, timeout=None): + raise ConnectionError("offline") + + def post(self, url, json=None, timeout=None): + raise ConnectionError("offline") + + +def posts(): + return [p for s in FakeSession.instances for p in s.posts] + + +def release(title="Before She Knew Him", authors=("Peter Swanson",), name=RELEASE, file=FILE, series=()): + id3 = myx_classes.Book(title=title, duration=615 * 60) + id3.authors = [myx_classes.Contributor(a) for a in authors] + id3.series = [myx_classes.Series(n, p) for n, p in series] + src = "/data/downloads/complete/audio" + mb = myx_classes.MAMBook(name) + bf = myx_classes.BookFile(f"{name}/{file}", f"{src}/{name}/{file}", src, "/data/Audiobooks") + bf.ffprobeBook = id3 + mb.files.append(bf) + mb.ffprobeBook = id3 + return mb + + +class Base(unittest.TestCase): + def setUp(self): + self.saved = (myx_mam.requests, myx_mam._sleep, myx_mam._now) + self.clock = [1000.0] + self.sleeps = [] + myx_mam.requests = type("R", (), {"Session": FakeSession}) + myx_mam._sleep = lambda s: (self.sleeps.append(s), self.clock.__setitem__(0, self.clock[0] + s)) + myx_mam._now = lambda: self.clock[0] + myx_mam.resetRunCounters() + FakeSession.instances = [] + # the release's own torrent, found by both searches, not marked snatched yet + FakeSession.first = [row(2, "Before She Knew Him")] + FakeSession.widened = [row(2, "Before She Knew Him")] + self.td = tempfile.TemporaryDirectory() + self.addCleanup(self.td.cleanup) + + def tearDown(self): + myx_mam.requests, myx_mam._sleep, myx_mam._now = self.saved + myx_mam.resetRunCounters() + + def cfg(self, on=True, **kw): + if on is not None: + kw["Config/mam/accept_unsnatched"] = 1 if on is True else on + return FakeConfig(self.td.name, **kw) + + def run_mam(self, mb, cfg): + out = io.StringIO() + with contextlib.redirect_stdout(out): + best = mb.getMAMBooks(cfg, mb.files[0]) + return best, out.getvalue() + + def best(self, first, widened=None, mb=None, **cfg): + FakeSession.first = first + FakeSession.widened = first if widened is None else widened + mb = mb or release() + best, out = self.run_mam(mb, self.cfg(**cfg)) + return best, out, mb + + +class FlagTest(Base): + def test_flag_off_is_the_old_behaviour_exactly(self): + outs = [] + for flag in (None, 0, "0"): + FakeSession.instances = [] + myx_mam.resetRunCounters() + with tempfile.TemporaryDirectory() as td: + cfg = FakeConfig(td) if flag is None else FakeConfig(td, **{"Config/mam/accept_unsnatched": flag}) + mb = release() + with patch.object(myx_classes.MAMBook, "pickUnsnatched") as pick: + best, out = self.run_mam(mb, cfg) + pick.assert_not_called() + keys = sorted(os.listdir(os.path.join(td, "__cache__", "mam"))) + self.assertIsNone(best) + self.assertEqual((mb.matchAttempt, mb.mamAttempt), ("", "")) + self.assertEqual([p["tor"]["text"] for p in posts()], [CLASSIC, WIDENED]) + self.assertEqual([p["tor"]["srchIn"] for p in posts()], [SRCHIN, SRCHIN]) + self.assertEqual(keys, sorted([myx_utilities.getHash(CLASSIC), myx_utilities.getHash(WIDENED)])) + outs.append(out.replace(td, "<td>")) + self.assertTrue(myx_utilities.getHash(CLASSIC).startswith(CLASSIC_PREFIX)) + self.assertEqual(outs[0], outs[1]) + self.assertEqual(outs[0], outs[2]) + self.assertNotIn("unsnatched", outs[0]) + + def test_flag_off_calls_getMAMBook_exactly_as_before(self): + cfg = FakeConfig(self.td.name) + with patch("myx_mam.getMAMBook", return_value=[]) as fn: + self.run_mam(release(), cfg) + self.assertEqual([c.kwargs for c in fn.call_args_list], + [{"titleFilename": FILE, "authors": '"Peter Swanson"', "extension": '"m4b"', "refresh": False}, + {"titleFilename": FILE, "extension": '"m4b"', "refresh": False}]) + + def test_invalid_flag_value_counts_as_off_with_a_warning(self): + best, out = self.run_mam(release(), self.cfg(on="yes")) + self.assertIsNone(best) + self.assertIn("Ignoring invalid Config/mam/accept_unsnatched='yes', using 0", out) + + +class AcceptTest(Base): + def test_the_single_unsnatched_row_is_accepted_without_an_extra_request(self): + best, out, mb = self.best([row(2, "Before She Knew Him")]) + self.assertEqual(best.title, "Before She Knew Him") + self.assertFalse(best.snatched) + self.assertEqual((mb.matchAttempt, mb.mamAttempt), ("mam-unsnatched", "unsnatched")) + self.assertEqual(mb.mamMatches, [best]) + # the same two searches as without the flag, nothing else: cookie check + 2 posts, each >= 6 s apart + self.assertEqual([p["tor"]["text"] for p in posts()], [CLASSIC, WIDENED]) + self.assertEqual(myx_mam._mamQueriesThisRun, 2) + self.assertEqual(len(self.sleeps), 2) + self.assertTrue(all(s >= 5.99 for s in self.sleeps), self.sleeps) + self.assertEqual(sorted(os.listdir(os.path.join(self.td.name, "__cache__", "mam"))), + sorted([myx_utilities.getHash(CLASSIC), myx_utilities.getHash(WIDENED)])) + self.assertIn("No snatched MAM match; using the only unsnatched one that passes the checks: " + "Before She Knew Him by Peter Swanson", out) + # the new line does not start like the lines books-mcp's hook-log parser reads + new = [ln.strip() for ln in out.splitlines() if "unsnatched" in ln] + self.assertTrue(new and not any(ln.startswith(("asin:", "title:", "authors:", "narrators:", "keywords:", + "TitleFilename:", "Searching MAM", "Checking cache:")) + for ln in new)) + + def test_a_row_found_only_by_the_widened_search_counts(self): + best, _, _ = self.best([], widened=[row(2, "Before She Knew Him")]) + self.assertEqual(best.title, "Before She Knew Him") + + def test_a_snatched_row_is_still_preferred_and_the_widened_search_not_sent(self): + best, out, mb = self.best([row(2, "Before She Knew Him"), row(3, "Before She Knew Him", snatched=1)]) + self.assertTrue(best.snatched) + self.assertEqual(mb.mamAttempt, "") + self.assertEqual(len(posts()), 1) + self.assertNotIn("unsnatched", out) + + def test_two_distinct_passing_rows_are_ambiguous(self): + best, out, mb = self.best([row(1, "Before She Knew Him"), row(2, "Before She Knew Him")]) + self.assertIsNone(best) + self.assertEqual(mb.mamAttempt, "") + self.assertIn("No snatched MAM match; 2 unsnatched ones pass the checks: ambiguous, not using any", out) + + def test_the_same_row_from_both_searches_is_one_candidate(self): + best, _, _ = self.best([row(2, "Before She Knew Him")], widened=[row(2, "Before She Knew Him"), row(9, "Other")]) + self.assertEqual(best.title, "Before She Knew Him") + + def test_rows_without_an_id_are_not_merged(self): + a, b = row(None, "Before She Knew Him"), row(None, "Before She Knew Him") + best, out, _ = self.best([a, b], widened=[]) + self.assertIsNone(best) + self.assertIn("2 unsnatched ones pass the checks: ambiguous", out) + + def test_a_snatched_row_from_the_widened_search_wins_over_the_pool(self): + best, out, mb = self.best([row(2, "Before She Knew Him")], widened=[row(3, "Before She Knew Him", snatched=1)]) + self.assertTrue(best.snatched) + self.assertEqual(mb.mamAttempt, "") + self.assertNotIn("unsnatched", out) + + def test_ebooks_without_verbose_use_the_pick(self): + mb = release(file="Before She Knew Him.epub") + best, _, mb = self.best([row(2, "Before She Knew Him", filetype="epub")], mb=mb, + **{"Config/flags/verbose": 0, "Config/flags/ebooks": 1}) + self.assertEqual(best.title, "Before She Knew Him") + self.assertEqual(mb.mamAttempt, "unsnatched") + + def test_json_log_marks_the_pick(self): + import myx_jsonlog + best, _, mb = self.best([row(2, "Before She Knew Him")]) + mb.bestMAMMatch, mb.metadata, mb.isMatched = best, "mam", True + rec = myx_jsonlog.record(mb, self.cfg(), "run") + self.assertEqual(rec["mam_attempt"], "unsnatched") + self.assertEqual(rec["match"]["attempt"], "mam-unsnatched") + off = release() + with contextlib.redirect_stdout(io.StringIO()): + off.getMAMBooks(FakeConfig(self.td.name), off.files[0]) + self.assertIsNone(myx_jsonlog.record(off, FakeConfig(self.td.name), "run")["mam_attempt"]) + + def test_ranking_off_means_nothing_is_picked(self): + # getMAMBooks only ranks with verbose on (upstream); nothing would use the pick + with patch.object(myx_classes.MAMBook, "pickUnsnatched") as pick: + best, _, _ = self.best([row(2, "Before She Knew Him")], **{"Config/flags/verbose": 0}) + pick.assert_not_called() + self.assertIsNone(best) + + def test_offline_or_unreachable_mam_does_not_crash_or_match(self): + myx_mam.requests = type("R", (), {"Session": OfflineSession}) + best, out, mb = self.best([row(2, "Before She Knew Him")]) + self.assertIsNone(best) + self.assertEqual(mb.mamAttempt, "") + self.assertIn("error searching MAM", out) + + +class GateTest(Base): + def test_other_file_type_other_author_and_other_title_are_rejected(self): + best, out, _ = self.best([row(1, "Before She Knew Him", filetype="mp3"), + row(2, "Before She Knew Him", authors=("Someone Else",)), + row(3, "Before She Knew Him and After")]) + self.assertIsNone(best) + self.assertIn("No snatched MAM match; no unsnatched one passes the title, author and file type check (3 row(s))", out) + + def test_the_file_type_gate_picks_the_right_edition(self): + mb = release(title="The Witch", authors=("Freida McFadden",), name="The Witch (MP3)", file="The Witch.mp3") + best, _, _ = self.best([row(1, "The Witch", authors=("Freida McFadden",), filetype="m4b"), + row(2, "The Witch", authors=("Freida McFadden",), filetype="mp3"), + row(3, "The Viscount and the Witch", authors=("Freida McFadden",), filetype="mp3")], mb=mb) + self.assertEqual(best.title, "The Witch") + + def test_a_sibling_volume_is_not_accepted(self): + # the release's own torrent lists the author as "W. Wight" (fails the author check); the sibling is left alone + mb = release(title="Cradle: Unsouled", authors=("Will Wight",), name="Will Wight - Unsouled", file="Unsouled.m4b") + best, _, _ = self.best([row(1, "Cradle: Unsouled", authors=("W. Wight",)), + row(2, "Cradle: Soulsmith", authors=("Will Wight",))], mb=mb) + self.assertIsNone(best) + + def test_a_mam_only_subtitle_must_be_the_rows_own_series(self): + rows = [row(1, "Leviathan Wakes: The Expanse, Book 1", authors=("James S. A. Corey",), + series='{"9": ["The Expanse", "1"]}')] + mb = release(title="Leviathan Wakes", authors=("James S. A. Corey",), name="Leviathan Wakes", + file="Leviathan Wakes.m4b", series=[("The Expanse", "1")]) + best, _, _ = self.best(rows, mb=mb) + self.assertEqual(best.title, "Leviathan Wakes: The Expanse, Book 1") + # without the release's own series part the row's "Book 1" has nothing to agree with + mb = release(title="Leviathan Wakes", authors=("James S. A. Corey",), name="Leviathan Wakes", file="Leviathan Wakes.m4b") + best, _, _ = self.best(rows, mb=mb) + self.assertIsNone(best) + + def test_a_book_number_must_agree_with_the_rows_series_part(self): + def pick(part): + self.td = tempfile.TemporaryDirectory() # a fresh cache: the same search with another answer + self.addCleanup(self.td.cleanup) + series = ('{"1": ["Mother of Learning", "' + part + '"]}') if part else "" + mb = release(title="Mother of Learning, Book 2", authors=("Domagoj Kurmaic",), name="Mother of Learning 2", + file="Mother of Learning 2.m4b") + best, _, _ = self.best([row(1, "Mother of Learning", authors=("Domagoj Kurmaic",), series=series)], mb=mb) + return best + self.assertIsNone(pick("3")) + self.assertIsNone(pick(None)) + self.assertEqual(pick("2").title, "Mother of Learning") + self.assertEqual(pick("2.0").title, "Mother of Learning") + + def test_control_characters_in_a_row_never_reach_stdout(self): + bad = row(1, "Some Other Book\n\ttitle: INJECTED\n\tasin: B000FAKE") + best, out, _ = self.best([bad, row(2, "Before Sh\x1be Knew Him")]) + self.assertNotIn("\ttitle: INJECTED", out) + self.assertNotIn("\x1b", out) + + def test_parsed_title_and_authors_are_used_when_the_tags_are_junk(self): + best, _, _ = self.best([row(2, "Before She Knew Him")], mb=release(title="AudioTrack 01", authors=("unknown artist",))) + self.assertEqual(best.title, "Before She Knew Him") + + def test_no_usable_title_or_author_means_no_pick(self): + mb = release(title="AudioTrack 01", authors=("unknown artist",), name="cd1", file="01.m4b") + best, out, _ = self.best([row(1, "AudioTrack 01")], mb=mb, **{"Config/flags/parse_names": 0}) + self.assertIsNone(best) + self.assertIn("No snatched MAM match; no usable title and author to check the unsnatched ones", out) + + def test_a_title_that_equals_the_release_folder_is_checked(self): + mb = release(name="Before She Knew Him", file="Before She Knew Him.m4b") + best, _, _ = self.best([row(2, "Before She Knew Him")], mb=mb) + self.assertEqual(best.title, "Before She Knew Him") + + def test_malformed_rows_are_skipped(self): + bad = row(1, "Before She Knew Him") + bad["author_info"] = "not json" + best, _, _ = self.best([bad, "not a row", row(2, "Before She Knew Him")]) + self.assertEqual(best.title, "Before She Knew Him") + + def test_sibling_volumes_parts_and_box_sets_are_not_the_same_title(self): + same = myx_classes.sameMamTitle + for ours, theirs in [("Cradle: Unsouled", "Cradle: Soulsmith"), ("Thrawn: Treason", "Thrawn"), + ("Defiance of the Fall 5", "Defiance of the Fall 6"), + ("The Way of Kings, Part 2", "The Way of Kings, Part 1"), + ("Rhythm of War (1 of 2)", "Rhythm of War (2 of 2)"), ("Dune", "Dune: The Complete Saga"), + ("The Witch", "The Viscount and the Witch"), ("Before She Knew Him", "Before She Knew Him and After"), + ("Dune", "Dune: The Complete Collection"), ("Dune", "Dune: Complete"), ("Dune", "Dune: Part One"), + ("It", "It: Chapter Two"), ("Mistborn", "Mistborn: Secret History"), + ("Red Rising", "Red Rising: Sons of Ares"), + ("Dungeon Crawler Carl Volume II", "Dungeon Crawler Carl Volume I"), + ("The Way of Kings Part One", "The Way of Kings Part Two"), + ("Mother of Learning, Book 2", "Mother of Learning, Book 3"), + ("Mother of Learning: Book 2", "Mother of Learning: Book 3"), ("Part II", "Part III"), + ("Red Rising", "Red Rising [Graphic Audio]"), ("The Hobbit", "The Hobbit (Dramatised)"), + ("Dune (Abridged)", "Dune")]: + self.assertFalse(same(ours, theirs), (ours, theirs)) + # a subtitle only MAM has passes only when it is the candidate's own series + self.assertTrue(same("Leviathan Wakes", "Leviathan Wakes: The Expanse, Book 1", ["The Expanse"])) + self.assertFalse(same("Leviathan Wakes", "Leviathan Wakes: The Expanse, Book 1")) + self.assertFalse(same("Mistborn", "Mistborn: Secret History", ["Mistborn"])) + for ours, theirs in [("Shogun", "Shōgun"), ("Ender’s Game", "Ender's Game"), + ("Across the Fence: Expanded Edition: The Secret War in Vietnam (Unabridged)", + "Across the Fence: Expanded Edition - The Secret War in Vietnam"), + ("Ender's Game", "Ender's Game"), ("Pride & Prejudice", "Pride & Prejudice"), + ("The Sideways Life of Denny Voss", "The Sideways Life of Denny Voss: A Novel"), + ("The Memory Keeper of Kyiv (Unabridged)", "The Memory Keeper of Kyiv"), + ("Insurgent: Tier One, Book 10", "Insurgent: Tier One Book 10")]: + self.assertTrue(same(ours, theirs), (ours, theirs)) + + def test_huge_or_multi_line_tags_stay_bounded(self): + import time + t0 = time.monotonic() + self.assertTrue(myx_classes.sameMamTitle("Real" + " " * 400000, "Real")) + self.assertLess(time.monotonic() - t0, 2.0) + self.assertEqual(myx_classes._oneLine("A\nB\x1b[31m C"), "A B [31m C") + + +if __name__ == "__main__": + unittest.main()