ExperionCrawler/dxf-graph/pid_parser_coding_plan.md

P&ID Parser 코딩 플랜 (v3 기반)

작성일: 2026-05-01
기준: End-to-End P&ID Graph Pipeline/PID_Parser_작업지시서_v3.md
목적: DXF/DWG/PDF 통합 파싱 → MCP 도구 추가 → C# 클라이언트 연동

---

📋 개요

현재 프로젝트는 extract_pid_tags와 match_pid_tags MCP 도구만 구현되어 있음.
작업 지시서 v3에 따라 다음 4개 도구를 추가해야 함:

도구	설명	포맷	정확도	처리 시간
parse_pid_dxf	ezdxf 기반 직독	DXF	95%+	<1초
parse_pid_dwg	ODA Converter → DXF	DWG	95%+	5~30초
parse_pid_pdf	PaddleOCR 기반	PDF	85~90%	5~90초
parse_pid_drawing	확장자 자동 라우팅	-	-	-

---

✅ 완료된 단계 (기존 구현)

단계	내용	상태
1	extract_pid_tags (raw text 모드)	✅ 완료
2	match_pid_tags	✅ 완료

---

📝 추가 작업 계획

단위 1: 시스템 도구 설치 (ODA File Converter) ✅

목적: DWG 파일을 DXF로 변환

작업 내용:

1. ODA File Converter 다운로드 및 설치
2. 경로 확인 및 server.py 상수 등록

검증 방법:

which ODAFileConverter
ODAFileConverter --help

완료 여부: ❌ (사용자 환경에 따라 설치 필요)

---

단위 2: Python 의존성 설치 ✅

목적: DXF/PDF 파싱 및 OCR 처리

작업 내용:

# DXF 처리
pip install ezdxf>=1.3.0

# OCR (PDF fallback용)
pip install paddlepaddle-gpu==2.6.* -i https://www.paddlepaddle.org.cn/packages/stable/cu118/
pip install paddleocr>=2.7.0

# 공통
pip install pymupdf>=1.24.0 scikit-learn>=1.3.0 numpy>=1.24.0 Pillow>=10.0.0

requirements.txt 업데이트:

ezdxf>=1.3.0
paddlepaddle-gpu>=2.6.0
paddleocr>=2.7.0
pymupdf>=1.24.0
scikit-learn>=1.3.0
numpy>=1.24.0
Pillow>=10.0.0

완료 여부: ❌ (사용자 환경에 따라 설치 필요)

---

단위 3: 공통 헬퍼 함수 추가 ✅

파일: mcp-server/server.py
위치: 기존 _llm() 근처 (58~62행)

추가 코드:

# ── PaddleOCR 싱글톤 (PDF fallback용) ──────────────────────────────────────────

@lru_cache(maxsize=1)
def _ocr():
    """PaddleOCR 인스턴스 (한/영, GPU). 첫 호출 시 ~50MB 모델 다운로드."""
    from paddleocr import PaddleOCR
    import os

    use_gpu = os.environ.get("PADDLE_USE_GPU", "true").lower() == "true"
    try:
        return PaddleOCR(
            use_angle_cls=True, lang="korean", use_gpu=use_gpu,
            show_log=False, det_db_box_thresh=0.3, det_db_unclip_ratio=1.6,
        )
    except Exception as e:
        logging.warning(f"PaddleOCR GPU 초기화 실패, CPU fallback: {e}")
        return PaddleOCR(use_angle_cls=True, lang="korean", use_gpu=False, show_log=False)


# ── ODA File Converter 경로 ────────────────────────────────────────────────────

ODA_CONVERTER_PATH = "/usr/bin/ODAFileConverter"  # 환경별 조정


def _parse_page_range(page_range: str, total: int) -> list[int]:
    """페이지 범위 문자열 → 0-based 인덱스 리스트.
    "" → 전체, "1" → [0], "1-3" → [0,1,2], "1,3" → [0,2], "2-" → [1..total-1]
    """
    if not page_range or not page_range.strip():
        return list(range(total))
    indices = set()
    for part in page_range.split(","):
        part = part.strip()
        if not part:
            continue
        if "-" in part:
            a, b = part.split("-", 1)
            start = int(a) - 1 if a.strip() else 0
            end = int(b) - 1 if b.strip() else total - 1
            for i in range(max(0, start), min(total - 1, end) + 1):
                indices.add(i)
        else:
            i = int(part) - 1
            if 0 <= i < total:
                indices.add(i)
    return sorted(indices)

완료 여부: ❌ (코드 추가 필요)

