메일 분석시스템 보완
This commit is contained in:
234
analyze.py
234
analyze.py
@@ -2,91 +2,165 @@ import os
|
||||
import re
|
||||
import unicodedata
|
||||
from pypdf import PdfReader
|
||||
try:
|
||||
import pytesseract
|
||||
from pdf2image import convert_from_path
|
||||
from PIL import Image
|
||||
TESSERACT_PATH = r'C:\Users\User\AppData\Local\Programs\Tesseract-OCR\tesseract.exe'
|
||||
POPPLER_PATH = r'D:\이태훈\00크롬다운로드\poppler-25.12.0\Library\bin'
|
||||
pytesseract.pytesseract.tesseract_cmd = TESSERACT_PATH
|
||||
OCR_AVAILABLE = True
|
||||
except ImportError:
|
||||
OCR_AVAILABLE = False
|
||||
import pytesseract
|
||||
from pdf2image import convert_from_path
|
||||
|
||||
def analyze_file_content(filename: str):
|
||||
file_path = os.path.join("sample", filename)
|
||||
if not os.path.exists(file_path):
|
||||
return {"error": "File not found"}
|
||||
|
||||
log_steps = []
|
||||
|
||||
# Layer 1: 제목 분석 (Quick)
|
||||
log_steps.append("1. 레이어: 파일 제목(Title) 스캔 중...")
|
||||
title_text = filename.lower().replace(" ", "")
|
||||
|
||||
# Layer 2: 텍스트 추출 (Fast)
|
||||
log_steps.append("2. 레이어: PDF 텍스트 엔진(Extraction) 가동...")
|
||||
text_content = ""
|
||||
try:
|
||||
if filename.lower().endswith(".pdf"):
|
||||
reader = PdfReader(file_path)
|
||||
for page in reader.pages[:5]: # 전체가 아닌 핵심 페이지 위주
|
||||
page_txt = page.extract_text()
|
||||
if page_txt: text_content += page_txt + "\n"
|
||||
text_content = unicodedata.normalize('NFC', text_content)
|
||||
log_steps.append(f" - 텍스트 데이터 확보 완료 ({len(text_content)}자)")
|
||||
except:
|
||||
log_steps.append(" - 텍스트 추출 실패")
|
||||
# 1. 시스템 설정
|
||||
TESSERACT_EXE = r'C:\Users\User\AppData\Local\Programs\Tesseract-OCR\tesseract.exe'
|
||||
TESSDATA_DIR = r'C:\Users\User\AppData\Local\Programs\Tesseract-OCR\tessdata'
|
||||
POPPLER_BIN = r'D:\이태훈\00크롬다운로드\poppler-25.12.0\Library\bin'
|
||||
|
||||
# Layer 3: OCR 정밀 분석 (Deep)
|
||||
log_steps.append("3. 레이어: OCR 이미지 스캔(Vision) 강제 실행...")
|
||||
ocr_content = ""
|
||||
if OCR_AVAILABLE and os.path.exists(TESSERACT_PATH):
|
||||
try:
|
||||
# 상징적인 첫 페이지 위주 OCR (성능과 정확도 타협)
|
||||
images = convert_from_path(file_path, first_page=1, last_page=2, poppler_path=POPPLER_PATH)
|
||||
for i, img in enumerate(images):
|
||||
page_ocr = pytesseract.image_to_string(img, lang='kor+eng')
|
||||
ocr_content += unicodedata.normalize('NFC', page_ocr) + "\n"
|
||||
log_steps.append(f" - OCR 스캔 완료 ({len(ocr_content)}자)")
|
||||
except Exception as e:
|
||||
log_steps.append(f" - OCR 오류: {str(e)[:20]}")
|
||||
pytesseract.pytesseract.tesseract_cmd = TESSERACT_EXE
|
||||
os.environ["TESSDATA_PREFIX"] = TESSDATA_DIR
|
||||
OCR_AVAILABLE = os.path.exists(TESSERACT_EXE)
|
||||
|
||||
SYSTEM_HIERARCHY = {
|
||||
"행정": {
|
||||
"계약": ["계약관리", "기성관리", "업무지시서", "인원관리"],
|
||||
"업무관리": ["업무일지(2025)", "업무일지(2025년 이전)", "발주처 정기보고", "본사업무보고", "공사감독일지", "양식서류"]
|
||||
},
|
||||
"설계성과품": {
|
||||
"시방서": ["공사시방서", "장비 반입허가 검토서"],
|
||||
"설계도면": ["공통", "토공", "비탈면안전공", "배수공", "교량공", "포장공", "교통안전시설공", "부대공", "용지공 & 기타공"],
|
||||
"수량산출서": ["토공", "비탈면안전공", "배수공", "교량공", "포장공", "교통안전시설공", "부대공", "용지공 & 기타공"],
|
||||
"내역서": ["단가산출서"],
|
||||
"보고서": ["실시설계보고서", "지반조사보고서", "구조계산서", "수리 및 전기계산서", "기타보고서", "기술자문 및 심의"],
|
||||
"측량계산부": ["측량계산부"],
|
||||
"설계단계 수행협의": ["회의·협의"]
|
||||
},
|
||||
"시공성과품": {
|
||||
"설계도면": ["공통", "토공", "비탈면안전공", "배수공", "교량공", "포장공", "교통안전시설공", "부대공", "용지공 & 기타공"]
|
||||
},
|
||||
"시공검측": {
|
||||
"토공": ["검측 (깨기)", "검측 (연약지반)", "검측 (발파)", "검측 (노체)", "검측 (노상)", "검측 (토취장)"],
|
||||
"배수공": ["검측 (V형측구)", "검측 (산마루측구)", "검측 (U형측구)", "검측 (U형측구)(안)", "검측 (L형측구, J형측구)", "검측 (도수로)", "검측 (도수로)(안)", "검측 (횡배수관)", "검측 (종배수관)", "검측 (맹암거)", "검측 (통로암거)", "검측 (수로암거)", "검측 (호안공)", "검측 (옹벽공)", "검측 (용수개거)"],
|
||||
"구조물공": ["검측 (평목교-거더, 부대공)", "검측 (평목교)(안)", "검측 (개착터널, 생태통로)"],
|
||||
"포장공": ["검측 (기층, 보조기층)"],
|
||||
"부대공": ["검측 (환경)", "검측 (지장가옥,건물 철거)", "검측 (방음벽 등)"],
|
||||
"비탈면안전공": ["검측 (식생보호공)", "검측 (구조물보호공)"],
|
||||
"교통안전시설공": ["검측 (낙석방지책)"],
|
||||
"검측 양식서류": ["검측 양식서류"]
|
||||
},
|
||||
"설계변경": {
|
||||
"실정보고(어천~공주)": ["토공", "배수공", "교량공(평목교)", "구조물공", "포장공", "교통안전공", "부대공", "전기공사", "미확정공", "안전관리", "환경관리", "품질관리", "자재관리", "지장물", "기타"],
|
||||
"실정보고(대술~정안)": ["토공", "배수공", "비탈면안전공", "포장공", "부대공", "안전관리", "환경관리", "자재관리", "기타"],
|
||||
"기술지원 검토": ["토공", "배수공", "교량공(평목교)", "구조물&부대공", "기타"],
|
||||
"시공계획(어천~공주)": ["토공", "배수공", "교량공(평목교)", "구조물&부대&포장&교통안전공", "환경 및 품질관리"]
|
||||
},
|
||||
"공사관리": {
|
||||
"공정·일정": ["공정표", "월간 공정보고", "작업일보"],
|
||||
"품질 관리": ["품질시험계획서", "품질시험 실적보고", "콘크리트 타설현황[어천~공주(4차)]", "품질관리비 사용내역", "균열관리", "품질관리 양식서류"],
|
||||
"안전 관리": ["안전관리계획서", "안전관리 실적보고", "위험성 평가", "사전작업허가서", "안전관리비 사용내역", "안전관리수준평가", "안전관리 양식서류"],
|
||||
"환경 관리": ["환경영향평가", "사전재해영향성검토", "유지관리 및 보수점검", "환경보전비 사용내역", "건설폐기물 관리"],
|
||||
"자재 관리 (관급)": ["자재구매요청 (레미콘, 철근)", "자재구매요청 (그 외)", "납품기한", "계약 변경", "자재 반입·수불 관리", "자재관리 양식서류"],
|
||||
"자재 관리 (사급)": ["자재공급원 승인", "자재 반입·수불 관리", "자재 검수·확인"],
|
||||
"점검 (정리중)": ["내부점검", "외부점검"],
|
||||
"공문": ["접수(수신)", "발송(발신)", "하도급", "인력", "방침"]
|
||||
},
|
||||
"민원관리": {
|
||||
"민원(어천~공주)": ["처리대장", "보상", "공사일반", "환경분쟁"],
|
||||
"실정보고(어천~공주)": ["민원"],
|
||||
"실정보고(대술~정안)": ["민원"]
|
||||
}
|
||||
}
|
||||
|
||||
def analyze_flow_reasoning(filename, all_text_list):
|
||||
"""
|
||||
본문의 전수 조사 결과에 파일명의 '의도 가중치'를 더해 최종 추론
|
||||
"""
|
||||
full_text = " ".join(all_text_list)
|
||||
clean_ctx = full_text.replace(" ", "").replace("\n", "").lower()
|
||||
fn_clean = filename.replace(" ", "").lower()
|
||||
|
||||
# 3중 레이어 데이터 통합
|
||||
full_pool = (title_text + " | " + text_content + " | " + ocr_content).lower().replace(" ", "").replace("\n", "")
|
||||
|
||||
# 분석 초기화
|
||||
result = {
|
||||
"suggested_path": "분석실패",
|
||||
"confidence": "Low",
|
||||
"log_steps": log_steps,
|
||||
"raw_text": f"--- TITLE ---\n{filename}\n\n--- TEXT ---\n{text_content[:1000]}\n\n--- OCR ---\n{ocr_content[:1000]}",
|
||||
"reason": "학습된 키워드 일치 항목 없음"
|
||||
# 1. 도메인별 기본 점수 (본문 전수 조사 - 평등하게)
|
||||
scores = {
|
||||
"official": sum(clean_ctx.count(k) for k in ["수신:", "발신:", "경유:", "시행일자", "귀하", "드립니다", "바랍니다"]),
|
||||
"contract": sum(clean_ctx.count(k) for k in ["계약서", "하도급", "외주", "도급", "인감", "사업자"]),
|
||||
"hr": sum(clean_ctx.count(k) for k in ["이탈계", "인력", "기술자", "안전관리자", "재직증명", "배치"]),
|
||||
"change": sum(clean_ctx.count(k) for k in ["실정보고", "설계변경", "변경보고", "추가반영"]),
|
||||
"technical": sum(clean_ctx.count(k) for k in ["일위대가", "산출근거", "집계표", "물량산출", "단가", "내역", "도면", "dwg"])
|
||||
}
|
||||
|
||||
# 최종 추천 로직 (합의 알고리즘)
|
||||
is_eocheon = any(k in full_pool for k in ["어천", "공주"])
|
||||
# 2. 파일명에 대한 '방향타' 가중치 부여 (Final Push)
|
||||
# 본문 데이터가 아무리 많아도 파일명의 의도를 존중하기 위해 7배 가중치
|
||||
if "실정" in fn_clean or "변경" in fn_clean: scores["change"] += 50 # 본문 50회 언급과 맞먹는 가중치
|
||||
if "계약" in fn_clean or "하도급" in fn_clean: scores["contract"] += 50
|
||||
if "인력" in fn_clean or "이탈" in fn_clean: scores["hr"] += 50
|
||||
if "단가" in fn_clean or "수량" in fn_clean or "도면" in fn_clean: scores["technical"] += 50
|
||||
if "제출" in fn_clean or "건" in fn_clean: scores["official"] += 30
|
||||
|
||||
# 3. 종합 농도에 따른 최종 도메인 선정
|
||||
dominant_domain = max(scores, key=scores.get)
|
||||
|
||||
if "실정보고" in full_pool or "실정" in full_pool:
|
||||
if is_eocheon:
|
||||
if "품질" in full_pool:
|
||||
result["suggested_path"] = "설계변경 > 실정보고(어천~공주) > 품질관리"
|
||||
result["reason"] = "3중 레이어 분석: 실정보고+어천공주+품질관리 키워드 통합 검출"
|
||||
elif any(k in full_pool for k in ["토지", "임대"]):
|
||||
result["suggested_path"] = "설계변경 > 실정보고(어천~공주) > 기타"
|
||||
result["reason"] = "3중 레이어 분석: 토지임대 관련 실정보고(어천-공주) 확인"
|
||||
else:
|
||||
result["suggested_path"] = "설계변경 > 실정보고(어천~공주) > 기타"
|
||||
result["reason"] = "3중 레이어 분석: 실정보고(어천-공주) 문서 판정"
|
||||
result["confidence"] = "100%"
|
||||
else:
|
||||
result["suggested_path"] = "설계변경 > 실정보고(어천~공주) > 기타" # 폴백
|
||||
result["confidence"] = "80%"
|
||||
result["reason"] = "실정보고 키워드는 발견되었으나 프로젝트명 교차 검증 실패 (기본값 제안)"
|
||||
# 프로젝트 식별 (Fuzzy 매칭 및 교차 검증)
|
||||
project_loc = "어천~공주" if any(k in clean_ctx or k in fn_clean for k in ["어천", "공주"]) else "대술~정안" if any(k in clean_ctx or k in fn_clean for k in ["대술", "정안"]) else "공통"
|
||||
|
||||
elif "품질" in full_pool:
|
||||
result["suggested_path"] = "공사관리 > 품질 관리 > 품질시험계획서"
|
||||
result["confidence"] = "90%"
|
||||
result["reason"] = "텍스트/OCR 레이어에서 품질 관리 지표 다수 식별"
|
||||
# --- [통합 추론 및 매칭] ---
|
||||
|
||||
return result
|
||||
# 시나리오 A: 실정보고/설계변경 (본문 데이터 + 파일명 의도 합성)
|
||||
if dominant_domain == "change" or (scores["change"] > 0 and scores["technical"] > 5):
|
||||
cat = f"실정보고({project_loc})"
|
||||
sub = "지장물" if any(k in clean_ctx for k in ["임대료", "토지", "보상"]) else "구조물공" if "구조물" in clean_ctx else "기타"
|
||||
return f"설계변경 > {cat} > {sub}", f"본문의 기술 데이터 밀도와 파일명의 '{dominant_domain}' 관련 의도를 종합하여 {project_loc} 프로젝트의 실정보고 본체로 판정."
|
||||
|
||||
# 시나리오 B: 행정 계약/하도급 (본체 중심)
|
||||
if dominant_domain == "contract":
|
||||
return "행정 > 계약 > 계약관리", "문서 전체에서 계약 및 하도급 업무 본질이 지배적으로 확인됨."
|
||||
|
||||
# 시나리오 C: 인사/인력 관리
|
||||
if dominant_domain == "hr":
|
||||
if len(all_text_list) <= 2: return "공사관리 > 공문 > 인력", "인력 사항을 간략히 보고하는 공문 형식임."
|
||||
return "행정 > 계약 > 인원관리", "다량의 인력 증빙 데이터가 포함된 행정 서류임."
|
||||
|
||||
# 시나리오 D: 순수 공문 (형식 우선)
|
||||
if dominant_domain == "official" or scores["official"] > scores["technical"]:
|
||||
tab, cat = "공사관리", "공문"
|
||||
sub = "접수(수신)"
|
||||
if "방침" in clean_ctx or "지침" in clean_ctx: sub = "방침"
|
||||
elif "발신" in clean_ctx[:500]: sub = "발송(발신)"
|
||||
return f"{tab} > {cat} > {sub}", "전체 맥락상 기술적 데이터보다 행정적 전달 행위(공문)가 핵심 정체성으로 판단됨."
|
||||
|
||||
# 시나리오 E: 기술 성과품
|
||||
if dominant_domain == "technical":
|
||||
if any(k in clean_ctx or k in fn_clean for k in ["단가", "내역"]): return "설계성과품 > 내역서 > 단가산출서", "내역/단가 산출 기술 데이터 확인."
|
||||
if any(k in clean_ctx or k in fn_clean for k in ["도면", "dwg"]): return "설계성과품 > 설계도면 > 공통", "도면/그래픽 데이터 확인."
|
||||
return "설계성과품 > 수량산출서 > 토공", "수량/물량 산출 데이터 확인."
|
||||
|
||||
return "행정 > 업무관리 > 양식서류", "일반 행정 및 기타 양식 서류로 분류함."
|
||||
|
||||
def analyze_file_content(filename: str):
|
||||
try:
|
||||
file_path = os.path.join("sample", filename)
|
||||
text_by_pages = []
|
||||
if filename.lower().endswith(".pdf"):
|
||||
reader = PdfReader(file_path)
|
||||
for i in range(len(reader.pages)):
|
||||
page_text = reader.pages[i].extract_text() or ""
|
||||
if OCR_AVAILABLE:
|
||||
try:
|
||||
images = convert_from_path(file_path, first_page=i+1, last_page=i+1, poppler_path=POPPLER_BIN, dpi=200)
|
||||
if images:
|
||||
ocr_result = pytesseract.image_to_string(images[0], lang='kor+eng')
|
||||
page_text += "\n" + ocr_result
|
||||
except: pass
|
||||
text_by_pages.append(page_text)
|
||||
elif filename.lower().endswith(('.xlsx', '.xls')):
|
||||
import pandas as pd
|
||||
df = pd.read_excel(file_path)
|
||||
text_by_pages.append(df.to_string())
|
||||
else: text_by_pages.append("")
|
||||
|
||||
path, reason = analyze_flow_reasoning(filename, text_by_pages)
|
||||
|
||||
return {
|
||||
"filename": filename,
|
||||
"total_pages": len(text_by_pages),
|
||||
"final_result": {
|
||||
"suggested_path": path,
|
||||
"confidence": "100%",
|
||||
"reason": reason,
|
||||
"snippet": " ".join(text_by_pages)[:1500]
|
||||
}
|
||||
}
|
||||
except Exception as e:
|
||||
return {"error": str(e), "filename": filename}
|
||||
|
||||
Reference in New Issue
Block a user