Inspect TF-IDF and signed chapter projections

from pathlib import Path
import base64, gzip, hashlib, json, platform
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from IPython.display import display, HTML
import re
from collections import Counter
from IPython.display import display as _display
def display(value):
    if isinstance(value,pd.DataFrame):
        table=value.to_html(max_rows=12,escape=True)
        style="<style>.vibeit-readable-table{max-width:100%;overflow-x:auto}.vibeit-readable-table table{width:max-content;max-width:none;border-collapse:collapse}.vibeit-readable-table td,.vibeit-readable-table th{white-space:nowrap!important;word-break:normal!important;overflow-wrap:normal!important}</style>"
        return _display(HTML(style+'<div class="vibeit-readable-table">'+table+'</div>'))
    return _display(value)
RECIPE_ID='hu05-text-projection'
OUTPUT_DIR=Path.cwd()/"results"/RECIPE_ID
plt.rcParams.update({"figure.dpi":120,"font.size":10,"axes.spines.top":False,"axes.spines.right":False,
                      "axes.prop_cycle":plt.cycler(color=["#18785f","#416aa6","#bc6541","#9974a5"])})
print("Python:",platform.python_version(),"; NumPy:",np.__version__,"; Pandas:",pd.__version__)
print("Working folder:",Path.cwd())
print("Core data mode: embedded snapshot; no network requests")

Python: 3.13.13 ; NumPy: 2.5.3 ; Pandas: 3.0.6
Working folder: /Users/zhiluo/Developer/vibeit-cookbook/cookbook/downloads
Core data mode: embedded snapshot; no network requests

def load_snapshot(recipe_id, expected_sha256, snapshot_key):
    """Read embedded data from this notebook, including a renamed copy."""
    for path in sorted(Path.cwd().glob("*.ipynb")):
        try:
            notebook = json.loads(path.read_text(encoding="utf-8"))
            metadata = notebook.get("metadata", {}).get("vibeit_cookbook", {})
            blob = metadata.get("snapshots", {}).get(snapshot_key)
            if metadata.get("recipe_id") != recipe_id or not blob:
                continue
            if blob.get("sha256") != expected_sha256:
                continue
            raw = gzip.decompress(base64.b64decode(blob["gzip_base64"], validate=True))
            if hashlib.sha256(raw).hexdigest() != expected_sha256:
                raise ValueError("Embedded snapshot checksum mismatch: " + snapshot_key)
            return json.loads(raw)
        except (OSError, json.JSONDecodeError):
            continue
    raise FileNotFoundError(
        "Save/open the complete .ipynb in the active working folder. "
        "The embedded snapshot could not be located for " + recipe_id)

SNAPSHOT_HASHES={'novels': 'c03763abb21b2380d2934b088f1c22a6d5a4675b17e23534e3f7343dc90084ec'}
def snapshot(key):
    return load_snapshot(RECIPE_ID,SNAPSHOT_HASHES[key],key)
print("Embedded snapshots:",list(SNAPSHOT_HASHES))

Embedded snapshots: ['novels']

def strip_ebook_wrapper(raw_text):
    """Return the analysis body while the unmodified source/complete license remain in metadata."""
    text=raw_text.replace('\r\n','\n').replace('\r','\n')
    start=re.search(r'\*\*\* START[^\n]*\*\*\*',text,re.I)
    end=re.search(r'\*\*\* END[^\n]*\*\*\*',text,re.I)
    if not start or not end or end.start()<=start.end():raise ValueError('Ebook start/end markers not found')
    return text[start.end():end.start()]

print("Method ready: strip_ebook_wrapper")

Method ready: strip_ebook_wrapper

def split_novel_chapters(body):
    """Declared two-ebook rule: repeated labels resolve to last heading, excluding their TOCs."""
    headings=list(re.finditer(r'(?im)^\s*chapter\s+([ivxlcdm]+|\d+)[.\]]*\s*$',body))
    if not headings:raise ValueError('No supported chapter headings')
    last={match.group(1).upper():match for match in headings}
    chosen=sorted(last.values(),key=lambda match:match.start());chapters=[]
    for j,match in enumerate(chosen):
        text=body[match.end():chosen[j+1].start() if j+1<len(chosen) else len(body)]
        text=re.sub(r'\[Illustration\b[^\]]*\]',' ',text,flags=re.I|re.S)
        text=text.replace('_','').replace('’',"'").strip()
        if len(text)<300:raise ValueError('Unexpected short chapter after declared TOC rule')
        chapters.append(dict(label=match.group(1).upper(),order=j+1,text=text))
    return chapters

print("Method ready: split_novel_chapters")

Method ready: split_novel_chapters

def tokenize_words(text):
    """Transparent English-letter tokenizer; punctuation/case decisions are part of the method."""
    return re.findall(r"[a-z]+(?:'[a-z]+)?",text.lower().replace('’',"'"))

print("Method ready: tokenize_words")

Method ready: tokenize_words