---

단위 4: parse_pid_dxf 도구 추가 ✅

파일: mcp-server/server.py
위치: # ── P&ID 추출 도구 ── 섹션 바로 위 (445행 직전)

추가 코드:

# ── P&ID DXF 파서 (ezdxf 기반, 가장 정확) ─────────────────────────────────────

@mcp.tool()
def parse_pid_dxf(
    dxf_path: str,
    instrument_layer: str = "INSTRUMENT",
    balloon_text_tolerance: float = 1.5,
) -> str:
    """DXF 도면을 직접 파싱해서 cluster JSON을 반환합니다.

    ezdxf로 TEXT/MTEXT/CIRCLE/INSERT 엔티티를 직독.
    INSTRUMENT 레이어의 CIRCLE을 ISA balloon으로 보고 내부 텍스트와 좌표 매칭하여
    'FIC' + '10101' → 'FIC-10101'을 LLM 호출 없이 사전 정규화한다.

    Args:
        dxf_path:                DXF 파일 절대 경로
        instrument_layer:        ISA balloon이 위치한 레이어명. 회사마다 다름.
                                 (예: 'INSTRUMENT', 'INST', 'INSTR', 'I-1')
        balloon_text_tolerance:  balloon 반지름 대비 텍스트 검색 범위 배수.
                                 1.5 = 반지름의 1.5배 거리까지 내부로 간주.

    Returns:
        통합 cluster JSON. pre_normalized_tag 필드에 자동 정규화 결과 포함.
    """
    try:
        import ezdxf
        import re as _re
        import os

        if not os.path.exists(dxf_path):
            return json.dumps(
                {"success": False, "error": f"파일을 찾을 수 없습니다: {dxf_path}"},
                ensure_ascii=False,
            )

        doc = ezdxf.readfile(dxf_path)
        msp = doc.modelspace()

        # 1. 모든 텍스트 추출
        all_texts = []
        for e in msp:
            etype = e.dxftype()
            if etype == "TEXT":
                all_texts.append({
                    "text": e.dxf.text.strip(),
                    "layer": e.dxf.layer,
                    "x": e.dxf.insert.x,
                    "y": e.dxf.insert.y,
                    "rotation": e.dxf.rotation,
                    "height": e.dxf.height,
                })
            elif etype == "MTEXT":
                all_texts.append({
                    "text": e.text.strip(),
                    "layer": e.dxf.layer,
                    "x": e.dxf.insert.x,
                    "y": e.dxf.insert.y,
                    "rotation": e.dxf.rotation,
                    "height": e.dxf.char_height,
                })
            elif etype == "ATTRIB":
                all_texts.append({
                    "text": e.dxf.text.strip(),
                    "layer": e.dxf.layer,
                    "x": e.dxf.insert.x,
                    "y": e.dxf.insert.y,
                    "rotation": e.dxf.rotation,
                    "height": e.dxf.height,
                })

        all_texts = [t for t in all_texts if t["text"]]

        # MTEXT 폰트 escape 코드 청소 ({\fMalgun Gothic|...; 한울} → 한울)
        font_escape = _re.compile(r'\{\\f[^;]*;([^}]*)\}')
        for t in all_texts:
            cleaned = font_escape.sub(r'\1', t["text"])
            cleaned = _re.sub(r'\\[A-Za-z]\d*[\.\d]*[xX]?;?', '', cleaned)
            t["text"] = cleaned.strip()

        # 2. INSTRUMENT 레이어 CIRCLE = ISA balloon
        balloons = [
            {
                "x": c.dxf.center.x, "y": c.dxf.center.y, "r": c.dxf.radius,
                "layer": c.dxf.layer,
            }
            for c in msp.query("CIRCLE")
            if c.dxf.layer == instrument_layer
        ]

        # 3. INSTRUMENT 레이어 LWPOLYLINE/SOLID 중 사각형 = DCS balloon
        squares = []
        for poly in msp.query("LWPOLYLINE"):
            if poly.dxf.layer != instrument_layer:
                continue
            if not poly.is_closed:
                continue
            points = list(poly.get_points())
            if len(points) == 4:
                xs = [p[0] for p in points]
                ys = [p[1] for p in points]
                squares.append({
                    "x_center": sum(xs) / 4,
                    "y_center": sum(ys) / 4,
                    "size": max(max(xs) - min(xs), max(ys) - min(ys)),
                })

        # 각 balloon이 사각형 안에 있는지 확인 → DCS vs FIELD 분류
        for b in balloons:
            b["balloon_type"] = "FIELD"  # 기본값
            for sq in squares:
                if (abs(sq["x_center"] - b["x"]) < sq["size"] / 2
                        and abs(sq["y_center"] - b["y"]) < sq["size"] / 2):
                    b["balloon_type"] = "DCS"
                    break

        # 4. 각 balloon에 대해 내부 텍스트 매칭
        inst_texts = [t for t in all_texts if t["layer"] == instrument_layer]
        balloon_clusters = []
        used_text_keys = set()

        tag_pat = _re.compile(r'^[A-Z]{1,4}$')
        loop_pat = _re.compile(r'^\d{2,5}[A-Z]?$')
        full_pat = _re.compile(r'^[A-Z]{1,4}-?\d{2,5}[A-Z]?$')

        for idx, b in enumerate(balloons):
            r_lim = b["r"] * balloon_text_tolerance
            inside = []
            for t in inst_texts:
                dx, dy = t["x"] - b["x"], t["y"] - b["y"]
                if (dx * dx + dy * dy) ** 0.5 <= r_lim:
                    inside.append(t)
                    used_text_keys.add((t["text"], round(t["x"], 2), round(t["y"], 2)))

            inside.sort(key=lambda t: -t["y"])

            pre_normalized = None
            confidence = 0.5
            if len(inside) >= 2:
                top, bot = inside[0]["text"], inside[1]["text"]
                if tag_pat.match(top) and loop_pat.match(bot):
                    pre_normalized = f"{top}-{bot}"
                    confidence = 0.95
                elif full_pat.match(top):
                    pre_normalized = top if "-" in top else top
                    confidence = 0.9
            elif len(inside) == 1 and full_pat.match(inside[0]["text"]):
                pre_normalized = inside[0]["text"]
                confidence = 0.85

            xs = [t["x"] for t in inside] + [b["x"] - b["r"], b["x"] + b["r"]]
            ys = [t["y"] for t in inside] + [b["y"] - b["r"], b["y"] + b["r"]]

            balloon_clusters.append({
                "id": f"p1c{idx}",
                "texts": [t["text"] for t in inside],
                "bbox": [round(min(xs), 2), round(min(ys), 2),
                         round(max(xs), 2), round(max(ys), 2)],
                "ocr_confidence": confidence,
                "has_korean": any(any('\uac00' <= ch <= '\ud7a3' for ch in t["text"])
                                  for t in inside),
                "layer": instrument_layer,
                "balloon_type": b["balloon_type"],
                "balloon_radius": round(b["r"], 3),
                "pre_normalized_tag": pre_normalized,
            })

        # 5. balloon에 못 잡힌 나머지 텍스트는 레이어별로 묶음
        remaining = [
            t for t in all_texts
            if (t["text"], round(t["x"], 2), round(t["y"], 2)) not in used_text_keys
        ]

        from collections import defaultdict
        from sklearn.cluster import DBSCAN
        import numpy as np

        layer_groups = defaultdict(list)
        for t in remaining:
            layer_groups[t["layer"]].append(t)

        non_balloon_clusters = []
        cluster_idx = len(balloon_clusters)

        for layer, items in layer_groups.items():
            if len(items) == 0:
                continue

            if any(skip in layer.upper() for skip in
                   ["REV", "BOM", "DIMENSION", "치수"]):
                continue

            if len(items) == 1:
                t = items[0]
                non_balloon_clusters.append({
                    "id": f"p1c{cluster_idx}",
                    "texts": [t["text"]],
                    "bbox": [round(t["x"], 2), round(t["y"], 2),
                             round(t["x"] + len(t["text"]) * t["height"] * 0.6, 2),
                             round(t["y"] + t["height"], 2)],
                    "ocr_confidence": 1.0,
                    "has_korean": any('\uac00' <= ch <= '\ud7a3' for ch in t["text"]),
                    "layer": layer,
                    "balloon_type": None,
                    "balloon_radius": None,
                    "pre_normalized_tag": (t["text"]
                                           if full_pat.match(t["text"]) else None),
                })
                cluster_idx += 1
                continue

            avg_h = sum(t["height"] for t in items) / len(items)
            eps = max(avg_h * 5, 5.0)
            centers = np.array([[t["x"], t["y"]] for t in items])
            labels = DBSCAN(eps=eps, min_samples=1).fit_predict(centers)

            for lbl in sorted(set(labels)):
                members = [items[i] for i, lb in enumerate(labels) if lb == lbl]
                xs = [m["x"] for m in members]
                ys = [m["y"] for m in members]
                pre_norm = None
                for m in members:
                    if full_pat.match(m["text"]):
                        pre_norm = m["text"]
                        break

                non_balloon_clusters.append({
                    "id": f"p1c{cluster_idx}",
                    "texts": [m["text"] for m in members],
                    "bbox": [round(min(xs), 2), round(min(ys), 2),
                             round(max(xs) + max(m["height"] * len(m["text"]) * 0.6
                                                 for m in members), 2),
                             round(max(ys) + max(m["height"] for m in members), 2)],
                    "ocr_confidence": 1.0,
                    "has_korean": any(any('\uac00' <= ch <= '\ud7a3' for ch in m["text"])
                                      for m in members),
                    "layer": layer,
                    "balloon_type": None,
                    "balloon_radius": None,
                    "pre_normalized_tag": pre_norm,
                })
                cluster_idx += 1

        # 6. 도면번호 추출 (TITLE 레이어 또는 파일명에서)
        drawing_no = None
        for t in all_texts:
            if t["layer"].upper() == "TITLE":
                if _re.match(r'^[A-Z]+-?\d+(-\d+)*$', t["text"]):
                    drawing_no = t["text"]
                    break
        if not drawing_no:
            base = os.path.splitext(os.path.basename(dxf_path))[0]
            drawing_no = base.upper()

        all_clusters = balloon_clusters + non_balloon_clusters

        return json.dumps({
            "success": True,
            "source": "dxf",
            "source_path": dxf_path,
            "total_pages": 1,
            "processed_pages": [1],
            "pages": [{
                "page": 1,
                "image_size": None,
                "drawing_no": drawing_no,
                "clusters": all_clusters,
            }],
            "stats": {
                "total_texts": len(all_texts),
                "balloons_detected": len(balloons),
                "balloons_with_tag": sum(1 for c in balloon_clusters
                                         if c["pre_normalized_tag"]),
                "dcs_balloons": sum(1 for c in balloon_clusters
                                    if c["balloon_type"] == "DCS"),
                "field_balloons": sum(1 for c in balloon_clusters
                                      if c["balloon_type"] == "FIELD"),
                "non_balloon_clusters": len(non_balloon_clusters),
            },
        }, ensure_ascii=False)

    except Exception as e:
        import traceback
        return json.dumps({
            "success": False,
            "error": f"DXF 파싱 실패: {e}",
            "trace": traceback.format_exc()[-500:],
        }, ensure_ascii=False)

