Standardisation 5
This commit is contained in:
+234
-191
@@ -1,223 +1,266 @@
|
||||
import fitz # PyMuPDF
|
||||
from pypdf import PdfWriter
|
||||
from pypdf import PdfReader
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
from collections import defaultdict
|
||||
from collections.abc import Sequence
|
||||
from pathlib import Path
|
||||
|
||||
from utils import read_all_labels
|
||||
import fitz
|
||||
from pypdf import PdfReader, PdfWriter
|
||||
|
||||
carreau = 1000 // 38
|
||||
import utils
|
||||
from copienator import (
|
||||
CliError,
|
||||
EvaluationWorkspace,
|
||||
ExitCode,
|
||||
execute,
|
||||
read_json,
|
||||
target_parser,
|
||||
workspace_from_target,
|
||||
)
|
||||
from copienator.filesystem import staged_directory
|
||||
|
||||
SQUARE = 1000 // 38
|
||||
Coordinate = tuple[str, int, int, int, int, int]
|
||||
ParsedCoordinate = tuple[str, str, int, int, int, int, int]
|
||||
|
||||
|
||||
def decode_json(pdf_file):
|
||||
file_path = Path(pdf_file)
|
||||
with open(file_path.with_suffix(".json"), "r") as f:
|
||||
json_result = json.load(f)
|
||||
|
||||
nb_pages = len(PdfReader(file_path).pages)
|
||||
|
||||
bb_list = json_result["list"]
|
||||
name = json_result["name"]
|
||||
column_width = 1000 // nb_pages
|
||||
|
||||
def page_number(b):
|
||||
return ((b[1] + b[3]) // 2) // column_width
|
||||
|
||||
result = []
|
||||
for d in bb_list:
|
||||
(b, label) = d["box_2d"], d["label"]
|
||||
pn = page_number(b)
|
||||
result.append((label, pn, b[0] - carreau, b[2]-carreau, b[1], b[3]))
|
||||
result.sort(key=lambda x: (x[1], x[2]))
|
||||
return (name, result)
|
||||
def decode_json(pdf_file: str | Path) -> tuple[str, list[Coordinate]]:
|
||||
"""Read verified label coordinates associated with one copy PDF."""
|
||||
pdf_path = Path(pdf_file)
|
||||
loaded = read_json(pdf_path.with_suffix(".json"))
|
||||
if not isinstance(loaded, dict):
|
||||
raise TypeError(f"Expected a JSON object for {pdf_path}")
|
||||
boxes = loaded.get("list")
|
||||
if not isinstance(boxes, list):
|
||||
raise TypeError(f"Expected a list of labels for {pdf_path}")
|
||||
page_count = len(PdfReader(pdf_path).pages)
|
||||
if page_count == 0:
|
||||
raise ValueError(f"PDF contains no pages: {pdf_path}")
|
||||
column_width = 1000 // page_count
|
||||
result: list[Coordinate] = []
|
||||
for entry in boxes:
|
||||
if not isinstance(entry, dict):
|
||||
raise TypeError(f"Malformed label entry for {pdf_path}: {entry!r}")
|
||||
box = entry["box_2d"]
|
||||
label = str(entry["label"])
|
||||
page_number = ((box[1] + box[3]) // 2) // column_width
|
||||
result.append(
|
||||
(label, page_number, box[0] - SQUARE, box[2] - SQUARE, box[1], box[3])
|
||||
)
|
||||
result.sort(key=lambda item: (item[1], item[2]))
|
||||
return str(loaded.get("name", "")), result
|
||||
|
||||
|
||||
def split_an_interro(base_dir, input_pdf, coords_list):
|
||||
doc = fitz.open(input_pdf)
|
||||
|
||||
output_dir = base_dir / "Copies" / input_pdf.stem
|
||||
generated_files = set()
|
||||
parts_by_label = defaultdict(list)
|
||||
|
||||
# 1. Parse labels to strip '|' and determine type: L (Left), R (Right), N (Normal)
|
||||
parsed_coords = []
|
||||
for item in coords_list:
|
||||
label, pn, y0, y1, x0, x1 = item
|
||||
def _parse_coordinates(coords_list: list[Coordinate]) -> list[ParsedCoordinate]:
|
||||
parsed: list[ParsedCoordinate] = []
|
||||
for label, page, y0, y1, x0, x1 in coords_list:
|
||||
if label.startswith("|"):
|
||||
c_type, clean_label = "L", label[1:]
|
||||
kind, clean_label = "L", label[1:]
|
||||
elif label.endswith("|"):
|
||||
c_type, clean_label = "R", label[:-1]
|
||||
kind, clean_label = "R", label[:-1]
|
||||
else:
|
||||
c_type, clean_label = "N", label
|
||||
parsed_coords.append((clean_label, c_type, pn, y0, y1, x0, x1))
|
||||
kind, clean_label = "N", label
|
||||
parsed.append((clean_label, kind, page, y0, y1, x0, x1))
|
||||
filtered: list[ParsedCoordinate] = []
|
||||
for item in parsed:
|
||||
if not filtered or item[0] != filtered[-1][0]:
|
||||
filtered.append(item)
|
||||
return filtered
|
||||
|
||||
# 2. Filter consecutive duplicate labels based on the cleaned name
|
||||
filtered_coords = []
|
||||
if parsed_coords:
|
||||
filtered_coords.append(parsed_coords[0])
|
||||
for item in parsed_coords[1:]:
|
||||
if item[0] != filtered_coords[-1][0]:
|
||||
filtered_coords.append(item)
|
||||
coords_list = filtered_coords
|
||||
|
||||
def scale_coord(y, page):
|
||||
"""Scale y from 0–1000 range to PDF points."""
|
||||
page_height = page.rect.height
|
||||
return (y / 1000) * page_height
|
||||
|
||||
def save_cropped_page(doc, page_num, x0, y0, x1, y1, out_path):
|
||||
"""Saves a cropped portion of a page as a new PDF."""
|
||||
page = doc[page_num]
|
||||
rotated_rect = page.rect * page.transformation_matrix
|
||||
visual_crop_rect = fitz.Rect(rotated_rect.x0 + x0, y0, rotated_rect.x0 + x1, y1)
|
||||
unrotated_clip_rect = visual_crop_rect * page.derotation_matrix
|
||||
|
||||
temp_doc = fitz.open()
|
||||
temp_page = temp_doc.new_page(
|
||||
width=visual_crop_rect.width,
|
||||
height=visual_crop_rect.height
|
||||
)
|
||||
temp_page.show_pdf_page(
|
||||
temp_page.rect,
|
||||
doc,
|
||||
page_num,
|
||||
def _save_cropped_page(
|
||||
document: fitz.Document,
|
||||
page_number: int,
|
||||
x0: float,
|
||||
y0: float,
|
||||
x1: float,
|
||||
y1: float,
|
||||
output_path: Path,
|
||||
) -> None:
|
||||
page = document[page_number]
|
||||
rotated_rectangle = page.rect * page.transformation_matrix
|
||||
visual_crop = fitz.Rect(
|
||||
rotated_rectangle.x0 + x0,
|
||||
y0,
|
||||
rotated_rectangle.x0 + x1,
|
||||
y1,
|
||||
)
|
||||
unrotated_clip = visual_crop * page.derotation_matrix
|
||||
cropped = fitz.open()
|
||||
try:
|
||||
target_page = cropped.new_page(width=visual_crop.width, height=visual_crop.height)
|
||||
target_page.show_pdf_page(
|
||||
target_page.rect,
|
||||
document,
|
||||
page_number,
|
||||
rotate=-page.rotation,
|
||||
clip=unrotated_clip_rect
|
||||
clip=unrotated_clip,
|
||||
)
|
||||
temp_doc.save(out_path)
|
||||
temp_doc.close()
|
||||
cropped.save(output_path)
|
||||
finally:
|
||||
cropped.close()
|
||||
|
||||
# Iterate through all labels
|
||||
for idx, (clean_label, c_type, start_page, y_start_raw, y_end_box, x0_raw, x1_raw) in enumerate(coords_list):
|
||||
if clean_label == "_":
|
||||
|
||||
def _render_split_outputs(
|
||||
input_pdf: Path,
|
||||
coords_list: list[Coordinate],
|
||||
staging: Path,
|
||||
) -> set[str]:
|
||||
"""Render every current answer into an otherwise empty staging directory."""
|
||||
document = fitz.open(input_pdf)
|
||||
try:
|
||||
parsed = _parse_coordinates(coords_list)
|
||||
parts_by_label: defaultdict[str, list[Path]] = defaultdict(list)
|
||||
with tempfile.TemporaryDirectory(prefix="copienator-split-") as temp_directory:
|
||||
temporary = Path(temp_directory)
|
||||
for index, item in enumerate(parsed):
|
||||
clean_label, kind, start_page, y_start, _y_end, x0_raw, _x1_raw = item
|
||||
if clean_label == "_":
|
||||
continue
|
||||
if not 0 <= start_page < document.page_count:
|
||||
raise ValueError(
|
||||
f"Invalid page {start_page} for {input_pdf.name}"
|
||||
)
|
||||
end_page = document.page_count - 1
|
||||
end_y = 1000
|
||||
for next_item in parsed[index + 1 :]:
|
||||
_next_label, next_kind, next_page, next_y, *_rest = next_item
|
||||
if (
|
||||
(kind == "L" and next_kind in {"L", "N"})
|
||||
or (kind == "R" and next_kind in {"R", "N"})
|
||||
or kind == "N"
|
||||
):
|
||||
end_page = next_page
|
||||
end_y = min(next_y + int(1.5 * SQUARE), 1000)
|
||||
break
|
||||
|
||||
column_width = 1000 / document.page_count
|
||||
if kind == "L":
|
||||
fraction_x0 = (x0_raw % column_width) / column_width
|
||||
fraction_x1 = 1.0
|
||||
end_y = min(1000, end_y + 40)
|
||||
elif kind == "R":
|
||||
fraction_x0 = 0.0
|
||||
left_labels = [entry for entry in parsed if entry[1] == "L"]
|
||||
if left_labels:
|
||||
closest = min(left_labels, key=lambda entry: abs(entry[3] - y_start))
|
||||
center = (closest[5] + closest[6]) / 2.0
|
||||
fraction_x1 = (center % column_width) / column_width
|
||||
if fraction_x1 <= fraction_x0:
|
||||
fraction_x1 = 1.0
|
||||
else:
|
||||
fraction_x1 = 1.0
|
||||
else:
|
||||
fraction_x0, fraction_x1 = 0.0, 1.0
|
||||
|
||||
for page_number in range(start_page, end_page + 1):
|
||||
page = document[page_number]
|
||||
y0 = (y_start / 1000) * page.rect.height if page_number == start_page else 0
|
||||
y1 = (end_y / 1000) * page.rect.height if page_number == end_page else page.rect.height
|
||||
if y1 <= y0 + 1:
|
||||
continue
|
||||
part_path = temporary / f"part-{index}-{page_number}.pdf"
|
||||
_save_cropped_page(
|
||||
document,
|
||||
page_number,
|
||||
fraction_x0 * page.rect.width,
|
||||
y0,
|
||||
fraction_x1 * page.rect.width,
|
||||
y1,
|
||||
part_path,
|
||||
)
|
||||
parts_by_label[clean_label].append(part_path)
|
||||
|
||||
generated: set[str] = set()
|
||||
for label, parts in parts_by_label.items():
|
||||
filename = f"{label}.pdf"
|
||||
merger = PdfWriter()
|
||||
try:
|
||||
for part in parts:
|
||||
merger.append(part)
|
||||
merger.write(staging / filename)
|
||||
finally:
|
||||
merger.close()
|
||||
generated.add(filename)
|
||||
return generated
|
||||
finally:
|
||||
document.close()
|
||||
|
||||
|
||||
def _preserve_previous_outputs(
|
||||
output_dir: Path,
|
||||
staging: Path,
|
||||
generated_files: set[str],
|
||||
) -> None:
|
||||
if not output_dir.is_dir():
|
||||
return
|
||||
for directory in (path for path in output_dir.iterdir() if path.is_dir()):
|
||||
shutil.copytree(directory, staging / directory.name, dirs_exist_ok=True)
|
||||
missing_dir = staging / "Missing"
|
||||
for item in (path for path in output_dir.iterdir() if path.is_file()):
|
||||
if item.name in generated_files:
|
||||
continue
|
||||
print(f"ALERT: File '{item.name}' not generated. Moving to {missing_dir}")
|
||||
missing_dir.mkdir(exist_ok=True)
|
||||
shutil.copy2(item, missing_dir / item.name)
|
||||
|
||||
temp_parts = []
|
||||
end_page = doc.page_count - 1
|
||||
end_y_target_raw = 1000
|
||||
|
||||
# RULE 2: Determine stopping label
|
||||
for next_item in coords_list[idx + 1:]:
|
||||
n_clean, n_type, n_pn, n_y_start, n_y_end, _, _ = next_item
|
||||
def split_an_interro(
|
||||
workspace: EvaluationWorkspace,
|
||||
input_pdf: Path,
|
||||
coords_list: list[Coordinate],
|
||||
) -> None:
|
||||
"""Regenerate one copy's answers and preserve obsolete ones under Missing."""
|
||||
output_dir = workspace.copies_dir / input_pdf.stem
|
||||
with staged_directory(output_dir) as staging:
|
||||
generated = _render_split_outputs(input_pdf, coords_list, staging)
|
||||
_preserve_previous_outputs(output_dir, staging, generated)
|
||||
|
||||
if c_type == "L":
|
||||
is_stop = (n_type in ("L", "N"))
|
||||
elif c_type == "R":
|
||||
is_stop = (n_type in ("R", "N"))
|
||||
else:
|
||||
is_stop = True # Normal labels stop at anything
|
||||
|
||||
if is_stop:
|
||||
end_page = n_pn
|
||||
# end_y_target_raw = n_y_start
|
||||
# On avait retiré un carreau précédemment inutilement, on le rajoute, plus un demi carreau
|
||||
end_y_target_raw = min(n_y_start + int(1.5 * carreau), 1000)
|
||||
break
|
||||
def _selected_pdfs(workspace: EvaluationWorkspace, target: Path) -> list[Path]:
|
||||
workspace.require_directories("Copies")
|
||||
if target.is_file():
|
||||
if target.suffix.casefold() != ".pdf":
|
||||
raise CliError(f"Target is not a PDF: {target}", ExitCode.INVALID_ARGUMENTS)
|
||||
return [target]
|
||||
return sorted(workspace.copies_dir.glob("*.pdf"), key=lambda path: path.name.casefold())
|
||||
|
||||
# RULES 3 & 4: Calculate horizontal boundaries (0.0 to 1.0 fraction of local page width)
|
||||
col_w = 1000 / doc.page_count
|
||||
if c_type == "L": # |name
|
||||
fraction_x0 = (x0_raw % col_w) / col_w
|
||||
fraction_x1 = 1.0
|
||||
end_y_target_raw = min(1000, end_y_target_raw + 40)
|
||||
elif c_type == "R": # name|
|
||||
fraction_x0 = 0.0
|
||||
# Find the closest 'L' label in y-distance
|
||||
L_labels = [it for it in parsed_coords if it[1] == "L"]
|
||||
if L_labels:
|
||||
closest_L = min(L_labels, key=lambda it: abs(it[3] - y_start_raw))
|
||||
closest_L_x_center = (closest_L[5] + closest_L[6]) / 2.0
|
||||
fraction_x1 = (closest_L_x_center % col_w) / col_w
|
||||
if fraction_x1 <= fraction_x0: fraction_x1 = 1.0 # Fallback
|
||||
else:
|
||||
fraction_x1 = 1.0
|
||||
else: # Normal
|
||||
fraction_x0 = 0.0
|
||||
fraction_x1 = 1.0
|
||||
|
||||
current_p = start_page
|
||||
while current_p <= end_page:
|
||||
page = doc[current_p]
|
||||
def run(workspace: EvaluationWorkspace, target: Path) -> ExitCode:
|
||||
workspace.require_files("labels")
|
||||
utils.read_all_labels(workspace.root)
|
||||
pdf_files = _selected_pdfs(workspace, target)
|
||||
status = ExitCode.SUCCESS
|
||||
for pdf_path in pdf_files:
|
||||
json_path = pdf_path.with_suffix(".json")
|
||||
if not json_path.is_file():
|
||||
print(f"Warning: No JSON found for {pdf_path.name}")
|
||||
status = ExitCode.PARTIAL
|
||||
continue
|
||||
name, coordinates = decode_json(pdf_path)
|
||||
print(f"Decoded name: {name}")
|
||||
split_an_interro(workspace, pdf_path, coordinates)
|
||||
if not pdf_files:
|
||||
print("No PDF copies found.")
|
||||
return status
|
||||
|
||||
y0 = scale_coord(y_start_raw, page) if current_p == start_page else 0
|
||||
y1 = scale_coord(end_y_target_raw, page) if current_p == end_page else page.rect.height
|
||||
|
||||
if y1 > y0 + 1:
|
||||
# Convert fractions to absolute PDF points
|
||||
x0_pdf = fraction_x0 * page.rect.width
|
||||
x1_pdf = fraction_x1 * page.rect.width
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
return target_parser("Split verified PDF copies into answers by label")
|
||||
|
||||
temp_path = f"_part_{idx}_{current_p}.pdf"
|
||||
save_cropped_page(doc, current_p, x0_pdf, y0, x1_pdf, y1, temp_path)
|
||||
temp_parts.append(temp_path)
|
||||
|
||||
current_p += 1
|
||||
def main(argv: Sequence[str] | None = None) -> int:
|
||||
parser = build_parser()
|
||||
|
||||
parts_by_label[clean_label].extend(temp_parts)
|
||||
def handle(args: argparse.Namespace) -> ExitCode:
|
||||
workspace, target = workspace_from_target(args)
|
||||
return run(workspace, target)
|
||||
|
||||
output_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Process aggregated parts by label
|
||||
for title, parts in parts_by_label.items():
|
||||
merger = PdfWriter()
|
||||
for part in parts:
|
||||
if os.path.exists(part):
|
||||
merger.append(part)
|
||||
|
||||
filename = f"{title}.pdf"
|
||||
merger.write(output_dir / filename)
|
||||
merger.close()
|
||||
generated_files.add(filename)
|
||||
|
||||
# Cleanup
|
||||
for part in parts:
|
||||
if os.path.exists(part):
|
||||
os.remove(part)
|
||||
|
||||
doc.close()
|
||||
|
||||
# Move files not generated in this run to 'Missing' folder
|
||||
if output_dir.exists():
|
||||
missing_dir = output_dir / "Missing"
|
||||
for item in output_dir.iterdir():
|
||||
if item.is_file() and item.name not in generated_files:
|
||||
print(f"ALERT: File '{item.name}' not generated. Moving to {missing_dir}")
|
||||
missing_dir.mkdir(exist_ok=True)
|
||||
item.rename(missing_dir / item.name)
|
||||
return execute(parser, argv, handle)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) < 2:
|
||||
print("Usage: python script.py <directory or pdf_file>")
|
||||
sys.exit(1)
|
||||
|
||||
input_arg = Path(sys.argv[1])
|
||||
|
||||
if input_arg.is_file():
|
||||
base_dir = input_arg.parent
|
||||
if base_dir.name == "Copies":
|
||||
base_dir = base_dir.parent
|
||||
pdf_files = [input_arg]
|
||||
elif input_arg.is_dir():
|
||||
base_dir = input_arg
|
||||
copies_dir = base_dir / "Copies"
|
||||
pdf_files = sorted(copies_dir.glob("*.pdf"))
|
||||
else:
|
||||
print(f"Error: {input_arg} is not a valid file or directory.")
|
||||
sys.exit(1)
|
||||
|
||||
read_all_labels(base_dir)
|
||||
|
||||
for pdf_path in pdf_files:
|
||||
json_path = pdf_path.with_suffix(".json")
|
||||
# print("Debug :", json_path)
|
||||
if json_path.exists():
|
||||
(name, coords) = decode_json(pdf_path)
|
||||
print("Decoded name : ", name)
|
||||
split_an_interro(base_dir, pdf_path, coords)
|
||||
else:
|
||||
print(f"Warning: No JSON found for {pdf_path.name}")
|
||||
raise SystemExit(main())
|
||||
|
||||
Reference in New Issue
Block a user