114 lines
4.0 KiB
Python
114 lines
4.0 KiB
Python
#!/usr/bin/env python3
|
|
"""수령한 O2O XLSX를 provenance SQLite 코퍼스로 적재한다."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from collections import Counter
|
|
from pathlib import Path
|
|
|
|
if __package__ in (None, ""):
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from app.engine.provenance import CorpusStore, DocumentRecord, SegmentRecord, stable_id
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
p = argparse.ArgumentParser(description=__doc__)
|
|
p.add_argument("xlsx", type=Path)
|
|
p.add_argument("--database", type=Path, required=True)
|
|
p.add_argument("--sheet", default=None, help="기본값: 첫 번째 시트")
|
|
p.add_argument("--book-column", default="book_name")
|
|
p.add_argument("--text-column", default="에피소드")
|
|
p.add_argument("--index-column", default="episode_index")
|
|
p.add_argument("--path-column", default="json_path")
|
|
return p.parse_args()
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
try:
|
|
import openpyxl
|
|
except ImportError:
|
|
print("openpyxl이 필요합니다: pip install openpyxl", file=sys.stderr)
|
|
return 2
|
|
if not args.xlsx.is_file():
|
|
print(f"파일을 찾을 수 없습니다: {args.xlsx}", file=sys.stderr)
|
|
return 2
|
|
|
|
wb = openpyxl.load_workbook(args.xlsx, read_only=True, data_only=True)
|
|
ws = wb[args.sheet] if args.sheet else wb.worksheets[0]
|
|
rows = ws.iter_rows(values_only=True)
|
|
headers = [str(v).strip() if v is not None else "" for v in next(rows)]
|
|
positions = {name: i for i, name in enumerate(headers)}
|
|
required = [args.book_column, args.text_column, args.index_column]
|
|
missing = [name for name in required if name not in positions]
|
|
if missing:
|
|
print(f"필수 열 없음: {missing}; 실제 열={headers}", file=sys.stderr)
|
|
return 2
|
|
|
|
store = CorpusStore(args.database)
|
|
store.initialize()
|
|
skipped = 0
|
|
book_counts: Counter[str] = Counter()
|
|
documents: dict[str, DocumentRecord] = {}
|
|
segments: list[SegmentRecord] = []
|
|
|
|
for row in rows:
|
|
book = str(row[positions[args.book_column]] or "").strip()
|
|
text = str(row[positions[args.text_column]] or "").strip()
|
|
ordinal = str(row[positions[args.index_column]] or "").strip()
|
|
if not book or not text:
|
|
skipped += 1
|
|
continue
|
|
source_path = None
|
|
if args.path_column in positions:
|
|
source_path = str(row[positions[args.path_column]] or "").strip() or None
|
|
document_id = stable_id("doc", book)
|
|
documents[document_id] = DocumentRecord(
|
|
document_id=document_id,
|
|
title=book,
|
|
source_path=source_path,
|
|
metadata={"import_source": args.xlsx.name},
|
|
)
|
|
segments.append(SegmentRecord(
|
|
segment_id=stable_id("seg", document_id, text),
|
|
document_id=document_id,
|
|
text=text,
|
|
ordinal=ordinal,
|
|
coordinate_scope="episode",
|
|
char_start=0,
|
|
char_end=len(text),
|
|
source_locator=f"{source_path or args.xlsx.name}#episode={ordinal}",
|
|
metadata={
|
|
"provenance_quality": "episode_only",
|
|
"page_offset_available": False,
|
|
},
|
|
))
|
|
book_counts[book] += 1
|
|
|
|
store.upsert_documents(documents.values())
|
|
inserted, duplicates = store.add_segments(segments)
|
|
|
|
report = {
|
|
"database": str(args.database),
|
|
"sheet": ws.title,
|
|
"inserted_segments": inserted,
|
|
"duplicate_segments": duplicates,
|
|
"skipped_rows": skipped,
|
|
"source_books": len(book_counts),
|
|
"store": store.stats(),
|
|
"location_warning": (
|
|
"수령 XLSX에는 원본 페이지/문단 offset이 없어 episode 좌표만 저장했습니다. "
|
|
"PDF/DOCX 재추출 전에는 페이지 근거를 표시할 수 없습니다."
|
|
),
|
|
}
|
|
print(json.dumps(report, ensure_ascii=False, indent=2))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|