완료 여부: ❌ (코드 추가 필요)

---

단위 5: parse_pid_dwg 도구 추가 ✅

파일: mcp-server/server.py
위치: parse_pid_dxf 바로 아래

추가 코드:

# ── P&ID DWG 파서 (ODA Converter로 DXF 변환 후 처리) ──────────────────────────

@mcp.tool()
def parse_pid_dwg(
    dwg_path: str,
    instrument_layer: str = "INSTRUMENT",
) -> str:
    """DWG 파일을 ODA File Converter로 DXF로 변환 후 parse_pid_dxf 호출.

    ODA File Converter가 시스템에 설치되어 있어야 함 (ODA_CONVERTER_PATH 상수).
    임시 디렉토리에 변환된 DXF 생성 후 처리, 완료 시 자동 삭제.

    Args:
        dwg_path:         DWG 파일 절대 경로
        instrument_layer: ISA balloon 레이어명 (parse_pid_dxf와 동일)

    Returns:
        parse_pid_dxf 결과와 동일하나 source="dwg", source_path는 원본 DWG.
    """
    try:
        import os
        import subprocess
        import tempfile
        import shutil

        if not os.path.exists(dwg_path):
            return json.dumps(
                {"success": False, "error": f"파일을 찾을 수 없습니다: {dwg_path}"},
                ensure_ascii=False,
            )

        if not os.path.exists(ODA_CONVERTER_PATH):
            return json.dumps({
                "success": False,
                "error": (f"ODA File Converter가 {ODA_CONVERTER_PATH}에 없습니다. "
                          f"https://www.opendesign.com/guestfiles/oda_file_converter "
                          f"에서 다운로드 후 설치하세요."),
            }, ensure_ascii=False)

        with tempfile.TemporaryDirectory(prefix="pid_dwg_") as in_dir, \
             tempfile.TemporaryDirectory(prefix="pid_dxf_") as out_dir:

            in_file = os.path.join(in_dir, os.path.basename(dwg_path))
            shutil.copy2(dwg_path, in_file)

            cmd = [
                ODA_CONVERTER_PATH,
                in_dir, out_dir,
                "ACAD2018",
                "DXF",
                "0",
                "1",
                "*.DWG",
            ]

            result = subprocess.run(
                cmd, capture_output=True, text=True, timeout=120,
            )

            if result.returncode != 0:
                return json.dumps({
                    "success": False,
                    "error": (f"DWG → DXF 변환 실패 (returncode={result.returncode}). "
                              f"stderr: {result.stderr[:500]}"),
                }, ensure_ascii=False)

            base_name = os.path.splitext(os.path.basename(dwg_path))[0]
            dxf_path = os.path.join(out_dir, base_name + ".dxf")
            if not os.path.exists(dxf_path):
                candidates = [f for f in os.listdir(out_dir) if f.endswith(".dxf")]
                if candidates:
                    dxf_path = os.path.join(out_dir, candidates[0])
                else:
                    return json.dumps({
                        "success": False,
                        "error": "DWG → DXF 변환은 성공했으나 출력 DXF를 찾을 수 없음",
                    }, ensure_ascii=False)

            dxf_result_json = parse_pid_dxf(dxf_path, instrument_layer)
            dxf_result = json.loads(dxf_result_json)

            dxf_result["source"] = "dwg"
            dxf_result["source_path"] = dwg_path
            dxf_result["converted_via"] = "ODA File Converter"

            return json.dumps(dxf_result, ensure_ascii=False)

    except subprocess.TimeoutExpired:
        return json.dumps(
            {"success": False, "error": "DWG → DXF 변환 시간 초과 (120초)"},
            ensure_ascii=False,
        )
    except Exception as e:
        import traceback
        return json.dumps({
            "success": False,
            "error": f"DWG 파싱 실패: {e}",
            "trace": traceback.format_exc()[-500:],
        }, ensure_ascii=False)

