Pdf to rag md.py
Jump to navigation
Jump to search
#!/usr/bin/env python3
"""
Convert a text-based PDF into RAG-friendly Markdown files, one file per topic.
Main features
-------------
- Uses the PDF table of contents when available.
- Falls back to font-size / bold / numbering heuristics for headings.
- Falls back again to page groups if no reliable headings exist.
- Ignores images and other non-text PDF blocks.
- Removes repeated headers and footers.
- Removes standalone page numbers.
- Removes likely footnotes conservatively.
- Writes YAML front matter containing traceable RAG metadata.
- Creates a JSON manifest for auditing and bulk ingestion.
Notes
-----
This script is intended for PDFs that already contain selectable text.
Scanned/image-only PDFs must be OCRed first.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import math
import re
import sys
import unicodedata
from collections import Counter, defaultdict
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable, Sequence
try:
import pymupdf # PyMuPDF >= modern releases
except ImportError: # compatibility with older PyMuPDF installations
try:
import fitz as pymupdf # type: ignore
except ImportError as exc:
raise SystemExit(
"PyMuPDF belum terpasang. Jalankan: pip install pymupdf pyyaml"
) from exc
try:
import yaml
except ImportError as exc:
raise SystemExit(
"PyYAML belum terpasang. Jalankan: pip install pymupdf pyyaml"
) from exc
PAGE_NUMBER_RE = re.compile(
r"^\s*(?:(?:page|halaman|hlm\.?|p\.)\s*)?"
r"[-–—]?\s*(?:\d{1,5}|[ivxlcdm]{1,12})\s*[-–—]?\s*$",
re.IGNORECASE,
)
NUMBERED_HEADING_RE = re.compile(
r"^\s*(?:"
r"(?:bab|chapter|bagian|part|section|seksi)\s+[A-Z0-9IVXLCDM]+"
r"|(?:\d+(?:\.\d+){0,4})[\s.)–—:-]+"
r")",
re.IGNORECASE,
)
FOOTNOTE_MARKER_RE = re.compile(
r"^\s*(?:\[?\d{1,3}\]?|[*†‡])(?:[.)\]:-]|\s)+"
)
BULLET_RE = re.compile(r"^\s*(?:[-*•◦▪‣]|\d+[.)]|[a-zA-Z][.)])\s+")
SKIP_TOPIC_RE = re.compile(
r"^\s*(?:"
r"daftar\s+isi|table\s+of\s+contents|contents|"
r"daftar\s+gambar|list\s+of\s+figures|"
r"daftar\s+tabel|list\s+of\s+tables"
r")\s*$",
re.IGNORECASE,
)
@dataclass(slots=True)
class TextBlock:
page_index: int
page_number: int
page_width: float
page_height: float
x0: float
y0: float
x1: float
y1: float
text: str
avg_font_size: float
max_font_size: float
bold_ratio: float
line_count: int
@property
def top_ratio(self) -> float:
return self.y0 / self.page_height if self.page_height else 0.0
@property
def bottom_ratio(self) -> float:
return self.y1 / self.page_height if self.page_height else 0.0
@dataclass(slots=True)
class HeadingInfo:
title: str
level: int
source: str
@dataclass
class Topic:
title: str
index: int
detection_method: str
blocks: list[TextBlock] = field(default_factory=list)
body_parts: list[str] = field(default_factory=list)
pages: set[int] = field(default_factory=set)
def add(self, block: TextBlock, markdown: str) -> None:
self.blocks.append(block)
self.body_parts.append(markdown)
self.pages.add(block.page_number)
@property
def body(self) -> str:
text = "\n\n".join(part.strip() for part in self.body_parts if part.strip())
return clean_markdown_spacing(text)
@dataclass(slots=True)
class Config:
topic_level: int
fallback_pages: int
top_band: float
bottom_band: float
repeat_ratio: float
drop_footnotes: bool
footnote_bottom_band: float
footnote_font_ratio: float
language: str
tags: list[str]
source_uri: str | None
include_page_comments: bool
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description=(
"Ubah PDF menjadi file Markdown per topik untuk RAG, lengkap "
"dengan YAML metadata dan pembersihan artefak PDF."
)
)
parser.add_argument("pdf", type=Path, help="File PDF sumber")
parser.add_argument(
"-o",
"--output-dir",
type=Path,
default=None,
help="Folder output. Default: <nama-pdf>_rag_md",
)
parser.add_argument(
"--topic-level",
type=int,
choices=(1, 2, 3, 4),
default=2,
help="Heading sampai level ini akan menjadi file topik baru. Default: 2",
)
parser.add_argument(
"--fallback-pages",
type=int,
default=5,
help="Jika heading tidak terdeteksi, buat satu topik per N halaman. Default: 5",
)
parser.add_argument(
"--language",
default="id",
help="Kode bahasa metadata, misalnya id atau en. Default: id",
)
parser.add_argument(
"--tags",
default="",
help="Tag metadata dipisahkan koma, misalnya: keamanan,soc,wazuh",
)
parser.add_argument(
"--source-uri",
default=None,
help="URI/URL sumber asli, jika ada",
)
parser.add_argument(
"--password",
default=None,
help="Password PDF terenkripsi, jika diperlukan",
)
parser.add_argument(
"--keep-footnotes",
action="store_true",
help="Jangan hapus blok yang terdeteksi sebagai footnote",
)
parser.add_argument(
"--page-comments",
action="store_true",
help="Tambahkan komentar HTML saat halaman berubah untuk trace lebih rinci",
)
parser.add_argument(
"--top-band",
type=float,
default=0.12,
help="Proporsi area atas untuk deteksi header berulang. Default: 0.12",
)
parser.add_argument(
"--bottom-band",
type=float,
default=0.12,
help="Proporsi area bawah untuk deteksi footer berulang. Default: 0.12",
)
parser.add_argument(
"--repeat-ratio",
type=float,
default=0.35,
help="Rasio halaman minimum agar teks dianggap header/footer berulang. Default: 0.35",
)
parser.add_argument(
"--footnote-bottom-band",
type=float,
default=0.18,
help="Proporsi area bawah untuk deteksi footnote. Default: 0.18",
)
parser.add_argument(
"--footnote-font-ratio",
type=float,
default=0.86,
help="Ukuran font footnote relatif terhadap font isi. Default: 0.86",
)
return parser.parse_args()
def utc_now_iso() -> str:
return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
def sha256_file(path: Path, chunk_size: int = 1024 * 1024) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
while chunk := handle.read(chunk_size):
digest.update(chunk)
return digest.hexdigest()
def normalize_unicode(text: str) -> str:
text = unicodedata.normalize("NFKC", text)
return (
text.replace("\u00ad", "")
.replace("\u200b", "")
.replace("\ufeff", "")
.replace("\xa0", " ")
)
def normalize_repeat_key(text: str) -> str:
text = normalize_unicode(text).lower()
text = re.sub(r"\b\d+\b", "#", text)
text = re.sub(r"\s+", " ", text).strip(" -–—|•\t")
return text
def normalize_heading_key(text: str) -> str:
text = normalize_unicode(text).casefold()
text = re.sub(r"[^\w\s]", " ", text, flags=re.UNICODE)
return re.sub(r"\s+", " ", text).strip()
def weighted_mode_font_size(samples: Iterable[tuple[float, int]]) -> float:
buckets: Counter[float] = Counter()
for size, weight in samples:
if size <= 0 or weight <= 0:
continue
bucket = round(size * 2) / 2.0
buckets[bucket] += weight
if not buckets:
return 11.0
return float(buckets.most_common(1)[0][0])
def join_pdf_lines(lines: Sequence[str]) -> str:
"""Join visual PDF lines into cleaner paragraphs while preserving list items."""
output: list[str] = []
current = ""
for raw_line in lines:
line = re.sub(r"\s+", " ", normalize_unicode(raw_line)).strip()
if not line:
continue
if BULLET_RE.match(line):
if current:
output.append(current.strip())
current = ""
bullet = re.sub(r"^\s*[•◦▪‣]\s+", "- ", line)
output.append(bullet)
continue
if not current:
current = line
continue
if current.endswith("-") and line[:1].islower():
current = current[:-1] + line
else:
current += " " + line
if current:
output.append(current.strip())
return "\n".join(output)
def extract_raw_blocks(doc: Any) -> tuple[list[TextBlock], list[tuple[float, int]]]:
blocks: list[TextBlock] = []
font_samples: list[tuple[float, int]] = []
for page_index in range(doc.page_count):
page = doc.load_page(page_index)
page_dict = page.get_text("dict", sort=True)
page_width = float(page.rect.width)
page_height = float(page.rect.height)
for raw_block in page_dict.get("blocks", []):
# type 0 is text. Image and drawing blocks are deliberately ignored.
if raw_block.get("type") != 0:
continue
line_texts: list[str] = []
block_font_weight = 0
block_font_total = 0.0
block_max_font = 0.0
bold_chars = 0
total_chars = 0
for line in raw_block.get("lines", []):
span_texts: list[str] = []
for span in line.get("spans", []):
text = normalize_unicode(str(span.get("text", "")))
if not text:
continue
size = float(span.get("size", 0.0) or 0.0)
chars = max(len(text.strip()), 1)
font_name = str(span.get("font", "")).casefold()
is_bold = "bold" in font_name or "black" in font_name or "semibold" in font_name
span_texts.append(text)
block_font_total += size * chars
block_font_weight += chars
block_max_font = max(block_max_font, size)
total_chars += chars
if is_bold:
bold_chars += chars
font_samples.append((size, min(chars, 200)))
line_text = "".join(span_texts).strip()
if line_text:
line_texts.append(line_text)
text = join_pdf_lines(line_texts).strip()
if not text:
continue
bbox = raw_block.get("bbox", (0.0, 0.0, 0.0, 0.0))
avg_font = block_font_total / block_font_weight if block_font_weight else 0.0
bold_ratio = bold_chars / total_chars if total_chars else 0.0
blocks.append(
TextBlock(
page_index=page_index,
page_number=page_index + 1,
page_width=page_width,
page_height=page_height,
x0=float(bbox[0]),
y0=float(bbox[1]),
x1=float(bbox[2]),
y1=float(bbox[3]),
text=text,
avg_font_size=avg_font,
max_font_size=block_max_font,
bold_ratio=bold_ratio,
line_count=len(line_texts),
)
)
return blocks, font_samples
def detect_repeated_margin_text(
blocks: Sequence[TextBlock],
page_count: int,
top_band: float,
bottom_band: float,
repeat_ratio: float,
) -> set[str]:
occurrences: dict[str, set[int]] = defaultdict(set)
for block in blocks:
is_top = block.top_ratio <= top_band
is_bottom = block.bottom_ratio >= 1.0 - bottom_band
if not (is_top or is_bottom):
continue
key = normalize_repeat_key(block.text)
if 2 <= len(key) <= 180:
occurrences[key].add(block.page_number)
threshold = max(3, math.ceil(page_count * repeat_ratio))
return {key for key, pages in occurrences.items() if len(pages) >= threshold}
def is_standalone_page_number(text: str) -> bool:
compact = re.sub(r"\s+", " ", text).strip()
return bool(PAGE_NUMBER_RE.fullmatch(compact))
def is_likely_footnote(block: TextBlock, body_font: float, config: Config) -> bool:
if not config.drop_footnotes:
return False
small_font = block.avg_font_size <= body_font * config.footnote_font_ratio
in_bottom_area = block.top_ratio >= 1.0 - config.footnote_bottom_band
marker = bool(FOOTNOTE_MARKER_RE.match(block.text))
shortish = len(block.text) <= 900
# Conservative: require both positional/font evidence, or a strong marker
# combined with small font near the lower half of the page.
if small_font and in_bottom_area and shortish:
return True
if marker and small_font and block.top_ratio >= 0.55 and shortish:
return True
return False
def clean_blocks(
blocks: Sequence[TextBlock],
repeated_margin_text: set[str],
body_font: float,
config: Config,
) -> list[TextBlock]:
cleaned: list[TextBlock] = []
for block in blocks:
key = normalize_repeat_key(block.text)
is_margin = (
block.top_ratio <= config.top_band
or block.bottom_ratio >= 1.0 - config.bottom_band
)
if is_margin and key in repeated_margin_text:
continue
if is_standalone_page_number(block.text):
continue
if is_likely_footnote(block, body_font, config):
continue
text = clean_block_text(block.text)
if not text:
continue
cleaned.append(
TextBlock(
page_index=block.page_index,
page_number=block.page_number,
page_width=block.page_width,
page_height=block.page_height,
x0=block.x0,
y0=block.y0,
x1=block.x1,
y1=block.y1,
text=text,
avg_font_size=block.avg_font_size,
max_font_size=block.max_font_size,
bold_ratio=block.bold_ratio,
line_count=block.line_count,
)
)
return cleaned
def clean_block_text(text: str) -> str:
text = normalize_unicode(text)
lines: list[str] = []
for line in text.splitlines():
line = re.sub(r"[ \t]+", " ", line).strip()
if not line or is_standalone_page_number(line):
continue
line = re.sub(r"^\s*[•◦▪‣]\s+", "- ", line)
lines.append(line)
return "\n".join(lines).strip()
def get_toc_map(doc: Any) -> tuple[dict[int, list[tuple[int, str]]], list[list[Any]]]:
toc = doc.get_toc(simple=True) or []
by_page: dict[int, list[tuple[int, str]]] = defaultdict(list)
for entry in toc:
if len(entry) < 3:
continue
level, title, page_number = entry[:3]
title = re.sub(r"\s+", " ", str(title)).strip()
if not title or not isinstance(page_number, int) or page_number < 1:
continue
by_page[page_number].append((int(level), title))
return by_page, toc
def match_toc_heading(
block: TextBlock,
toc_by_page: dict[int, list[tuple[int, str]]],
) -> HeadingInfo | None:
block_key = normalize_heading_key(block.text)
if not block_key:
return None
candidates: list[tuple[int, str]] = []
for page_number in (block.page_number - 1, block.page_number, block.page_number + 1):
candidates.extend(toc_by_page.get(page_number, []))
for level, title in candidates:
title_key = normalize_heading_key(title)
if not title_key:
continue
exact = block_key == title_key
contained = (
len(title_key) >= 5
and (block_key.startswith(title_key) or title_key.startswith(block_key))
and min(len(block_key), len(title_key)) / max(len(block_key), len(title_key)) >= 0.72
)
if exact or contained:
return HeadingInfo(title=title, level=max(1, min(level, 6)), source="pdf_toc")
return None
def looks_like_heading_text(text: str) -> bool:
one_line = re.sub(r"\s+", " ", text).strip()
if not one_line or len(one_line) > 180:
return False
if one_line.count(".") >= 4 and len(one_line) > 90:
return False
if len(one_line.split()) > 22:
return False
if one_line.endswith((".", ";", ",")) and not NUMBERED_HEADING_RE.match(one_line):
return False
return True
def infer_heading_level(text: str, font_ratio: float) -> int:
compact = re.sub(r"\s+", " ", text).strip()
lowered = compact.casefold()
if re.match(r"^(bab|chapter|bagian|part)\s+", lowered):
return 1
numbered = re.match(r"^(\d+(?:\.\d+){0,4})[\s.)–—:-]+", compact)
if numbered:
depth = numbered.group(1).count(".") + 1
return min(depth, 4)
if font_ratio >= 1.65:
return 1
if font_ratio >= 1.38:
return 2
return 3
def detect_heading(
block: TextBlock,
body_font: float,
toc_by_page: dict[int, list[tuple[int, str]]],
) -> HeadingInfo | None:
toc_match = match_toc_heading(block, toc_by_page)
if toc_match:
return toc_match
if not looks_like_heading_text(block.text):
return None
font_ratio = block.avg_font_size / body_font if body_font else 1.0
compact = re.sub(r"\s+", " ", block.text).strip()
numbered = bool(NUMBERED_HEADING_RE.match(compact))
mostly_bold = block.bold_ratio >= 0.55
mostly_upper = len(compact) >= 4 and compact.upper() == compact and any(c.isalpha() for c in compact)
strong_size = font_ratio >= 1.20
modest_size_with_signal = font_ratio >= 1.04 and (mostly_bold or numbered or mostly_upper)
if not (strong_size or modest_size_with_signal):
return None
level = infer_heading_level(compact, font_ratio)
return HeadingInfo(title=compact, level=level, source="font_heuristic")
def unique_toc_starts_for_page(
page_number: int,
toc_by_page: dict[int, list[tuple[int, str]]],
topic_level: int,
) -> list[tuple[int, str]]:
seen: set[str] = set()
result: list[tuple[int, str]] = []
for level, title in toc_by_page.get(page_number, []):
if level > topic_level:
continue
key = normalize_heading_key(title)
if not key or key in seen:
continue
seen.add(key)
result.append((level, title))
return result
def topic_default_title(doc_title: str, pdf_stem: str) -> str:
title = re.sub(r"\s+", " ", doc_title or "").strip()
return title if title else pdf_stem.replace("_", " ").replace("-", " ").strip().title()
def build_topics_from_headings(
blocks: Sequence[TextBlock],
body_font: float,
toc_by_page: dict[int, list[tuple[int, str]]],
default_title: str,
config: Config,
) -> tuple[list[Topic], int]:
topics: list[Topic] = []
current: Topic | None = None
detected_splits = 0
last_page: int | None = None
page_toc_started: set[int] = set()
last_body_page: int | None = None
def start_topic(title: str, method: str) -> Topic:
nonlocal detected_splits
cleaned_title = re.sub(r"\s+", " ", title).strip() or default_title
topic = Topic(
title=cleaned_title,
index=len(topics) + 1,
detection_method=method,
)
topics.append(topic)
detected_splits += 1
return topic
for block in blocks:
if block.page_number != last_page:
last_page = block.page_number
toc_starts = unique_toc_starts_for_page(
block.page_number, toc_by_page, config.topic_level
)
if toc_starts and block.page_number not in page_toc_started:
_, toc_title = toc_starts[0]
if current is None or normalize_heading_key(current.title) != normalize_heading_key(toc_title):
current = start_topic(toc_title, "pdf_toc")
page_toc_started.add(block.page_number)
heading = detect_heading(block, body_font, toc_by_page)
if heading and heading.level <= config.topic_level:
same_as_current = (
current is not None
and normalize_heading_key(current.title) == normalize_heading_key(heading.title)
)
if not same_as_current:
current = start_topic(heading.title, heading.source)
# The topic title is already emitted as H1 during writing.
continue
if current is None:
current = Topic(
title=default_title,
index=1,
detection_method="document_intro",
)
topics.append(current)
if config.include_page_comments and last_body_page != block.page_number:
current.body_parts.append(f"<!-- source_page: {block.page_number} -->")
last_body_page = block.page_number
if heading:
markdown_level = max(2, min(heading.level + 1, 6))
current.add(block, f"{'#' * markdown_level} {heading.title}")
else:
current.add(block, block.text)
# Remove empty topics and fix indexes.
topics = [topic for topic in topics if topic.body.strip()]
for index, topic in enumerate(topics, start=1):
topic.index = index
return topics, detected_splits
def best_fallback_title(blocks: Sequence[TextBlock], default: str) -> str:
for block in blocks:
compact = re.sub(r"\s+", " ", block.text).strip()
if looks_like_heading_text(compact) and len(compact) <= 110:
if block.bold_ratio >= 0.45 or block.line_count == 1:
return compact
return default
def build_topics_by_page_groups(
blocks: Sequence[TextBlock],
page_count: int,
fallback_pages: int,
default_title: str,
include_page_comments: bool,
) -> list[Topic]:
by_page: dict[int, list[TextBlock]] = defaultdict(list)
for block in blocks:
by_page[block.page_number].append(block)
topics: list[Topic] = []
fallback_pages = max(1, fallback_pages)
for start_page in range(1, page_count + 1, fallback_pages):
end_page = min(page_count, start_page + fallback_pages - 1)
group_blocks: list[TextBlock] = []
for page_number in range(start_page, end_page + 1):
group_blocks.extend(by_page.get(page_number, []))
if not group_blocks:
continue
fallback_title = f"{default_title} — Halaman {start_page}-{end_page}"
title = best_fallback_title(group_blocks, fallback_title)
topic = Topic(
title=title,
index=len(topics) + 1,
detection_method="page_group_fallback",
)
last_page: int | None = None
for block in group_blocks:
if include_page_comments and last_page != block.page_number:
topic.body_parts.append(f"<!-- source_page: {block.page_number} -->")
last_page = block.page_number
topic.add(block, block.text)
topics.append(topic)
return topics
def clean_markdown_spacing(text: str) -> str:
text = normalize_unicode(text)
text = re.sub(r"[ \t]+\n", "\n", text)
text = re.sub(r"\n{3,}", "\n\n", text)
return text.strip()
def slugify(text: str, max_length: int = 100) -> str:
text = normalize_unicode(text).casefold()
text = unicodedata.normalize("NFKD", text)
text = "".join(char for char in text if not unicodedata.combining(char))
text = re.sub(r"[^a-z0-9]+", "-", text)
text = re.sub(r"-+", "-", text).strip("-")
return (text[:max_length].rstrip("-") or "topik")
def ensure_unique_filename(base_name: str, used: set[str]) -> str:
candidate = base_name
counter = 2
while candidate in used:
stem = Path(base_name).stem
suffix = Path(base_name).suffix
candidate = f"{stem}-{counter}{suffix}"
counter += 1
used.add(candidate)
return candidate
def metadata_value(value: Any) -> Any:
if isinstance(value, str):
return re.sub(r"\s+", " ", value).strip()
return value
def build_front_matter(
*,
topic: Topic,
filename: str,
source_pdf: Path,
source_sha256: str,
document_id: str,
pdf_metadata: dict[str, Any],
config: Config,
body: str,
generated_at: str,
body_font: float,
) -> dict[str, Any]:
pages = sorted(topic.pages)
page_start = pages[0] if pages else None
page_end = pages[-1] if pages else None
topic_id = f"{document_id}-topic-{topic.index:04d}"
return {
"schema_version": "1.0",
"document_id": document_id,
"topic_id": topic_id,
"chunk_id": topic_id,
"title": topic.title,
"topic": topic.title,
"topic_index": topic.index,
"language": config.language,
"tags": config.tags,
"content_type": "text/markdown",
"source": {
"file": source_pdf.name,
"uri": config.source_uri,
"sha256": source_sha256,
"page_start": page_start,
"page_end": page_end,
"pages": pages,
"trace": (
f"{source_pdf.name}#page={page_start}-{page_end}"
if page_start is not None
else source_pdf.name
),
"pdf_title": metadata_value(pdf_metadata.get("title", "")) or None,
"pdf_author": metadata_value(pdf_metadata.get("author", "")) or None,
"pdf_subject": metadata_value(pdf_metadata.get("subject", "")) or None,
"pdf_keywords": metadata_value(pdf_metadata.get("keywords", "")) or None,
},
"rag": {
"unit": "topic",
"topic_detection": topic.detection_method,
"word_count": len(re.findall(r"\b\w+\b", body, flags=re.UNICODE)),
"character_count": len(body),
"suggested_splitter": "markdown_headers_then_token_window",
},
"extraction": {
"engine": "PyMuPDF",
"body_font_size_estimate": round(body_font, 2),
"images": "ignored",
"repeated_header_footer_filter": "position_and_frequency",
"standalone_page_number_filter": "enabled",
"footnote_filter": (
"bottom_position_and_small_font"
if config.drop_footnotes
else "disabled"
),
"page_comments_in_body": config.include_page_comments,
},
"output_file": filename,
"generated_at": generated_at,
}
def write_topic_files(
topics: Sequence[Topic],
output_dir: Path,
source_pdf: Path,
source_sha256: str,
document_id: str,
pdf_metadata: dict[str, Any],
config: Config,
body_font: float,
) -> list[dict[str, Any]]:
output_dir.mkdir(parents=True, exist_ok=True)
generated_at = utc_now_iso()
used_filenames: set[str] = set()
manifest_topics: list[dict[str, Any]] = []
for topic in topics:
if SKIP_TOPIC_RE.fullmatch(topic.title.strip()):
continue
body = topic.body
if not body:
continue
base_filename = f"{topic.index:03d}-{slugify(topic.title)}.md"
filename = ensure_unique_filename(base_filename, used_filenames)
metadata = build_front_matter(
topic=topic,
filename=filename,
source_pdf=source_pdf,
source_sha256=source_sha256,
document_id=document_id,
pdf_metadata=pdf_metadata,
config=config,
body=body,
generated_at=generated_at,
body_font=body_font,
)
front_matter = yaml.safe_dump(
metadata,
allow_unicode=True,
sort_keys=False,
default_flow_style=False,
width=120,
).strip()
markdown = f"---\n{front_matter}\n---\n\n# {topic.title}\n\n{body}\n"
destination = output_dir / filename
destination.write_text(markdown, encoding="utf-8")
manifest_topics.append(
{
"topic_index": topic.index,
"topic_id": metadata["topic_id"],
"title": topic.title,
"file": filename,
"page_start": metadata["source"]["page_start"],
"page_end": metadata["source"]["page_end"],
"word_count": metadata["rag"]["word_count"],
"detection": topic.detection_method,
}
)
return manifest_topics
def write_manifest(
output_dir: Path,
source_pdf: Path,
source_sha256: str,
document_id: str,
page_count: int,
body_font: float,
removed_repeated_items: int,
topics: list[dict[str, Any]],
config: Config,
) -> None:
manifest = {
"schema_version": "1.0",
"document_id": document_id,
"source_file": source_pdf.name,
"source_sha256": source_sha256,
"page_count": page_count,
"body_font_size_estimate": round(body_font, 2),
"repeated_header_footer_patterns_removed": removed_repeated_items,
"topic_count": len(topics),
"language": config.language,
"tags": config.tags,
"generated_at": utc_now_iso(),
"topics": topics,
}
(output_dir / "manifest.json").write_text(
json.dumps(manifest, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
def open_pdf(path: Path, password: str | None) -> Any:
try:
doc = pymupdf.open(path)
except Exception as exc:
raise RuntimeError(f"Gagal membuka PDF: {exc}") from exc
if getattr(doc, "needs_pass", False):
if not password:
doc.close()
raise RuntimeError("PDF terenkripsi. Gunakan opsi --password.")
if not doc.authenticate(password):
doc.close()
raise RuntimeError("Password PDF salah.")
return doc
def main() -> int:
args = parse_args()
pdf_path = args.pdf.expanduser().resolve()
if not pdf_path.is_file():
print(f"ERROR: File tidak ditemukan: {pdf_path}", file=sys.stderr)
return 2
if pdf_path.suffix.casefold() != ".pdf":
print("ERROR: Input harus berupa file .pdf", file=sys.stderr)
return 2
output_dir = (
args.output_dir.expanduser().resolve()
if args.output_dir
else pdf_path.with_name(f"{pdf_path.stem}_rag_md")
)
tags = [item.strip() for item in args.tags.split(",") if item.strip()]
config = Config(
topic_level=args.topic_level,
fallback_pages=max(1, args.fallback_pages),
top_band=min(max(args.top_band, 0.02), 0.30),
bottom_band=min(max(args.bottom_band, 0.02), 0.30),
repeat_ratio=min(max(args.repeat_ratio, 0.10), 0.95),
drop_footnotes=not args.keep_footnotes,
footnote_bottom_band=min(max(args.footnote_bottom_band, 0.05), 0.40),
footnote_font_ratio=min(max(args.footnote_font_ratio, 0.50), 1.0),
language=args.language.strip() or "und",
tags=tags,
source_uri=args.source_uri,
include_page_comments=args.page_comments,
)
source_sha256 = sha256_file(pdf_path)
document_id = f"pdf-{source_sha256[:20]}"
try:
doc = open_pdf(pdf_path, args.password)
except RuntimeError as exc:
print(f"ERROR: {exc}", file=sys.stderr)
return 2
try:
raw_blocks, font_samples = extract_raw_blocks(doc)
if not raw_blocks:
print(
"ERROR: Tidak ditemukan selectable text. PDF kemungkinan hasil scan; lakukan OCR terlebih dahulu.",
file=sys.stderr,
)
return 3
body_font = weighted_mode_font_size(font_samples)
repeated_margin_text = detect_repeated_margin_text(
raw_blocks,
doc.page_count,
config.top_band,
config.bottom_band,
config.repeat_ratio,
)
blocks = clean_blocks(raw_blocks, repeated_margin_text, body_font, config)
if not blocks:
print("ERROR: Semua teks terhapus oleh filter. Coba gunakan --keep-footnotes.", file=sys.stderr)
return 3
toc_by_page, toc = get_toc_map(doc)
pdf_metadata = dict(doc.metadata or {})
default_title = topic_default_title(
str(pdf_metadata.get("title", "")), pdf_path.stem
)
topics, detected_splits = build_topics_from_headings(
blocks,
body_font,
toc_by_page,
default_title,
config,
)
# If the PDF has no usable TOC/headings, a single very large topic is not
# ideal for retrieval. Split it into deterministic page groups instead.
if detected_splits == 0 or (len(topics) <= 1 and doc.page_count > config.fallback_pages):
topics = build_topics_by_page_groups(
blocks,
doc.page_count,
config.fallback_pages,
default_title,
config.include_page_comments,
)
manifest_topics = write_topic_files(
topics,
output_dir,
pdf_path,
source_sha256,
document_id,
pdf_metadata,
config,
body_font,
)
write_manifest(
output_dir,
pdf_path,
source_sha256,
document_id,
doc.page_count,
body_font,
len(repeated_margin_text),
manifest_topics,
config,
)
print(f"Selesai: {len(manifest_topics)} file topik dibuat di {output_dir}")
print(f"Manifest: {output_dir / 'manifest.json'}")
if toc:
print(f"Struktur PDF: TOC tersedia ({len(toc)} entri)")
else:
print("Struktur PDF: TOC tidak tersedia; memakai heading/font atau fallback halaman")
return 0
finally:
doc.close()
if __name__ == "__main__":
raise SystemExit(main())