import os import re import unicodedata from pypdf import PdfReader import pytesseract from pdf2image import convert_from_path # 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' 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() # 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"]) } # 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) # 프로젝트 식별 (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 "공통" # --- [통합 추론 및 매칭] --- # 시나리오 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 Exception as ocr_err: print(f"OCR Error on page {i+1}: {ocr_err}") 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}