완료 여부: ❌ (코드 추가 필요)

---

단위 6: parse_pid_pdf 도구 추가 ✅

파일: mcp-server/server.py
위치: parse_pid_dwg 바로 아래

추가 코드:

# ── P&ID PDF 파서 (OCR 기반 fallback) ─────────────────────────────────────────

@mcp.tool()
def parse_pid_pdf(
    pdf_path: str,
    dpi: int = 300,
    cluster_eps: float = 50.0,
    page_range: str = "",
    min_confidence: float = 0.5,
) -> str:
    """P&ID PDF를 OCR로 파싱. SHX path 변환된 PDF 대응 (DXF/DWG 없을 때 fallback).

    출력 스키마는 parse_pid_dxf와 통합. 단 layer/balloon_type/pre_normalized_tag는 null.

    Args:
        pdf_path, dpi, cluster_eps, page_range, min_confidence: 기존과 동일
    """
    try:
        import fitz
        import numpy as np
        from PIL import Image
        from sklearn.cluster import DBSCAN
        import os
        import io

        if not os.path.exists(pdf_path):
            return json.dumps(
                {"success": False, "error": f"파일을 찾을 수 없습니다: {pdf_path}"},
                ensure_ascii=False,
            )

        doc = fitz.open(pdf_path)
        total_pages = len(doc)
        target_pages = _parse_page_range(page_range, total_pages)

        ocr = _ocr()
        pages_out = []
        processed = []

        for page_idx in target_pages:
            page = doc[page_idx]

            zoom = dpi / 72.0
            mat = fitz.Matrix(zoom, zoom)
            pix = page.get_pixmap(matrix=mat, alpha=False)
            img = Image.open(io.BytesIO(pix.tobytes("png"))).convert("RGB")
            img_np = np.array(img)
            img_w, img_h = img.size

            ocr_result = ocr.ocr(img_np, cls=True)
            if ocr_result and isinstance(ocr_result[0], list):
                ocr_items = ocr_result[0] or []
            else:
                ocr_items = ocr_result or []

            boxes = []
            for item in ocr_items:
                if not item or len(item) < 2:
                    continue
                box_pts, (text, conf) = item[0], item[1]
                text = text.strip()
                if not text or conf < min_confidence:
                    continue
                xs = [p[0] for p in box_pts]
                ys = [p[1] for p in box_pts]
                bbox = [min(xs), min(ys), max(xs), max(ys)]
                has_korean = any('\uac00' <= ch <= '\ud7a3' for ch in text)
                boxes.append({
                    "text": text,
                    "bbox": [round(c, 1) for c in bbox],
                    "conf": round(float(conf), 3),
                    "has_korean": has_korean,
                })

            if not boxes:
                pages_out.append({
                    "page": page_idx + 1,
                    "image_size": [img_w, img_h],
                    "drawing_no": None,
                    "clusters": [],
                })
                processed.append(page_idx + 1)
                continue

            centers = np.array([
                [(b["bbox"][0] + b["bbox"][2]) / 2,
                 (b["bbox"][1] + b["bbox"][3]) / 2]
                for b in boxes
            ])
            eps_scaled = cluster_eps * (dpi / 300.0)
            labels = DBSCAN(eps=eps_scaled, min_samples=1).fit_predict(centers)

            clusters_map = {}
            for box, lbl in zip(boxes, labels):
                clusters_map.setdefault(int(lbl), []).append(box)

            cluster_list = []
            for lbl, members in sorted(clusters_map.items()):
                xs = ([m["bbox"][0] for m in members]
                      + [m["bbox"][2] for m in members])
                ys = ([m["bbox"][1] for m in members]
                      + [m["bbox"][3] for m in members])
                avg_conf = sum(m["conf"] for m in members) / len(members)
                cluster_list.append({
                    "id": f"p{page_idx + 1}c{lbl}",
                    "texts": [m["text"] for m in members],
                    "bbox": [round(min(xs), 1), round(min(ys), 1),
                             round(max(xs), 1), round(max(ys), 1)],
                    "ocr_confidence": round(avg_conf, 3),
                    "has_korean": any(m["has_korean"] for m in members),
                    "layer": None,
                    "balloon_type": None,
                    "balloon_radius": None,
                    "pre_normalized_tag": None,
                })

            pages_out.append({
                "page": page_idx + 1,
                "image_size": [img_w, img_h],
                "drawing_no": None,
                "clusters": cluster_list,
            })
            processed.append(page_idx + 1)

        doc.close()
        return json.dumps({
            "success": True,
            "source": "pdf",
            "source_path": pdf_path,
            "total_pages": total_pages,
            "processed_pages": processed,
            "pages": pages_out,
        }, ensure_ascii=False)

    except Exception as e:
        import traceback
        return json.dumps({
            "success": False,
            "error": f"PDF 파싱 실패: {e}",
            "trace": traceback.format_exc()[-500:],
        }, ensure_ascii=False)