def tfidf_svd(documents,stopwords,min_df=3,max_df_fraction=.9,max_terms=500,components=3):
    """Exploratory TF-IDF + uncentered SVD; signed components are not topic probabilities."""
    if len(documents)<3 or components<1 or max_terms<2:raise ValueError('Need documents and valid model sizes')
    counts=[Counter(word for word in tokenize_words(text) if word not in stopwords) for text in documents]
    frequency=Counter();totals=Counter()
    for count in counts:frequency.update(count.keys());totals.update(count)
    eligible=[word for word in frequency if min_df<=frequency[word]<=max_df_fraction*len(documents)]
    terms=sorted(eligible,key=lambda word:(-totals[word],word))[:max_terms]
    if len(terms)<components:raise ValueError('Vocabulary too small for the requested components')
    matrix=np.array([[count.get(word,0) for word in terms] for count in counts],dtype=float)
    idf=np.log((1+len(documents))/(1+np.count_nonzero(matrix,axis=0)))+1
    matrix*=idf
    norms=np.linalg.norm(matrix,axis=1)
    if np.any(norms==0):raise ValueError('A document has no retained terms')
    matrix/=norms[:,None]
    left,singular,right=np.linalg.svd(matrix,full_matrices=False)
    for j in range(min(components,len(singular))):
        if right[j,np.argmax(abs(right[j]))]<0:right[j]*=-1;left[:,j]*=-1
    return dict(terms=terms,tfidf=matrix,idf=idf,scores=left[:,:components]*singular[:components],components=right[:components],singular_values=singular)

print("Method ready: tfidf_svd")

Method ready: tfidf_svd

novels=snapshot("novels");corpus={};chapter_rows=[]
for book in novels["books"]:
 chapters=split_novel_chapters(strip_ebook_wrapper(book["raw_text"]))
 tokens=tokenize_words(" ".join(chapter["text"] for chapter in chapters))
 corpus[book["id"]]={"title":book["title"],"author":book["author"],"chapters":chapters,"tokens":tokens}
 for chapter in chapters:chapter_rows.append({"book_id":book["id"],"book":book["title"],"chapter":chapter["order"],"source_label":chapter["label"],"characters":len(chapter["text"]),"tokens":len(tokenize_words(chapter["text"])),"text":chapter["text"]})
chapter_table=pd.DataFrame(chapter_rows)
assert len(corpus[1342]["chapters"])==61 and len(corpus[84]["chapters"])==24
print(novels["source_notice"])
print("Declared chapter counts:",{b:len(corpus[b]["chapters"]) for b in corpus})
display(chapter_table.drop(columns="text").head(8))
stopwords=set("a an and are as at be been being but by can could did do does for from had has have he her hers him his i if in is it its me more most my no not of on one or our she so some than that the their them then there these they this those to too was we were what when where which who why will with would you your".split())
documents=chapter_table.text.tolist();model=tfidf_svd(documents,stopwords,min_df=3,max_df_fraction=.9,max_terms=500,components=3)
print("Documents:",len(documents),"; retained terms:",len(model["terms"]),"; TF-IDF row normalization: L2")

These source ebooks are available free of charge. Complete unmodified source texts and their licenses are included in this snapshot. Project Gutenberg is acknowledged as the transcription source; its name is not used as a course title or endorsement.
Declared chapter counts: {1342: 61, 84: 24}

Documents: 85 ; retained terms: 500 ; TF-IDF row normalization: L2

component=1;loading=model["components"][component]
selected=np.argsort(abs(loading))[-15:];selected=selected[np.argsort(loading[selected])]
fig,ax=plt.subplots(figsize=(8,5));ax.barh([model["terms"][j] for j in selected],loading[selected])
ax.axvline(0,color="black",lw=1);ax.set(xlabel="Signed SVD coefficient",title="Second uncentered text direction; not topic probabilities");plt.tight_layout();plt.show()

fig,ax=plt.subplots(figsize=(8,5))
for book_id in corpus:
 mask=chapter_table.book_id.eq(book_id).to_numpy();ax.scatter(model["scores"][mask,0],model["scores"][mask,1],label=corpus[book_id]["title"],s=25)
ax.set(xlabel="SVD1 score (uncentered)",ylabel="SVD2 score",title="Exploratory chapter geometry");ax.legend();plt.tight_layout();plt.show()

similarity=model["tfidf"]@model["tfidf"].T
fig,ax=plt.subplots(figsize=(7,6));im=ax.imshow(similarity,vmin=0,vmax=1,cmap="viridis")
ax.set(xlabel="Source chapter-document order",ylabel="Source chapter-document order",title="TF-IDF cosine similarity");fig.colorbar(im,ax=ax,label="Cosine similarity");plt.tight_layout();plt.show()

assert np.allclose(np.linalg.norm(model["tfidf"],axis=1),1)
assert np.allclose(similarity,similarity.T) and np.allclose(np.diag(similarity),1)
assert np.allclose(model["components"]@model["components"].T,np.eye(3))
OUTPUT_DIR.mkdir(parents=True,exist_ok=True)
pd.DataFrame(model["scores"],columns=["SVD1","SVD2","SVD3"]).assign(book=chapter_table.book,chapter=chapter_table.chapter).to_csv(OUTPUT_DIR/"chapter-projections.csv",index=False)
pd.DataFrame(model["components"].T,index=model["terms"],columns=["SVD1","SVD2","SVD3"]).to_csv(OUTPUT_DIR/"signed-term-directions.csv")
print("Normalization and orthogonality checked; exports:",OUTPUT_DIR)

Normalization and orthogonality checked; exports: /Users/zhiluo/Developer/vibeit-cookbook/cookbook/downloads/results/hu05-text-projection

RUN_ONLINE_UPDATE=False
if RUN_ONLINE_UPDATE:
    import requests
    try:
        response=requests.get('https://www.gutenberg.org/ebooks/1342',timeout=20)
        response.raise_for_status()
        print("Source:",response.url)
        print(response.text[:700])
    except requests.RequestException as error:
        print("Online extension unavailable:",type(error).__name__,str(error)[:160])
else:
    print("Online extension skipped; embedded core data unchanged")

Online extension skipped; embedded core data unchanged