guardia-mes/scripts/rag_ingest_mes.py

#!/usr/bin/env python3
"""
GUARDiA MES — 중앙 guardia-rag 색인 훅 (대표 도메인 메타 → mes 컬렉션).

불량 RCA·SPC 해석·예지보전/예측 해석의 근거 검색을 위해, MES 의 대표 도메인
**메타데이터(테이블/컬럼/판정규칙/방법 정의)** 만 중앙 RAG 의 `mes` 컬렉션에 적재한다.
실제 생산/검사 데이터 행·거래처/인사 PII·자격증명은 색인하지 않는다(보안 불변).

사용:
    python scripts/rag_ingest_mes.py                       # 기본(http://127.0.0.1:8020)
    RAG_URL=http://127.0.0.1:8020 python scripts/rag_ingest_mes.py

설계 메모:
- 외부 API 0. 중앙 guardia-rag(온프레미스)만 호출.
- 멱등: doc_id 안정 → 재실행 시 upsert(중앙이 처리). 실패해도 MES 동작에 영향 없음(보조 스크립트).
- PII/자격증명 미색인: 거래처 연락처·담당자명·SSH·비밀번호·IP 는 문서에 포함하지 않는다.
- 수치 결정론 불변: SPC 관리한계·Cp/Cpk·예측치는 MES Java 로직이 산출. 본 문서는 "방법·판정규칙 정의"만
  제공해 AI 가 수치를 재계산하지 않고 해석만 하도록 grounding 한다.
"""
import json
import os
import sys
import urllib.request
import urllib.error

RAG_URL = os.environ.get("RAG_URL", "http://127.0.0.1:8020").rstrip("/")
SOLUTION = "mes"