완료 여부: ❌ (코드 추가 필요)

---

단위 7: parse_pid_drawing 통합 디스패처 추가 ✅

파일: mcp-server/server.py
위치: parse_pid_pdf 바로 아래

추가 코드:

# ── 통합 디스패처: 확장자로 자동 라우팅 ───────────────────────────────────────

@mcp.tool()
def parse_pid_drawing(file_path: str, **kwargs) -> str:
    """파일 확장자로 적절한 파서를 자동 선택.

    .dxf → parse_pid_dxf  (가장 정확, 추천)
    .dwg → parse_pid_dwg  (DXF로 변환 후 처리)
    .pdf → parse_pid_pdf  (OCR fallback, 정확도 낮음)

    Args:
        file_path: 도면 파일 경로
        **kwargs:  하위 파서로 전달되는 옵션 (instrument_layer, dpi, page_range 등)
    """
    import os
    ext = os.path.splitext(file_path)[1].lower()

    if ext == ".dxf":
        return parse_pid_dxf(
            file_path,
            instrument_layer=kwargs.get("instrument_layer", "INSTRUMENT"),
            balloon_text_tolerance=kwargs.get("balloon_text_tolerance", 1.5),
        )
    elif ext == ".dwg":
        return parse_pid_dwg(
            file_path,
            instrument_layer=kwargs.get("instrument_layer", "INSTRUMENT"),
        )
    elif ext == ".pdf":
        return parse_pid_pdf(
            file_path,
            dpi=kwargs.get("dpi", 300),
            cluster_eps=kwargs.get("cluster_eps", 50.0),
            page_range=kwargs.get("page_range", ""),
            min_confidence=kwargs.get("min_confidence", 0.5),
        )
    else:
        return json.dumps({
            "success": False,
            "error": f"지원하지 않는 확장자: {ext} (지원: .dxf, .dwg, .pdf)",
        }, ensure_ascii=False)

