Metadata never fails a drop: deterministic title/author detection (metadata, first-page text, OCR of scanned first pages, file name), sources recorded in book.json; decrypt scratch copy before OCR
This commit is contained in:
Audiobook Producer
@@ -0,0 +1,348 @@
|
||||
"""Title and author for a book. Never fails: whatever cannot be found is replaced and logged.
|
||||
|
||||
Deterministic rules, in this order (the first that gives a value wins). Every value records where it came from.
|
||||
|
||||
1. Explicit: command-line arguments, then the optional sidecar <pdf name>.json {"title": "...", "author": "..."}.
|
||||
2. TITLE : PDF metadata Title ("pdf-metadata"; junk such as "Microsoft Word - x.doc" or "untitled" is ignored)
|
||||
-> first-page text ("first-page-text", or "first-page-ocr" when the first pages have no text layer)
|
||||
-> file name ("file-name") -> "Untitled".
|
||||
AUTHOR : a "By <Name>" line on the first pages (text layer, else OCR) [explicit, so it beats metadata]
|
||||
-> PDF metadata Author (names such as a typesetter or "admin" are ignored)
|
||||
-> a name line under the title on the first page ("first-page-name-line")
|
||||
-> "Title - by Author" in the file name -> "Unknown".
|
||||
3. The file name is read the same way for a downloaded PDF (the name from the URL), then the name of the .url file.
|
||||
|
||||
Missing author -> "Unknown" (folder "unknown") and a warning. Nothing found at all -> title from the cleaned file name
|
||||
and author "Unknown". The detection from the PDF itself is cached in the scratch folder, so asking twice costs nothing."""
|
||||
import difflib
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import tempfile
|
||||
|
||||
import bookio
|
||||
import pdftext
|
||||
import settings
|
||||
|
||||
UNKNOWN_AUTHOR = "Unknown"
|
||||
JUNK_AUTHORS = {"admin", "administrator", "user", "owner", "unknown", "scanner", "scan", "bluepen", "pdf", "adobe", "acrobat",
|
||||
"microsoft", "windows user", "author", "anonymous", "none", "n/a", "ocr", "internet archive"}
|
||||
NOT_NAME_WORDS = {"the", "of", "and", "a", "an", "in", "on", "for", "to", "with", "from", "press", "society", "publishing",
|
||||
"publishers", "publication", "publications", "company", "co", "edition", "volume", "vol", "university",
|
||||
"library", "books", "book", "chapter", "contents", "copyright", "printed", "published", "illustrated",
|
||||
"manual", "philosophy", "science", "complete", "new", "york", "london", "chicago", "boston", "all", "rights"}
|
||||
PUBLISHER_WORDS = {"press", "society", "publishing", "publishers", "publication", "company", "edition", "copyright", "printed",
|
||||
"published", "rights", "reserved", "library", "university", "isbn", "volume"}
|
||||
|
||||
OCR_FUNC = None # tests replace this: f(pdf, first_page, last_page, lang) -> [(page_number, text)]
|
||||
|
||||
|
||||
def clean_text(s):
|
||||
return re.sub(r"\s+", " ", s or "").strip()
|
||||
|
||||
|
||||
def junk_title(t):
|
||||
t = clean_text(t)
|
||||
if len(re.sub(r"[^A-Za-z]", "", t)) < 3:
|
||||
return True
|
||||
low = t.lower()
|
||||
return bool(re.match(r"^(microsoft word|untitled|document\d*|scan|new document|title|noname)\b", low) or re.search(r"\.(docx?|pdf|tiff?|indd|qxd|rtf|txt)$", low))
|
||||
|
||||
|
||||
def junk_author(a):
|
||||
a = clean_text(a)
|
||||
return len(re.sub(r"[^A-Za-z]", "", a)) < 3 or a.lower() in JUNK_AUTHORS or bool(re.search(r"\d{3,}", a))
|
||||
|
||||
|
||||
def display_name(name):
|
||||
"""'YOGI RAMACHARAKA' / 'yogi ramacharaka' -> 'Yogi Ramacharaka'. Mixed case is kept (McDonald)."""
|
||||
out = []
|
||||
for w in clean_text(name).split(" "):
|
||||
if w.isupper() or w.islower():
|
||||
w = "-".join(p[:1].upper() + p[1:].lower() for p in w.split("-"))
|
||||
out.append(w)
|
||||
return " ".join(out)
|
||||
|
||||
|
||||
def looks_like_name(tokens):
|
||||
if not 1 <= len(tokens) <= 5:
|
||||
return False
|
||||
letters = 0
|
||||
for t in tokens:
|
||||
if t.lower().strip(".,") in NOT_NAME_WORDS:
|
||||
return False
|
||||
if not re.match(r"^[A-Z][A-Za-z'’\-]*\.?,?$", t): # Capitalized, ALL CAPS, or an initial
|
||||
return False
|
||||
letters += len(re.sub(r"[^A-Za-z]", "", t))
|
||||
return letters >= 4
|
||||
|
||||
|
||||
def first_author(text):
|
||||
"""'Glenn Clark and Walter Russell' / 'A. B, C. D' -> first person; None if it does not look like a name."""
|
||||
text = re.split(r"\s+(?:and|&|with)\s+|,\s+(?=[A-Z])", clean_text(text).strip(" .,;:"))[0]
|
||||
tokens = text.split(" ")
|
||||
return display_name(text) if looks_like_name(tokens) else None
|
||||
|
||||
|
||||
def page_lines(text):
|
||||
"""Non-empty lines with a paragraph index: [(paragraph, line)]. OCR noise lines (under 2 letters) are dropped."""
|
||||
out, para, blank = [], 0, False
|
||||
for raw in text.splitlines():
|
||||
s = clean_text(raw)
|
||||
if not s:
|
||||
blank = True
|
||||
continue
|
||||
if len(re.sub(r"[^A-Za-z]", "", s)) < 2:
|
||||
continue
|
||||
if blank and out:
|
||||
para += 1
|
||||
blank = False
|
||||
out.append((para, s))
|
||||
return out
|
||||
|
||||
|
||||
BY_RE = re.compile(r"^(?:written\s+|translated\s+|compiled\s+|edited\s+)?by[\s:]+(.+)$", re.I)
|
||||
|
||||
|
||||
def norm_words(s):
|
||||
return re.sub(r"[^a-z0-9 ]", "", s.lower().replace("-", " ")).split()
|
||||
|
||||
|
||||
def match_hint(line, hint):
|
||||
"""How well a page line matches the title suggested by the file name, 0..1."""
|
||||
a, b = " ".join(norm_words(line)), " ".join(norm_words(hint))
|
||||
return difflib.SequenceMatcher(None, a, b).ratio() if a and b else 0.0
|
||||
|
||||
|
||||
def read_title_page(pages, hint=None):
|
||||
"""pages: [(page_number, text)]. Returns {'author','author_source','title','page'} from the first page that has a
|
||||
'By <Name>' line; else, from the first page with a plausible title block, {'title','page'} plus an 'author' when a
|
||||
name stands alone right under the title; else {}."""
|
||||
for number, text in pages:
|
||||
lines = page_lines(text)
|
||||
for i, (para, s) in enumerate(lines):
|
||||
m = BY_RE.match(s)
|
||||
if not m or len(s.split()) > 7:
|
||||
continue
|
||||
author = first_author(m.group(1))
|
||||
if author:
|
||||
return {"author": author, "author_source": "by-line", "title": title_block(lines[:i], hint)[0], "page": number}
|
||||
for number, text in pages:
|
||||
lines = page_lines(text)
|
||||
title, last = title_block(lines, hint)
|
||||
if not title:
|
||||
continue
|
||||
para = lines[last][0]
|
||||
mine = [i for i in range(last + 1) if lines[i][0] == para]
|
||||
tail = first_author(lines[last][1]) if len(mine) >= 2 and title_block(lines, hint)[0] == title else None
|
||||
if tail and len(mine) >= 2 and " " in lines[last][1]: # "Title\nJane Doe" in one block: the last line is the name
|
||||
return {"author": tail, "author_source": "name-line", "title": " ".join(lines[i][1] for i in mine[:-1]), "page": number}
|
||||
for _, nxt in lines[last + 1:last + 3]: # a name on its own line under the title, e.g. "Jane Doe"
|
||||
a = first_author(nxt)
|
||||
if a and not any(w in PUBLISHER_WORDS for w in norm_words(nxt)):
|
||||
return {"author": a, "author_source": "name-line", "title": title, "page": number}
|
||||
return {"title": title, "page": number}
|
||||
return {}
|
||||
|
||||
|
||||
def title_block(lines, hint=None):
|
||||
"""(title, index of its last line) from the given lines: the line that matches the file-name hint (score 0.7 or more),
|
||||
else the first paragraph. Publisher lines and lines with a year are skipped."""
|
||||
idx = [i for i, (p, s) in enumerate(lines)
|
||||
if not any(w in PUBLISHER_WORDS for w in norm_words(s)) and not re.search(r"\b1[0-9]{3}\b|\b20[0-9]{2}\b", s)]
|
||||
if not idx:
|
||||
return "", -1
|
||||
if hint:
|
||||
score, i = max((match_hint(lines[i][1], hint), i) for i in idx)
|
||||
if score >= 0.7:
|
||||
return lines[i][1], i
|
||||
para = lines[idx[0]][0]
|
||||
mine = [i for i in idx if lines[i][0] == para]
|
||||
return " ".join(lines[i][1] for i in mine)[:200], mine[-1]
|
||||
|
||||
|
||||
def clean_filename_title(stem):
|
||||
"""'2015.47927.Science-Of-Breath' -> 'Science Of Breath'; 'my_book_final' -> 'my book final' words, ids removed."""
|
||||
s = re.sub(r"\([^)]*\)|\[[^\]]*\]", " ", stem)
|
||||
s = re.sub(r"[_\-.+]+", " ", s)
|
||||
words = [w for w in s.split() if not re.fullmatch(r"[0-9v]+", w, re.I) and not re.fullmatch(r"[0-9a-f]{8,}", w, re.I)]
|
||||
words = [w for w in words if not w.lower() in ("pdf", "final", "scan", "scanned", "ocr", "copy")] or words
|
||||
return " ".join(words)
|
||||
|
||||
|
||||
def guess_from_filename(name):
|
||||
"""'The Title - by Glenn Clark (Extra).pdf' / 'Title by Jane Doe.pdf' -> (title, author), else (None, None)."""
|
||||
stem = os.path.splitext(os.path.basename(name))[0]
|
||||
m = re.match(r"^(?P<title>.+?)[\s\-–—]+by\s+(?P<author>[^()]+?)\s*(\(.*\))?\s*$", stem, re.I)
|
||||
if m:
|
||||
return m.group("title").strip(" -–—"), m.group("author").strip()
|
||||
return None, None
|
||||
|
||||
|
||||
def filename_candidates(names):
|
||||
"""From file names (the PDF's own, then others such as a .url file), the first that carries a usable title:
|
||||
(title, author_or_None, name). The title is the 'Title - by Author' part, else the cleaned name."""
|
||||
for name in names:
|
||||
if not name:
|
||||
continue
|
||||
t, a = guess_from_filename(name)
|
||||
if t:
|
||||
return t, a, os.path.basename(name)
|
||||
cleaned = clean_filename_title(os.path.splitext(os.path.basename(name))[0])
|
||||
if len(re.sub(r"[^A-Za-z]", "", cleaned)) >= 4:
|
||||
return cleaned, None, os.path.basename(name)
|
||||
return None, None, None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- reading the PDF
|
||||
|
||||
def pdf_info(pdf):
|
||||
out = subprocess.run(["pdfinfo", pdf], capture_output=True, text=True).stdout
|
||||
get = lambda k: (re.search(r"^%s:\s+(.+)$" % k, out, re.M) or [None, ""])[1].strip()
|
||||
return {"title": get("Title"), "author": get("Author")}
|
||||
|
||||
|
||||
def ocr_first_pages(pdf, first, last, lang):
|
||||
"""OCR of single pages with pdftoppm + tesseract, for pages that have no text layer. Returns [(page, text)]."""
|
||||
if OCR_FUNC:
|
||||
return OCR_FUNC(pdf, first, last, lang)
|
||||
if shutil.which("tesseract") is None or shutil.which("pdftoppm") is None:
|
||||
return None
|
||||
out = []
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
subprocess.run(["pdftoppm", "-r", "200", "-gray", "-f", str(first), "-l", str(last), "-png", pdf, os.path.join(tmp, "p")],
|
||||
capture_output=True, timeout=600)
|
||||
for name in sorted(os.listdir(tmp)):
|
||||
m = re.match(r"p-0*(\d+)\.png$", name)
|
||||
if not m:
|
||||
continue
|
||||
r = subprocess.run(["tesseract", os.path.join(tmp, name), "stdout", "-l", lang, "--psm", "3"],
|
||||
capture_output=True, text=True, timeout=300)
|
||||
out.append((int(m.group(1)), r.stdout))
|
||||
return out
|
||||
|
||||
|
||||
def first_pages_text(cfg, pdf, n, notes):
|
||||
"""[(page, text, source)] for the first n pages: the text layer where there is one, OCR where there is none."""
|
||||
pages = []
|
||||
try:
|
||||
units = pdftext.units(pdf, first_pages=n)
|
||||
except Exception as e:
|
||||
notes.append("could not read the text layer of the first pages: %s" % e)
|
||||
units = []
|
||||
have = {}
|
||||
for u in units:
|
||||
have.setdefault(u["pdf_page"], []).append(u["text"])
|
||||
min_chars = cfg["ocr"]["min_chars_per_page"]
|
||||
blank = []
|
||||
for p in range(1, n + 1):
|
||||
text = "\n".join(have.get(p, []))
|
||||
if len(re.sub(r"\s", "", text)) >= min_chars:
|
||||
pages.append((p, text, "first-page-text"))
|
||||
elif p in have or not units:
|
||||
blank.append(p)
|
||||
return pages, blank
|
||||
|
||||
|
||||
def detect_from_pdf(cfg, pdf, hint=None):
|
||||
"""What the PDF itself says. {'title','title_source','author','author_source','notes':[...]} (values may be empty)."""
|
||||
n = int(cfg.get("meta", {}).get("first_pages", 10))
|
||||
notes = []
|
||||
try:
|
||||
n = min(n, len(pdftext.page_geometry(pdf)))
|
||||
except Exception:
|
||||
pass
|
||||
info = pdf_info(pdf)
|
||||
res = {"title": "", "title_source": "", "author": "", "author_source": "", "notes": notes}
|
||||
if info["title"] and not junk_title(info["title"]):
|
||||
res["title"], res["title_source"] = bookio.title_case(clean_text(info["title"])), "pdf-metadata"
|
||||
info_author = info["author"] if info["author"] and not junk_author(info["author"]) else ""
|
||||
|
||||
pages, blank = first_pages_text(cfg, pdf, n, notes)
|
||||
page = read_title_page([(p, t) for p, t, _ in pages], hint)
|
||||
ocr_used = False
|
||||
if not page.get("author") and blank and cfg["ocr"]["enabled"]:
|
||||
if not (res["title"] and info_author): # nothing more to learn if metadata already gives both
|
||||
ocr = ocr_first_pages(pdf, min(blank), max(blank), cfg["ocr"]["language"])
|
||||
if ocr is None:
|
||||
notes.append("the first pages have no text layer and tesseract is not installed, so they could not be read")
|
||||
else:
|
||||
got = dict(ocr)
|
||||
allpages = sorted([(p, t) for p, t, _ in pages] + [(p, got[p]) for p in blank if p in got])
|
||||
page = read_title_page(allpages, hint)
|
||||
ocr_used = True
|
||||
src = "first-page-ocr" if ocr_used and page.get("page") in blank else "first-page-text"
|
||||
if page.get("author"):
|
||||
res["author"] = page["author"]
|
||||
res["author_source"] = ("%s (%s, pdf page %s)" % (src, page["author_source"], page["page"]))
|
||||
elif info_author:
|
||||
res["author"], res["author_source"] = display_name(info_author), "pdf-metadata"
|
||||
if not res["title"] and page.get("title") and page.get("author"):
|
||||
res["title"], res["title_source"] = bookio.title_case(page["title"]), "%s (pdf page %s)" % (src, page["page"])
|
||||
res["_page_title"] = bookio.title_case(page["title"]) if page.get("title") else ""
|
||||
res["_page_src"] = "%s (pdf page %s)" % (src, page.get("page"))
|
||||
return res
|
||||
|
||||
|
||||
def cached_detect(cfg, pdf, hint=None):
|
||||
h = hashlib.sha256()
|
||||
with open(pdf, "rb") as f:
|
||||
for block in iter(lambda: f.read(1 << 20), b""):
|
||||
h.update(block)
|
||||
h.update((hint or "").encode())
|
||||
path = os.path.join(settings.resolve(cfg, "work_dir"), "meta-%s.json" % h.hexdigest()[:20])
|
||||
if os.path.exists(path):
|
||||
with open(path, encoding="utf-8") as f:
|
||||
return json.load(f)
|
||||
res = detect_from_pdf(cfg, pdf, hint)
|
||||
os.makedirs(os.path.dirname(path), exist_ok=True)
|
||||
with open(path, "w", encoding="utf-8") as f:
|
||||
json.dump(res, f)
|
||||
return res
|
||||
|
||||
|
||||
def resolve(cfg, pdf, title=None, author=None, sidecar=None, name_hints=None):
|
||||
"""The one entry point. Returns {'title','author','title_source','author_source','warnings':[(kind, where, message)], 'log':[lines]}.
|
||||
Never raises for missing metadata."""
|
||||
sidecar = sidecar or {}
|
||||
names = [os.path.basename(pdf)] + [n for n in (name_hints or []) if n]
|
||||
ft, fa, fname = filename_candidates(names)
|
||||
t, a, ts, as_ = title or sidecar.get("title") or "", author or sidecar.get("author") or "", "", ""
|
||||
if t:
|
||||
ts = "argument" if title else "sidecar-json"
|
||||
if a:
|
||||
as_ = "argument" if author else "sidecar-json"
|
||||
det = {"notes": []}
|
||||
if not (t and a):
|
||||
try:
|
||||
det = cached_detect(cfg, pdf, ft)
|
||||
except Exception as e: # an unreadable PDF must not stop metadata; later stages report it properly
|
||||
det = {"notes": ["could not read the PDF to find title and author: %s" % e]}
|
||||
warnings = [("metadata-note", os.path.basename(pdf), n) for n in det.get("notes", [])]
|
||||
if not t:
|
||||
if det.get("title"):
|
||||
t, ts = det["title"], det["title_source"]
|
||||
elif ft:
|
||||
t, ts = bookio.title_case(ft) if ft.islower() else ft, "file-name (%s)" % fname
|
||||
if det.get("_page_title") and not det.get("author"):
|
||||
warnings.append(("metadata-title", os.path.basename(pdf), "a first-page text block %r was not used because no author line confirmed it" % det["_page_title"]))
|
||||
elif det.get("_page_title"):
|
||||
t, ts = det["_page_title"], det["_page_src"] + ", unconfirmed"
|
||||
else:
|
||||
stem = clean_filename_title(os.path.splitext(os.path.basename(pdf))[0])
|
||||
t, ts = (stem or "Untitled"), "file-name (cleaned)"
|
||||
if ts.startswith("file-name") or "unconfirmed" in ts:
|
||||
warnings.append(("metadata-title", os.path.basename(pdf), "title %r taken from %s" % (t, ts)))
|
||||
if not a:
|
||||
if det.get("author"):
|
||||
a, as_ = det["author"], det["author_source"]
|
||||
elif fa:
|
||||
a, as_ = fa, "file-name (%s)" % fname
|
||||
else:
|
||||
a, as_ = UNKNOWN_AUTHOR, "default"
|
||||
warnings.append(("metadata-author", os.path.basename(pdf), "no author found in the PDF or its file name; using 'Unknown' (folder 'unknown'). Add %s.json with {\"author\": \"...\"} next to the PDF to set it" % os.path.splitext(os.path.basename(pdf))[0]))
|
||||
log = ["metadata: title %r from %s" % (t, ts), "metadata: author %r from %s" % (a, as_)] + ["metadata: " + n for n in det.get("notes", [])]
|
||||
return {"title": t, "author": a, "title_source": ts, "author_source": as_, "warnings": warnings, "log": log}
|
||||
Reference in New Issue
Block a user