# 대표 도메인 메타 문서 — 불량 RCA·SPC·예측 해석 근거용.
# (PII/자격증명 컬럼 제외. 판정규칙·방법론·논리지표↔물리컬럼 중심.)
DOCS = [
    {
        "doc_id": "mes_qms_ncr",
        "source": "MES-SCHEMA-QMS",
        "text": (
            "부적합(NCR) 도메인: 검사/공정에서 발견된 부적합 기록. 핵심 필드 - ncr_no(부적합번호), "
            "defect_code(불량코드), lot_no(LOT), item_code(품목), process_code(공정), "
            "equipment_code(설비), qty_defect(불량수량), root_cause(추정원인), status. "
            "불량코드 접두 규칙: DIM=치수, SCR/외관=표면손상, MAT=자재편차. "
            "원인분석은 4M(Man/Machine/Material/Method)·5Why·파레토 관점으로 수행한다."
        ),
    },
    {
        "doc_id": "mes_qms_capa",
        "source": "MES-SCHEMA-QMS",
        "text": (
            "시정조치(CAPA) 도메인: 부적합에 대한 시정·예방 조치. 필드 - capa_no, ncr_no(연결 부적합), "
            "action_type(CORRECTIVE/PREVENTIVE), action_desc(조치내용), owner_role, due_date, status, "
            "effectiveness(유효성 검증결과). 유사 불량의 과거 CAPA 이력은 RCA 근거로 활용한다."
        ),
    },
    {
        "doc_id": "mes_qms_inspection",
        "source": "MES-SCHEMA-QMS",
        "text": (
            "검사(inspection) 도메인: 수입/공정/출하 검사 결과. 필드 - inspection_no, item_code, lot_no, "
            "characteristic(검사항목), measured(실측값), lsl(하한규격), usl(상한규격), target(목표), "
            "judgement(PASS/FAIL/REINSPECT), aql. 판정규칙: measured<lsl 또는 measured>usl 이면 FAIL. "
            "deviation = measured - target. 최종 판정 권한은 사람에게 있으며 AI 는 초안만 제시한다."
        ),
    },
    {
        "doc_id": "mes_method_spc",
        "source": "MES-METHOD-SPC",
        "text": (
            "SPC(통계적공정관리) 방법 정의 — 관리도 X̄-R, p, np, c, u. 런 규칙(Western Electric/Nelson): "
            "RULE1 관리한계 이탈(>UCL 또는 <LCL), RULE2 연속 9점이 중심선 한쪽, RULE3 연속 6점 증가/감소 추세. "
            "공정능력: Cp=(USL-LSL)/(6σ), Cpk=min((USL-μ),(μ-LSL))/(3σ). "
            "★수치(평균·UCL·LCL·Cp/Cpk·이탈점)는 MES 결정론 로직이 계산한다. AI 는 이탈 의미·원인·조치만 해석한다(재계산 금지)."
        ),
    },
    {
        "doc_id": "mes_equipment_oee",
        "source": "MES-SCHEMA-MES",
        "text": (
            "설비/OEE 도메인: 설비 가동·비가동·종합효율. 필드 - equipment_code, availability(가동률 0~1), "
            "performance, quality, oee(=가용성*성능*품질), downtime_count(비가동 횟수), mtbf_hours(평균고장간격), "
            "mttr_hours(평균수리시간). 예지보전 위험 규칙: 가동률 낮고(<0.7) 비가동 잦으며 MTBF 짧을수록(<100h) 고위험. "
            "★위험점수·정비시점은 MES 결정론 로직이 산출하고 AI 는 추세 해석·우선순위만 제시한다."
        ),
    },
    {
        "doc_id": "mes_production_forecast",
        "source": "MES-METHOD-FORECAST",
        "text": (
            "수요/생산 예측 방법 정의 — 생산실적(job)·작업지시·출하 시계열 기반. 베이스라인: 가중 이동평균 + 선형추세 "
            "(예측 = (마지막값 + 기울기*i)*0.6 + 평균*0.4, 음수 클램프). horizon 기본 7일, 최대 90일. "
            "★예측 수치는 MES 결정론 로직이 산출(재현 가능). AI 는 추세 해석·이상신호 강조·생산계획 조정 권고만 한다."
        ),
    },
    {
        "doc_id": "mes_inventory_optimize",
        "source": "MES-METHOD-INVENTORY",
        "text": (
            "재고 최적화 방법 정의 — 안전재고 = Z * 수요표준편차 * sqrt(리드타임), 발주점(ROP) = 평균일수요*리드타임 + 안전재고. "
            "기본 서비스수준 95%(Z=1.65). LOT 만료 위험은 lotserial 의 expiry_date 로 감지. "
            "★수치는 MES 결정론 로직이 산출. AI 는 과부족·만료 위험 해석과 발주 권고만 제시한다."
        ),
    },
    {
        "doc_id": "mes_rag_contract",
        "source": "MES-RAG",
        "text": (
            "MES 최신 AI 기법 적용(중앙 guardia-rag): 대표 2기능 — (1) 불량 RCA+SPC 이상감지: SPC 수치는 결정론, "
            "원인서술은 /rag/agent(tool-use)+/rag/structured(JSON 강제). (2) 예지보전/수요·생산 예측: 수치는 결정론 베이스라인, "
            "해석은 /rag/agent. 기법 토글(rerank·graphrag·hybrid·tool_use·structured·stream)은 mes_setting 의 rag.* 키로 격리 저장. "
            "무거운 기법은 서버 RAM 제약상 기본 off, 미가용 시 결정론 폴백(degraded:true). 외부 API 0(Ollama 온프레미스)."
        ),
    },
]


def ingest():
    payload = {"solution": SOLUTION, "documents": DOCS}
    data = json.dumps(payload).encode("utf-8")
    req = urllib.request.Request(
        f"{RAG_URL}/rag/ingest",
        data=data,
        headers={"Content-Type": "application/json", "X-Solution-Key": SOLUTION},
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=120) as resp:
            body = resp.read().decode("utf-8")
            print(f"[OK] ingested {len(DOCS)} docs into '{SOLUTION}' collection")
            print(f"     response: {body[:300]}")
            return 0
    except urllib.error.HTTPError as e:
        print(f"[WARN] ingest HTTP {e.code} — 중앙 RAG 응답 이상(스크립트는 보조). 본 MES 동작 영향 없음.")
        return 1
    except urllib.error.URLError as e:
        print(f"[WARN] 중앙 guardia-rag 미가용({RAG_URL}) — 색인 보류. 라이브 가동 후 재실행. (사유 요약: {str(e.reason)[:80]})")
        return 1
    except Exception:
        print("[WARN] 색인 중 예기치 못한 오류 — 보류(MES 동작 영향 없음).")
        return 1


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