generated from john/python-template
Misc updates I don't remember making
This commit is contained in:
Jim Lancaster
@@ -0,0 +1,166 @@
|
||||
import csv
|
||||
from pathlib import Path
|
||||
import re
|
||||
import unicodedata
|
||||
import pdfplumber
|
||||
|
||||
# Configuration
|
||||
SOURCE_DIR = Path(r"C:\Proton Drive\bbchops\My files\Archive\FamilySearch")
|
||||
OUTPUT_FILE = SOURCE_DIR / "familysearch_people_with_links.csv"
|
||||
|
||||
# Regex pattern for FamilySearch Person IDs
|
||||
FS_ID_PATTERN = re.compile(r"([A-Z0-9]{4}-[A-Z0-9]{3,4})")
|
||||
|
||||
# Badge/Artifact labels to strip
|
||||
BADGE_LABELS = {
|
||||
"MAYFLOWER",
|
||||
"DESCENDANT",
|
||||
"PATRIOT",
|
||||
"DEAD",
|
||||
"END",
|
||||
"FAMILYSEARCH",
|
||||
}
|
||||
|
||||
|
||||
def clean_text(text: str) -> str:
|
||||
"""Normalizes dashes, unicode artifacts, and collapses whitespace."""
|
||||
if not text:
|
||||
return ""
|
||||
# Replace non-breaking spaces, en-dashes, em-dashes, and special hyphens
|
||||
text = unicodedata.normalize("NFKD", text)
|
||||
text = re.sub(r"[\u2010-\u2015\u2212\uFE58\uFE63\uFF0D]", "-", text)
|
||||
return " ".join(text.split()).strip()
|
||||
|
||||
|
||||
def extract_people_from_pdf(pdf_path: Path):
|
||||
extracted_records = []
|
||||
|
||||
try:
|
||||
with pdfplumber.open(str(pdf_path)) as pdf:
|
||||
for page in pdf.pages:
|
||||
hyperlinks = []
|
||||
for annot in page.hyperlinks:
|
||||
uri = annot.get("uri", "")
|
||||
match = FS_ID_PATTERN.search(uri)
|
||||
if match:
|
||||
hyperlinks.append(
|
||||
{
|
||||
"fs_id": match.group(1),
|
||||
"uri": uri,
|
||||
"x0": annot.get("x0", 0),
|
||||
"top": annot.get("top", 0),
|
||||
"x1": annot.get("x1", 0),
|
||||
"bottom": annot.get("bottom", 0),
|
||||
}
|
||||
)
|
||||
|
||||
words = page.extract_words(
|
||||
keep_blank_chars=False, x_tolerance=2, y_tolerance=2
|
||||
)
|
||||
|
||||
for link in hyperlinks:
|
||||
lx_center = (link["x0"] + link["x1"]) / 2
|
||||
|
||||
# Narrow horizontal span (+/- 28px) to isolate this specific node column
|
||||
# Search from link top down to +75px
|
||||
matched_words = [
|
||||
w
|
||||
for w in words
|
||||
if abs(((w["x0"] + w["x1"]) / 2) - lx_center) <= 28
|
||||
and (link["top"] - 5) <= w["top"] <= (link["bottom"] + 75)
|
||||
]
|
||||
|
||||
if not matched_words:
|
||||
continue
|
||||
|
||||
# Group words into lines by vertical position
|
||||
matched_words.sort(key=lambda w: (w["top"], w["x0"]))
|
||||
lines = []
|
||||
current_line = []
|
||||
current_y = None
|
||||
|
||||
for w in matched_words:
|
||||
if current_y is None or abs(w["top"] - current_y) <= 4:
|
||||
current_line.append(w["text"])
|
||||
current_y = w["top"]
|
||||
else:
|
||||
lines.append(" ".join(current_line))
|
||||
current_line = [w["text"]]
|
||||
current_y = w["top"]
|
||||
if current_line:
|
||||
lines.append(" ".join(current_line))
|
||||
|
||||
# Parse dates and clean names line-by-line
|
||||
name_parts = []
|
||||
date_range = ""
|
||||
|
||||
for line in lines:
|
||||
cleaned_line = clean_text(line)
|
||||
|
||||
# Filter out badge text
|
||||
if (
|
||||
cleaned_line.upper() in BADGE_LABELS
|
||||
or "MAYFLOWER" in cleaned_line.upper()
|
||||
):
|
||||
continue
|
||||
|
||||
# Check for Date Range: e.g. 1890-1975 or 1715-Deceased
|
||||
date_match = re.search(
|
||||
r"(\b\d{4}\s*-\s*(?:\d{4}|Deceased)\b)",
|
||||
cleaned_line,
|
||||
re.IGNORECASE,
|
||||
)
|
||||
if date_match:
|
||||
date_range = date_match.group(1).replace(" ", "")
|
||||
# If text exists before the date on the same line, retain it
|
||||
prefix = cleaned_line[: date_match.start()].strip()
|
||||
if prefix:
|
||||
name_parts.append(prefix)
|
||||
else:
|
||||
name_parts.append(cleaned_line)
|
||||
|
||||
person_name = clean_text(" ".join(name_parts))
|
||||
|
||||
extracted_records.append(
|
||||
{
|
||||
"File": pdf_path.name,
|
||||
"Name": person_name,
|
||||
"Date_Range": date_range,
|
||||
"FS_ID": link["fs_id"],
|
||||
"URL": link["uri"],
|
||||
}
|
||||
)
|
||||
|
||||
except Exception as err:
|
||||
print(f"[ERROR] Failed reading {pdf_path.name}: {err}")
|
||||
|
||||
return extracted_records
|
||||
|
||||
|
||||
def batch_extract(source_dir: Path, output_csv: Path):
|
||||
pdf_files = sorted(source_dir.glob("*.pdf"))
|
||||
if not pdf_files:
|
||||
print(f"No PDF files found in: {source_dir}")
|
||||
return
|
||||
|
||||
print(f"Processing {len(pdf_files)} PDF(s)...")
|
||||
|
||||
all_records = []
|
||||
for pdf_path in pdf_files:
|
||||
records = extract_people_from_pdf(pdf_path)
|
||||
all_records.extend(records)
|
||||
print(f" - {pdf_path.name}: {len(records)} person/link pair(s)")
|
||||
|
||||
output_csv.parent.mkdir(parents=True, exist_ok=True)
|
||||
# Use utf-8-sig so Excel automatically reads hyphens and Unicode without Mojibake
|
||||
with open(output_csv, "w", newline="", encoding="utf-8-sig") as f:
|
||||
fieldnames = ["File", "Name", "Date_Range", "FS_ID", "URL"]
|
||||
writer = csv.DictWriter(f, fieldnames=fieldnames)
|
||||
writer.writeheader()
|
||||
writer.writerows(all_records)
|
||||
|
||||
print(f"\nDone. Saved to:\n{output_csv}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
batch_extract(SOURCE_DIR, OUTPUT_FILE)
|
||||
Reference in New Issue
Block a user