Project Gutenberg Is a Free 90,000-Book Inventory Feed for Kindle Publishing — This Tested Scanner Finds the Books Amazon Forgot

#!/usr/bin/env python3
"""
gutenberg_kdp_scanner.py -- find neglected public-domain books worth
republishing on Amazon KDP.

Every text in the Project Gutenberg catalog is US public domain; that is
PG's whole mission. So the legal question ("is this PD?") is answered for
you. The money question is different: has anyone published a *good modern
Kindle edition* of this exact book?

Our proxy for "probably not": the PG ebook's own Release date. A 2003-2012
release + fiction-ish subject = an old OCR transcription that likely has
zero decent competing Kindle editions.

Two free data sources, no API key:
  1. https://www.gutenberg.org/cache/epub/feeds/pg_catalog.csv.gz
  2. HTTP Range GET of the first 1,200 bytes of pg<id>.txt (release date)

Usage:
  python3 gutenberg_kdp_scanner.py [--sample N] [--top N]
"""
import csv
import gzip
import json
import os
import re
import sys
import time
import urllib.error
import urllib.request
from datetime import datetime, timezone

PG_CATALOG_URL = "https://www.gutenberg.org/cache/epub/feeds/pg_catalog.csv.gz"
TXT_URL = "https://www.gutenberg.org/cache/epub/{id}/pg{id}.txt"
UA = {"User-Agent": "Mozilla/5.0 (X11; Linux x86_64) kdp-gap-scanner/1.2"}
CACHE = "pg_catalog.csv.gz"
MIN_FILE_KB = 120            # under this it is a pamphlet, not a Kindle book
POLITE_DELAY = 0.12          # seconds between probes (~8 req/s, polite)
ROYALTY_RATE = 0.70          # KDP 70% tier
DELIVERY_FEE_PER_MB = 0.15   # KDP delivery fee, amazon.com
LIST_PRICES = [2.99, 4.99, 9.99]
RELEASE_RE = re.compile(r"Release\s+date:\s*([A-Z][a-z]+ \d{1,2}, \d{4})", re.I)

GOOD_SUBJECTS = ("fiction", "novels", "poetry", "fantasy", "detective",
                 "science fiction", "gothic", "romance", "adventure",
                 "humor", "short stories", "suspense", "horror")
BAD_SUBJECTS = ("periodicals", "laws", "statutes", "congress", "speeches",
                "correspondence", "journals", "essays", "biography",
                "almanacs", "encyclopedias", "dictionaries", "textbooks")


def fetch_catalog(path=CACHE):
    """Download the PG catalog once; reuse the cached copy afterwards."""
    if os.path.exists(path) and os.path.getsize(path) > 1_000_000:
        print("[cache] reusing %s (%s bytes)" % (path, f"{os.path.getsize(path):,}"))
        return path
    print("[net] downloading catalog ...")
    req = urllib.request.Request(PG_CATALOG_URL, headers=UA)
    with urllib.request.urlopen(req, timeout=90) as r, open(path, "wb") as f:
        f.write(r.read())
    return path


def load_candidates(path):
    """English fiction-biased candidates. All rows are PD-US by definition."""
    out = []
    with gzip.open(path, "rt", encoding="utf-8", errors="replace") as f:
        for row in csv.DictReader(f):
            if (row.get("Type") or "") != "Text":
                continue
            if (row.get("Language") or "").strip().lower() != "en":
                continue
            subjects = (row.get("Subjects") or "").lower()
            score = sum(3 for k in GOOD_SUBJECTS if k in subjects)
            score -= sum(5 for k in BAD_SUBJECTS if k in subjects)
            if score > 0:
                out.append({
                    "id": int(row["Text#"]),
                    "title": " ".join((row.get("Title") or "").split()),
                    "authors": (row.get("Authors") or "").strip(),
                })
    return out


def probe_release_date(book_id):
    """The first 1,200 bytes contain 'Release date: <Month D, YYYY>'.
    Returns (release_datetime, size_bytes) or None."""
    req = urllib.request.Request(
        TXT_URL.format(id=book_id), headers=dict(UA, **{"Range": "bytes=0-1200"}))
    try:
        with urllib.request.urlopen(req, timeout=30) as r:
            head = r.read(1200).decode("utf-8", errors="replace")
            crange = r.headers.get("Content-Range")
            m = RELEASE_RE.search(head)
            if m is None:
                return None
            dt = datetime.strptime(m.group(1), "%B %d, %Y")
            size_bytes = int(crange.split("/")[-1]) if crange else None
            return dt.replace(tzinfo=timezone.utc), size_bytes
    except (urllib.error.URLError, ValueError, OSError):
        return None


def kdp_math(size_mb):
    """Real 70%-tier royalty per list price after the per-MB delivery fee."""
    fee = round(size_mb * DELIVERY_FEE_PER_MB, 2)
    out = {"size_mb": round(size_mb, 2), "delivery_fee": fee}
    for p in LIST_PRICES:
        out["royalty_at_%.2f" % p] = round(p * ROYALTY_RATE - fee, 2)
    return out


