데이터 준비

# UI: Catalog > Volumes > Upload Files
# CLI: databricks fs cp ./documents/ dbfs:/Volumes/catalog/schema/docs/ --recursive

df = spark.read.format("binaryFile") \
    .option("pathGlobFilter", "*.pdf") \
    .load("s3://my-bucket/documents/")  # External Location 설정 필요

import requests
docs = requests.get("https://api.example.com/documents").json()
spark.createDataFrame(docs).write.mode("overwrite").saveAsTable("catalog.schema.raw_documents")

from pyspark.sql.functions import col

# ai_parse_document으로 PDF/이미지/Office 문서 파싱
parsed_df = spark.sql("""
    SELECT
        path,
        ai_parse_document(content, 'markdown') AS parsed_text
    FROM read_files(
        '/Volumes/catalog/schema/docs/',
        format => 'binaryFile'
    )
""")

from pypdf import PdfReader

def parse_pdf(file_path):
    reader = PdfReader(file_path)
    return "\n".join([page.extract_text() for page in reader.pages])

# ai_parse_document의 주요 장점
# 1. 레이아웃 인식: 다단 레이아웃에서도 올바른 읽기 순서 유지
# 2. 테이블 추출: 표를 Markdown 테이블 형식으로 변환
# 3. OCR 내장: 스캔된 문서도 텍스트 추출 가능
# 4. 이미지 설명: 이미지에 대한 텍스트 설명 생성

# 다양한 출력 포맷 지원
parsed_md = spark.sql("""
    SELECT path, ai_parse_document(content, 'markdown') AS parsed_text
    FROM read_files('/Volumes/catalog/schema/docs/', format => 'binaryFile')
""")

# JSON 형태로 구조화된 출력도 가능
parsed_json = spark.sql("""
    SELECT path, ai_parse_document(content, 'json') AS parsed_json
    FROM read_files('/Volumes/catalog/schema/docs/', format => 'binaryFile')
""")

import re
from datetime import datetime

def extract_metadata(file_path: str, parsed_text: str) -> dict:
    """파싱된 문서에서 메타데이터를 자동 추출"""
    metadata = {
        "source": file_path,
        "file_type": file_path.split(".")[-1].lower(),
        "parsed_at": datetime.now().isoformat(),
        "char_count": len(parsed_text),
        "word_count": len(parsed_text.split()),
    }

    # 문서 제목 추출 (첫 번째 헤딩)
    title_match = re.search(r'^#\s+(.+)', parsed_text, re.MULTILINE)
    if title_match:
        metadata["title"] = title_match.group(1).strip()

    # 날짜 패턴 추출 (YYYY-MM-DD 또는 YYYY.MM.DD)
    date_match = re.search(r'(\d{4}[-./]\d{1,2}[-./]\d{1,2})', parsed_text)
    if date_match:
        metadata["document_date"] = date_match.group(1)

    # 카테고리 추론 (키워드 기반)
    text_lower = parsed_text.lower()
    if any(kw in text_lower for kw in ["정책", "규정", "지침"]):
        metadata["category"] = "policy"
    elif any(kw in text_lower for kw in ["api", "sdk", "코드", "함수"]):
        metadata["category"] = "technical"
    elif any(kw in text_lower for kw in ["faq", "자주 묻는"]):
        metadata["category"] = "faq"

    return metadata

import re

def clean_parsed_text(text: str) -> str:
    """파싱된 텍스트에서 노이즈를 제거하는 정제 파이프라인"""

    # 1. 반복되는 헤더/푸터 제거 (페이지 번호, 회사명 등)
    text = re.sub(r'Page \d+ of \d+', '', text)
    text = re.sub(r'- \d+ -', '', text)  # 페이지 번호 패턴

    # 2. 과도한 공백 정리
    text = re.sub(r'\n{3,}', '\n\n', text)  # 3줄 이상 빈 줄 → 2줄로
    text = re.sub(r' {2,}', ' ', text)      # 연속 공백 → 단일 공백

    # 3. 특수 문자 정리 (유니코드 제어 문자 등)
    text = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f]', '', text)

    # 4. URL 보존하되 트래킹 파라미터 제거
    text = re.sub(r'(\?utm_[^\s]+)', '', text)

    # 5. 빈 테이블 행 제거
    text = re.sub(r'\|[\s|]*\|', '', text)

    return text.strip()

from langchain.text_splitter import CharacterTextSplitter

splitter = CharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separator="\n"
)
chunks = splitter.split_text(document_text)

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", ".", " "]
)
chunks = splitter.split_text(document_text)

from langchain_experimental.text_splitter import SemanticChunker
from langchain_community.embeddings import DatabricksEmbeddings

embeddings = DatabricksEmbeddings(endpoint="databricks-gte-large-en")
splitter = SemanticChunker(embeddings, breakpoint_threshold_type="percentile")
chunks = splitter.create_documents([document_text])

import pandas as pd

chunk_data = [{"chunk_id": f"doc_{d}_chunk_{i}", "content": c, "source": doc_sources[d]}
              for d, doc_chunks in enumerate(all_chunks) for i, c in enumerate(doc_chunks)]

chunks_df = spark.createDataFrame(pd.DataFrame(chunk_data))

# Delta Table로 저장 (Change Data Feed 활성화 필수)
chunks_df.write.format("delta") \
    .option("delta.enableChangeDataFeed", "true") \
    .mode("overwrite").saveAsTable("catalog.schema.document_chunks")

from pyspark.sql.functions import col, current_timestamp

# 마지막 처리 시점 이후에 추가/변경된 파일만 읽기
new_files = spark.sql(f"""
    SELECT path, content, modificationTime
    FROM read_files('/Volumes/catalog/schema/docs/', format => 'binaryFile')
    WHERE modificationTime > '{last_processed_time}'
""")

# 파싱 및 청킹 수행
# ... (기존 파싱/청킹 코드)

# 증분 저장 (append 모드)
new_chunks_df.write.format("delta") \
    .option("delta.enableChangeDataFeed", "true") \
    .mode("append").saveAsTable("catalog.schema.document_chunks")