cyber-valley/cve/tools/redact.py

#!/usr/bin/env python3
"""Black out personal data of natural persons on scanned deeds, line by line.

For every page PNG in ocr/<doc>-<n>.png with its tesseract TSV, a line is
blacked out when it carries an ID marker (NIK / KTP / paspor / passport /
nomor induk / 16-digit number), a home address ("bertempat tinggal", plus
the two lines after it), or a passport validity ("berlaku sampai"). The
lines right after an ID marker are also covered: numbers wrap in notarial
layout. Output: redacted/<doc>.pdf, and redacted/<doc>-<n>.png for a re-OCR.

    /tmp/redact-venv/bin/python redact.py SALINAN_NO_13_PKR_PT._CYBER_VALLEY_ESTATE deed_of_sale_and_purchase …
"""
import csv
import re
import sys
from collections import defaultdict
from pathlib import Path

from PIL import Image, ImageDraw

OCR = Path(__file__).parent / "ocr"
OUT = Path(__file__).parent / "redacted"
OUT.mkdir(exist_ok=True)

import os
EXTRA = os.environ.get("REDACT_ALSO", "")
MARK = re.compile(r"\b(nik|ktp|paspor|passport|nomor\s*induk|kartu\s*tanda\s*penduduk|berlaku\s*sampai|izin\s*tinggal|kitas|kitap|tinggal\s*terbatas" + ("|" + EXTRA if EXTRA else "") + r")\b|\d{16}", re.I)
ADDRESS = re.compile(os.environ.get("REDACT_ADDRESS") or r"berte?mpat\s*ting?g?al|beralamat|tinggal\s*di\b|\bjalan\b|\bjl\.|pemegang\s*izin|ungasan|kuta\s*selatan|badung|wijaya\s*kusuma|pura\s*masuka|utan\s*kayu|matraman|jakarta\s*timur|galur\s*sari", re.I)
# a number that looks like an ID: 7+ digits, or passport style letters+digits
IDNUM = re.compile(r"\b\d[\d\s.-]{6,}\d\b|\b[A-Z]{1,2}\s?\d{6,9}\b")


def lines_of(tsv: Path):
    """line key → (text, bbox) from a tesseract TSV."""
    rows = list(csv.DictReader(tsv.open(), delimiter="\t", quoting=csv.QUOTE_NONE))
    words = defaultdict(list)
    for r in rows:
        if r["level"] != "5" or not r["text"].strip():
            continue
        key = (int(r["page_num"]), int(r["block_num"]), int(r["par_num"]), int(r["line_num"]))
        words[key].append(r)
    out = []
    for key in sorted(words):
        ws = words[key]
        text = " ".join(w["text"] for w in ws)
        x0 = min(int(w["left"]) for w in ws)
        y0 = min(int(w["top"]) for w in ws)
        x1 = max(int(w["left"]) + int(w["width"]) for w in ws)
        y1 = max(int(w["top"]) + int(w["height"]) for w in ws)
        out.append((text, (x0, y0, x1, y1)))
    # reading order: by top then left
    out.sort(key=lambda t: (t[1][1], t[1][0]))
    return out


def targets(lines):
    """Indexes of lines to cover."""
    hit = set()
    for i, (text, _) in enumerate(lines):
        if MARK.search(text):
            hit.update({i, i + 1})
            # a bare number on the next line after "Nomor" wraps the ID
            if i + 2 < len(lines) and IDNUM.search(lines[i + 2][0]):
                hit.add(i + 2)
        if ADDRESS.search(text):
            hit.update({i, i + 1, i + 2})
    return {i for i in hit if 0 <= i < len(lines)}


def redact_doc(doc: str) -> int:
    pages = sorted(OCR.glob(f"{doc}-*.png"), key=lambda p: int(p.stem.rsplit("-", 1)[1]))
    if not pages:
        print(f"no pages for {doc}", file=sys.stderr)
        return 0
    images, covered = [], 0
    carry = 0  # an address cut by the page break continues at the top of the next page
    for png in pages:
        tsv = png.with_suffix(".tsv")
        im = Image.open(png).convert("RGB")
        if tsv.exists():
            lines = lines_of(tsv)
            draw = ImageDraw.Draw(im)
            hit = targets(lines)
            hit.update(range(min(carry, len(lines))))
            carry = 2 if lines and (len(lines) - 1) in hit else 0
            for i in hit:
                x0, y0, x1, y1 = lines[i][1]
                # the whole text line across the page, with a margin
                draw.rectangle((max(0, x0 - 8), max(0, y0 - 4), im.width - 1, y1 + 4), fill="black")
                covered += 1
        im.save(OUT / png.name)
        images.append(im)
    images[0].save(OUT / f"{doc}.pdf", save_all=True, append_images=images[1:], resolution=150.0)
    print(f"{doc}: {len(pages)} pages, {covered} lines covered")
    return covered


if __name__ == "__main__":
    for d in sys.argv[1:]:
        redact_doc(d)

Graph