def main():
    args = sys.argv[1:]
    sample_n = int(args[args.index("--sample") + 1]) if "--sample" in args else 300
    top_n = int(args[args.index("--top") + 1]) if "--top" in args else 15

    path = fetch_catalog()
    candidates = load_candidates(path)
    print("[*] %s English fiction-biased PD candidates" % f"{len(candidates):,}")

    step = max(1, len(candidates) // sample_n)
    sample = candidates[::step][:sample_n]
    print("[*] probing %d books' release dates (%.2fs delay between probes)"
          % (len(sample), POLITE_DELAY))

    results = []
    failures = 0
    for i, b in enumerate(sample, 1):
        probe = probe_release_date(b["id"])
        if probe is None:
            failures += 1
            continue
        release_dt, size_bytes = probe
        size_mb = size_bytes / 1_048_576 if size_bytes else 0
        if size_mb * 1024 < MIN_FILE_KB:
            continue  # pamphlet, skip
        b["release_date"] = release_dt.date().isoformat()
        b["ebook_age_years"] = round(
            (datetime.now(timezone.utc) - release_dt).days / 365.25, 1)
        b.update(kdp_math(size_mb))
        results.append(b)
        if i % 25 == 0:
            print("    ... %d/%d probed, %d valid, %d failed"
                  % (i, len(sample), len(results), failures))
        time.sleep(POLITE_DELAY)

    results.sort(key=lambda b: b["release_date"])
    print()
    print("=== TOP %d OLDEST-RELEASE PD BOOKS (2003-2012 era transcriptions) ===" % top_n)
    hdr = ("%7s %10s %5s %6s %6s %6s %6s  title"
           % ("PG#", "released", "age", "MB", "$2.99", "$4.99", "$9.99"))
    print(hdr)
    print("-" * len(hdr))
    for b in results[:top_n]:
        print("%7d %10s %4.0fy %6.2f %6.2f %6.2f %6.2f  %s" % (
            b["id"], b["release_date"], b["ebook_age_years"], b["size_mb"],
            b["royalty_at_2.99"], b["royalty_at_4.99"], b["royalty_at_9.99"],
            b["title"][:48]))

    with open("kdp_candidates.json", "w", encoding="utf-8") as f:
        json.dump(results, f, indent=2, ensure_ascii=False)
    print()
    print("[ok] wrote kdp_candidates.json (%d books)" % len(results))
    print("[!] Manual step REQUIRED: search each title+author in the Amazon")
    print("    Kindle store before publishing anything.")
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except Exception as exc:
        print("FATAL: %s" % exc, file=sys.stderr)
        sys.exit(1)

#!/usr/bin/env python3
"""
pg_to_kindle_batch.py -- batch-convert PG plain text to KDP-ready EPUB 2
files, one EPUB per book, from a candidates JSON (e.g. kdp_candidates.json).

EPUB is a zip of XHTML + a tiny OPF manifest + NCX toc. Nothing exotic.
KDP accepts EPUB directly; Amazon converts it to KF8 on ingest.

Dependency: standard library only (zipfile + html-ish escaping).
Convert a *sample* of the text; the text itself is Project Gutenberg PD.

Usage:  python3 pg_to_kindle_batch.py kdp_candidates.json --limit 3
"""
import html
import json
import re
import sys
import time
import urllib.request
import zipfile
from pathlib import Path

TXT_URL = "https://www.gutenberg.org/cache/epub/{id}/pg{id}.txt"
UA = {"User-Agent": "Mozilla/5.0 (X11; Linux x86_64) pg-to-kindle/1.0"}
CHAPTER_CHARS = 18_000   # split chapters roughly every 18k characters


def download_text(book_id):
    req = urllib.request.Request(TXT_URL.format(id=book_id), headers=UA)
    with urllib.request.urlopen(req, timeout=90) as r:
        return r.read().decode("utf-8", errors="replace")


def strip_gutenberg_header_footer(text):
    """Cut the PG boilerplate so Amazon does not flag the file as a re-upload
    of a PG file (KDP does reject files containing the PG license header)."""
    start = text.find("*** START OF THE PROJECT GUTENBERG EBOOK")
    end = text.find("*** END OF THE PROJECT GUTENBERG EBOOK")
    if start != -1 and end != -1:
        start = text.find("\n", start)
        return text[start:end].strip()
    return text.strip()


def split_chapters(text):
    """Split on blank lines into ~CHAPTER_CHARS-sized pseudo-chapters."""
    paras = [p.strip() for p in text.split("\n\r\n") if p.strip()]
    if not paras:
        paras = [p.strip() for p in text.split("\n\n") if p.strip()]
    chapters, buf, size = [], [], 0
    for p in paras:
        buf.append(p)
        size += len(p)
        if size >= CHAPTER_CHARS:
            chapters.append("\n\n".join(buf))
            buf, size = [], 0
    if buf:
        chapters.append("\n\n".join(buf))
    return chapters or [text]


def make_xhtml(title, author, chapters):
    pages = []
    for i, ch in enumerate(chapters, 1):
        body = "".join(
            "<p>%s</p>" % html.escape(line) for line in ch.split("\n\n") if line.strip())
        pages.append(
            '<?xml version="1.0" encoding="utf-8"?>\n'
            '<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.1//EN" '
            '"http://www.w3.org/TR/xhtml11/DTD/xhtml11.dtd">\n'
            '<html xmlns="http://www.w3.org/1999/xhtml">\n<head>'
            '<title>%s</title>\n'
            '<meta http-equiv="Content-Type" content="text/html; charset=utf-8"/>\n'
            '</head>\n<body>\n%s\n</body>\n</html>' % (html.escape(title), body))
    return pages


def make_epub(title, author, chapters, out_path):
    pages = make_xhtml(title, author, chapters)
    uid = "urn:uuid:%s" % __import__("uuid").uuid4()
    opf = (
        '<?xml version="1.0" encoding="utf-8"?>\n'
        '<package xmlns="http://www.idpf.org/2007/opf" xmlns:opf="http://www.idpf.org/2007/opf" unique-identifier="bookid" version="2.0">\n'
        '  <metadata xmlns:dc="http://purl.org/dc/elements/1.1/">\n'
        '    <dc:title>%s</dc:title>\n'
        '    <dc:creator opf:role="aut">%s</dc:creator>\n'
        '    <dc:language>en</dc:language>\n'
        '    <dc:identifier id="bookid">%s</dc:identifier>\n'
        '  </metadata>\n'
        '  <manifest>\n'
        '    <item id="ncx" href="toc.ncx" media-type="application/x-dtbncx+xml"/>\n'
        '%s\n'
        '  </manifest>\n'
        '  <spine toc="ncx">\n%s\n  </spine>\n'
        '</package>' % (
            html.escape(title), html.escape(author), uid,
            "\n".join('    <item id="ch%d" href="ch%d.xhtml" media-type="application/xhtml+xml"/>'
                       % (i, i) for i in range(1, len(pages) + 1)),
            "\n".join('    <itemref idref="ch%d"/>' % i for i in range(1, len(pages) + 1))))
    navpoints = "\n".join(
        '    <navPoint id="np%d" playOrder="%d"><navLabel><text>Chapter %d</text></navLabel>'
        '<content src="ch%d.xhtml"/></navPoint>' % (i, i, i, i)
        for i in range(1, len(pages) + 1))
    ncx = (
        '<?xml version="1.0" encoding="utf-8"?>\n'
        '<ncx xmlns="http://www.daisy.org/z3986/2005/ncx/" version="2005-1">\n'
        '  <head><meta name="dtb:uid" content="%s"/></head>\n'
        '  <docTitle><text>%s</text></docTitle>\n'
        '  <navMap>\n%s\n  </navMap>\n</ncx>' % (uid, html.escape(title), navpoints))
    container = (
        '<?xml version="1.0" encoding="utf-8"?>\n'
        '<container version="1.0" xmlns="urn:oasis:names:tc:opendocument:xmlns:container">\n'
        '  <rootfiles><rootfile full-path="OEBPS/content.opf" media-type="application/oebps-package+xml"/></rootfiles>\n'
        '</container>')
    with zipfile.ZipFile(out_path, "w") as z:
        z.writestr("mimetype", "application/epub+zip", compress_type=zipfile.ZIP_STORED)
        z.writestr("META-INF/container.xml", container)
        z.writestr("OEBPS/content.opf", opf)
        z.writestr("OEBPS/toc.ncx", ncx)
        for i, p in enumerate(pages, 1):
            z.writestr("OEBPS/ch%d.xhtml" % i, p)


def main():
    if len(sys.argv) < 2:
        print("usage: pg_to_kindle_batch.py <candidates.json> [--limit N]")
        return 1
    books = json.load(open(sys.argv[1], encoding="utf-8"))
    limit = 3
    if "--limit" in sys.argv:
        limit = int(sys.argv[sys.argv.index("--limit") + 1])
    outdir = Path("epub_output")
    outdir.mkdir(exist_ok=True)
    print("[*] converting %d of %d candidate books to EPUB" % (min(limit, len(books)), len(books)))
    ok = 0
    for b in books[:limit]:
        bid, title, author = b["id"], b["title"], b["authors"] or "Unknown"
        print("  - #%d %s (by %s) ..." % (bid, title[:50], author[:30]))
        try:
            text = download_text(bid)
            core = strip_gutenberg_header_footer(text)
            chapters = split_chapters(core)
            out = outdir / ("%d_%s.epub" % (bid, re.sub(r"[^a-z0-9]+", "_", title.lower())[:40]))
            make_epub(title, author, chapters, str(out))
            print("      ok: %s (%d chapters)" % (out.name, len(chapters)))
            ok += 1
        except Exception as exc:
            print("      FAILED: %s" % exc)
        time.sleep(0.5)
    print("[ok] %d/%d EPUBs written to %s/" % (ok, min(limit, len(books)), outdir))
    return 0 if ok else 1


if __name__ == "__main__":
    sys.exit(main())