Upload converters/pipeline/step3_domain.py

03.Code/업로드용/converters/pipeline/step3_domain.py

@@ -0,0 +1,224 @@
+# -*- coding: utf-8 -*-
+from dotenv import load_dotenv
+load_dotenv()
+"""
+domain_prompt.py
+
+기능:
+- D:\\test\\report 내의 텍스트 파일들을 읽어 내용 요약을 생성합니다.
+- 추출된 정보를 바탕으로, 해당 문서 묶음이 어떤 도메인/업종의 문서인지 분석합니다.
+- 분석 결과는 텍스트 형태의 프롬프트(domain_prompt.txt)로 저장됩니다.
+- 향후 단계에서 모든 GPT 호출(system role)에 이 분석 내용을 주입하여 사용합니다.
+"""
+import os
+import sys
+import json
+from pathlib import Path
+import pdfplumber
+import fitz  # PyMuPDF
+import pandas as pd
+from openai import OpenAI
+
+# ===== OpenAI 설정 (구조 유지, 현재 단계에서는 실제 키가 없어도 코드 작성 가능) =====
+OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY")
+client = OpenAI(api_key=OPENAI_API_KEY)
+GPT_MODEL      = "gpt-5-2025-08-07"
+SKIP_DIR_NAMES = {"System Volume Information", "$RECYCLE.BIN", ".git", "__pycache__"}
+
+
+def log(msg: str):
+    print(msg, flush=True)
+    with (LOG_DIR / "domain_prompt_log.txt").open("a", encoding="utf-8") as f:
+        f.write(msg + "\n")
+
+
+def safe_rel(p: Path) -> str:
+    try:
+        return str(p.relative_to(DATA_ROOT))
+    except Exception:
+        return str(p)
+
+
+def sample_from_pdf(p: Path, max_chars: int = 1000) -> str:
+    texts = []
+    try:
+        with pdfplumber.open(str(p)) as pdf:
+            # 상단 몇 페이지만 샘플링
+            for page in pdf.pages[:3]:
+                t = page.extract_text() or ""
+                if t:
+                    texts.append(t)
+                if sum(len(x) for x in texts) >= max_chars:
+                    break
+    except Exception as e:
+        log(f"[WARN] PDF 샘플 추출 실패: {safe_rel(p)} | {e}")
+    joined = "\n".join(texts)
+    return joined[:max_chars]
+
+
+def sample_from_xlsx(p: Path, max_chars: int = 1000) -> str:
+    texts = [f"[파일명] {p.name}"]
+    try:
+        xls = pd.ExcelFile(str(p))
+        for sheet_name in xls.sheet_names[:3]:
+            try:
+                df = xls.parse(sheet_name)
+            except Exception as e:
+                log(f"[WARN] 엑셀 로드 실패: {safe_rel(p)} | {sheet_name} | {e}")
+                continue
+            texts.append(f"\n[시트명] {sheet_name}")
+            texts.append("컬럼: " + ", ".join(map(str, df.columns)))
+            head = df.head(5)
+            texts.append(head.to_string(index=False))
+            if sum(len(x) for x in texts) >= max_chars:
+                break
+    except Exception as e:
+        log(f"[WARN] XLSX 샘플 추출 실패: {safe_rel(p)} | {e}")
+    joined = "\n".join(texts)
+    return joined[:max_chars]
+
+
+def sample_from_text_file(p: Path, max_chars: int = 1000) -> str:
+    try:
+        t = p.read_text(encoding="utf-8", errors="ignore")
+    except Exception:
+        t = p.read_text(encoding="cp949", errors="ignore")
+    return t[:max_chars]
+
+
+def gather_file_samples(
+    max_files_per_type: int = 100,
+    max_total_samples: int = 300,
+    max_chars_per_sample: int = 1000,
+):
+
+    file_names = []
+    samples = []
+
+    count_pdf = 0
+    count_xlsx = 0
+    count_img = 0
+    count_txt = 0
+
+    for root, dirs, files in os.walk(DATA_ROOT):
+        dirs[:] = [d for d in dirs if d not in SKIP_DIR_NAMES and not d.startswith(".")]
+        cur_dir = Path(root)
+
+        for fname in files:
+            fpath = cur_dir / fname
+            ext = fpath.suffix.lower()
+
+            # 전체 문서 목록은 모두 수집하여 리스트업
+            file_names.append(safe_rel(fpath))
+
+            if len(samples) >= max_total_samples:
+                continue
+
+            try:
+                if ext == ".pdf" and count_pdf < max_files_per_type:
+                    s = sample_from_pdf(fpath, max_chars=max_chars_per_sample)
+                    if s.strip():
+                        samples.append(f"[PDF] {safe_rel(fpath)}\n{s}")
+                        count_pdf += 1
+                    continue
+
+                if ext in {".xlsx", ".xls"} and count_xlsx < max_files_per_type:
+                    s = sample_from_xlsx(fpath, max_chars=max_chars_per_sample)
+                    if s.strip():
+                        samples.append(f"[XLSX] {safe_rel(fpath)}\n{s}")
+                        count_xlsx += 1
+                    continue
+
+                if ext in {".txt", ".md"} and count_txt < max_files_per_type:
+                    s = sample_from_text_file(fpath, max_chars=max_chars_per_sample)
+                    if s.strip():
+                        samples.append(f"[TEXT] {safe_rel(fpath)}\n{s}")
+                        count_txt += 1
+                    continue
+
+            except Exception as e:
+                log(f"[WARN] 파일 샘플 추출 실패: {safe_rel(fpath)} | {e}")
+                continue
+
+    return file_names, samples
+
+
+def build_domain_prompt():
+    """
+    파일명 + 샘플 내용을 바탕으로 GPT에 질의하여
+    '본 시스템은 ~~ 분야의 전문 문서를 다루는...' 형태의 프롬프트를 생성합니다.
+    """
+    log("도메인 정보 수집 및 분석 시작...")
+    file_names, samples = gather_file_samples()
+
+    if not file_names and not samples:
+        log("분석할 수 있는 문서 데이터가 없습니다.")
+        sys.exit(1)
+
+    file_names_text = "\n".join(file_names[:80])
+    sample_text = "\n\n".join(samples[:30])
+
+    prompt = f"""
+다음은 특정 기술 분야의 문서 데이터 샘플입니다.
+
+[문서 리스트/내용 샘플]
+{sample_text}
+
+이 정보들을 바탕으로 이 문서 묶음이 어떤 산업, 업무, 분야의 것인지,
+주요 용어와 특징은 무엇인지 요약한 2~3문장 정도의 시스템 가이드 문구를 작성하세요.
+또한, 이 분야의 전문가가 사용하는 AI로서의 페르소나(persona)를 정의하세요.
+
+작성 규칙:
+   - 첫 문장: "본 시스템은 ~~ 분야의 전문 문서를 다룹니다." 형식으로, 이 문서 묶음의 성격을 정의하세요.
+   - 두 번째 문장 이후: "주요 내용은 ~~를 포함합니다.", "우리는 ~~ 분야의 전문가를 돕기 위한 보고서 자동 생성 가이드를 제공합니다." 형식으로
+     사용자 및 AI의 전문성과 목적을 서술하세요.
+   - 출력은 반드시 한글로 작성하세요.
+   - 이 내용이 프롬프트(보고서 요약, RAG, 보고서 작성 등)의 앞단에 연결될 것이므로 역할(role), 목적, 기준(측량 기준, 사용 기법 등)이
+     잘 포함되어야 합니다.
+
+출력은 반드시 설명이나 머리말 없이 분석 내용만 텍스트로 작성하세요.
+이 내용 전체를 domain_prompt.txt로 저장할 것입니다.
+"""
+
+    resp = client.chat.completions.create(
+        model=GPT_MODEL,
+        messages=[
+            {
+                "role": "system",
+                "content": "당신은 문서 묶음의 성격을 분석하여, 향후 AI 시스템의 시스템 프롬프트를 구성하는 도메인 분석 전문가입니다."
+            },
+            {
+                "role": "user",
+                "content": prompt
+            }
+        ],
+    )
+
+    content = (resp.choices[0].message.content or "").strip()
+    out_path = CONTEXT_DIR / "domain_prompt.txt"
+    out_path.write_text(content, encoding="utf-8")
+
+    log(f"도메인 프롬프트 생성 완료: {out_path}")
+    return content
+
+
+def main(input_dir, output_dir):
+    global DATA_ROOT, OUTPUT_ROOT, CONTEXT_DIR, LOG_DIR
+    DATA_ROOT   = Path(input_dir)
+    OUTPUT_ROOT = Path(output_dir)
+    CONTEXT_DIR = OUTPUT_ROOT / "context"
+    LOG_DIR     = OUTPUT_ROOT / "logs"
+    for d in [OUTPUT_ROOT, CONTEXT_DIR, LOG_DIR]:
+        d.mkdir(parents=True, exist_ok=True)
+    log("=== 도메인 프롬프트 생성 시작 ===")
+    out_path = CONTEXT_DIR / "domain_prompt.txt"
+    if out_path.exists():
+        log(f"이미 존재함: {out_path}")
+        log("기존 정보를 사용하려면 건너뛰고, 새로 생성하려면 파일을 삭제 후 다시 실행하십시오.")
+    else:
+        build_domain_prompt()
+    log("=== 도메인 프롬프트 생성 완료 ===")
+
+
+if __name__ == "__main__":
+    main()