완료 여부: ❌ (코드 추가 필요)

---

단위 8: extract_pid_tags 확장 (clusters 모드) ✅

파일: mcp-server/server.py
위치: 기존 extract_pid_tags 함수 교체

기존 코드 대체:

@mcp.tool()
def extract_pid_tags(text: str, source_type: str) -> str:
    """P&ID 도면에서 태그 정보를 추출합니다.

    Args:
        text:        parse_pid_drawing/dxf/dwg/pdf의 JSON 결과(문자열) 또는 raw text
        source_type: 'clusters' (권장) | 'pdf' | 'dxf' (레거시)
    """
    if source_type == "clusters":
        try:
            import json as json_module
            parsed = json_module.loads(text)
            if parsed.get("source") in ("dxf", "dwg"):
                return _extract_from_dxf_clusters(parsed)
        except Exception:
            pass

        system = (
            "You are a P&ID expert. The input is clusters from a P&ID drawing.\n"
            "Each cluster represents ONE equipment/instrument with associated nearby texts.\n"
            "Texts may include both English (tags, codes) and Korean (descriptions).\n"
            "\n"
            "If a cluster has 'pre_normalized_tag', use it directly as tagNo.\n"
            "Otherwise, normalize ISA-5.1 two-line tags:\n"
            "    Top:    function code (FIC, PT, LT, ...)\n"
            "    Bottom: loop number (10101, 6113, 201A)\n"
            "    Merge: 'FIC-10101'\n"
            "\n"
            "If cluster has 'balloon_type', use it as instrumentLocation:\n"
            "    'DCS'   → DCS point\n"
            "    'FIELD' → field instrument\n"
            "    null    → unknown\n"
            "\n"
            "Return ONLY a JSON array:\n"
            '[{"tagNo":"FIC-10101","equipmentName":"...","equipmentNameKo":"...",\n'
            '  "instrumentType":"FIC","instrumentLocation":"DCS","layer":"INSTRUMENT",\n'
            '  "lineNumber":"L-101","pidDrawingNo":"P-9100","clusterId":"p1c3",\n'
            '  "confidence":0.95},...]\n'
            "\n"
            "OCR 오인식 보정 (PDF source인 경우):\n"
            "    'O' ↔ '0',  'I' ↔ '1' ↔ 'l',  'S' ↔ '5',  'B' ↔ '8'\n"
            "    보정 시 confidence 0.6 이하.\n"
            "\n"
            "- Skip clusters with only descriptive text and no tag pattern.\n"
            "- Skip pipe line numbers (4\"-CS-1234-A1A) unless clearly an instrument.\n"
            "- Skip title block / drawing metadata (DWG NO., SCALE, dates, names).\n"
            "- If no tags found, return [].\n"
            "- Do NOT include explanation, only JSON array."
        )
        truncated = text[:30000] if len(text) > 30000 else text
        user_content = f"Drawing clusters:\n{truncated}"
    else:
        system = (
            "You are a P&ID (Piping and Instrumentation Diagram) expert.\n"
            "Extract instrument and equipment tags from the provided text.\n"
            "Return ONLY a JSON array of objects with the following structure:\n"
            '[{"tagNo":"FT-101","equipmentName":"Flow Transmitter",'
            '"instrumentType":"FT","lineNumber":"L-101","pidDrawingNo":"P&ID-001",'
            '"confidence":0.95},...]\n'
            "Rules:\n"
            "- tagNo: Standard tag format\n"
            "- instrumentType: First 2-4 letters\n"
            "- equipmentName, lineNumber, pidDrawingNo: null if not present\n"
            "- confidence: 0.0~1.0\n"
            "- Empty array if no tags."
        )
        truncated = text[:12000] if len(text) > 12000 else text
        user_content = f"Source: {source_type}\n\nText:\n{truncated}"

    try:
        resp = _llm().chat.completions.create(
            model=VLLM_MODEL,
            messages=[
                {"role": "system", "content": system},
                {"role": "user", "content": user_content},
            ],
            max_tokens=4096,
            temperature=0.1,
        )

        raw = (resp.choices[0].message.content or "").strip()
        if raw.startswith("```"):
            lines = raw.splitlines()
            raw = "\n".join(
                lines[1:-1] if lines and lines[-1].strip() == "```" else lines[1:]
            ).strip()

        import re
        match = re.search(r'\[.*\]', raw, re.DOTALL)
        if match:
            raw = match.group(0)

        import json as json_module
        data = json_module.loads(raw)

        return json_module.dumps({
            "success": True,
            "count": len(data),
            "tags": data,
        }, ensure_ascii=False, indent=2)

    except Exception as e:
        return json.dumps(
            {"success": False, "error": f"P&ID 태그 추출 실패: {e}"},
            ensure_ascii=False,
        )


