corrective-rag-data-assistant / scripts /build_data_assets.py
nvikou's picture
Upgrade CRAG: embeddings, SQL SQLite, pro UI, 60-eval, local data
98aa813 verified
Raw History Blame Contribute Delete
3.11 kB
"""Génère PDF minimaux lisibles par pypdf + seed SQLite."""
from __future__ import annotations
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
from pypdf import PdfReader # noqa: E402
from app.core.config import get_settings # noqa: E402
from app.sql.seed import seed_company_db # noqa: E402
def _esc(text: str) -> str:
return (
text.replace("\\", "\\\\")
.replace("(", "\\(")
.replace(")", "\\)")
)
def write_simple_pdf(path: Path, title: str, body: str) -> None:
lines = [title, ""]
for para in body.splitlines():
while len(para) > 88:
lines.append(para[:88])
para = para[88:]
lines.append(para)
safe = [
line.encode("latin-1", "replace").decode("latin-1")
for line in lines[:70]
]
content = ["BT", "/F1 10 Tf", "40 780 Td", "12 TL"]
first = True
for line in safe:
if not first:
content.append("T*")
content.append(f"({_esc(line)}) Tj")
first = False
content.append("ET")
stream = "\n".join(content).encode("latin-1")
objects = [
b"1 0 obj<< /Type /Catalog /Pages 2 0 R >>endobj\n",
b"2 0 obj<< /Type /Pages /Kids [3 0 R] /Count 1 >>endobj\n",
(
b"3 0 obj<< /Type /Page /Parent 2 0 R "
b"/MediaBox [0 0 612 792] /Contents 4 0 R "
b"/Resources << /Font << /F1 5 0 R >> >> >>endobj\n"
),
(
f"4 0 obj<< /Length {len(stream)} >>stream\n".encode("ascii")
+ stream
+ b"\nendstream\nendobj\n"
),
b"5 0 obj<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>endobj\n",
]
out = bytearray(b"%PDF-1.4\n")
offsets = [0]
for obj in objects:
offsets.append(len(out))
out.extend(obj)
xref = len(out)
out.extend(f"xref\n0 {len(offsets)}\n".encode("ascii"))
out.extend(b"0000000000 65535 f \n")
for off in offsets[1:]:
out.extend(f"{off:010d} 00000 n \n".encode("ascii"))
out.extend(
(
f"trailer<< /Size {len(offsets)} /Root 1 0 R >>\n"
f"startxref\n{xref}\n%%EOF\n"
).encode("ascii")
)
path.write_bytes(out)
def main() -> None:
pairs = [
(
ROOT / "data/policies/refund_policy.md",
ROOT / "data/policies/refund_policy.pdf",
),
(
ROOT / "data/reports/profitability_report.md",
ROOT / "data/reports/profitability_report.pdf",
),
]
for src, dst in pairs:
text = src.read_text(encoding="utf-8")
write_simple_pdf(dst, src.stem, text)
reader = PdfReader(str(dst))
extracted = "".join(
(page.extract_text() or "") for page in reader.pages
)
print(f"PDF ok {dst.name}: {len(extracted)} chars")
get_settings.cache_clear()
db = seed_company_db(ROOT / "data/database/company.db", ROOT / "data")
print(f"DB ok {db} ({db.stat().st_size} octets)")
if __name__ == "__main__":
main()