poison-flower/pf-ocr
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(ocr): unify config, add manga translation pipeline and context lookahead

Browse Source 
- Consolidate module configs into root config.example.json with ocr, vision, and epub sections

- Split LLM OCR workflows into novel_ocr.py (prose) and manga_ocr_llm.py (manga)

- Remove gemini_direct_ocr.py in favor of OpenAI-compatible API endpoints

- Support direct manga translation via --translate, --target-lang, and glossary.md

- Add bidirectional context support: past translations (--context-pages) and lookahead Japanese text (--context-pages-ahead)

- Add per-page JSON audit logging under logs/ and expose OpenAI sampling parameters
This commit is contained in:
Poison Flower
2026-09-06 15:42:29 +03:00
parent 41c6e97cd4
commit 35b7be680e
23 changed files with 1811 additions and 494 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ocr/google_vision_ocr.py
+185 -10
View File
@@ -4,7 +4,11 @@ Batch-OCR for scanned pages of a Japanese light novel (vertical text) via
Google Cloud Vision.
Usage:
python google_vision_ocr.py --input ./pages --output ./out --credentials ./key.json
python google_vision_ocr.py --input ./pages --output ./out
# (credentials path comes from config.json's vision.credentials, or pass --credentials)
# for manga instead of a light novel:
python google_vision_ocr.py --input ./pages --output ./out --mode manga
Requirements:
pip install google-cloud-vision natsort tqdm
@@ -24,16 +28,23 @@ the free tier (1000 requests/month covers ~600 pages comfortably). No charge
should occur unless you exceed that quota.
This is a classic OCR engine (not an LLM) — generally solid for image
quality, but it doesn't understand context the way a multimodal model does.
For light novel pages with dense vertical prose, openrouter_ocr.py /
gemini_direct_ocr.py usually give better results with less setup friction.
quality, but it doesn't understand context the way a multimodal model does,
and (unlike the LLM-based scripts) it can't translate. For light novel
pages with dense vertical prose, novel_ocr.py usually gives better results
with less setup friction; for manga, see manga_ocr_llm.py.
Every request/response is logged as one JSON file under logs/ at the repo
root (--log-dir to change, --no-log to disable).
"""
import argparse
import io
import json
import os
import re
import sys
import time
from datetime import datetime
from pathlib import Path
from google.cloud import vision
@@ -41,10 +52,43 @@ from natsort import natsorted
from tqdm import tqdm
IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".webp", ".tif", ".tiff", ".bmp"}
SCRIPT_DIR = Path(__file__).resolve().parent
ROOT_DIR = SCRIPT_DIR.parent
def ocr_image(client: vision.ImageAnnotatorClient, path: Path, retries: int = 3) -> str:
"""OCRs a single scan, returning text in reading order."""
def load_config(config_path: Path) -> dict:
"""Loads the shared config.json and returns its "vision" section
(Google Cloud Vision has nothing to do with the OpenAI-compatible "ocr"
section used by novel_ocr.py / manga_ocr_llm.py)."""
if not config_path.exists():
return {}
try:
data = json.loads(config_path.read_text(encoding="utf-8"))
except json.JSONDecodeError as e:
print(f"Failed to parse {config_path}: {e}", file=sys.stderr)
sys.exit(1)
return data.get("vision", data) if isinstance(data, dict) else {}
def write_log(log_dir: Path, page_name: str, entry: dict) -> None:
"""Writes one JSON log file per request/response (see novel_ocr.py
for the rationale — same format, minus the image payload itself)."""
log_dir.mkdir(parents=True, exist_ok=True)
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
safe_name = re.sub(r"[^A-Za-z0-9_.-]", "_", page_name)
log_path = log_dir / f"{timestamp}_{safe_name}.json"
log_path.write_text(json.dumps(entry, ensure_ascii=False, indent=2, default=str), encoding="utf-8")
def _response_to_dict(response) -> dict:
try:
return vision.AnnotateImageResponse.to_dict(response)
except Exception: # noqa: BLE001
return {"full_text_annotation_text": getattr(getattr(response, "full_text_annotation", None), "text", None)}
def ocr_image_novel(client: vision.ImageAnnotatorClient, path: Path, retries: int = 3, log_entry: dict = None) -> str:
"""OCRs a single scan of running prose, returning text in reading order."""
with io.open(path, "rb") as f:
content = f.read()
@@ -52,6 +96,14 @@ def ocr_image(client: vision.ImageAnnotatorClient, path: Path, retries: int = 3)
# The "ja" language hint helps the model handle vertical Japanese more accurately
image_context = vision.ImageContext(language_hints=["ja"])
if log_entry is not None:
log_entry["request"] = {
"feature": "document_text_detection",
"language_hints": ["ja"],
"image": f"<omitted: {path.name}, {path.stat().st_size} bytes>",
}
log_entry["attempts"] = []
last_err = None
for attempt in range(retries):
try:
@@ -60,24 +112,132 @@ def ocr_image(client: vision.ImageAnnotatorClient, path: Path, retries: int = 3)
)
if response.error.message:
raise RuntimeError(response.error.message)
return response.full_text_annotation.text
text = response.full_text_annotation.text
if log_entry is not None:
log_entry["attempts"].append(
{"attempt": attempt + 1, "success": True, "response": _response_to_dict(response)}
)
return text
except Exception as e: # noqa: BLE001
last_err = e
if log_entry is not None:
log_entry["attempts"].append({"attempt": attempt + 1, "success": False, "error": str(e)})
time.sleep(2 * (attempt + 1))
raise RuntimeError(f"Failed to OCR {path.name} after {retries} attempts: {last_err}")
def ocr_image_manga(client: vision.ImageAnnotatorClient, path: Path, retries: int = 3, log_entry: dict = None) -> str:
"""OCRs a single manga page.
document_text_detection assumes a running paragraph flow, which falls
apart on manga: bubbles are scattered blocks, not one paragraph. Instead
this groups Vision's per-paragraph bounding boxes into text-block
"clusters" and orders them in manga reading order: clusters right-to-left
by their rightmost edge, breaking ties top-to-bottom.
"""
with io.open(path, "rb") as f:
content = f.read()
image = vision.Image(content=content)
image_context = vision.ImageContext(language_hints=["ja"])
if log_entry is not None:
log_entry["request"] = {
"feature": "document_text_detection",
"language_hints": ["ja"],
"image": f"<omitted: {path.name}, {path.stat().st_size} bytes>",
}
log_entry["attempts"] = []
last_err = None
for attempt in range(retries):
try:
response = client.document_text_detection(
image=image, image_context=image_context
)
if response.error.message:
raise RuntimeError(response.error.message)
if log_entry is not None:
log_entry["attempts"].append(
{"attempt": attempt + 1, "success": True, "response": _response_to_dict(response)}
)
break
except Exception as e: # noqa: BLE001
last_err = e
if log_entry is not None:
log_entry["attempts"].append({"attempt": attempt + 1, "success": False, "error": str(e)})
time.sleep(2 * (attempt + 1))
else:
raise RuntimeError(f"Failed to OCR {path.name} after {retries} attempts: {last_err}")
blocks = []
for page in response.full_text_annotation.pages:
for block in page.blocks:
xs = [v.x for v in block.bounding_box.vertices]
ys = [v.y for v in block.bounding_box.vertices]
text = ""
for paragraph in block.paragraphs:
words = []
for word in paragraph.words:
words.append("".join(s.text for s in word.symbols))
text += "".join(words)
if text.strip():
blocks.append({"text": text.strip(), "x_max": max(xs), "y_min": min(ys)})
if not blocks:
result = "[NO_TEXT]"
else:
# Manga reading order: right-to-left, breaking ties top-to-bottom. This
# is a coarse heuristic (true panel/bubble order can't be recovered from
# plain bounding boxes) — always spot-check against the page.
blocks.sort(key=lambda b: (-b["x_max"], b["y_min"]))
result = "\n".join(f"{i}. {b['text']}" for i, b in enumerate(blocks, start=1))
if log_entry is not None:
log_entry["reading_order_result"] = result
return result
def main():
parser = argparse.ArgumentParser(description="Batch-OCR a light novel via Google Cloud Vision")
parser.add_argument("--input", required=True, help="Folder with scanned page images (jpg/png)")
parser.add_argument("--output", required=True, help="Folder for the OCR results")
parser.add_argument("--credentials", required=True, help="Path to the service-account key.json")
parser.add_argument(
"--credentials", default=None,
help="Path to the service-account key.json (defaults to config.json's vision.credentials)"
)
parser.add_argument(
"--config", default=str(ROOT_DIR / "config.json"),
help="Path to config.json with a \"vision\" section. Defaults to config.json at the repo root."
)
parser.add_argument(
"--mode", choices=["novel", "manga"], default="novel",
help="novel: dense running prose, full-page reading order (default). "
"manga: scattered bubbles, grouped and sorted in manga reading order."
)
parser.add_argument(
"--start-page", type=int, default=1, help="Page number to start the header numbering from"
)
parser.add_argument(
"--log-dir", default=str(ROOT_DIR / "logs"),
help="Folder for per-page request/response logs (one JSON file per page). "
"Defaults to logs/ at the repo root."
)
parser.add_argument(
"--no-log", action="store_true",
help="Disable request/response logging entirely"
)
args = parser.parse_args()
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = args.credentials
config = load_config(Path(args.config))
credentials = args.credentials or config.get("credentials")
if not credentials:
print(
f"No credentials found. Set vision.credentials in {args.config} or pass --credentials.",
file=sys.stderr,
)
sys.exit(1)
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = credentials
input_dir = Path(args.input)
output_dir = Path(args.output)
@@ -94,6 +254,8 @@ def main():
print(f"Pages found: {len(images)}")
log_dir = None if args.no_log else Path(args.log_dir)
client = vision.ImageAnnotatorClient()
combined_path = output_dir / "combined.md"
@@ -107,21 +269,34 @@ def main():
if txt_out.exists():
text = txt_out.read_text(encoding="utf-8")
else:
log_entry = {
"timestamp": datetime.now().isoformat(),
"backend": "google_vision",
"page": img_path.name,
"mode": args.mode,
} if log_dir is not None else None
try:
text = ocr_image(client, img_path)
ocr_fn = ocr_image_manga if args.mode == "manga" else ocr_image_novel
text = ocr_fn(client, img_path, log_entry=log_entry)
txt_out.write_text(text, encoding="utf-8")
except Exception as e: # noqa: BLE001
print(f"\nError on {img_path.name}: {e}", file=sys.stderr)
failed.append(img_path.name)
text = ""
if log_entry is not None:
log_entry["error"] = str(e)
# Do NOT write a file to disk on failure — otherwise the next
# run would see the file exists and skip retrying it.
if log_entry is not None:
write_log(log_dir, img_path.stem, log_entry)
combined_f.write(f"\n\n<!-- page {idx}: {img_path.name} -->\n\n")
combined_f.write(text)
print(f"\nDone. Combined file: {combined_path}")
print(f"Per-page files: {pages_dir}")
if log_dir is not None:
print(f"Request/response logs: {log_dir}")
if failed:
print(f"\nFailed to OCR {len(failed)} page(s):")
for name in failed: