ExperionCrawler/mcp-server/test_pid_worker_dxf.py

#!/usr/bin/env python3
"""No-10_Plant_PID.dxf 파일을 이용한 pid_worker 테스트 스크립트"""

import sys
import os
import json
import asyncio
import requests

# 테스트 설정
SERVER_URL = "http://localhost:5001"
DXF_FILE_PATH = "/home/windpacer/projects/ExperionCrawler/futurePlan/End-to-End P&ID Graph Pipeline/No-10_Plant_PID.dxf"

def test_health():
    """헬스 체크"""
    print("=" * 60)
    print("1. 헬스 체크")
    print("=" * 60)
    try:
        resp = requests.get(f"{SERVER_URL}/health")
        print(f"응답: {resp.status_code}")
        print(json.dumps(resp.json(), indent=2, ensure_ascii=False))
        return resp.status_code == 200
    except Exception as e:
        print(f"오류: {e}")
        return False

def test_parse_pid_drawing():
    """parse_pid_drawing 도구 테스트"""
    print("\n" + "=" * 60)
    print("2. parse_pid_drawing 도구 테스트")
    print("=" * 60)
    print(f"DXF 파일: {DXF_FILE_PATH}")
    
    payload = {
        "tool": "parse_pid_drawing",
        "params": {
            "filepath": DXF_FILE_PATH
        }
    }
    
    try:
        resp = requests.post(f"{SERVER_URL}/execute", json=payload)
        print(f"응답: {resp.status_code}")
        # 응답 본문을 문자열로 받고 JSON 파싱
        response_text = resp.text
        print(f"응답 길이: {len(response_text)}")
        
        # 응답이 JSON 문자열로 감싸져 있는 경우 처리
        if response_text.startswith('"') and response_text.endswith('"'):
            try:
                # JSON 문자열을 파싱
                result = json.loads(json.loads(response_text))
            except json.JSONDecodeError:
                print(f"중첩 JSON 파싱 실패. 응답 텍스트:\n{response_text[:500]}")
                return False
        else:
            try:
                result = json.loads(response_text)
            except json.JSONDecodeError:
                print(f"JSON 파싱 실패. 응답 텍스트:\n{response_text[:500]}")
                return False
        
        print(json.dumps(result, indent=2, ensure_ascii=False))
        
        if result.get("success"):
            print(f"\n✅ 추출된 태그 수: {result.get('count', 0)}")
            if result.get("tags"):
                print("\n첫 5개 태그:")
                for tag in result["tags"][:5]:
                    print(f"  - {tag.get('tagNo')} ({tag.get('instrumentType')})")
        
        return resp.status_code == 200 and result.get("success")
    except Exception as e:
        print(f"오류: {e}")
        import traceback
        traceback.print_exc()
        return False

def test_extract_pid_tags():
    """extract_pid_tags 도구 테스트 (직접 텍스트 추출)"""
    print("\n" + "=" * 60)
    print("3. extract_pid_tags 도구 테스트")
    print("=" * 60)
    
    # DXF에서 텍스트 추출
    try:
        import ezdxf
        doc = ezdxf.readfile(DXF_FILE_PATH)
        msp = doc.modelspace()
        texts = []
        for entity in msp:
            if entity.dxftype() == "TEXT":
                texts.append(entity.dxf.text)
            elif entity.dxftype() == "MTEXT":
                from ezdxf.tools.text import plain_mtext
                try:
                    plain = plain_mtext(entity.dxf.text)
                    if plain.strip():
                        texts.append(plain)
                except Exception:
                    pass
        
        text_content = "\n".join(texts)[:5000]  # 제한
        print(f"DXF에서 추출한 텍스트 길이: {len(text_content)}")
        print(f"첫 200자: {text_content[:200]}...")
        
        payload = {
            "tool": "extract_pid_tags",
            "params": {
                "text": text_content,
                "source_type": "dxf"
            }
        }
        
        resp = requests.post(f"{SERVER_URL}/execute", json=payload)
        print(f"\n응답: {resp.status_code}")
        result = resp.json()
        print(json.dumps(result, indent=2, ensure_ascii=False))
        
        return resp.status_code == 200
    except Exception as e:
        print(f"오류: {e}")
        import traceback
        traceback.print_exc()
        return False

async def test_build_pid_graph_parallel():
    """build_pid_graph_parallel 도구 테스트"""
    print("\n" + "=" * 60)
    print("4. build_pid_graph_parallel 도구 테스트")
    print("=" * 60)
    print(f"DXF 파일: {DXF_FILE_PATH}")
    
    payload = {
        "tool": "build_pid_graph_parallel",
        "params": {
            "filepath": DXF_FILE_PATH
        }
    }
    
    try:
        resp = requests.post(f"{SERVER_URL}/execute", json=payload)
        print(f"응답: {resp.status_code}")
        result = resp.json()
        print(json.dumps(result, indent=2, ensure_ascii=False))
        
        if result.get("success"):
            print(f"\n✅ 그래프 ID: {result.get('graph_id')}")
            print(f"✅ 노드 수: {result.get('nodes', 0)}")
            print(f"✅ 엣지 수: {result.get('edges', 0)}")
        
        return resp.status_code == 200 and result.get("success")
    except Exception as e:
        print(f"오류: {e}")
        import traceback
        traceback.print_exc()
        return False


def main():
    print("No-10_Plant_PID.dxf를 이용한 pid_worker 테스트")
    print(f"서버 URL: {SERVER_URL}")
    print(f"DXF 파일: {DXF_FILE_PATH}")
    
    # 테스트 실행
    results = []
    results.append(("헬스 체크", test_health()))
    results.append(("parse_pid_drawing", test_parse_pid_drawing()))
    results.append(("extract_pid_tags", test_extract_pid_tags()))
    results.append(("build_pid_graph_parallel", asyncio.run(test_build_pid_graph_parallel())))
    
    # 요약
    print("\n" + "=" * 60)
    print("테스트 요약")
    print("=" * 60)
    for name, passed in results:
        status = "✅ 통과" if passed else "❌ 실패"
        print(f"{name}: {status}")
    
    all_passed = all(r[1] for r in results)
    if all_passed:
        print("\n🎉 모든 테스트 통과!")
        return 0
    else:
        print("\n⚠️ 일부 테스트 실패")
        return 1

if __name__ == "__main__":
    sys.exit(main())