def _extract_from_dxf_clusters(parsed: dict) -> str:
    """DXF/DWG cluster에서 pre_normalized_tag를 활용해 LLM 호출 없이 태그 추출."""
    import json as json_module
    import re as _re

    tags = []
    for page in parsed.get("pages", []):
        drawing_no = page.get("drawing_no")
        for c in page.get("clusters", []):
            tag_no = c.get("pre_normalized_tag")
            if not tag_no:
                continue

            m = _re.match(r'^([A-Z]+)-?(\d+[A-Z]?)$', tag_no)
            if not m:
                continue
            inst_type = m.group(1)

            ko_desc = None
            for t in c.get("texts", []):
                if any('\uac00' <= ch <= '\ud7a3' for ch in t):
                    ko_desc = t
                    break

            tags.append({
                "tagNo": tag_no,
                "equipmentName": None,
                "equipmentNameKo": ko_desc,
                "instrumentType": inst_type,
                "instrumentLocation": c.get("balloon_type"),
                "layer": c.get("layer"),
                "lineNumber": None,
                "pidDrawingNo": drawing_no,
                "clusterId": c.get("id"),
                "confidence": c.get("ocr_confidence", 0.95),
            })

    return json_module.dumps({
        "success": True,
        "count": len(tags),
        "tags": tags,
        "method": "coordinate_matching",
    }, ensure_ascii=False, indent=2)

