testing out OCR viewer of OCR'd text
This commit is contained in:
nathan
33
scripts/check_eod.py
Normal file
33
scripts/check_eod.py
Normal file
@@ -0,0 +1,33 @@
|
||||
import json, pathlib
|
||||
from collections import Counter, defaultdict
|
||||
|
||||
data = json.loads(pathlib.Path('outputs/Calgary-Highlanders_Sep44_positions.json').read_text())
|
||||
|
||||
cats = Counter(p['category'] for p in data)
|
||||
print('=== CATEGORIES ===')
|
||||
for k, v in sorted(cats.items()):
|
||||
print(f' {k}: {v}')
|
||||
|
||||
eod = [p for p in data if p['is_end_of_day']]
|
||||
print(f'\n=== EOD entries: {len(eod)} ===')
|
||||
for p in eod:
|
||||
print(f" {str(p['date']):<16} cat={p['category']:<14} grid={str(p['grid']):<8} place={p['place_name']}")
|
||||
|
||||
eod_by_date = defaultdict(list)
|
||||
for p in eod:
|
||||
eod_by_date[p['date']].append(p)
|
||||
multi = {d: ps for d, ps in eod_by_date.items() if len(ps) > 1}
|
||||
if multi:
|
||||
print('\nWARNING - multiple EOD on same date:')
|
||||
for d, ps in multi.items():
|
||||
print(f' {d}: {len(ps)} entries')
|
||||
else:
|
||||
print('\nOK: exactly one EOD per date')
|
||||
|
||||
# List dates with no EOD
|
||||
all_dates = sorted({p['date'] for p in data if p['date']})
|
||||
eod_dates = set(eod_by_date.keys())
|
||||
missing = [d for d in all_dates if d not in eod_dates]
|
||||
if missing:
|
||||
print(f'\nDates with NO EOD: {missing}')
|
||||
|
||||
721
scripts/extract_positions.py
Normal file
721
scripts/extract_positions.py
Normal file
@@ -0,0 +1,721 @@
|
||||
"""
|
||||
Extract positional data from Calgary Highlanders War Diary Sep 44 (pages 7-57).
|
||||
Outputs a JSON array of position objects.
|
||||
"""
|
||||
|
||||
import re
|
||||
import json
|
||||
from html.parser import HTMLParser
|
||||
|
||||
INPUT_FILE = r"C:\Users\natha\IdeaProjects\AI-Prototype\Inputs\ocr-output\Calgary-Highlanders_War-Diary_Sep44_olmocr.md"
|
||||
OUTPUT_FILE = r"C:\Users\natha\IdeaProjects\AI-Prototype\outputs\Calgary-Highlanders_Sep44_positions.json"
|
||||
|
||||
# ── helpers ──────────────────────────────────────────────────────────────────
|
||||
|
||||
def expand_grid(raw: str) -> tuple[str, bool]:
|
||||
"""
|
||||
Return (6-figure-string, inferred).
|
||||
4-figure AABB → centre of 1km square = AA5 BB5 (grid_inferred=True)
|
||||
6-figure AAABBB → returned as-is (grid_inferred=False)
|
||||
8-figure AAAABBBB → truncate to AAA BBB (grid_inferred=False)
|
||||
"""
|
||||
s = re.sub(r'[^0-9]', '', raw)
|
||||
if len(s) == 4:
|
||||
# e.g. "8450" → easting 84, northing 50 → centre 845, 505
|
||||
e = s[0:2] + '5'
|
||||
n = s[2:4] + '5'
|
||||
return e + n, True
|
||||
if len(s) == 6:
|
||||
return s, False
|
||||
if len(s) == 8:
|
||||
# AAAABBBB: easting = AAAA (take first 3), northing = BBBB (take first 3)
|
||||
# e.g. 15618050 → E=1561 → 156, N=8050 → 805 → "156805"
|
||||
e = s[0:3]
|
||||
n = s[4:7]
|
||||
return e + n, False
|
||||
return s, False
|
||||
|
||||
|
||||
_GRID_RE = re.compile(
|
||||
r'\b'
|
||||
r'(?:MR\s*|GR\s*)?' # optional prefix
|
||||
r'([0-9]{3,4}\s*[0-9]{3,4})' # 6 or 4+4 digit grid
|
||||
r'\b',
|
||||
re.IGNORECASE
|
||||
)
|
||||
|
||||
def find_grids(text: str):
|
||||
"""Return list of (raw_match, cleaned_digits)."""
|
||||
found = []
|
||||
for m in _GRID_RE.finditer(text):
|
||||
raw = m.group(1)
|
||||
digits = re.sub(r'\s', '', raw)
|
||||
if len(digits) in (4, 6, 8):
|
||||
found.append((m.group(0).strip(), digits))
|
||||
# also pick up standalone 6-digit runs not already caught
|
||||
for m in re.finditer(r'\b([0-9]{6})\b', text):
|
||||
digits = m.group(1)
|
||||
already = any(d == digits for _, d in found)
|
||||
if not already:
|
||||
found.append((digits, digits))
|
||||
return found
|
||||
|
||||
|
||||
# ── HTML strip ────────────────────────────────────────────────────────────────
|
||||
|
||||
class _Stripper(HTMLParser):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.parts = []
|
||||
def handle_data(self, data):
|
||||
self.parts.append(data)
|
||||
def get_text(self):
|
||||
return ' '.join(p for p in self.parts if p.strip())
|
||||
|
||||
def strip_html(html_str: str) -> str:
|
||||
s = _Stripper()
|
||||
# replace <br> with spaces for readability
|
||||
html_str = re.sub(r'<br\s*/?>', ' ', html_str, flags=re.IGNORECASE)
|
||||
s.feed(html_str)
|
||||
return s.get_text()
|
||||
|
||||
|
||||
# ── categorise ───────────────────────────────────────────────────────────────
|
||||
|
||||
_SUBUNIT_RE = re.compile(
|
||||
r'\b(Able\s+Coy|Baker\s+Coy|Charlie\s+Coy|Dog\s+Coy|'
|
||||
r'"A"\s+Coy|"B"\s+Coy|"C"\s+Coy|"D"\s+Coy|'
|
||||
r"'A'\s+Coy|'B'\s+Coy|'C'\s+Coy|'D'\s+Coy|'Able'\s+Coy|'Baker'\s+Coy|'Charlie'\s+Coy|'Dog'\s+Coy|"
|
||||
r'Carrier\s+Platoon|carriers|Pioneer\s+Platoon|pioneers|scouts?|Scout\s+Platoon|'
|
||||
r'Support\s+Coy|Anti[-\s]?[Tt]ank\s+Pl(?:atoon)?|mortar\s+platoon|'
|
||||
r'\d+\s*Pl(?:atoon)?|'
|
||||
r'\d+\s*Sec(?:tion)?)\b',
|
||||
re.IGNORECASE
|
||||
)
|
||||
_ENEMY_RE = re.compile(
|
||||
r'\b(enemy|Jerry|Hun|Boche|MG\s*pos|MMG\s*pos|sniper|'
|
||||
r'counter.attack|strongpoint|mortar\s*pos|machine\s*gun|'
|
||||
r'block\s*house|pill\s*box|88mm|S\.S\.|SS troop)\b',
|
||||
re.IGNORECASE
|
||||
)
|
||||
_PATROL_RE = re.compile(
|
||||
r'\b(patrol|recce\s*patrol|fighting\s*patrol|'
|
||||
r'standing\s*patrol|OP|O\.P\.|observation\s*post)\b',
|
||||
re.IGNORECASE
|
||||
)
|
||||
_FRIENDLY_UNITS = re.compile(
|
||||
r'\b(R\.H\.C\.|Black\s*Watch|RHC|R\s*de\s*Mais|'
|
||||
r'Fus\s*M\.R\.|F\.M\.R\.|R\.R\.C\.|'
|
||||
r'4\s*S\.S\.|Royal\s*Regt|S\.Sask|'
|
||||
r'Tor\s*Scots|Toronto\s*Scots|'
|
||||
r'White\s*Brig(?:ade)?|F\.F\.I\.|Maquis)\b',
|
||||
re.IGNORECASE
|
||||
)
|
||||
# Tac HQ and Bn HQ in narrative
|
||||
_TAC_HQ_RE = re.compile(
|
||||
r'\bTac\s*H(?:Q)?\b',
|
||||
re.IGNORECASE
|
||||
)
|
||||
_BN_HQ_RE = re.compile(
|
||||
r'\b(?:Bn\.?\s*H\.?Q\.?|Battalion\s*H\.?Q\.?|'
|
||||
r'Battle\s*H\.?Q\.?|Command\s*Post|'
|
||||
r'moved\s+(?:his\s+)?H\.?Q\.|'
|
||||
r'set\s+up\s+H\.?Q\.|'
|
||||
r'took\s+up\s+(?:a\s+)?H\.?Q\.)\b',
|
||||
re.IGNORECASE
|
||||
)
|
||||
|
||||
def categorise(sentence: str, place_name: str | None, subunit: str | None,
|
||||
friendly: str | None, is_place_col: bool = False) -> str:
|
||||
"""
|
||||
Categories (in priority order):
|
||||
TAC_HQ – Tactical HQ position
|
||||
BN_HQ – Battalion HQ position (incl. all place-column entries)
|
||||
FRIENDLY – another friendly unit's position
|
||||
ENEMY – enemy position / feature
|
||||
PATROL – patrol route/endpoint
|
||||
SUBUNIT – company or platoon position
|
||||
UNIT_MOVEMENT – general battalion move/position
|
||||
MISC – catch-all
|
||||
"""
|
||||
if is_place_col:
|
||||
# Place column always records where Bn HQ was
|
||||
if _TAC_HQ_RE.search(sentence):
|
||||
return "TAC_HQ"
|
||||
return "BN_HQ"
|
||||
if _TAC_HQ_RE.search(sentence):
|
||||
return "TAC_HQ"
|
||||
if _BN_HQ_RE.search(sentence):
|
||||
return "BN_HQ"
|
||||
if friendly:
|
||||
return "FRIENDLY"
|
||||
if _ENEMY_RE.search(sentence):
|
||||
return "ENEMY"
|
||||
if _PATROL_RE.search(sentence):
|
||||
return "PATROL"
|
||||
if subunit:
|
||||
return "SUBUNIT"
|
||||
return "UNIT_MOVEMENT"
|
||||
|
||||
|
||||
def extract_subunit(sentence: str) -> str | None:
|
||||
m = _SUBUNIT_RE.search(sentence)
|
||||
if m:
|
||||
return m.group(0).strip()
|
||||
return None
|
||||
|
||||
def extract_friendly(sentence: str) -> str | None:
|
||||
m = _FRIENDLY_UNITS.search(sentence)
|
||||
if m:
|
||||
return m.group(0).strip()
|
||||
return None
|
||||
|
||||
|
||||
# ── split sentences ──────────────────────────────────────────────────────────
|
||||
|
||||
def split_sentences(text: str):
|
||||
"""Rough sentence splitter – split on '. ' or '<br>'."""
|
||||
# normalise
|
||||
text = re.sub(r'\s+', ' ', text).strip()
|
||||
parts = re.split(r'(?<=[.!?])\s+(?=[A-Z"\'(])', text)
|
||||
return [p.strip() for p in parts if p.strip()]
|
||||
|
||||
|
||||
# ── page extractor ────────────────────────────────────────────────────────────
|
||||
|
||||
_PAGE_RE = re.compile(r'^## Page (\d+)\s*$', re.MULTILINE)
|
||||
|
||||
def extract_pages(text: str, first: int, last: int) -> str:
|
||||
pages = list(_PAGE_RE.finditer(text))
|
||||
start_idx = None
|
||||
end_idx = len(text)
|
||||
for i, m in enumerate(pages):
|
||||
n = int(m.group(1))
|
||||
if n == first and start_idx is None:
|
||||
start_idx = m.start()
|
||||
if n == last + 1 and start_idx is not None:
|
||||
end_idx = m.start()
|
||||
break
|
||||
if start_idx is None:
|
||||
return ""
|
||||
return text[start_idx:end_idx]
|
||||
|
||||
|
||||
# ── table row parser ──────────────────────────────────────────────────────────
|
||||
|
||||
_ROW_RE = re.compile(r'<tr>(.*?)</tr>', re.DOTALL | re.IGNORECASE)
|
||||
_CELL_RE = re.compile(r'<t[dh][^>]*>(.*?)</t[dh]>', re.DOTALL | re.IGNORECASE)
|
||||
|
||||
def parse_table_rows(table_html: str) -> list[dict]:
|
||||
"""Parse a single HTML table into list of {place, date, hour, summary} dicts."""
|
||||
rows = []
|
||||
for row_m in _ROW_RE.finditer(table_html):
|
||||
cells = [strip_html(c.group(1)).strip()
|
||||
for c in _CELL_RE.finditer(row_m.group(1))]
|
||||
if len(cells) < 4:
|
||||
continue
|
||||
# skip header rows
|
||||
if re.match(r'place|date|hour|summary|no\.', cells[0], re.IGNORECASE):
|
||||
continue
|
||||
place = cells[0] if cells[0] else None
|
||||
date = cells[1] if len(cells) > 1 else None
|
||||
hour = cells[2] if len(cells) > 2 else None
|
||||
summary = cells[3] if len(cells) > 3 else ""
|
||||
rows.append(dict(place=place, date=date, hour=hour, summary=summary))
|
||||
return rows
|
||||
|
||||
|
||||
# ── sheet ref extractor ───────────────────────────────────────────────────────
|
||||
_SHEET_RE = re.compile(
|
||||
# must start with a digit to avoid matching "Sheet Ste. Foy" etc.
|
||||
r'Sheet\s+(\d[A-Z0-9]*(?:\s*[&\-]\s*\d[A-Z0-9]*)?)',
|
||||
re.IGNORECASE
|
||||
)
|
||||
|
||||
def extract_sheet(place_text: str) -> str | None:
|
||||
if not place_text:
|
||||
return None
|
||||
# find the LAST occurrence (most specific sheet reference in column)
|
||||
matches = list(_SHEET_RE.finditer(place_text))
|
||||
if matches:
|
||||
return "Sheet " + matches[-1].group(1).strip()
|
||||
return None
|
||||
|
||||
# also pull MR grid from place column
|
||||
_PLACE_MR_RE = re.compile(r'MR\s*([0-9]{4,8}|\d{2,3}\s*\d{2,3})', re.IGNORECASE)
|
||||
|
||||
def extract_place_name(place_text: str) -> str | None:
|
||||
"""
|
||||
Extract the primary place name from the Place column.
|
||||
Strip country, bare MR tokens, Sheet refs, standalone digits,
|
||||
and Tac H / Fort coord suffixes.
|
||||
"""
|
||||
if not place_text:
|
||||
return None
|
||||
s = place_text
|
||||
# remove country names
|
||||
s = re.sub(r'\b(France|Belgium|Holland|Netherlands)\b', '', s, flags=re.IGNORECASE)
|
||||
# remove Sheet + value (digit-led)
|
||||
s = re.sub(r'\bSheet\s+\d[\w\s&\-]*', ' ', s, flags=re.IGNORECASE)
|
||||
# remove bare "Sheet" not followed by a digit
|
||||
s = re.sub(r'\bSheet\b', ' ', s, flags=re.IGNORECASE)
|
||||
# remove MR + optional digits/spaces
|
||||
s = re.sub(r'\bMR\b\s*[\d\s]*', ' ', s, flags=re.IGNORECASE)
|
||||
# remove Tac H + coords
|
||||
s = re.sub(r'\bTac\s*H\s*[\d]+\b', ' ', s, flags=re.IGNORECASE)
|
||||
# remove bare 4-8 digit grid refs
|
||||
s = re.sub(r'\b\d{4,8}\b', ' ', s)
|
||||
# remove "X Pub" and similar codes
|
||||
s = re.sub(r'\bX[\s\-]?Pub\b', ' ', s, flags=re.IGNORECASE)
|
||||
# remove orphan punctuation / collapse whitespace
|
||||
s = re.sub(r'[/\\|]', ' ', s)
|
||||
s = re.sub(r'\s+', ' ', s).strip().strip('.,-()')
|
||||
# If multiple tokens, take first meaningful phrase (up to the first double space separator)
|
||||
parts = [p.strip() for p in re.split(r'\s{2,}', s) if p.strip()]
|
||||
result = parts[0] if parts else s.strip()
|
||||
# Trim trailing stray words like "(outskirts...)" parenthetical noise
|
||||
result = re.sub(r'\s*\(.*\)\s*$', '', result).strip()
|
||||
return result if result else None
|
||||
|
||||
|
||||
def clean_date(raw: str) -> str | None:
|
||||
if not raw:
|
||||
return None
|
||||
d = re.sub(r'\s+', ' ', raw).strip()
|
||||
# Remove parenthetical notes like "(Cont)"
|
||||
d = re.sub(r'\(.*?\)', '', d).strip()
|
||||
# Extract DD Mon [YY] — ignore anything else in the cell (e.g. embedded grid refs)
|
||||
m = re.search(
|
||||
r'(\d{1,2})\s+(Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)\s*(44|1944)?',
|
||||
d, re.IGNORECASE
|
||||
)
|
||||
if m:
|
||||
day = m.group(1)
|
||||
mon = m.group(2).capitalize()
|
||||
# always use "44" as year for this diary
|
||||
return f"{day} {mon} 44"
|
||||
return None # unrecognisable — don't inherit garbage
|
||||
|
||||
|
||||
def clean_hour(raw: str) -> str | None:
|
||||
if not raw:
|
||||
return None
|
||||
raw = raw.strip()
|
||||
# Reject empty, the bare year "44", and plain digit-strings ≥4 chars (grids)
|
||||
if not raw or raw in ('0', '44', '1944'):
|
||||
return None
|
||||
if re.match(r'^\d{4,6}$', raw):
|
||||
return None # grid ref leaked into hour column, not a time
|
||||
# Extract a valid HHMM time
|
||||
m = re.search(r'\b([012]\d[0-5]\d)\b', raw)
|
||||
if m:
|
||||
return m.group(1)
|
||||
return None
|
||||
|
||||
|
||||
def hour_col_grid(raw: str) -> str | None:
|
||||
"""
|
||||
If the Hour column contains a plain 4-digit grid ref (not a time),
|
||||
return the digit string so it can be added as a BN_HQ entry.
|
||||
"""
|
||||
if not raw:
|
||||
return None
|
||||
raw = raw.strip()
|
||||
if re.match(r'^\d{4}$', raw):
|
||||
# Confirm it can't be a time (hours > 23 or minutes > 59)
|
||||
h, m2 = int(raw[:2]), int(raw[2:])
|
||||
if h > 23 or m2 > 59:
|
||||
return raw
|
||||
return None
|
||||
|
||||
|
||||
# ── place-column MR grid ──────────────────────────────────────────────────────
|
||||
|
||||
def place_col_positions(place_text: str, date: str, sheet: str) -> list[dict]:
|
||||
"""
|
||||
Extract grids from the Place column.
|
||||
Handles:
|
||||
(a) NAME + MR + GRID (e.g. 'Ste. Foy MR 2553')
|
||||
(b) NAME + bare 6-digit grid (e.g. 'Chateau Helleputte 769969')
|
||||
Returns one entry per unique grid found.
|
||||
"""
|
||||
if not place_text:
|
||||
return []
|
||||
|
||||
results = []
|
||||
seen_digits = set()
|
||||
|
||||
def _clean_label(raw: str) -> str | None:
|
||||
s = raw
|
||||
s = re.sub(r'\b(France|Belgium|Holland|Netherlands)\b', '', s, flags=re.IGNORECASE)
|
||||
s = re.sub(r'\bSheet\b[\w\s&\-]*', '', s, flags=re.IGNORECASE)
|
||||
s = re.sub(r'\bMR\b', '', s, flags=re.IGNORECASE)
|
||||
s = re.sub(r'\b\d{4,8}\b', '', s)
|
||||
s = re.sub(r'\s+', ' ', s).strip().strip('.,-()')
|
||||
# take rightmost meaningful word group
|
||||
parts = [p.strip() for p in re.split(r'\s{2,}', s) if p.strip()]
|
||||
r = parts[-1] if parts else s.strip()
|
||||
return r if r else None
|
||||
|
||||
# ── (a) NAME MR GRID ─────────────────────────────────────────────────────
|
||||
# Require digits immediately after MR (captures 4-digit and 6-digit grids)
|
||||
mr_pattern = re.compile(
|
||||
r'([A-Z][A-Za-zÀ-ÿ\s\.\-\']*?)\s+MR\s*(\d[\d\s]+\d)',
|
||||
re.IGNORECASE
|
||||
)
|
||||
for m in mr_pattern.finditer(place_text):
|
||||
label_raw = m.group(1)
|
||||
raw_digits = re.sub(r'\s', '', m.group(2))
|
||||
if len(raw_digits) not in (4, 6, 8):
|
||||
continue
|
||||
if raw_digits in seen_digits:
|
||||
continue
|
||||
seen_digits.add(raw_digits)
|
||||
|
||||
grid, inferred = expand_grid(raw_digits)
|
||||
if len(grid) != 6:
|
||||
grid = None
|
||||
|
||||
label = _clean_label(label_raw)
|
||||
# Determine category: place column = BN_HQ unless Tac H explicit
|
||||
cat = categorise(place_text, label, None, None, is_place_col=True)
|
||||
results.append(dict(
|
||||
date=date, hour=None,
|
||||
grid=grid, grid_inferred=inferred,
|
||||
place_name=label,
|
||||
sheet_ref=sheet,
|
||||
category=cat,
|
||||
subunit=None, friendly_unit=None, is_end_of_day=False,
|
||||
context=f"Place column: {label or '?'} MR {raw_digits}"
|
||||
))
|
||||
|
||||
# ── (b) NAME followed directly by bare 6-digit grid ──────────────────────
|
||||
bare_pattern = re.compile(
|
||||
r'([A-Z][A-Za-zÀ-ÿ\s\.\-\']+?)\s+(\d{6})\b'
|
||||
)
|
||||
for m in bare_pattern.finditer(place_text):
|
||||
label_raw = m.group(1)
|
||||
raw_digits = m.group(2)
|
||||
if raw_digits in seen_digits:
|
||||
continue
|
||||
# Skip if label contains only noise words
|
||||
clean = re.sub(r'\b(France|Belgium|Holland|Netherlands|Sheet|MR)\b', '',
|
||||
label_raw, flags=re.IGNORECASE).strip()
|
||||
if not clean:
|
||||
continue
|
||||
seen_digits.add(raw_digits)
|
||||
|
||||
grid, inferred = expand_grid(raw_digits)
|
||||
label = _clean_label(label_raw)
|
||||
cat = categorise(place_text, label, None, None, is_place_col=True)
|
||||
results.append(dict(
|
||||
date=date, hour=None,
|
||||
grid=grid, grid_inferred=inferred,
|
||||
place_name=label,
|
||||
sheet_ref=sheet,
|
||||
category=cat,
|
||||
subunit=None, friendly_unit=None, is_end_of_day=False,
|
||||
context=f"Place column: {label or '?'} {raw_digits}"
|
||||
))
|
||||
|
||||
# ── (c) NAME followed by bare 4-digit grid (no MR prefix) ────────────────
|
||||
bare_4_pattern = re.compile(
|
||||
r'([A-Z][A-Za-zÀ-ÿ\s\.\-\']+?)\s+(\d{4})\b'
|
||||
)
|
||||
for m in bare_4_pattern.finditer(place_text):
|
||||
label_raw = m.group(1)
|
||||
raw_digits = m.group(2)
|
||||
if raw_digits in seen_digits:
|
||||
continue
|
||||
# Only treat as grid if NOT a valid time (i.e. HH>23 or MM>59)
|
||||
h_val, mn_val = int(raw_digits[:2]), int(raw_digits[2:])
|
||||
if h_val <= 23 and mn_val <= 59:
|
||||
continue
|
||||
clean = re.sub(r'\b(France|Belgium|Holland|Netherlands|Sheet|MR)\b', '',
|
||||
label_raw, flags=re.IGNORECASE).strip()
|
||||
if not clean:
|
||||
continue
|
||||
seen_digits.add(raw_digits)
|
||||
|
||||
grid, inferred = expand_grid(raw_digits)
|
||||
label = _clean_label(label_raw)
|
||||
cat = categorise(label_raw, label, None, None, is_place_col=True)
|
||||
results.append(dict(
|
||||
date=date, hour=None,
|
||||
grid=grid if len(grid) == 6 else None,
|
||||
grid_inferred=inferred,
|
||||
place_name=label,
|
||||
sheet_ref=sheet,
|
||||
category=cat,
|
||||
subunit=None, friendly_unit=None, is_end_of_day=False,
|
||||
context=f"Place column: {label or '?'} {raw_digits}"
|
||||
))
|
||||
|
||||
return results
|
||||
|
||||
|
||||
# ── main extraction ───────────────────────────────────────────────────────────
|
||||
|
||||
def extract_positions(md_text: str) -> list[dict]:
|
||||
section = extract_pages(md_text, 7, 57)
|
||||
|
||||
# find all tables
|
||||
table_re = re.compile(r'<table>(.*?)</table>', re.DOTALL | re.IGNORECASE)
|
||||
|
||||
all_positions = []
|
||||
last_date = None
|
||||
last_sheet = None
|
||||
|
||||
for tbl_m in table_re.finditer(section):
|
||||
tbl_html = tbl_m.group(0)
|
||||
rows = parse_table_rows(tbl_html)
|
||||
if not rows:
|
||||
continue
|
||||
|
||||
for row in rows:
|
||||
raw_date = clean_date(row.get('date', '') or '')
|
||||
raw_hour = clean_hour(row.get('hour', '') or '')
|
||||
place_text = row.get('place', '') or ''
|
||||
summary = row.get('summary', '') or ''
|
||||
|
||||
# update tracking state
|
||||
if raw_date:
|
||||
last_date = raw_date
|
||||
cur_date = last_date
|
||||
|
||||
sheet = extract_sheet(place_text) or last_sheet
|
||||
if sheet and re.search(r'\d', sheet):
|
||||
last_sheet = sheet
|
||||
|
||||
# ── place-column positions ──
|
||||
for pos in place_col_positions(place_text, cur_date, sheet):
|
||||
pos['hour'] = raw_hour
|
||||
all_positions.append(pos)
|
||||
|
||||
# ── hour-column grid (OCR sometimes puts MR ref here) ──
|
||||
hr_raw = row.get('hour', '') or ''
|
||||
hcg = hour_col_grid(hr_raw)
|
||||
if hcg:
|
||||
grid, inferred = expand_grid(hcg)
|
||||
col_place = extract_place_name(place_text)
|
||||
all_positions.append(dict(
|
||||
date=cur_date, hour=None,
|
||||
grid=grid if len(grid) == 6 else None,
|
||||
grid_inferred=inferred,
|
||||
place_name=col_place,
|
||||
sheet_ref=last_sheet,
|
||||
category="BN_HQ",
|
||||
subunit=None, friendly_unit=None, is_end_of_day=False,
|
||||
context=f"Place column (hour field): {col_place or '?'} {hcg}"
|
||||
))
|
||||
|
||||
# ── summary / narrative positions ──
|
||||
sentences = split_sentences(summary)
|
||||
# track whether this is the last sentence in the entry
|
||||
last_sentence_idx = len(sentences) - 1
|
||||
|
||||
# collect all (sentence_idx, grid_str, raw_context) tuples for this row
|
||||
row_matches = []
|
||||
for s_idx, sent in enumerate(sentences):
|
||||
grids = find_grids(sent)
|
||||
if grids:
|
||||
for raw_match, digits in grids:
|
||||
row_matches.append((s_idx, sent, digits, raw_match))
|
||||
else:
|
||||
# named positions without grids: look for capitalised place names
|
||||
named = re.findall(
|
||||
r'\b(Fme\s+\w+|Chateau\s+\w+|Fort\s+\w+|Casino|'
|
||||
r'Distillery\s+\w+|Brickworks|blockhouse|'
|
||||
r'Mardick|Dunkerque|Dunkirk|Brecht|Loon\s*Plage|'
|
||||
r'Bourbourgville|St\.\s*Folquin|Nordamsques|Montreuil|'
|
||||
r'Wommelg[ea]h?m|Antwerp|Ypres|Pasch[aeo]nd[ae]?le|'
|
||||
r'Gravelines|Le\s*Clipon|Coppenaxfort|'
|
||||
r'Lochtenberg|Eindhoven|Sternhoven|Ryckevorsel|'
|
||||
r'St\.\s*Leonard|Bindhoven|Schilde|Schelde)\b',
|
||||
sent, re.IGNORECASE
|
||||
)
|
||||
for name in named:
|
||||
row_matches.append((s_idx, sent, None, name))
|
||||
|
||||
# determine is_end_of_day: last grid-bearing sentence in last row of date?
|
||||
for i, (s_idx, sent, digits, raw_match) in enumerate(row_matches):
|
||||
is_last = (i == len(row_matches) - 1)
|
||||
|
||||
# extract hour from sentence if not in column
|
||||
hour = raw_hour
|
||||
if not hour:
|
||||
h_m = re.search(r'\b([012]\d[0-5]\d)\s*h(?:r|our|s)?', sent, re.IGNORECASE)
|
||||
if h_m:
|
||||
hour = h_m.group(1)
|
||||
else:
|
||||
h_m2 = re.search(r'\b([012]\d[0-5]\d)[A-Z]?\b', sent)
|
||||
if h_m2:
|
||||
candidate = h_m2.group(1)
|
||||
# make sure it looks like a time not a grid
|
||||
if int(candidate[:2]) <= 23 and int(candidate[2:]) <= 59:
|
||||
hour = candidate
|
||||
|
||||
# grid
|
||||
if digits:
|
||||
grid, inferred = expand_grid(digits)
|
||||
if len(grid) != 6:
|
||||
grid = None
|
||||
inferred = False
|
||||
else:
|
||||
grid = None
|
||||
inferred = False
|
||||
|
||||
# place name from sentence context
|
||||
pn_m = re.search(
|
||||
r'\b(Fme\s+\w+[\w\s]+?(?=\s+\d|\s+MR|\.|,|$)|'
|
||||
r'Chateau\s+\w+|Fort\s+\d+|Casino|Distillery\s+\w+|'
|
||||
r'Brickworks|blockhouse|moated\s+farm|'
|
||||
r'Mardick|Dunkerque|Dunkirk|Brecht|Loon\s*Plage|'
|
||||
r'Bourbourgville|St\.\s*Folquin|Nordamsques|Montreuil|'
|
||||
r'Wommelg[ea]h?m|Antwerp|Ypres|Pasch[aeo]nd[ae]?le|'
|
||||
r'Gravelines|Le\s*Clipon|Coppenaxfort|'
|
||||
r'Lochtenberg|Eindhoven|Sternhoven|Ryckevorsel|'
|
||||
r'St\.\s*Leonard|Bindhoven|Schilde|Schelde|'
|
||||
r'cross.?roads?|road\s+junction|road\s+junc\.?|'
|
||||
r'start\s+line|bridge|lock\s+gates|railway\s+st[na]|'
|
||||
r'windmill|windpump|pier|beach|canal)\b',
|
||||
sent, re.IGNORECASE)
|
||||
place_name = pn_m.group(0).strip() if pn_m else None
|
||||
|
||||
# also check the place column label
|
||||
col_pn = extract_place_name(place_text)
|
||||
if not place_name and col_pn:
|
||||
place_name = col_pn
|
||||
|
||||
subunit = extract_subunit(sent)
|
||||
friendly = extract_friendly(sent)
|
||||
category = categorise(sent, place_name, subunit, friendly, is_place_col=False)
|
||||
|
||||
# shorten context to 2 sentences max
|
||||
context = sent.strip()
|
||||
if len(context) > 300:
|
||||
context = context[:297] + "..."
|
||||
|
||||
all_positions.append(dict(
|
||||
date=cur_date,
|
||||
hour=hour,
|
||||
grid=grid,
|
||||
grid_inferred=inferred,
|
||||
place_name=place_name,
|
||||
sheet_ref=last_sheet,
|
||||
category=category,
|
||||
subunit=subunit,
|
||||
friendly_unit=friendly,
|
||||
is_end_of_day=False, # will be set in post-processing
|
||||
context=context
|
||||
))
|
||||
|
||||
return all_positions
|
||||
|
||||
|
||||
# ── EOD post-processing ───────────────────────────────────────────────────────
|
||||
|
||||
def assign_end_of_day(positions: list[dict]) -> list[dict]:
|
||||
"""
|
||||
For each date, set is_end_of_day=True on exactly ONE entry — the last
|
||||
recorded HQ position for that date (in document order).
|
||||
|
||||
Priority (highest first):
|
||||
1. Last place-column TAC_HQ with grid (context starts "Place column:")
|
||||
2. Last place-column BN_HQ with grid
|
||||
3. Last place-column BN_HQ or TAC_HQ without grid
|
||||
4. Last narrative TAC_HQ with grid
|
||||
5. Last narrative BN_HQ with grid
|
||||
6. Last UNIT_MOVEMENT with grid
|
||||
7. Last any entry with grid
|
||||
"""
|
||||
from collections import defaultdict
|
||||
|
||||
for p in positions:
|
||||
p['is_end_of_day'] = False
|
||||
|
||||
date_indices: dict[str, list[int]] = defaultdict(list)
|
||||
for i, p in enumerate(positions):
|
||||
if p['date']:
|
||||
date_indices[p['date']].append(i)
|
||||
|
||||
def _is_place_col(p):
|
||||
return (p.get('context') or '').startswith('Place column')
|
||||
|
||||
for date, indices in date_indices.items():
|
||||
def _last(cats, require_grid=True, place_col_only=False):
|
||||
matches = [
|
||||
i for i in indices
|
||||
if positions[i]['category'] in cats
|
||||
and (not require_grid or positions[i]['grid'])
|
||||
and (not place_col_only or _is_place_col(positions[i]))
|
||||
]
|
||||
return matches[-1] if matches else None
|
||||
|
||||
winner = (
|
||||
_last({'TAC_HQ'}, require_grid=True, place_col_only=True) or
|
||||
_last({'BN_HQ'}, require_grid=True, place_col_only=True) or
|
||||
_last({'TAC_HQ', 'BN_HQ'}, require_grid=False, place_col_only=True) or
|
||||
_last({'TAC_HQ'}, require_grid=True, place_col_only=False) or
|
||||
_last({'BN_HQ'}, require_grid=True, place_col_only=False) or
|
||||
_last({'UNIT_MOVEMENT'}, require_grid=True) or
|
||||
next((i for i in reversed(indices) if positions[i]['grid']), None)
|
||||
)
|
||||
|
||||
if winner is not None:
|
||||
positions[winner]['is_end_of_day'] = True
|
||||
|
||||
return positions
|
||||
|
||||
|
||||
# ── run ───────────────────────────────────────────────────────────────────────
|
||||
|
||||
if __name__ == "__main__":
|
||||
import pathlib
|
||||
|
||||
print("Reading source file …")
|
||||
text = pathlib.Path(INPUT_FILE).read_text(encoding="utf-8")
|
||||
|
||||
print("Extracting positions …")
|
||||
positions = extract_positions(text)
|
||||
|
||||
# deduplicate identical (date + grid + context[:60]) entries
|
||||
seen = set()
|
||||
unique = []
|
||||
for p in positions:
|
||||
key = (p["date"], p["grid"], p["context"][:60])
|
||||
if key not in seen:
|
||||
seen.add(key)
|
||||
unique.append(p)
|
||||
|
||||
print("Assigning end-of-day flags …")
|
||||
unique = assign_end_of_day(unique)
|
||||
|
||||
pathlib.Path(OUTPUT_FILE).parent.mkdir(parents=True, exist_ok=True)
|
||||
pathlib.Path(OUTPUT_FILE).write_text(
|
||||
json.dumps(unique, indent=2, ensure_ascii=False),
|
||||
encoding="utf-8"
|
||||
)
|
||||
print(f"Done. {len(unique)} positions written to {OUTPUT_FILE}")
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
220
scripts/ocr_wardiaries.py
Normal file
220
scripts/ocr_wardiaries.py
Normal file
@@ -0,0 +1,220 @@
|
||||
"""
|
||||
ocr_wardiaries.py
|
||||
-----------------
|
||||
Converts war diary PDFs to Markdown using the olmOCR model hosted on DeepInfra.
|
||||
Bypasses the olmOCR pipeline GPU check by calling the API directly.
|
||||
|
||||
Requirements:
|
||||
pip install requests pillow
|
||||
|
||||
Poppler must be on PATH (pdftoppm command must work).
|
||||
|
||||
Usage:
|
||||
python ocr_wardiaries.py --input_dir "C:\path\to\pdfs" --output_dir "C:\path\to\output" --api_key "YOUR_KEY"
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import base64
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
import requests
|
||||
from PIL import Image
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Configuration
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
DEEPINFRA_API_URL = "https://api.deepinfra.com/v1/openai/chat/completions"
|
||||
MODEL = "allenai/olmOCR-2-7B-1025"
|
||||
DPI = 150 # Higher = better quality but slower/more expensive. 150 is good for typewritten text.
|
||||
MAX_RETRIES = 3 # Retries per page on API error
|
||||
RETRY_DELAY = 5 # Seconds between retries
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def pdf_to_images(pdf_path: Path, output_dir: Path, dpi: int = DPI) -> list[Path]:
|
||||
"""Render each page of a PDF to a PNG image using pdftoppm (poppler)."""
|
||||
prefix = output_dir / pdf_path.stem
|
||||
cmd = [
|
||||
"pdftoppm",
|
||||
"-r", str(dpi),
|
||||
"-png",
|
||||
str(pdf_path),
|
||||
str(prefix),
|
||||
]
|
||||
result = subprocess.run(cmd, capture_output=True, text=True)
|
||||
if result.returncode != 0:
|
||||
raise RuntimeError(f"pdftoppm failed for {pdf_path.name}:\n{result.stderr}")
|
||||
|
||||
# pdftoppm names files like prefix-1.png, prefix-2.png etc (zero-padded)
|
||||
images = sorted(output_dir.glob(f"{pdf_path.stem}-*.png"))
|
||||
return images
|
||||
|
||||
|
||||
def image_to_base64(image_path: Path) -> str:
|
||||
"""Convert an image file to a base64 string."""
|
||||
with open(image_path, "rb") as f:
|
||||
return base64.b64encode(f.read()).decode("utf-8")
|
||||
|
||||
|
||||
def ocr_page(image_path: Path, api_key: str, page_num: int) -> str:
|
||||
"""Send one page image to DeepInfra olmOCR and return the extracted text."""
|
||||
b64 = image_to_base64(image_path)
|
||||
|
||||
headers = {
|
||||
"Content-Type": "application/json",
|
||||
"Authorization": f"Bearer {api_key}",
|
||||
}
|
||||
|
||||
payload = {
|
||||
"model": MODEL,
|
||||
"messages": [
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{
|
||||
"type": "image_url",
|
||||
"image_url": {
|
||||
"url": f"data:image/png;base64,{b64}"
|
||||
},
|
||||
},
|
||||
{
|
||||
"type": "text",
|
||||
"text": (
|
||||
"Below is a scanned page from a Canadian Army WWII war diary or supplementary document, "
|
||||
"circa 1944-1945. Extract all text exactly as it appears, preserving layout, "
|
||||
"column structure, dates, grid references, unit names, and abbreviations. "
|
||||
"Output clean Markdown. Do not add commentary or summaries."
|
||||
),
|
||||
},
|
||||
],
|
||||
}
|
||||
],
|
||||
"max_tokens": 4096,
|
||||
"temperature": 0.0,
|
||||
}
|
||||
|
||||
for attempt in range(1, MAX_RETRIES + 1):
|
||||
try:
|
||||
response = requests.post(DEEPINFRA_API_URL, headers=headers, json=payload, timeout=120)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
return data["choices"][0]["message"]["content"]
|
||||
except requests.exceptions.HTTPError as e:
|
||||
print(f" HTTP error on page {page_num}, attempt {attempt}/{MAX_RETRIES}: {e}")
|
||||
if attempt < MAX_RETRIES:
|
||||
time.sleep(RETRY_DELAY)
|
||||
else:
|
||||
return f"[OCR FAILED - page {page_num} - HTTP error: {e}]"
|
||||
except Exception as e:
|
||||
print(f" Error on page {page_num}, attempt {attempt}/{MAX_RETRIES}: {e}")
|
||||
if attempt < MAX_RETRIES:
|
||||
time.sleep(RETRY_DELAY)
|
||||
else:
|
||||
return f"[OCR FAILED - page {page_num} - Error: {e}]"
|
||||
|
||||
|
||||
def ocr_pdf(pdf_path: Path, output_dir: Path, api_key: str) -> Path:
|
||||
"""OCR a single PDF and write output to a Markdown file."""
|
||||
print(f"\n{'='*60}")
|
||||
print(f"Processing: {pdf_path.name}")
|
||||
print(f"{'='*60}")
|
||||
|
||||
output_md = output_dir / f"{pdf_path.stem}_olmocr.md"
|
||||
|
||||
# Skip if already done
|
||||
if output_md.exists():
|
||||
print(f" Already processed — skipping. Delete {output_md.name} to reprocess.")
|
||||
return output_md
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
tmp_path = Path(tmp_dir)
|
||||
|
||||
# Render PDF pages to images
|
||||
print(f" Rendering PDF pages to images at {DPI} DPI...")
|
||||
try:
|
||||
images = pdf_to_images(pdf_path, tmp_path)
|
||||
except RuntimeError as e:
|
||||
print(f" ERROR: {e}")
|
||||
return output_md
|
||||
|
||||
total_pages = len(images)
|
||||
print(f" Found {total_pages} pages.")
|
||||
|
||||
all_text = [
|
||||
f"# {pdf_path.stem}\n",
|
||||
f"*OCR'd by olmOCR ({MODEL}) via DeepInfra*\n",
|
||||
f"*Source: {pdf_path.name} — {total_pages} pages*\n",
|
||||
"---\n",
|
||||
]
|
||||
|
||||
for i, image_path in enumerate(images, start=1):
|
||||
print(f" Page {i}/{total_pages}...", end=" ", flush=True)
|
||||
page_text = ocr_page(image_path, api_key, i)
|
||||
all_text.append(f"\n---\n## Page {i}\n\n{page_text}\n")
|
||||
print("done")
|
||||
|
||||
# Small delay to avoid hammering the API
|
||||
if i < total_pages:
|
||||
time.sleep(0.5)
|
||||
|
||||
# Write output
|
||||
output_dir.mkdir(parents=True, exist_ok=True)
|
||||
with open(output_md, "w", encoding="utf-8") as f:
|
||||
f.write("\n".join(all_text))
|
||||
|
||||
print(f" Saved: {output_md}")
|
||||
return output_md
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Main
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="OCR war diary PDFs using olmOCR via DeepInfra.")
|
||||
parser.add_argument("--input_dir", required=True, help="Folder containing PDF files to process.")
|
||||
parser.add_argument("--output_dir", required=True, help="Folder to write Markdown output files.")
|
||||
parser.add_argument("--api_key", required=True, help="Your DeepInfra API key.")
|
||||
parser.add_argument("--single", default=None, help="Process a single PDF file instead of a folder.")
|
||||
args = parser.parse_args()
|
||||
|
||||
input_dir = Path(args.input_dir)
|
||||
output_dir = Path(args.output_dir)
|
||||
output_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
if args.single:
|
||||
pdfs = [Path(args.single)]
|
||||
else:
|
||||
pdfs = sorted(input_dir.glob("*.pdf"))
|
||||
|
||||
if not pdfs:
|
||||
print(f"No PDF files found in {input_dir}")
|
||||
sys.exit(1)
|
||||
|
||||
print(f"Found {len(pdfs)} PDF(s) to process.")
|
||||
print(f"Output directory: {output_dir}")
|
||||
|
||||
results = []
|
||||
for pdf_path in pdfs:
|
||||
output_md = ocr_pdf(pdf_path, output_dir, args.api_key)
|
||||
results.append(output_md)
|
||||
|
||||
print(f"\n{'='*60}")
|
||||
print(f"Complete. {len(results)} file(s) processed.")
|
||||
print(f"Output files:")
|
||||
for r in results:
|
||||
print(f" {r}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user