138 lines
5.2 KiB
Python
138 lines
5.2 KiB
Python
"""자유 양식 표 파일(엑셀/CSV) → 텍스트 변환 (P4, 2026-07-13 한신대 회의).
|
|
|
|
연구팀이 교수자 페이지에 올리는 자유 양식 엑셀(파란 라벨 방식 포함)을
|
|
페르소나 저작 KB에 넣을 수 있는 평문으로 결정론 변환한다.
|
|
|
|
원칙:
|
|
- 업로드 원본 바이트는 여기서 파싱만 하고 어디에도 저장하지 않는다(원본 파기 원칙).
|
|
파생 텍스트만 기존 `/personas/sources` 마스킹·hash-only 증거 경로로 넘어간다.
|
|
- 양식 변형에 견디도록 셀 색/서식에 의존하지 않는다 — 비어있지 않은 셀만 행 단위로 평탄화한다.
|
|
- openpyxl 은 선택 의존성이다(presidio 패턴). 없으면 명확한 한국어 오류로 안내한다.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import io
|
|
|
|
# 변환 상한 — 파일럿 추고록 기준 여유값. LLM/KB 입력 상한(120k)과 정합.
|
|
MAX_TEXT_CHARS = 120_000
|
|
MAX_CELLS = 40_000
|
|
MAX_UPLOAD_BYTES = 8 * 1024 * 1024 # 8MB
|
|
|
|
|
|
class TabularIngestError(ValueError):
|
|
"""사용자에게 그대로 보여줄 수 있는 한국어 사유를 담는다."""
|
|
|
|
|
|
def _try_load_openpyxl():
|
|
try:
|
|
import openpyxl # noqa: PLC0415 — 선택 의존성 지연 로드
|
|
|
|
return openpyxl
|
|
except ImportError as exc: # pragma: no cover - 설치 환경에선 도달하지 않음
|
|
raise TabularIngestError(
|
|
"엑셀 변환 모듈(openpyxl)이 설치되어 있지 않습니다. 관리자에게 API 의존성 설치를 요청하세요."
|
|
) from exc
|
|
|
|
|
|
def _cell_text(value: object) -> str:
|
|
if value is None:
|
|
return ""
|
|
if isinstance(value, float) and value.is_integer():
|
|
return str(int(value))
|
|
text = str(value).strip()
|
|
return " ".join(text.split())
|
|
|
|
|
|
def _rows_to_lines(rows: list[list[str]]) -> list[str]:
|
|
lines: list[str] = []
|
|
for cells in rows:
|
|
filled = [cell for cell in cells if cell]
|
|
if not filled:
|
|
continue
|
|
if len(filled) == 2:
|
|
# 자유 양식에서 가장 흔한 "라벨 | 값" 행 — 읽기 좋은 쌍으로 변환.
|
|
lines.append(f"{filled[0]}: {filled[1]}")
|
|
else:
|
|
lines.append(" | ".join(filled))
|
|
return lines
|
|
|
|
|
|
def _extract_xlsx(data: bytes) -> str:
|
|
openpyxl = _try_load_openpyxl()
|
|
try:
|
|
workbook = openpyxl.load_workbook(
|
|
io.BytesIO(data), read_only=True, data_only=True
|
|
)
|
|
except Exception as exc:
|
|
raise TabularIngestError(
|
|
"엑셀 파일을 열지 못했습니다. 손상되지 않은 .xlsx 파일인지 확인해 주세요."
|
|
) from exc
|
|
try:
|
|
sections: list[str] = []
|
|
cell_budget = MAX_CELLS
|
|
for sheet in workbook.worksheets:
|
|
rows: list[list[str]] = []
|
|
for row in sheet.iter_rows(values_only=True):
|
|
if cell_budget <= 0:
|
|
break
|
|
cell_budget -= len(row)
|
|
rows.append([_cell_text(value) for value in row])
|
|
lines = _rows_to_lines(rows)
|
|
if lines:
|
|
sections.append(f"## 시트: {sheet.title}\n" + "\n".join(lines))
|
|
if cell_budget <= 0:
|
|
sections.append("(셀 수 상한에 도달해 이후 내용은 생략했습니다)")
|
|
break
|
|
return "\n\n".join(sections)
|
|
finally:
|
|
workbook.close()
|
|
|
|
|
|
def _extract_csv(data: bytes) -> str:
|
|
text: str | None = None
|
|
for encoding in ("utf-8-sig", "cp949", "utf-8"):
|
|
try:
|
|
text = data.decode(encoding)
|
|
break
|
|
except UnicodeDecodeError:
|
|
continue
|
|
if text is None:
|
|
raise TabularIngestError(
|
|
"CSV 인코딩을 해석하지 못했습니다. UTF-8 또는 엑셀(xlsx)로 저장해 다시 올려 주세요."
|
|
)
|
|
rows = [[_cell_text(cell) for cell in row] for row in csv.reader(io.StringIO(text))]
|
|
return "\n".join(_rows_to_lines(rows[: MAX_CELLS // 8]))
|
|
|
|
|
|
def extract_tabular_text(*, filename: str, data: bytes) -> str:
|
|
"""업로드 파일을 KB 등록용 평문으로 변환한다. 실패 사유는 TabularIngestError."""
|
|
if not data:
|
|
raise TabularIngestError("업로드된 파일이 비어 있습니다.")
|
|
if len(data) > MAX_UPLOAD_BYTES:
|
|
raise TabularIngestError("파일이 8MB를 넘습니다. 시트를 나눠 다시 올려 주세요.")
|
|
|
|
lowered = (filename or "").lower()
|
|
if lowered.endswith(".xlsx") or lowered.endswith(".xlsm"):
|
|
text = _extract_xlsx(data)
|
|
elif lowered.endswith(".xls"):
|
|
raise TabularIngestError(
|
|
"구형 엑셀(.xls)은 지원하지 않습니다. 엑셀에서 '다른 이름으로 저장 → .xlsx'로 변환해 올려 주세요."
|
|
)
|
|
elif lowered.endswith(".csv"):
|
|
text = _extract_csv(data)
|
|
else:
|
|
raise TabularIngestError(
|
|
"지원하지 않는 파일 형식입니다. 엑셀(.xlsx) 또는 CSV 파일을 올려 주세요."
|
|
)
|
|
|
|
text = text.strip()
|
|
if len(text) < 20:
|
|
raise TabularIngestError(
|
|
"표에서 읽을 수 있는 텍스트가 거의 없습니다. 내용이 있는 시트인지 확인해 주세요."
|
|
)
|
|
return text[:MAX_TEXT_CHARS]
|
|
|
|
|
|
__all__ = ["TabularIngestError", "extract_tabular_text", "MAX_UPLOAD_BYTES"]
|