309 lines
10 KiB
Python
309 lines
10 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
テンプレート(赤枠=氏名, 青枠=日付)を基準に、対象フォルダ内のスキャンPDFから
|
||
氏名・日付をOCR抽出し、"output/氏名/YYYYMMDD_氏名.pdf" にリネーム・移動する。
|
||
|
||
使い方:
|
||
python extract_and_rename.py \
|
||
--template template.png \
|
||
--input ./inbox \
|
||
--output ./output \
|
||
--failed ./failed \
|
||
--margin 0.10
|
||
|
||
必要ライブラリ: opencv-python, pdf2image, pytesseract, pillow, numpy
|
||
必要な外部ツール: tesseract-ocr, tesseract-ocr-jpn, poppler-utils(pdftoppm)
|
||
"""
|
||
|
||
import argparse
|
||
import re
|
||
import shutil
|
||
import sys
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
|
||
import pytesseract
|
||
from pdf2image import convert_from_path
|
||
from PIL import Image
|
||
|
||
from box_detector import detect_template_fields, to_ratio_box
|
||
|
||
DPI = 300 # PDF→画像変換の解像度。テンプレートも同じDPIで作成/スキャンしてください
|
||
|
||
import importlib
|
||
|
||
TEMPLATE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".bmp"}
|
||
|
||
|
||
def select_template_file(template_dir: Path) -> Path:
|
||
candidates = sorted(
|
||
p for p in template_dir.iterdir()
|
||
if p.is_file() and p.suffix.lower() in TEMPLATE_EXTENSIONS
|
||
)
|
||
if not candidates:
|
||
raise FileNotFoundError(
|
||
f"テンプレートフォルダに画像ファイルが見つかりません: {template_dir}"
|
||
)
|
||
return candidates[0]
|
||
|
||
|
||
REQUIRED_MODULES = ("cv2", "numpy", "PIL", "pytesseract", "pdf2image")
|
||
|
||
|
||
def check_dependencies(modules: tuple[str, ...] = REQUIRED_MODULES) -> list[str]:
|
||
missing = []
|
||
for mod in modules:
|
||
try:
|
||
importlib.import_module(mod)
|
||
except ImportError:
|
||
missing.append(mod)
|
||
return missing
|
||
|
||
|
||
def check_external_tools(tesseract_exe: Path, poppler_exe: Path) -> list[str]:
|
||
missing = []
|
||
if not tesseract_exe.exists():
|
||
missing.append(str(tesseract_exe))
|
||
if not poppler_exe.exists():
|
||
missing.append(str(poppler_exe))
|
||
return missing
|
||
|
||
|
||
import time
|
||
import traceback
|
||
|
||
from file_ops import copy_with_unique_name, move_with_unique_name, wait_until_stable
|
||
from logger import append_log
|
||
|
||
|
||
def key_pressed() -> bool:
|
||
import msvcrt
|
||
|
||
return msvcrt.kbhit()
|
||
|
||
|
||
@dataclass
|
||
class FieldBox:
|
||
label: str
|
||
x_ratio: float
|
||
y_ratio: float
|
||
w_ratio: float
|
||
h_ratio: float
|
||
|
||
def to_pixel_box(self, img_w: int, img_h: int, margin: float) -> tuple[int, int, int, int]:
|
||
"""比率座標を対象画像の実ピクセルに変換し、上下左右にmargin(比率)だけ広げる"""
|
||
x = self.x_ratio * img_w
|
||
y = self.y_ratio * img_h
|
||
w = self.w_ratio * img_w
|
||
h = self.h_ratio * img_h
|
||
|
||
mx = w * margin
|
||
my = h * margin
|
||
|
||
x0 = max(0, int(x - mx))
|
||
y0 = max(0, int(y - my))
|
||
x1 = min(img_w, int(x + w + mx))
|
||
y1 = min(img_h, int(y + h + my))
|
||
return x0, y0, x1, y1
|
||
|
||
|
||
def load_field_boxes(template_path: str) -> list[FieldBox]:
|
||
fields = detect_template_fields(template_path)
|
||
image_size = fields.pop("_image_size")
|
||
|
||
boxes = []
|
||
for label, box in fields.items():
|
||
xr, yr, wr, hr = to_ratio_box(box, image_size)
|
||
boxes.append(FieldBox(label, xr, yr, wr, hr))
|
||
|
||
if not boxes:
|
||
raise ValueError("テンプレートから赤枠・青枠が検出できませんでした。枠の色・太さを確認してください。")
|
||
return boxes
|
||
|
||
|
||
def ocr_region(page_img: Image.Image, box: tuple[int, int, int, int], lang: str = "jpn") -> str:
|
||
x0, y0, x1, y1 = box
|
||
cropped = page_img.crop((x0, y0, x1, y1))
|
||
text = pytesseract.image_to_string(cropped, lang=lang)
|
||
return text.strip()
|
||
|
||
|
||
def clean_name(raw: str) -> str:
|
||
"""OCR結果から氏名らしき文字列を抽出（改行・空白・記号ノイズを除去）"""
|
||
text = re.sub(r"[\s ]+", "", raw)
|
||
text = re.sub(r"[^\w一-龠ぁ-んァ-ヶー]", "", text)
|
||
return text
|
||
|
||
|
||
def clean_date(raw: str) -> str | None:
|
||
"""OCR結果から日付を検出し YYYYMMDD 形式で返す。見つからなければNone。"""
|
||
text = raw.replace(" ", "").replace(" ", "")
|
||
|
||
patterns = [
|
||
r"(\d{4})[年/\-\.](\d{1,2})[月/\-\.](\d{1,2})", # 2026年08月02日 / 2026/08/02 等
|
||
r"(\d{2})[年/\-\.](\d{1,2})[月/\-\.](\d{1,2})", # 26/08/02 のような2桁年
|
||
]
|
||
for pat in patterns:
|
||
m = re.search(pat, text)
|
||
if m:
|
||
y, mo, d = m.groups()
|
||
if len(y) == 2:
|
||
y = "20" + y # 2桁年は20XX年と仮定。運用に応じて要調整
|
||
return f"{int(y):04d}{int(mo):02d}{int(d):02d}"
|
||
return None
|
||
|
||
|
||
def process_pdf(pdf_path: Path, boxes: list[FieldBox], margin: float) -> tuple[str | None, str | None]:
|
||
"""PDFの1ページ目を画像化し、氏名・日付を抽出して返す"""
|
||
pages = convert_from_path(str(pdf_path), dpi=DPI, first_page=1, last_page=1)
|
||
if not pages:
|
||
return None, None
|
||
page_img = pages[0]
|
||
img_w, img_h = page_img.size
|
||
|
||
name_text, date_text = None, None
|
||
for box in boxes:
|
||
pixel_box = box.to_pixel_box(img_w, img_h, margin)
|
||
raw = ocr_region(page_img, pixel_box, lang="jpn")
|
||
if box.label == "name":
|
||
name_text = clean_name(raw)
|
||
elif box.label == "date":
|
||
date_text = clean_date(raw)
|
||
|
||
return name_text, date_text
|
||
|
||
|
||
def run_queue(
|
||
scan_dir: Path,
|
||
output_dir: Path,
|
||
success_dir: Path,
|
||
failed_dir: Path,
|
||
log_dir: Path,
|
||
boxes: list[FieldBox],
|
||
margin: float,
|
||
stable_wait_sec: float,
|
||
stable_retries: int,
|
||
process_pdf_func=process_pdf,
|
||
key_check=key_pressed,
|
||
sleep_func=time.sleep,
|
||
) -> None:
|
||
while True:
|
||
queue = sorted(scan_dir.glob("*.pdf"))
|
||
if not queue:
|
||
break
|
||
|
||
pdf_path = queue[0]
|
||
|
||
if key_check():
|
||
append_log(
|
||
log_dir, "エラー", pdf_path.name, None, None,
|
||
"ユーザー操作により停止しました",
|
||
)
|
||
break
|
||
|
||
if not pdf_path.exists():
|
||
continue
|
||
|
||
if not wait_until_stable(pdf_path, stable_wait_sec, stable_retries, sleep_func):
|
||
append_log(
|
||
log_dir, "スキップ", pdf_path.name, None, None,
|
||
"サイズ不安定のためスキップ",
|
||
)
|
||
continue
|
||
|
||
try:
|
||
name, date = process_pdf_func(pdf_path, boxes, margin)
|
||
except Exception as e:
|
||
append_log(
|
||
log_dir, "エラー", pdf_path.name, None, None,
|
||
f"{type(e).__name__}: {e}\n{traceback.format_exc()}",
|
||
)
|
||
move_with_unique_name(pdf_path, failed_dir, pdf_path.stem, pdf_path.suffix)
|
||
continue
|
||
|
||
if not name or not date:
|
||
append_log(log_dir, "失敗", pdf_path.name, name, date, "")
|
||
move_with_unique_name(pdf_path, failed_dir, pdf_path.stem, pdf_path.suffix)
|
||
continue
|
||
|
||
copy_with_unique_name(pdf_path, output_dir / name, f"{name}_{date}", ".pdf")
|
||
move_with_unique_name(pdf_path, success_dir, pdf_path.stem, pdf_path.suffix)
|
||
append_log(log_dir, "成功", pdf_path.name, name, date, "")
|
||
|
||
|
||
def safe_move(src: Path, dest_dir: Path, filename: str) -> Path:
|
||
dest_dir.mkdir(parents=True, exist_ok=True)
|
||
dest_path = dest_dir / filename
|
||
# 同名ファイルが既にある場合は連番を付けて衝突を回避
|
||
counter = 1
|
||
while dest_path.exists():
|
||
stem = Path(filename).stem
|
||
suffix = Path(filename).suffix
|
||
dest_path = dest_dir / f"{stem}_{counter}{suffix}"
|
||
counter += 1
|
||
shutil.move(str(src), str(dest_path))
|
||
return dest_path
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="テンプレート枠を基準にPDFから氏名・日付を抽出してリネーム・移動する")
|
||
parser.add_argument("--template", required=True, help="赤枠(氏名)・青枠(日付)付きテンプレート画像")
|
||
parser.add_argument("--input", required=True, help="処理対象PDFが入ったフォルダ")
|
||
parser.add_argument("--output", required=True, help="成功時の出力先ルートフォルダ（氏名ごとにサブフォルダ作成）")
|
||
parser.add_argument("--failed", required=True, help="抽出失敗時の退避先フォルダ")
|
||
parser.add_argument("--margin", type=float, default=0.10, help="枠の許容誤差率（デフォルト 0.10 = 10%%）")
|
||
args = parser.parse_args()
|
||
|
||
template_path = Path(args.template)
|
||
input_dir = Path(args.input)
|
||
output_dir = Path(args.output)
|
||
failed_dir = Path(args.failed)
|
||
|
||
if not template_path.exists():
|
||
sys.exit(f"テンプレート画像が見つかりません: {template_path}")
|
||
if not input_dir.exists():
|
||
sys.exit(f"入力フォルダが見つかりません: {input_dir}")
|
||
|
||
boxes = load_field_boxes(str(template_path))
|
||
print(f"検出フィールド: {[b.label for b in boxes]}")
|
||
|
||
pdf_files = sorted(input_dir.glob("*.pdf"))
|
||
if not pdf_files:
|
||
print("対象PDFが見つかりませんでした。")
|
||
return
|
||
|
||
results = []
|
||
for pdf_path in pdf_files:
|
||
print(f"処理中: {pdf_path.name}")
|
||
try:
|
||
name, date = process_pdf(pdf_path, boxes, args.margin)
|
||
except Exception as e:
|
||
print(f" エラー: {e}")
|
||
safe_move(pdf_path, failed_dir, pdf_path.name)
|
||
results.append((pdf_path.name, None, None, "error", str(e)))
|
||
continue
|
||
|
||
if not name or not date:
|
||
print(f" 抽出不十分（氏名={name!r}, 日付={date!r}）→ failed へ退避")
|
||
safe_move(pdf_path, failed_dir, pdf_path.name)
|
||
results.append((pdf_path.name, name, date, "failed", ""))
|
||
continue
|
||
|
||
new_filename = f"{date}_{name}.pdf"
|
||
dest_dir = output_dir / name
|
||
moved_path = safe_move(pdf_path, dest_dir, new_filename)
|
||
print(f" → {moved_path}")
|
||
results.append((pdf_path.name, name, date, "success", str(moved_path)))
|
||
|
||
# サマリー出力
|
||
print("\n=== 処理結果サマリー ===")
|
||
for orig, name, date, status, note in results:
|
||
print(f"{status:8s} | {orig:30s} | 氏名={name} 日付={date} | {note}")
|
||
|
||
success_count = sum(1 for r in results if r[3] == "success")
|
||
print(f"\n合計 {len(results)} 件中 {success_count} 件成功")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|