완료 여부: ❌ (코드 교체 필요)

---

단위 9: 빌드 검증 및 테스트 ✅

작업 내용:

1. dotnet build src/Web/ExperionCrawler.csproj --no-restore -v q 실행
2. MCP 서버 재시작: python mcp-server/server.py --http
3. 단위 테스트 실행: python test_parse_pid.py

테스트 스크립트 (test_parse_pid.py):

"""parse_pid_drawing 통합 테스트 (DXF/DWG/PDF)."""
import json
import sys
import time
sys.path.insert(0, ".")
from server import (
    parse_pid_drawing, parse_pid_dxf, parse_pid_dwg, parse_pid_pdf,
    extract_pid_tags,
)

DXF_PATH = "test_data/sample.dxf"

print("[DXF] 파싱 시작...")
t0 = time.time()
dxf_json = parse_pid_dxf(DXF_PATH, instrument_layer="INSTRUMENT")
elapsed = time.time() - t0
result = json.loads(dxf_json)

assert result["success"], f"DXF 파싱 실패: {result.get('error')}"
print(f"[OK] {elapsed:.2f}초 (OCR 대비 100배+ 빠름)")
print(f"      통계: {result['stats']}")

stats = result["stats"]
assert stats["balloons_detected"] > 50, "balloon이 너무 적음"
match_rate = stats["balloons_with_tag"] / stats["balloons_detected"]
assert match_rate > 0.85, f"매칭률 부족: {match_rate:.1%}"
print(f"      ISA balloon 매칭률: {match_rate:.1%}")

t0 = time.time()
tags_json = extract_pid_tags(dxf_json, "clusters")
print(f"[OK] 태그 추출 {time.time() - t0:.2f}초")
tags = json.loads(tags_json)
assert tags["success"]
assert tags.get("method") == "coordinate_matching", "DXF는 LLM 미사용 경로여야 함"
print(f"      추출 태그: {tags['count']}개")
for t in tags["tags"][:5]:
    print(f"      - {t['tagNo']:<15} ({t['instrumentType']}, "
          f"loc={t['instrumentLocation']}, conf={t['confidence']})")

print("\n✅ 모든 테스트 통과")

완료 여부: ❌ (테스트 필요)

---

📊 완료 체크리스트

단위	내용	상태	검증 방법
1	ODA File Converter 설치	❌	which ODAFileConverter
2	Python 의존성 설치	❌	pip list | grep ezdxf
3	공통 헬퍼 추가	❌	server.py 확인
4	parse_pid_dxf 추가	❌	server.py 확인
5	parse_pid_dwg 추가	❌	server.py 확인
6	parse_pid_pdf 추가	❌	server.py 확인
7	parse_pid_drawing 추가	❌	server.py 확인
8	extract_pid_tags 확장	❌	server.py 확인
9	빌드 검증	❌	dotnet build 성공

---

📝 참고 사항

DXF 처리 시 주의사항

• 레이어명은 회사마다 다름: INSTRUMENT / INST / INSTR / I-1
• MTEXT 폰트 escape 코드 청소 필요
• 블록 참조(ATTRIB)는 현재 익명 블록 무시

성능 지표 (사용자 제공 p-9100.dxf)

• 텍스트 엔티티: 3,925개
• ISA balloon: 215개
• 좌표 매칭 태그 정규화: 197개 (91.6%)
• 처리 시간: <1초 (LLM 미사용)

---

작성일: 2026-05-01
버전: v1 (초안)