from fastapi import APIRouter, Depends, HTTPException, UploadFile, File, Form from sqlalchemy.orm import Session from sqlalchemy import text from typing import List, Optional import os import shutil from datetime import datetime import uuid import pandas as pd import re from pathlib import Path from ..database import get_db from app.services.material_classifier import classify_material from app.services.bolt_classifier import classify_bolt from app.services.flange_classifier import classify_flange from app.services.fitting_classifier import classify_fitting from app.services.gasket_classifier import classify_gasket from app.services.instrument_classifier import classify_instrument from app.services.pipe_classifier import classify_pipe from app.services.valve_classifier import classify_valve router = APIRouter() UPLOAD_DIR = Path("uploads") UPLOAD_DIR.mkdir(exist_ok=True) ALLOWED_EXTENSIONS = {".xlsx", ".xls", ".csv"} @router.get("/") async def get_files_info(): return { "message": "파일 관리 API", "allowed_extensions": list(ALLOWED_EXTENSIONS), "upload_directory": str(UPLOAD_DIR) } @router.get("/test") async def test_endpoint(): return {"status": "파일 API가 정상 작동합니다!"} @router.post("/add-missing-columns") async def add_missing_columns(db: Session = Depends(get_db)): """누락된 컬럼들 추가""" try: db.execute(text("ALTER TABLE files ADD COLUMN IF NOT EXISTS parsed_count INTEGER DEFAULT 0")) db.execute(text("ALTER TABLE materials ADD COLUMN IF NOT EXISTS row_number INTEGER")) db.commit() return { "success": True, "message": "누락된 컬럼들이 추가되었습니다", "added_columns": ["files.parsed_count", "materials.row_number"] } except Exception as e: db.rollback() return {"success": False, "error": f"컬럼 추가 실패: {str(e)}"} def validate_file_extension(filename: str) -> bool: return Path(filename).suffix.lower() in ALLOWED_EXTENSIONS def generate_unique_filename(original_filename: str) -> str: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") unique_id = str(uuid.uuid4())[:8] stem = Path(original_filename).stem suffix = Path(original_filename).suffix return f"{stem}_{timestamp}_{unique_id}{suffix}" def parse_dataframe(df): df = df.dropna(how='all') df.columns = df.columns.str.strip().str.lower() column_mapping = { 'description': ['description', 'item', 'material', '품명', '자재명'], 'quantity': ['qty', 'quantity', 'ea', '수량'], 'main_size': ['main_nom', 'nominal_diameter', 'nd', '주배관'], 'red_size': ['red_nom', 'reduced_diameter', '축소배관'], 'length': ['length', 'len', '길이'], 'weight': ['weight', 'wt', '중량'], 'dwg_name': ['dwg_name', 'drawing', '도면명'], 'line_num': ['line_num', 'line_number', '라인번호'] } mapped_columns = {} for standard_col, possible_names in column_mapping.items(): for possible_name in possible_names: if possible_name in df.columns: mapped_columns[standard_col] = possible_name break materials = [] for index, row in df.iterrows(): description = str(row.get(mapped_columns.get('description', ''), '')) quantity_raw = row.get(mapped_columns.get('quantity', ''), 0) try: quantity = float(quantity_raw) if pd.notna(quantity_raw) else 0 except: quantity = 0 material_grade = "" if "ASTM" in description.upper(): astm_match = re.search(r'ASTM\s+([A-Z0-9\s]+)', description.upper()) if astm_match: material_grade = astm_match.group(0).strip() main_size = str(row.get(mapped_columns.get('main_size', ''), '')) red_size = str(row.get(mapped_columns.get('red_size', ''), '')) if main_size != 'nan' and red_size != 'nan' and red_size != '': size_spec = f"{main_size} x {red_size}" elif main_size != 'nan' and main_size != '': size_spec = main_size else: size_spec = "" if description and description not in ['nan', 'None', '']: materials.append({ 'original_description': description, 'quantity': quantity, 'unit': "EA", 'size_spec': size_spec, 'material_grade': material_grade, 'line_number': index + 1, 'row_number': index + 1 }) return materials def parse_file_data(file_path): file_extension = Path(file_path).suffix.lower() try: if file_extension == ".csv": df = pd.read_csv(file_path, encoding='utf-8') elif file_extension in [".xlsx", ".xls"]: df = pd.read_excel(file_path, sheet_name=0) else: raise HTTPException(status_code=400, detail="지원하지 않는 파일 형식") return parse_dataframe(df) except Exception as e: raise HTTPException(status_code=400, detail=f"파일 파싱 실패: {str(e)}") @router.post("/upload") async def upload_file( file: UploadFile = File(...), job_no: str = Form(...), revision: str = Form("Rev.0"), db: Session = Depends(get_db) ): if not validate_file_extension(file.filename): raise HTTPException( status_code=400, detail=f"지원하지 않는 파일 형식입니다. 허용된 확장자: {', '.join(ALLOWED_EXTENSIONS)}" ) if file.size and file.size > 10 * 1024 * 1024: raise HTTPException(status_code=400, detail="파일 크기는 10MB를 초과할 수 없습니다") unique_filename = generate_unique_filename(file.filename) file_path = UPLOAD_DIR / unique_filename try: print("파일 저장 시작") with open(file_path, "wb") as buffer: shutil.copyfileobj(file.file, buffer) print(f"파일 저장 완료: {file_path}") except Exception as e: raise HTTPException(status_code=500, detail=f"파일 저장 실패: {str(e)}") try: print("파일 저장 시작") materials_data = parse_file_data(str(file_path)) parsed_count = len(materials_data) print(f"파싱 완료: {parsed_count}개 자재") # 파일 정보 저장 (project_id 제거) print("DB 저장 시작") file_insert_query = text(""" INSERT INTO files (filename, original_filename, file_path, job_no, revision, description, file_size, parsed_count, is_active) VALUES (:filename, :original_filename, :file_path, :job_no, :revision, :description, :file_size, :parsed_count, :is_active) RETURNING id """) file_result = db.execute(file_insert_query, { "filename": unique_filename, "original_filename": file.filename, "file_path": str(file_path), "job_no": job_no, "revision": revision, "description": f"BOM 파일 - {parsed_count}개 자재", "file_size": file.size, "parsed_count": parsed_count, "is_active": True }) file_id = file_result.fetchone()[0] print(f"파일 저장 완료: file_id = {file_id}") # 자재 데이터 저장 (분류 포함) print("자재 분류 및 저장 시작") materials_inserted = 0 classification_stats = { 'BOLT': 0, 'FLANGE': 0, 'FITTING': 0, 'GASKET': 0, 'INSTRUMENT': 0, 'PIPE': 0, 'VALVE': 0, 'MATERIAL': 0, 'OTHER': 0 } for material_data in materials_data: # 자재 분류 실행 classification_result = classify_material_item( material_data["original_description"], material_data["size_spec"] ) # 분류 통계 업데이트 category = classification_result.get('category', 'OTHER') if category in classification_stats: classification_stats[category] += 1 material_insert_query = text(""" INSERT INTO materials ( file_id, original_description, quantity, unit, size_spec, material_grade, line_number, row_number, classified_category, classification_confidence, is_verified, created_at, subcategory, standard, grade ) VALUES ( :file_id, :original_description, :quantity, :unit, :size_spec, :material_grade, :line_number, :row_number, :classified_category, :classification_confidence, :is_verified, :created_at, :subcategory, :standard, :grade ) """) db.execute(material_insert_query, { "file_id": file_id, "original_description": material_data["original_description"], "quantity": material_data["quantity"], "unit": material_data["unit"], "size_spec": material_data["size_spec"], "material_grade": material_data["material_grade"], "line_number": material_data["line_number"], "row_number": material_data["row_number"], "classified_category": classification_result.get('category', 'OTHER'), "classification_confidence": classification_result.get('confidence', 0.0), "is_verified": False, "created_at": datetime.now(), "subcategory": classification_result.get('subcategory', ''), "standard": classification_result.get('standard', ''), "grade": classification_result.get('grade', '') }) materials_inserted += 1 print(f"자재 저장 완료: {materials_inserted}개") print("커밋 직전") db.commit() print("커밋 완료") return { "success": True, "message": f"완전한 DB 저장 성공! {materials_inserted}개 자재 저장됨", "original_filename": file.filename, "file_id": file_id, "parsed_materials_count": parsed_count, "saved_materials_count": materials_inserted, "classification_stats": classification_stats, "sample_materials": materials_data[:3] if materials_data else [] } except Exception as e: db.rollback() if os.path.exists(file_path): os.remove(file_path) import traceback print(traceback.format_exc()) # 에러 전체 로그 출력 raise HTTPException(status_code=500, detail=f"파일 처리 실패: {str(e)}") @router.get("/materials") async def get_materials( project_id: Optional[int] = None, job_id: Optional[int] = None, revision: Optional[str] = None, grouping: Optional[str] = None, search: Optional[str] = None, search_value: Optional[str] = None, item_type: Optional[str] = None, material_grade: Optional[str] = None, size_spec: Optional[str] = None, file_filter: Optional[str] = None, sort_by: Optional[str] = None, skip: int = 0, limit: int = 100, db: Session = Depends(get_db) ): """자재 목록 조회 (개선된 버전)""" try: # 기본 쿼리 구성 base_query = """ SELECT m.id, m.original_description, m.quantity, m.unit, m.size_spec, m.material_grade, m.line_number, m.row_number, m.classified_category, m.classification_confidence, m.is_verified, m.created_at, f.job_no as job_number, f.revision, f.original_filename, f.project_id, p.project_name, COUNT(*) OVER() as total_count FROM materials m LEFT JOIN files f ON m.file_id = f.id LEFT JOIN projects p ON f.project_id = p.id WHERE 1=1 """ params = {} conditions = [] # 프로젝트 필터 if project_id: conditions.append("f.project_id = :project_id") params["project_id"] = project_id # Job ID 필터 if job_id: conditions.append("f.job_no = (SELECT job_no FROM jobs WHERE id = :job_id)") params["job_id"] = job_id # 리비전 필터 if revision: conditions.append("f.revision = :revision") params["revision"] = revision # 검색 필터 (개선된 버전) if search and search_value: try: if search == "project": conditions.append("p.project_name ILIKE :search_value") elif search == "job": conditions.append("f.job_no ILIKE :search_value") elif search == "material": conditions.append("m.original_description ILIKE :search_value") elif search == "description": conditions.append("m.original_description ILIKE :search_value") elif search == "grade": conditions.append("m.material_grade ILIKE :search_value") elif search == "size": conditions.append("m.size_spec ILIKE :search_value") elif search == "filename": conditions.append("f.original_filename ILIKE :search_value") else: # 기본 검색 (기존 방식) conditions.append("(m.original_description ILIKE :search_value OR m.material_grade ILIKE :search_value)") params["search_value"] = f"%{search_value}%" except Exception as e: print(f"검색 필터 처리 오류: {e}") # 오류 발생 시 기본 검색으로 fallback conditions.append("(m.original_description ILIKE :search_value OR m.material_grade ILIKE :search_value)") params["search_value"] = f"%{search_value}%" # 품목 타입 필터 if item_type: conditions.append("m.classified_category = :item_type") params["item_type"] = item_type # 재질 필터 if material_grade: conditions.append("m.material_grade ILIKE :material_grade") params["material_grade"] = f"%{material_grade}%" # 사이즈 필터 if size_spec: conditions.append("m.size_spec ILIKE :size_spec") params["size_spec"] = f"%{size_spec}%" # 파일명 필터 if file_filter: conditions.append("f.original_filename ILIKE :file_filter") params["file_filter"] = f"%{file_filter}%" # 조건 추가 if conditions: base_query += " AND " + " AND ".join(conditions) # 그룹핑 처리 if grouping: if grouping == "item": base_query += " GROUP BY m.classified_category, m.original_description, m.size_spec, m.material_grade, m.id, m.quantity, m.unit, m.line_number, m.row_number, m.classification_confidence, m.is_verified, m.created_at, f.job_no, f.revision, f.original_filename, f.project_id, p.project_name" elif grouping == "material": base_query += " GROUP BY m.material_grade, m.original_description, m.size_spec, m.id, m.quantity, m.unit, m.line_number, m.row_number, m.classified_category, m.classification_confidence, m.is_verified, m.created_at, f.job_no, f.revision, f.original_filename, f.project_id, p.project_name" elif grouping == "size": base_query += " GROUP BY m.size_spec, m.original_description, m.material_grade, m.id, m.quantity, m.unit, m.line_number, m.row_number, m.classified_category, m.classification_confidence, m.is_verified, m.created_at, f.job_no, f.revision, f.original_filename, f.project_id, p.project_name" elif grouping == "job": base_query += " GROUP BY f.job_no, m.original_description, m.size_spec, m.material_grade, m.id, m.quantity, m.unit, m.line_number, m.row_number, m.classified_category, m.classification_confidence, m.is_verified, m.created_at, f.revision, f.original_filename, f.project_id, p.project_name" elif grouping == "revision": base_query += " GROUP BY f.revision, m.original_description, m.size_spec, m.material_grade, m.id, m.quantity, m.unit, m.line_number, m.row_number, m.classified_category, m.classification_confidence, m.is_verified, m.created_at, f.job_no, f.original_filename, f.project_id, p.project_name" # 정렬 if sort_by: if sort_by == "quantity_desc": base_query += " ORDER BY SUM(m.quantity) DESC" elif sort_by == "quantity_asc": base_query += " ORDER BY SUM(m.quantity) ASC" elif sort_by == "name_asc": base_query += " ORDER BY m.original_description ASC" elif sort_by == "name_desc": base_query += " ORDER BY m.original_description DESC" elif sort_by == "created_desc": base_query += " ORDER BY m.created_at DESC" elif sort_by == "created_asc": base_query += " ORDER BY m.created_at ASC" else: base_query += " ORDER BY m.id DESC" else: base_query += " ORDER BY m.id DESC" # 페이징 base_query += " LIMIT :limit OFFSET :skip" params["limit"] = limit params["skip"] = skip result = db.execute(text(base_query), params) materials = result.fetchall() # 리비전 비교 데이터 생성 revision_comparison = None if revision and revision != "Rev.0": comparison_query = """ SELECT m.original_description, m.size_spec, m.material_grade, SUM(CASE WHEN f.revision = :current_revision THEN m.quantity ELSE 0 END) as current_qty, SUM(CASE WHEN f.revision = :prev_revision THEN m.quantity ELSE 0 END) as prev_qty FROM materials m LEFT JOIN files f ON m.file_id = f.id WHERE f.project_id = :project_id AND f.revision IN (:current_revision, :prev_revision) GROUP BY m.original_description, m.size_spec, m.material_grade HAVING SUM(CASE WHEN f.revision = :current_revision THEN m.quantity ELSE 0 END) != SUM(CASE WHEN f.revision = :prev_revision THEN m.quantity ELSE 0 END) """ comparison_params = { "project_id": project_id, "current_revision": revision, "prev_revision": f"Rev.{int(revision.split('.')[-1]) - 1}" } comparison_result = db.execute(text(comparison_query), comparison_params) comparison_data = comparison_result.fetchall() if comparison_data: changes = [] for row in comparison_data: change = row.current_qty - row.prev_qty if change != 0: changes.append({ "description": row.original_description, "size_spec": row.size_spec, "material_grade": row.material_grade, "current_qty": row.current_qty, "prev_qty": row.prev_qty, "change": change }) revision_comparison = { "summary": f"{revision}에서 {len(changes)}개 항목이 변경되었습니다", "changes": changes } # 결과 포맷팅 formatted_materials = [] for material in materials: # 라인 번호 문자열 생성 line_numbers = [material.line_number] if material.line_number else [] line_numbers_str = ", ".join(map(str, line_numbers)) if line_numbers else "" # 수량 변경 계산 (리비전 비교) quantity_change = None if revision_comparison: for change in revision_comparison["changes"]: if (change["description"] == material.original_description and change["size_spec"] == material.size_spec and change["material_grade"] == material.material_grade): quantity_change = change["change"] break formatted_material = { "id": material.id, "original_description": material.original_description, "quantity": float(material.quantity) if material.quantity else 0, "unit": material.unit or "EA", "size_spec": material.size_spec or "", "material_grade": material.material_grade or "", "line_number": material.line_number, "line_numbers_str": line_numbers_str, "line_count": len(line_numbers), "classified_category": material.classified_category or "OTHER", "classification_confidence": float(material.classification_confidence) if material.classification_confidence else 0, "is_verified": material.is_verified or False, "created_at": material.created_at.isoformat() if material.created_at else None, "job_number": material.job_number, "revision": material.revision or "Rev.0", "original_filename": material.original_filename, "project_id": material.project_id, "project_name": material.project_name, "quantity_change": quantity_change } formatted_materials.append(formatted_material) total_count = materials[0].total_count if materials else 0 return { "materials": formatted_materials, "total_count": total_count, "revision_comparison": revision_comparison } except Exception as e: raise HTTPException(status_code=500, detail=f"자재 조회 실패: {str(e)}") @router.get("/materials/summary") async def get_materials_summary( job_no: Optional[str] = None, file_id: Optional[str] = None, db: Session = Depends(get_db) ): """자재 요약 통계""" try: query = """ SELECT COUNT(*) as total_items, COUNT(DISTINCT m.original_description) as unique_descriptions, COUNT(DISTINCT m.size_spec) as unique_sizes, COUNT(DISTINCT m.material_grade) as unique_materials, SUM(m.quantity) as total_quantity, AVG(m.quantity) as avg_quantity, MIN(m.created_at) as earliest_upload, MAX(m.created_at) as latest_upload FROM materials m LEFT JOIN files f ON m.file_id = f.id WHERE 1=1 """ params = {} if job_no: query += " AND f.job_no = :job_no" params["job_no"] = job_no if file_id: query += " AND m.file_id = :file_id" params["file_id"] = file_id result = db.execute(text(query), params) summary = result.fetchone() return { "success": True, "summary": { "total_items": summary.total_items, "unique_descriptions": summary.unique_descriptions, "unique_sizes": summary.unique_sizes, "unique_materials": summary.unique_materials, "total_quantity": float(summary.total_quantity) if summary.total_quantity else 0, "avg_quantity": round(float(summary.avg_quantity), 2) if summary.avg_quantity else 0, "earliest_upload": summary.earliest_upload, "latest_upload": summary.latest_upload } } except Exception as e: raise HTTPException(status_code=500, detail=f"요약 조회 실패: {str(e)}") # Job 검증 함수 (파일 끝에 추가할 예정) async def validate_job_exists(job_no: str, db: Session): """Job 존재 여부 및 활성 상태 확인""" try: query = text("SELECT job_no, job_name, status FROM jobs WHERE job_no = :job_no AND is_active = true") job = db.execute(query, {"job_no": job_no}).fetchone() if not job: return {"valid": False, "error": f"Job No. '{job_no}'를 찾을 수 없습니다"} if job.status == '완료': return {"valid": False, "error": f"완료된 Job '{job.job_name}'에는 파일을 업로드할 수 없습니다"} return { "valid": True, "job": { "job_no": job.job_no, "job_name": job.job_name, "status": job.status } } except Exception as e: return {"valid": False, "error": f"Job 검증 실패: {str(e)}"} def get_major_category(description): """간단한 키워드 기반 대분류""" desc_upper = description.upper() if 'PIPE' in desc_upper or 'TUBE' in desc_upper: return 'pipe' elif any(word in desc_upper for word in ['ELBOW', 'ELL', 'TEE', 'REDUCER', 'CAP', 'COUPLING']): return 'fitting' elif 'VALVE' in desc_upper: return 'valve' elif 'FLANGE' in desc_upper or 'FLG' in desc_upper: return 'flange' elif any(word in desc_upper for word in ['GAUGE', 'SENSOR', 'INSTRUMENT', 'TRANSMITTER']): return 'instrument' elif 'GASKET' in desc_upper or 'GASK' in desc_upper: return 'gasket' elif any(word in desc_upper for word in ['BOLT', 'STUD', 'NUT', 'SCREW']): return 'bolt' else: return 'other' def classify_material_item(description: str, size_spec: str = "") -> dict: """ 자재를 각 분류기로 보내서 분류하는 통합 함수 Args: description: 자재 설명 size_spec: 사이즈 정보 Returns: 분류 결과 딕셔너리 """ desc_upper = description.upper().strip() # 1. 볼트 분류 bolt_result = classify_bolt(description) if bolt_result.get('confidence', 0) > 0.7: return { 'category': 'BOLT', 'subcategory': bolt_result.get('bolt_type', 'UNKNOWN'), 'standard': bolt_result.get('standard', ''), 'grade': bolt_result.get('grade', ''), 'confidence': bolt_result.get('confidence', 0), 'details': bolt_result } # 2. 플랜지 분류 flange_result = classify_flange(description) if flange_result.get('confidence', 0) > 0.7: return { 'category': 'FLANGE', 'subcategory': flange_result.get('flange_type', 'UNKNOWN'), 'standard': flange_result.get('standard', ''), 'grade': flange_result.get('grade', ''), 'confidence': flange_result.get('confidence', 0), 'details': flange_result } # 3. 피팅 분류 fitting_result = classify_fitting(description) if fitting_result.get('confidence', 0) > 0.7: return { 'category': 'FITTING', 'subcategory': fitting_result.get('fitting_type', 'UNKNOWN'), 'standard': fitting_result.get('standard', ''), 'grade': fitting_result.get('grade', ''), 'confidence': fitting_result.get('confidence', 0), 'details': fitting_result } # 4. 가스켓 분류 gasket_result = classify_gasket(description) if gasket_result.get('confidence', 0) > 0.7: return { 'category': 'GASKET', 'subcategory': gasket_result.get('gasket_type', 'UNKNOWN'), 'standard': gasket_result.get('standard', ''), 'grade': gasket_result.get('grade', ''), 'confidence': gasket_result.get('confidence', 0), 'details': gasket_result } # 5. 계기 분류 instrument_result = classify_instrument(description) if instrument_result.get('confidence', 0) > 0.7: return { 'category': 'INSTRUMENT', 'subcategory': instrument_result.get('instrument_type', 'UNKNOWN'), 'standard': instrument_result.get('standard', ''), 'grade': instrument_result.get('grade', ''), 'confidence': instrument_result.get('confidence', 0), 'details': instrument_result } # 6. 파이프 분류 pipe_result = classify_pipe(description) if pipe_result.get('confidence', 0) > 0.7: return { 'category': 'PIPE', 'subcategory': pipe_result.get('pipe_type', 'UNKNOWN'), 'standard': pipe_result.get('standard', ''), 'grade': pipe_result.get('grade', ''), 'confidence': pipe_result.get('confidence', 0), 'details': pipe_result } # 7. 밸브 분류 valve_result = classify_valve(description) if valve_result.get('confidence', 0) > 0.7: return { 'category': 'VALVE', 'subcategory': valve_result.get('valve_type', 'UNKNOWN'), 'standard': valve_result.get('standard', ''), 'grade': valve_result.get('grade', ''), 'confidence': valve_result.get('confidence', 0), 'details': valve_result } # 8. 재질 분류 (기본) material_result = classify_material(description) if material_result.get('confidence', 0) > 0.5: return { 'category': 'MATERIAL', 'subcategory': material_result.get('material_type', 'UNKNOWN'), 'standard': material_result.get('standard', ''), 'grade': material_result.get('grade', ''), 'confidence': material_result.get('confidence', 0), 'details': material_result } # 9. 기본 분류 (키워드 기반) category = get_major_category(description) return { 'category': category.upper(), 'subcategory': 'UNKNOWN', 'standard': '', 'grade': '', 'confidence': 0.3, 'details': {} }