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 import json from ..database import get_db from app.services.material_classifier import classify_material from app.services.integrated_classifier import classify_material_integrated, should_exclude_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') # 원본 컬럼명 출력 print(f"원본 컬럼들: {list(df.columns)}") df.columns = df.columns.str.strip().str.lower() print(f"소문자 변환 후: {list(df.columns)}") 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 print(f"찾은 컬럼 매핑: {mapped_columns}") 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 표준과 등급만 추출, end_preparation(BOE, POE, BBE 등)은 제외 astm_match = re.search(r'ASTM\s+([A-Z0-9]+(?:\s+GR\s+[A-Z0-9]+)?)', 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', ''), '')) # main_nom과 red_nom 별도 저장 (원본 값 유지) main_nom = main_size if main_size != 'nan' and main_size != '' else None red_nom = red_size if red_size != 'nan' and red_size != '' else None # 기존 size_spec도 유지 (호환성을 위해) 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 = "" # LENGTH 정보 추출 length_raw = row.get(mapped_columns.get('length', ''), '') length_value = None if pd.notna(length_raw) and str(length_raw).strip() != '': try: length_value = float(str(length_raw).strip()) except (ValueError, TypeError): length_value = None if description and description not in ['nan', 'None', '']: materials.append({ 'original_description': description, 'quantity': quantity, 'unit': "EA", 'size_spec': size_spec, 'main_nom': main_nom, # 추가 'red_nom': red_nom, # 추가 'material_grade': material_grade, 'length': length_value, '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"), # 기본값은 Rev.0 (새 BOM) parent_file_id: Optional[int] = Form(None), # 리비전 업로드 시 부모 파일 ID bom_name: Optional[str] = Form(None), # BOM 이름 (사용자 입력) db: Session = Depends(get_db) ): print(f"📥 업로드 요청 받음:") print(f" - 파일명: {file.filename}") print(f" - job_no: {job_no}") print(f" - revision: {revision}") print(f" - parent_file_id: {parent_file_id}") print(f" - bom_name: {bom_name}") print(f" - parent_file_id 타입: {type(parent_file_id)}") 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}개 자재") # 리비전 업로드인 경우만 자동 리비전 생성 if parent_file_id is not None: print(f"리비전 업로드 모드: parent_file_id = {parent_file_id}") # 부모 파일의 정보 조회 parent_query = text(""" SELECT original_filename, revision, bom_name FROM files WHERE id = :parent_file_id AND job_no = :job_no """) parent_result = db.execute(parent_query, { "parent_file_id": parent_file_id, "job_no": job_no }) parent_file = parent_result.fetchone() if not parent_file: raise HTTPException(status_code=404, detail="부모 파일을 찾을 수 없습니다.") # 해당 BOM의 최신 리비전 확인 (bom_name 기준) bom_name_to_use = parent_file[2] or parent_file[0] # bom_name 우선, 없으면 original_filename latest_revision_query = text(""" SELECT revision FROM files WHERE job_no = :job_no AND (bom_name = :bom_name OR (bom_name IS NULL AND original_filename = :bom_name)) ORDER BY revision DESC LIMIT 1 """) latest_result = db.execute(latest_revision_query, { "job_no": job_no, "bom_name": bom_name_to_use }) latest_revision = latest_result.fetchone() if latest_revision: latest_rev = latest_revision[0] if latest_rev.startswith("Rev."): try: rev_num = int(latest_rev.replace("Rev.", "")) revision = f"Rev.{rev_num + 1}" except ValueError: revision = "Rev.1" else: revision = "Rev.1" print(f"리비전 업로드: {latest_rev} → {revision}") else: revision = "Rev.1" print(f"첫 번째 리비전: {revision}") # 파일명을 부모와 동일하게 유지 file.filename = parent_file[0] else: # 일반 업로드 (새 BOM) print(f"일반 업로드 모드: 새 BOM 파일 (Rev.0)") # 파일 정보 저장 print("DB 저장 시작") file_insert_query = text(""" INSERT INTO files (filename, original_filename, file_path, job_no, revision, bom_name, description, file_size, parsed_count, is_active) VALUES (:filename, :original_filename, :file_path, :job_no, :revision, :bom_name, :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, "bom_name": bom_name or file.filename, # bom_name 우선, 없으면 파일명 "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}") # 자재 데이터 저장 (분류 포함) - 배치 처리로 성능 개선 materials_to_insert = [] pipe_details_to_insert = [] fitting_details_to_insert = [] bolt_details_to_insert = [] gasket_details_to_insert = [] flange_details_to_insert = [] materials_inserted = 0 for material_data in materials_data: # 자재 타입 분류기 적용 (PIPE, FITTING, VALVE 등) description = material_data["original_description"] size_spec = material_data["size_spec"] # 각 분류기로 시도 (올바른 매개변수 사용) print(f"분류 시도: {description}") # LENGTH 정보 추출 length_value = None if "length" in material_data: try: length_value = float(material_data["length"]) except (ValueError, TypeError): length_value = None # main_nom과 red_nom 추출 main_nom = material_data.get("main_nom") red_nom = material_data.get("red_nom") # 1. 통합 분류기로 자재 타입 결정 integrated_result = classify_material_integrated(description, main_nom or "", red_nom or "", length_value) print(f"[분류] {description}") print(f"통합 분류 결과: {integrated_result.get('category', 'UNKNOWN')} (신뢰도: {integrated_result.get('confidence', 0)}, Level: {integrated_result.get('classification_level', 'NONE')})") # 2. 제외 대상 확인 if should_exclude_material(description): classification_result = { "category": "EXCLUDE", "overall_confidence": 0.95, "reason": "제외 대상 자재" } else: # 3. 타입별 상세 분류기 실행 material_type = integrated_result.get('category', 'UNKNOWN') if material_type == "PIPE": classification_result = classify_pipe("", description, main_nom or "", length_value) elif material_type == "FITTING": classification_result = classify_fitting("", description, main_nom or "", red_nom) elif material_type == "FLANGE": classification_result = classify_flange("", description, main_nom or "", red_nom) elif material_type == "VALVE": classification_result = classify_valve("", description, main_nom or "") elif material_type == "BOLT": classification_result = classify_bolt("", description, main_nom or "") elif material_type == "GASKET": classification_result = classify_gasket("", description, main_nom or "") elif material_type == "INSTRUMENT": classification_result = classify_instrument("", description, main_nom or "") else: # UNKNOWN 처리 classification_result = { "category": "UNKNOWN", "overall_confidence": integrated_result.get('confidence', 0.0), "reason": f"분류 불가: {integrated_result.get('evidence', [])}" } # 통합 분류기의 신뢰도가 더 낮으면 조정 if integrated_result.get('confidence', 0) < 0.5: classification_result['overall_confidence'] = min( classification_result.get('overall_confidence', 1.0), integrated_result.get('confidence', 0.0) + 0.2 ) print(f"최종 분류 결과: {classification_result.get('category', 'UNKNOWN')}") # 기본 자재 정보 저장 material_insert_query = text(""" INSERT INTO materials ( file_id, original_description, quantity, unit, size_spec, main_nom, red_nom, material_grade, line_number, row_number, classified_category, classification_confidence, is_verified, created_at ) VALUES ( :file_id, :original_description, :quantity, :unit, :size_spec, :main_nom, :red_nom, :material_grade, :line_number, :row_number, :classified_category, :classification_confidence, :is_verified, :created_at ) RETURNING id """) # 첫 번째 자재에 대해서만 디버그 출력 if materials_inserted == 0: print(f"첫 번째 자재 저장:") print(f" size_spec: '{material_data['size_spec']}'") print(f" original_description: {material_data['original_description']}") print(f" category: {classification_result.get('category', 'UNKNOWN')}") material_result = 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"], "main_nom": material_data.get("main_nom"), # 추가 "red_nom": material_data.get("red_nom"), # 추가 "material_grade": material_data["material_grade"], "line_number": material_data["line_number"], "row_number": material_data["row_number"], "classified_category": classification_result.get("category", "UNKNOWN"), "classification_confidence": classification_result.get("overall_confidence", 0.0), "is_verified": False, "created_at": datetime.now() }) material_id = material_result.fetchone()[0] materials_inserted += 1 # PIPE 분류 결과인 경우 상세 정보 저장 if classification_result.get("category") == "PIPE": print("PIPE 상세 정보 저장 시작") # 길이 정보 추출 - material_data에서 직접 가져옴 length_mm = material_data.get("length", 0.0) if material_data.get("length") else None # material_id도 함께 저장하도록 수정 pipe_detail_insert_query = text(""" INSERT INTO pipe_details ( material_id, file_id, outer_diameter, schedule, material_spec, manufacturing_method, end_preparation, length_mm ) VALUES ( :material_id, :file_id, :outer_diameter, :schedule, :material_spec, :manufacturing_method, :end_preparation, :length_mm ) """) # 재질 정보 material_info = classification_result.get("material", {}) manufacturing_info = classification_result.get("manufacturing", {}) end_prep_info = classification_result.get("end_preparation", {}) schedule_info = classification_result.get("schedule", {}) size_info = classification_result.get("size_info", {}) # main_nom을 outer_diameter로 활용 outer_diameter = material_data.get("main_nom") or material_data.get("size_spec", "") # end_preparation 정보 추출 (분류 결과에서) end_prep = "" if isinstance(end_prep_info, dict): end_prep = end_prep_info.get("type", "") else: end_prep = str(end_prep_info) if end_prep_info else "" # 재질 정보 - 분류 결과에서 상세 정보 추출 material_grade_from_classifier = "" if isinstance(material_info, dict): material_grade_from_classifier = material_info.get("grade", "") # 분류기에서 더 상세한 재질 정보가 나왔으면 업데이트 if material_grade_from_classifier and material_grade_from_classifier != "UNKNOWN": material_spec = material_grade_from_classifier # materials 테이블의 material_grade도 업데이트 db.execute(text(""" UPDATE materials SET material_grade = :new_material_grade WHERE id = :material_id """), { "new_material_grade": material_grade_from_classifier, "material_id": material_id }) print(f"PIPE material_grade 업데이트: {material_grade_from_classifier}") else: # 기존 파싱 결과 사용 material_spec = material_data.get("material_grade", "") # 제조방법 추출 manufacturing_method = "" if isinstance(manufacturing_info, dict): manufacturing_method = manufacturing_info.get("method", "UNKNOWN") else: manufacturing_method = str(manufacturing_info) if manufacturing_info else "UNKNOWN" # 스케줄 정보 추출 schedule = "" if isinstance(schedule_info, dict): schedule = schedule_info.get("schedule", "UNKNOWN") else: schedule = str(schedule_info) if schedule_info else "UNKNOWN" db.execute(pipe_detail_insert_query, { "material_id": material_id, "file_id": file_id, "outer_diameter": outer_diameter, "schedule": schedule, "material_spec": material_spec, "manufacturing_method": manufacturing_method, "end_preparation": end_prep, "length_mm": material_data.get("length", 0.0) if material_data.get("length") else 0.0 }) print("PIPE 상세 정보 저장 완료") # FITTING 분류 결과인 경우 상세 정보 저장 elif classification_result.get("category") == "FITTING": print("FITTING 상세 정보 저장 시작") # 피팅 정보 추출 fitting_type_info = classification_result.get("fitting_type", {}) connection_info = classification_result.get("connection_method", {}) pressure_info = classification_result.get("pressure_rating", {}) material_info = classification_result.get("material", {}) # 피팅 타입 및 서브타입 fitting_type = fitting_type_info.get("type", "UNKNOWN") fitting_subtype = fitting_type_info.get("subtype", "UNKNOWN") # 연결 방식 connection_method = connection_info.get("method", "UNKNOWN") # 압력 등급 pressure_rating = pressure_info.get("rating", "UNKNOWN") # 재질 정보 material_standard = material_info.get("standard", "") material_grade = material_info.get("grade", "") # 분류기에서 더 상세한 재질 정보가 나왔으면 업데이트 if material_grade and material_grade != "UNKNOWN": db.execute(text(""" UPDATE materials SET material_grade = :new_material_grade WHERE id = :material_id """), { "new_material_grade": material_grade, "material_id": material_id }) print(f"FITTING material_grade 업데이트: {material_grade}") # main_size와 reduced_size main_size = material_data.get("main_nom") or material_data.get("size_spec", "") reduced_size = material_data.get("red_nom", "") # NIPPLE인 경우 길이와 스케줄 정보 추가 length_mm = None schedule = "UNKNOWN" if fitting_type == "NIPPLE": # 길이 정보 추출 length_mm = material_data.get("length", 0.0) if material_data.get("length") else None # 스케줄 정보 추출 (분류 결과에서) schedule_info = classification_result.get("schedule_info", {}) schedule = schedule_info.get("schedule", "UNKNOWN") schedule_info = classification_result.get("schedule", {}) if isinstance(schedule_info, dict): schedule = schedule_info.get("schedule", "UNKNOWN") else: schedule = str(schedule_info) if schedule_info else "UNKNOWN" db.execute(text(""" INSERT INTO fitting_details ( material_id, file_id, fitting_type, fitting_subtype, connection_method, pressure_rating, material_standard, material_grade, main_size, reduced_size, length_mm, schedule ) VALUES ( :material_id, :file_id, :fitting_type, :fitting_subtype, :connection_method, :pressure_rating, :material_standard, :material_grade, :main_size, :reduced_size, :length_mm, :schedule ) """), { "material_id": material_id, "file_id": file_id, "fitting_type": fitting_type, "fitting_subtype": fitting_subtype, "connection_method": connection_method, "pressure_rating": pressure_rating, "material_standard": material_standard, "material_grade": material_grade, "main_size": main_size, "reduced_size": reduced_size, "length_mm": length_mm, "schedule": schedule }) print(f"FITTING 상세 정보 저장 완료: {fitting_type} - {fitting_subtype}") # FLANGE 분류 결과인 경우 상세 정보 저장 if classification_result.get("category") == "FLANGE": print("FLANGE 상세 정보 저장 시작") # 플랜지 타입 정보 flange_type_info = classification_result.get("flange_type", {}) pressure_info = classification_result.get("pressure_rating", {}) face_finish_info = classification_result.get("face_finish", {}) material_info = classification_result.get("material", {}) # 플랜지 타입 (WN, BL, SO 등) flange_type = "" if isinstance(flange_type_info, dict): flange_type = flange_type_info.get("type", "UNKNOWN") else: flange_type = str(flange_type_info) if flange_type_info else "UNKNOWN" # 압력 등급 (150LB, 300LB 등) pressure_rating = "" if isinstance(pressure_info, dict): pressure_rating = pressure_info.get("rating", "UNKNOWN") else: pressure_rating = str(pressure_info) if pressure_info else "UNKNOWN" # 면 가공 (RF, FF, RTJ 등) facing_type = "" if isinstance(face_finish_info, dict): facing_type = face_finish_info.get("finish", "UNKNOWN") else: facing_type = str(face_finish_info) if face_finish_info else "UNKNOWN" # 재질 정보 material_standard = "" material_grade = "" if isinstance(material_info, dict): material_standard = material_info.get("standard", "") material_grade = material_info.get("grade", "") # 분류기에서 더 상세한 재질 정보가 나왔으면 업데이트 if material_grade and material_grade != "UNKNOWN": db.execute(text(""" UPDATE materials SET material_grade = :new_material_grade WHERE id = :material_id """), { "new_material_grade": material_grade, "material_id": material_id }) print(f"FLANGE material_grade 업데이트: {material_grade}") # 사이즈 정보 size_inches = material_data.get("main_nom") or material_data.get("size_spec", "") db.execute(text(""" INSERT INTO flange_details ( material_id, file_id, flange_type, pressure_rating, facing_type, material_standard, material_grade, size_inches ) VALUES ( :material_id, :file_id, :flange_type, :pressure_rating, :facing_type, :material_standard, :material_grade, :size_inches ) """), { "material_id": material_id, "file_id": file_id, "flange_type": flange_type, "pressure_rating": pressure_rating, "facing_type": facing_type, "material_standard": material_standard, "material_grade": material_grade, "size_inches": size_inches }) print(f"FLANGE 상세 정보 저장 완료: {flange_type} - {pressure_rating}") # GASKET 분류 결과인 경우 상세 정보 저장 if classification_result.get("category") == "GASKET": print("GASKET 상세 정보 저장 시작") # 가스켓 타입 정보 gasket_type_info = classification_result.get("gasket_type", {}) gasket_material_info = classification_result.get("gasket_material", {}) pressure_info = classification_result.get("pressure_rating", {}) # 가스켓 타입 (SPIRAL_WOUND, O_RING 등) gasket_type = "" if isinstance(gasket_type_info, dict): gasket_type = gasket_type_info.get("type", "UNKNOWN") else: gasket_type = str(gasket_type_info) if gasket_type_info else "UNKNOWN" # 가스켓 소재 (GRAPHITE, PTFE 등) material_type = "" if isinstance(gasket_material_info, dict): material_type = gasket_material_info.get("material", "UNKNOWN") else: material_type = str(gasket_material_info) if gasket_material_info else "UNKNOWN" # 압력 등급 pressure_rating = "" if isinstance(pressure_info, dict): pressure_rating = pressure_info.get("rating", "UNKNOWN") else: pressure_rating = str(pressure_info) if pressure_info else "UNKNOWN" # 사이즈 정보 size_inches = material_data.get("main_nom") or material_data.get("size_spec", "") # SWG 상세 정보 추출 swg_details = gasket_material_info.get("swg_details", {}) if isinstance(gasket_material_info, dict) else {} thickness = swg_details.get("thickness", None) if swg_details else None filler_material = swg_details.get("filler", "") if swg_details else "" # additional_info에 SWG 상세 정보 저장 additional_info = "" if swg_details: face_type = swg_details.get("face_type", "") outer_ring = swg_details.get("outer_ring", "") inner_ring = swg_details.get("inner_ring", "") construction = swg_details.get("detailed_construction", "") # JSON 형태로 additional_info 생성 additional_info = { "face_type": face_type, "construction": construction, "outer_ring": outer_ring, "inner_ring": inner_ring, "filler": swg_details.get("filler", ""), "thickness": swg_details.get("thickness", None) } additional_info_json = json.dumps(additional_info, ensure_ascii=False) db.execute(text(""" INSERT INTO gasket_details ( material_id, file_id, gasket_type, material_type, pressure_rating, size_inches, thickness, filler_material, additional_info ) VALUES ( :material_id, :file_id, :gasket_type, :material_type, :pressure_rating, :size_inches, :thickness, :filler_material, :additional_info ) """), { "material_id": material_id, "file_id": file_id, "gasket_type": gasket_type, "material_type": material_type, "pressure_rating": pressure_rating, "size_inches": size_inches, "thickness": thickness, "filler_material": filler_material, "additional_info": additional_info_json }) print(f"GASKET 상세 정보 저장 완료: {gasket_type} - {material_type}") # BOLT 분류 결과인 경우 상세 정보 저장 if classification_result.get("category") == "BOLT": print("BOLT 상세 정보 저장 시작") # 볼트 타입 정보 fastener_type_info = classification_result.get("fastener_type", {}) thread_spec_info = classification_result.get("thread_specification", {}) dimensions_info = classification_result.get("dimensions", {}) material_info = classification_result.get("material", {}) # 볼트 타입 (STUD_BOLT, HEX_BOLT 등) bolt_type = "" if isinstance(fastener_type_info, dict): bolt_type = fastener_type_info.get("type", "UNKNOWN") else: bolt_type = str(fastener_type_info) if fastener_type_info else "UNKNOWN" # 나사 타입 (METRIC, INCH 등) thread_type = "" if isinstance(thread_spec_info, dict): thread_type = thread_spec_info.get("standard", "UNKNOWN") else: thread_type = str(thread_spec_info) if thread_spec_info else "UNKNOWN" # 치수 정보 diameter = material_data.get("main_nom", "") length = "" if isinstance(dimensions_info, dict): length = dimensions_info.get("length", "") if not length and "70.0000 LG" in description: # 원본 설명에서 길이 추출 import re length_match = re.search(r'(\d+(?:\.\d+)?)\s*LG', description.upper()) if length_match: length = f"{length_match.group(1)}mm" # 재질 정보 material_standard = "" material_grade = "" if isinstance(material_info, dict): material_standard = material_info.get("standard", "") material_grade = material_info.get("grade", "") # 분류기에서 더 상세한 재질 정보가 나왔으면 업데이트 if material_grade and material_grade != "UNKNOWN": db.execute(text(""" UPDATE materials SET material_grade = :new_material_grade WHERE id = :material_id """), { "new_material_grade": material_grade, "material_id": material_id }) print(f"BOLT material_grade 업데이트: {material_grade}") # 압력 등급 (150LB 등) pressure_rating = "" if "150LB" in description.upper(): pressure_rating = "150LB" elif "300LB" in description.upper(): pressure_rating = "300LB" elif "600LB" in description.upper(): pressure_rating = "600LB" # 코팅 타입 (ELEC.GALV 등) coating_type = "" if "ELEC.GALV" in description.upper() or "ELEC GALV" in description.upper(): coating_type = "ELECTRO_GALVANIZED" elif "HOT.GALV" in description.upper() or "HOT GALV" in description.upper(): coating_type = "HOT_DIP_GALVANIZED" elif "GALV" in description.upper(): coating_type = "GALVANIZED" elif "ZINC" in description.upper(): coating_type = "ZINC_PLATED" elif "DACROMET" in description.upper(): coating_type = "DACROMET" elif "SS" in description.upper() or "STAINLESS" in description.upper(): coating_type = "STAINLESS" elif "PLAIN" in description.upper() or "BLACK" in description.upper(): coating_type = "PLAIN" db.execute(text(""" INSERT INTO bolt_details ( material_id, file_id, bolt_type, thread_type, diameter, length, material_standard, material_grade, coating_type, pressure_rating, classification_confidence ) VALUES ( :material_id, :file_id, :bolt_type, :thread_type, :diameter, :length, :material_standard, :material_grade, :coating_type, :pressure_rating, :classification_confidence ) """), { "material_id": material_id, "file_id": file_id, "bolt_type": bolt_type, "thread_type": thread_type, "diameter": diameter, "length": length, "material_standard": material_standard, "material_grade": material_grade, "coating_type": coating_type, "pressure_rating": pressure_rating, "classification_confidence": classification_result.get("overall_confidence", 0.0) }) print(f"BOLT 상세 정보 저장 완료: {bolt_type} - {material_standard} {material_grade}") # VALVE 분류 결과인 경우 상세 정보 저장 if classification_result.get("category") == "VALVE": print("VALVE 상세 정보 저장 시작") # 밸브 타입 정보 valve_type_info = classification_result.get("valve_type", {}) connection_info = classification_result.get("connection_method", {}) pressure_info = classification_result.get("pressure_rating", {}) material_info = classification_result.get("material", {}) # 밸브 타입 (GATE_VALVE, BALL_VALVE 등) valve_type = "" if isinstance(valve_type_info, dict): valve_type = valve_type_info.get("type", "UNKNOWN") else: valve_type = str(valve_type_info) if valve_type_info else "UNKNOWN" # 밸브 서브타입 (특수 기능) valve_subtype = "" special_features = classification_result.get("special_features", []) if special_features: valve_subtype = ", ".join(special_features) # 작동 방식 actuator_type = "MANUAL" # 기본값 actuation_info = classification_result.get("actuation", {}) if isinstance(actuation_info, dict): actuator_type = actuation_info.get("method", "MANUAL") # 연결 방식 (FLANGED, THREADED 등) connection_method = "" if isinstance(connection_info, dict): connection_method = connection_info.get("method", "UNKNOWN") else: connection_method = str(connection_info) if connection_info else "UNKNOWN" # 압력 등급 (150LB, 300LB 등) pressure_rating = "" if isinstance(pressure_info, dict): pressure_rating = pressure_info.get("rating", "UNKNOWN") else: pressure_rating = str(pressure_info) if pressure_info else "UNKNOWN" # 재질 정보 body_material = "" trim_material = "" if isinstance(material_info, dict): body_material = material_info.get("grade", "") # 트림 재질은 일반적으로 바디와 동일하거나 별도 명시 trim_material = body_material # 분류기에서 더 상세한 재질 정보가 나왔으면 업데이트 if body_material and body_material != "UNKNOWN": db.execute(text(""" UPDATE materials SET material_grade = :new_material_grade WHERE id = :material_id """), { "new_material_grade": body_material, "material_id": material_id }) print(f"VALVE material_grade 업데이트: {body_material}") # 사이즈 정보 size_inches = material_data.get("main_nom") or material_data.get("size_spec", "") # 특수 기능 (Fire Safe, Anti-Static 등) fire_safe = any("FIRE" in feature.upper() for feature in special_features) low_temp_service = any("CRYO" in feature.upper() or "LOW" in feature.upper() for feature in special_features) # 추가 정보 JSON 생성 additional_info = { "characteristics": valve_type_info.get("characteristics", "") if isinstance(valve_type_info, dict) else "", "typical_connections": valve_type_info.get("typical_connections", []) if isinstance(valve_type_info, dict) else [], "special_features": special_features, "manufacturing": classification_result.get("manufacturing", {}), "evidence": valve_type_info.get("evidence", []) if isinstance(valve_type_info, dict) else [] } additional_info_json = json.dumps(additional_info, ensure_ascii=False) db.execute(text(""" INSERT INTO valve_details ( material_id, file_id, valve_type, valve_subtype, actuator_type, connection_method, pressure_rating, body_material, trim_material, size_inches, fire_safe, low_temp_service, classification_confidence, additional_info ) VALUES ( :material_id, :file_id, :valve_type, :valve_subtype, :actuator_type, :connection_method, :pressure_rating, :body_material, :trim_material, :size_inches, :fire_safe, :low_temp_service, :classification_confidence, :additional_info ) """), { "material_id": material_id, "file_id": file_id, "valve_type": valve_type, "valve_subtype": valve_subtype, "actuator_type": actuator_type, "connection_method": connection_method, "pressure_rating": pressure_rating, "body_material": body_material, "trim_material": trim_material, "size_inches": size_inches, "fire_safe": fire_safe, "low_temp_service": low_temp_service, "classification_confidence": classification_result.get("overall_confidence", 0.0), "additional_info": additional_info_json }) print(f"VALVE 상세 정보 저장 완료: {valve_type} - {connection_method} - {pressure_rating}") db.commit() print(f"자재 저장 완료: {materials_inserted}개") return { "success": True, "message": f"업로드 성공! {materials_inserted}개 자재가 분류되었습니다.", "original_filename": file.filename, "file_id": file_id, "materials_count": materials_inserted, "saved_materials_count": materials_inserted, "revision": revision, # 생성된 리비전 정보 추가 "parsed_count": parsed_count } except Exception as e: db.rollback() if os.path.exists(file_path): os.remove(file_path) raise HTTPException(status_code=500, detail=f"파일 처리 실패: {str(e)}") @router.get("/files") async def get_files( job_no: Optional[str] = None, db: Session = Depends(get_db) ): """파일 목록 조회""" try: query = """ SELECT id, filename, original_filename, job_no, revision, description, file_size, parsed_count, created_at, is_active FROM files WHERE is_active = TRUE """ params = {} if job_no: query += " AND job_no = :job_no" params["job_no"] = job_no query += " ORDER BY created_at DESC" result = db.execute(text(query), params) files = result.fetchall() return [ { "id": file.id, "filename": file.filename, "original_filename": file.original_filename, "job_no": file.job_no, "revision": file.revision, "description": file.description, "file_size": file.file_size, "parsed_count": file.parsed_count, "created_at": file.created_at, "is_active": file.is_active } for file in files ] except Exception as e: raise HTTPException(status_code=500, detail=f"파일 목록 조회 실패: {str(e)}") @router.delete("/files/{file_id}") async def delete_file(file_id: int, db: Session = Depends(get_db)): """파일 삭제""" try: # 자재 먼저 삭제 db.execute(text("DELETE FROM materials WHERE file_id = :file_id"), {"file_id": file_id}) # 파일 삭제 result = db.execute(text("DELETE FROM files WHERE id = :file_id"), {"file_id": file_id}) if result.rowcount == 0: raise HTTPException(status_code=404, detail="파일을 찾을 수 없습니다") db.commit() return { "success": True, "message": "파일이 삭제되었습니다" } except Exception as e: db.rollback() raise HTTPException(status_code=500, detail=f"파일 삭제 실패: {str(e)}") @router.get("/materials") async def get_materials( project_id: Optional[int] = None, file_id: Optional[int] = None, job_no: Optional[str] = None, filename: Optional[str] = None, revision: Optional[str] = None, skip: int = 0, limit: int = 100, search: 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, db: Session = Depends(get_db) ): """ 저장된 자재 목록 조회 (job_no, filename, revision 3가지로 필터링 가능) """ try: query = """ SELECT m.id, m.file_id, m.original_description, m.quantity, m.unit, m.size_spec, m.main_nom, m.red_nom, m.material_grade, m.line_number, m.row_number, m.created_at, m.classified_category, m.classification_confidence, m.classification_details, f.original_filename, f.project_id, f.job_no, f.revision, p.official_project_code, p.project_name, pd.outer_diameter, pd.schedule, pd.material_spec, pd.manufacturing_method, pd.end_preparation, pd.length_mm FROM materials m LEFT JOIN files f ON m.file_id = f.id LEFT JOIN projects p ON f.project_id = p.id LEFT JOIN pipe_details pd ON m.id = pd.material_id WHERE 1=1 """ params = {} if project_id: query += " AND f.project_id = :project_id" params["project_id"] = project_id if file_id: query += " AND m.file_id = :file_id" params["file_id"] = file_id if job_no: query += " AND f.job_no = :job_no" params["job_no"] = job_no if filename: query += " AND f.original_filename = :filename" params["filename"] = filename if revision: query += " AND f.revision = :revision" params["revision"] = revision if search: query += " AND (m.original_description ILIKE :search OR m.material_grade ILIKE :search)" params["search"] = f"%{search}%" if item_type: query += " AND m.classified_category = :item_type" params["item_type"] = item_type if material_grade: query += " AND m.material_grade ILIKE :material_grade" params["material_grade"] = f"%{material_grade}%" if size_spec: query += " AND m.size_spec ILIKE :size_spec" params["size_spec"] = f"%{size_spec}%" if file_filter: query += " AND f.original_filename ILIKE :file_filter" params["file_filter"] = f"%{file_filter}%" # 정렬 처리 if sort_by: if sort_by == "quantity_desc": query += " ORDER BY m.quantity DESC" elif sort_by == "quantity_asc": query += " ORDER BY m.quantity ASC" elif sort_by == "name_asc": query += " ORDER BY m.original_description ASC" elif sort_by == "name_desc": query += " ORDER BY m.original_description DESC" elif sort_by == "created_desc": query += " ORDER BY m.created_at DESC" elif sort_by == "created_asc": query += " ORDER BY m.created_at ASC" else: query += " ORDER BY m.line_number ASC" else: query += " ORDER BY m.line_number ASC" query += " LIMIT :limit OFFSET :skip" params["limit"] = limit params["skip"] = skip result = db.execute(text(query), params) materials = result.fetchall() # 전체 개수 조회 count_query = """ SELECT COUNT(*) as total FROM materials m LEFT JOIN files f ON m.file_id = f.id WHERE 1=1 """ count_params = {} if project_id: count_query += " AND f.project_id = :project_id" count_params["project_id"] = project_id if file_id: count_query += " AND m.file_id = :file_id" count_params["file_id"] = file_id if search: count_query += " AND (m.original_description ILIKE :search OR m.material_grade ILIKE :search)" count_params["search"] = f"%{search}%" if item_type: count_query += " AND m.classified_category = :item_type" count_params["item_type"] = item_type if material_grade: count_query += " AND m.material_grade ILIKE :material_grade" count_params["material_grade"] = f"%{material_grade}%" if size_spec: count_query += " AND m.size_spec ILIKE :size_spec" count_params["size_spec"] = f"%{size_spec}%" if file_filter: count_query += " AND f.original_filename ILIKE :file_filter" count_params["file_filter"] = f"%{file_filter}%" count_result = db.execute(text(count_query), count_params) total_count = count_result.fetchone()[0] # 각 자재의 상세 정보도 가져오기 material_list = [] for m in materials: material_dict = { "id": m.id, "file_id": m.file_id, "filename": m.original_filename, "project_id": m.project_id, "project_code": m.official_project_code, "project_name": m.project_name, "original_description": m.original_description, "quantity": float(m.quantity) if m.quantity else 0, "unit": m.unit, "size_spec": m.size_spec, "main_nom": m.main_nom, # 추가 "red_nom": m.red_nom, # 추가 "material_grade": m.material_grade, "line_number": m.line_number, "row_number": m.row_number, "classified_category": m.classified_category, "classification_confidence": float(m.classification_confidence) if m.classification_confidence else 0.0, "classification_details": m.classification_details, "created_at": m.created_at } # 카테고리별 상세 정보 추가 (JOIN 결과 사용) if m.classified_category == 'PIPE': # JOIN된 결과에서 pipe_details 정보 가져오기 if hasattr(m, 'outer_diameter') and m.outer_diameter is not None: material_dict['pipe_details'] = { "outer_diameter": m.outer_diameter, "schedule": m.schedule, "material_spec": m.material_spec, "manufacturing_method": m.manufacturing_method, "end_preparation": m.end_preparation, "length_mm": float(m.length_mm) if m.length_mm else None } elif m.classified_category == 'FITTING': fitting_query = text("SELECT * FROM fitting_details WHERE material_id = :material_id") fitting_result = db.execute(fitting_query, {"material_id": m.id}) fitting_detail = fitting_result.fetchone() if fitting_detail: material_dict['fitting_details'] = { "fitting_type": fitting_detail.fitting_type, "fitting_subtype": fitting_detail.fitting_subtype, "connection_method": fitting_detail.connection_method, "pressure_rating": fitting_detail.pressure_rating, "material_standard": fitting_detail.material_standard, "material_grade": fitting_detail.material_grade, "main_size": fitting_detail.main_size, "reduced_size": fitting_detail.reduced_size, "length_mm": float(fitting_detail.length_mm) if fitting_detail.length_mm else None, "schedule": fitting_detail.schedule } elif m.classified_category == 'FLANGE': flange_query = text("SELECT * FROM flange_details WHERE material_id = :material_id") flange_result = db.execute(flange_query, {"material_id": m.id}) flange_detail = flange_result.fetchone() if flange_detail: material_dict['flange_details'] = { "flange_type": flange_detail.flange_type, "facing_type": flange_detail.facing_type, "pressure_rating": flange_detail.pressure_rating, "material_standard": flange_detail.material_standard, "material_grade": flange_detail.material_grade, "size_inches": flange_detail.size_inches } elif m.classified_category == 'GASKET': gasket_query = text("SELECT * FROM gasket_details WHERE material_id = :material_id") gasket_result = db.execute(gasket_query, {"material_id": m.id}) gasket_detail = gasket_result.fetchone() if gasket_detail: material_dict['gasket_details'] = { "gasket_type": gasket_detail.gasket_type, "material_type": gasket_detail.material_type, "pressure_rating": gasket_detail.pressure_rating, "size_inches": gasket_detail.size_inches, "thickness": gasket_detail.thickness, "temperature_range": gasket_detail.temperature_range } elif m.classified_category == 'VALVE': valve_query = text("SELECT * FROM valve_details WHERE material_id = :material_id") valve_result = db.execute(valve_query, {"material_id": m.id}) valve_detail = valve_result.fetchone() if valve_detail: material_dict['valve_details'] = { "valve_type": valve_detail.valve_type, "valve_subtype": valve_detail.valve_subtype, "actuator_type": valve_detail.actuator_type, "connection_method": valve_detail.connection_method, "pressure_rating": valve_detail.pressure_rating, "body_material": valve_detail.body_material, "size_inches": valve_detail.size_inches } elif m.classified_category == 'BOLT': bolt_query = text("SELECT * FROM bolt_details WHERE material_id = :material_id") bolt_result = db.execute(bolt_query, {"material_id": m.id}) bolt_detail = bolt_result.fetchone() if bolt_detail: material_dict['bolt_details'] = { "bolt_type": bolt_detail.bolt_type, "thread_type": bolt_detail.thread_type, "diameter": bolt_detail.diameter, "length": bolt_detail.length, "material_standard": bolt_detail.material_standard, "material_grade": bolt_detail.material_grade, "coating_type": bolt_detail.coating_type, "pressure_rating": bolt_detail.pressure_rating } material_list.append(material_dict) return { "success": True, "total_count": total_count, "returned_count": len(materials), "skip": skip, "limit": limit, "materials": material_list } except Exception as e: raise HTTPException(status_code=500, detail=f"자재 조회 실패: {str(e)}") @router.get("/materials/summary") async def get_materials_summary( project_id: Optional[int] = None, file_id: Optional[int] = 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 project_id: query += " AND f.project_id = :project_id" params["project_id"] = project_id 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)}") @router.get("/materials/compare-revisions") async def compare_revisions( job_no: str, filename: str, old_revision: str, new_revision: str, db: Session = Depends(get_db) ): """ 리비전 간 자재 비교 """ try: # 기존 리비전 자재 조회 old_materials_query = text(""" SELECT m.original_description, m.quantity, m.unit, m.size_spec, m.material_grade, m.classified_category, m.classification_confidence FROM materials m JOIN files f ON m.file_id = f.id WHERE f.job_no = :job_no AND f.original_filename = :filename AND f.revision = :old_revision """) old_result = db.execute(old_materials_query, { "job_no": job_no, "filename": filename, "old_revision": old_revision }) old_materials = old_result.fetchall() # 새 리비전 자재 조회 new_materials_query = text(""" SELECT m.original_description, m.quantity, m.unit, m.size_spec, m.material_grade, m.classified_category, m.classification_confidence FROM materials m JOIN files f ON m.file_id = f.id WHERE f.job_no = :job_no AND f.original_filename = :filename AND f.revision = :new_revision """) new_result = db.execute(new_materials_query, { "job_no": job_no, "filename": filename, "new_revision": new_revision }) new_materials = new_result.fetchall() # 자재 키 생성 함수 def create_material_key(material): return f"{material.original_description}_{material.size_spec}_{material.material_grade}" # 기존 자재를 딕셔너리로 변환 old_materials_dict = {} for material in old_materials: key = create_material_key(material) old_materials_dict[key] = { "original_description": material.original_description, "quantity": float(material.quantity) if material.quantity else 0, "unit": material.unit, "size_spec": material.size_spec, "material_grade": material.material_grade, "classified_category": material.classified_category, "classification_confidence": material.classification_confidence } # 새 자재를 딕셔너리로 변환 new_materials_dict = {} for material in new_materials: key = create_material_key(material) new_materials_dict[key] = { "original_description": material.original_description, "quantity": float(material.quantity) if material.quantity else 0, "unit": material.unit, "size_spec": material.size_spec, "material_grade": material.material_grade, "classified_category": material.classified_category, "classification_confidence": material.classification_confidence } # 변경 사항 분석 all_keys = set(old_materials_dict.keys()) | set(new_materials_dict.keys()) added_items = [] removed_items = [] changed_items = [] for key in all_keys: old_item = old_materials_dict.get(key) new_item = new_materials_dict.get(key) if old_item and not new_item: # 삭제된 항목 removed_items.append({ "key": key, "item": old_item, "change_type": "removed" }) elif not old_item and new_item: # 추가된 항목 added_items.append({ "key": key, "item": new_item, "change_type": "added" }) elif old_item and new_item: # 수량 변경 확인 if old_item["quantity"] != new_item["quantity"]: changed_items.append({ "key": key, "old_item": old_item, "new_item": new_item, "quantity_change": new_item["quantity"] - old_item["quantity"], "change_type": "quantity_changed" }) # 분류별 통계 def calculate_category_stats(items): stats = {} for item in items: category = item.get("item", {}).get("classified_category", "OTHER") if category not in stats: stats[category] = {"count": 0, "total_quantity": 0} stats[category]["count"] += 1 stats[category]["total_quantity"] += item.get("item", {}).get("quantity", 0) return stats added_stats = calculate_category_stats(added_items) removed_stats = calculate_category_stats(removed_items) changed_stats = calculate_category_stats(changed_items) return { "success": True, "comparison": { "old_revision": old_revision, "new_revision": new_revision, "filename": filename, "job_no": job_no, "summary": { "added_count": len(added_items), "removed_count": len(removed_items), "changed_count": len(changed_items), "total_changes": len(added_items) + len(removed_items) + len(changed_items) }, "changes": { "added": added_items, "removed": removed_items, "changed": changed_items }, "category_stats": { "added": added_stats, "removed": removed_stats, "changed": changed_stats } } } except Exception as e: raise HTTPException(status_code=500, detail=f"리비전 비교 실패: {str(e)}") @router.get("/pipe-details") async def get_pipe_details( file_id: Optional[int] = None, job_no: Optional[str] = None, db: Session = Depends(get_db) ): """ PIPE 상세 정보 조회 """ try: query = """ SELECT pd.*, f.original_filename, f.job_no, f.revision, m.original_description, m.quantity, m.unit FROM pipe_details pd LEFT JOIN files f ON pd.file_id = f.id LEFT JOIN materials m ON pd.file_id = m.file_id AND m.classified_category = 'PIPE' WHERE 1=1 """ params = {} if file_id: query += " AND pd.file_id = :file_id" params["file_id"] = file_id if job_no: query += " AND f.job_no = :job_no" params["job_no"] = job_no query += " ORDER BY pd.created_at DESC" result = db.execute(text(query), params) pipe_details = result.fetchall() return [ { "id": pd.id, "file_id": pd.file_id, "original_filename": pd.original_filename, "job_no": pd.job_no, "revision": pd.revision, "original_description": pd.original_description, "quantity": pd.quantity, "unit": pd.unit, "material_spec": pd.material_spec, "manufacturing_method": pd.manufacturing_method, "end_preparation": pd.end_preparation, "schedule": pd.schedule, "outer_diameter": pd.outer_diameter, "length_mm": pd.length_mm, "created_at": pd.created_at, "updated_at": pd.updated_at } for pd in pipe_details ] except Exception as e: raise HTTPException(status_code=500, detail=f"PIPE 상세 정보 조회 실패: {str(e)}") @router.get("/fitting-details") async def get_fitting_details( file_id: Optional[int] = None, job_no: Optional[str] = None, db: Session = Depends(get_db) ): """ FITTING 상세 정보 조회 """ try: query = """ SELECT fd.*, f.original_filename, f.job_no, f.revision, m.original_description, m.quantity, m.unit FROM fitting_details fd LEFT JOIN files f ON fd.file_id = f.id LEFT JOIN materials m ON fd.material_id = m.id WHERE 1=1 """ params = {} if file_id: query += " AND fd.file_id = :file_id" params["file_id"] = file_id if job_no: query += " AND f.job_no = :job_no" params["job_no"] = job_no query += " ORDER BY fd.created_at DESC" result = db.execute(text(query), params) fitting_details = result.fetchall() return [ { "id": fd.id, "file_id": fd.file_id, "fitting_type": fd.fitting_type, "fitting_subtype": fd.fitting_subtype, "connection_method": fd.connection_method, "pressure_rating": fd.pressure_rating, "material_standard": fd.material_standard, "material_grade": fd.material_grade, "main_size": fd.main_size, "reduced_size": fd.reduced_size, "classification_confidence": fd.classification_confidence, "original_description": fd.original_description, "quantity": fd.quantity } for fd in fitting_details ] except Exception as e: raise HTTPException(status_code=500, detail=f"FITTING 상세 정보 조회 실패: {str(e)}") @router.get("/valve-details") async def get_valve_details( file_id: Optional[int] = None, job_no: Optional[str] = None, db: Session = Depends(get_db) ): """ VALVE 상세 정보 조회 """ try: query = """ SELECT vd.*, f.original_filename, f.job_no, f.revision, m.original_description, m.quantity, m.unit FROM valve_details vd LEFT JOIN files f ON vd.file_id = f.id LEFT JOIN materials m ON vd.material_id = m.id WHERE 1=1 """ params = {} if file_id: query += " AND vd.file_id = :file_id" params["file_id"] = file_id if job_no: query += " AND f.job_no = :job_no" params["job_no"] = job_no query += " ORDER BY vd.created_at DESC" result = db.execute(text(query), params) valve_details = result.fetchall() return [ { "id": vd.id, "file_id": vd.file_id, "valve_type": vd.valve_type, "valve_subtype": vd.valve_subtype, "actuator_type": vd.actuator_type, "connection_method": vd.connection_method, "pressure_rating": vd.pressure_rating, "body_material": vd.body_material, "size_inches": vd.size_inches, "fire_safe": vd.fire_safe, "classification_confidence": vd.classification_confidence, "original_description": vd.original_description, "quantity": vd.quantity } for vd in valve_details ] except Exception as e: raise HTTPException(status_code=500, detail=f"VALVE 상세 정보 조회 실패: {str(e)}") @router.get("/user-requirements") async def get_user_requirements( file_id: Optional[int] = None, job_no: Optional[str] = None, status: Optional[str] = None, db: Session = Depends(get_db) ): """ 사용자 요구사항 조회 """ try: query = """ SELECT ur.*, f.original_filename, f.job_no, f.revision, rt.type_name, rt.category FROM user_requirements ur LEFT JOIN files f ON ur.file_id = f.id LEFT JOIN requirement_types rt ON ur.requirement_type = rt.type_code WHERE 1=1 """ params = {} if file_id: query += " AND ur.file_id = :file_id" params["file_id"] = file_id if job_no: query += " AND f.job_no = :job_no" params["job_no"] = job_no if status: query += " AND ur.status = :status" params["status"] = status query += " ORDER BY ur.created_at DESC" result = db.execute(text(query), params) requirements = result.fetchall() return [ { "id": req.id, "file_id": req.file_id, "original_filename": req.original_filename, "job_no": req.job_no, "revision": req.revision, "requirement_type": req.requirement_type, "type_name": req.type_name, "category": req.category, "requirement_title": req.requirement_title, "requirement_description": req.requirement_description, "requirement_spec": req.requirement_spec, "status": req.status, "priority": req.priority, "assigned_to": req.assigned_to, "due_date": req.due_date, "created_at": req.created_at, "updated_at": req.updated_at } for req in requirements ] except Exception as e: raise HTTPException(status_code=500, detail=f"사용자 요구사항 조회 실패: {str(e)}") @router.post("/user-requirements") async def create_user_requirement( file_id: int, requirement_type: str, requirement_title: str, requirement_description: Optional[str] = None, requirement_spec: Optional[str] = None, priority: str = "NORMAL", assigned_to: Optional[str] = None, due_date: Optional[str] = None, db: Session = Depends(get_db) ): """ 사용자 요구사항 생성 """ try: insert_query = text(""" INSERT INTO user_requirements ( file_id, requirement_type, requirement_title, requirement_description, requirement_spec, priority, assigned_to, due_date ) VALUES ( :file_id, :requirement_type, :requirement_title, :requirement_description, :requirement_spec, :priority, :assigned_to, :due_date ) RETURNING id """) result = db.execute(insert_query, { "file_id": file_id, "requirement_type": requirement_type, "requirement_title": requirement_title, "requirement_description": requirement_description, "requirement_spec": requirement_spec, "priority": priority, "assigned_to": assigned_to, "due_date": due_date }) requirement_id = result.fetchone()[0] db.commit() return { "success": True, "message": "요구사항이 생성되었습니다", "requirement_id": requirement_id } except Exception as e: db.rollback() raise HTTPException(status_code=500, detail=f"요구사항 생성 실패: {str(e)}")