debug(bot): classify_and_route 상세 로그 추가 + 파이프라인 검증 스크립트

_test_pipeline.py

@@ -0,0 +1,148 @@
+"""전수 검증: unified prompt → JSON parse → NC handler."""
+
+import asyncio
+import io
+import json
+import sys
+sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
+sys.path.insert(0, ".")
+
+
+# ── Step 1: _parse_unified_response 검증 ──
+
+def _parse_unified_response(raw: str) -> dict:
+    import re
+    m = re.search(r"```json\s*\n(.+?)```", raw, re.DOTALL)
+    if m:
+        try:
+            return json.loads(m.group(1))
+        except json.JSONDecodeError:
+            pass
+
+    start = raw.find("{")
+    if start != -1:
+        depth = 0
+        for i in range(start, len(raw)):
+            if raw[i] == "{":
+                depth += 1
+            elif raw[i] == "}":
+                depth -= 1
+                if depth == 0:
+                    try:
+                        return json.loads(raw[start:i + 1])
+                    except json.JSONDecodeError as e:
+                        print(f"  JSON decode error: {e}")
+                        print(f"  Attempted: {raw[start:i+1][:200]}")
+                    break
+
+    return {"mode": "chat", "response": raw}
+
+
+# 테스트 케이스들
+test_cases = [
+    # (1) 일반 JSON
+    '{\n  "mode": "nextcloud",\n  "tool": "files",\n  "op": "list",\n  "params": {\n    "path": "/"\n  },\n  "summary": "test"\n}',
+    # (2) 끝에 ``` 붙은 경우
+    '{\n  "mode": "nextcloud",\n  "tool": "files",\n  "op": "list",\n  "params": {\n    "path": "/"\n  },\n  "summary": "test"\n}\n```',
+    # (3) ```json 블록
+    '```json\n{\n  "mode": "chat",\n  "response": "안녕하세요"\n}\n```',
+    # (4) 텍스트 + JSON
+    'Here is the result:\n{\n  "mode": "nextcloud",\n  "tool": "mail",\n  "op": "unread",\n  "params": {"limit": 5},\n  "summary": "test"\n}',
+    # (5) 빈 문자열
+    '',
+    # (6) mode 없는 JSON
+    '{"error": "something"}',
+]
+
+print("=== Step 1: JSON 파서 테스트 ===\n")
+for i, tc in enumerate(test_cases, 1):
+    result = _parse_unified_response(tc)
+    mode = result.get("mode", "?")
+    status = "✅" if mode != "chat" or "response" in result else "❌"
+    print(f"  [{i}] {status} mode={mode} keys={list(result.keys())}")
+    if mode == "nextcloud":
+        print(f"      tool={result.get('tool')} op={result.get('op')} params={result.get('params')}")
+
+
+# ── Step 2: NC handler 검증 (실제 API 호출) ──
+
+async def test_nc_handler():
+    print("\n=== Step 2: NC handler 메서드 검증 ===\n")
+
+    from handlers.nc_handler import NCHandler
+    handler = NCHandler()
+
+    # files.list_dir
+    print("  [1] files.list_dir('')...")
+    try:
+        files = await handler.files.list_dir("")
+        print(f"  ✅ {len(files)}건: {', '.join(f.name for f in files[:5])}")
+    except Exception as e:
+        print(f"  ❌ {e}")
+
+    # files.search
+    print("  [2] files.search('pdf')...")
+    try:
+        files = await handler.files.search("pdf")
+        print(f"  ✅ {len(files)}건: {', '.join(f.name for f in files[:5])}")
+    except Exception as e:
+        print(f"  ❌ {e}")
+
+    # calendar.get_today (method exists?)
+    print("  [3] calendar 메서드 확인...")
+    cal = handler.calendar
+    has_today = hasattr(cal, "get_today")
+    has_week = hasattr(cal, "get_week")
+    has_events = hasattr(cal, "get_events")
+    print(f"      get_today={has_today} get_week={has_week} get_events={has_events}")
+
+    # mail.get_unread
+    print("  [4] mail.get_unread(3)...")
+    try:
+        msgs = await handler.mail.get_unread(3)
+        print(f"  ✅ {len(msgs)}건")
+    except Exception as e:
+        print(f"  ❌ {e}")
+
+
+# ── Step 3: NC handler.handle() 모의 호출 검증 ──
+
+async def test_handle_dispatch():
+    print("\n=== Step 3: handle() 디스패치 검증 ===\n")
+
+    from handlers.nc_handler import NCHandler
+    handler = NCHandler()
+
+    # handle() 내부 _handle_files 호출 경로 확인
+    action = {"mode": "nextcloud", "tool": "files", "op": "list", "params": {"path": ""}}
+
+    class FakeChannel:
+        """Discord channel 모의 객체."""
+        sent = []
+
+        async def send(self, content=None, embed=None):
+            if embed:
+                self.sent.append(f"[EMBED] title={embed.title}, desc_len={len(embed.description or '')}")
+            elif content:
+                self.sent.append(f"[TEXT] {content[:100]}")
+            print(f"      → send: {self.sent[-1]}")
+
+    ch = FakeChannel()
+    try:
+        await handler.handle(action, ch)
+        if ch.sent:
+            print(f"  ✅ {len(ch.sent)}건 전송됨")
+        else:
+            print(f"  ❌ 전송 없음")
+    except Exception as e:
+        print(f"  ❌ handle() 오류: {e}")
+        import traceback
+        traceback.print_exc()
+
+
+async def main():
+    await test_nc_handler()
+    await test_handle_dispatch()
+
+
+asyncio.run(main())