refactor(#28): IMP-28 L4 _parse_json dedup (4 modules -> src/json_utils)

Consolidate duplicate _parse_json helpers from content_editor.py /
design_director.py / kei_client.py (fuller form) and pipeline.py (simple form)
into shared src/json_utils.parse_json (strict superset). All 18 call-sites
preserved via `parse_json as _parse_json` alias import; no behavior change.

- src/json_utils.py (new): shared helper, fenced/plain-fence/bare-brace patterns
  + list-prefix cleanup fallback.
- tests/test_json_utils.py (new): 9 unit tests pinning parser semantics.
- src/content_editor.py / design_director.py: remove local helper +
  unused `import json` / `import re`.
- src/kei_client.py / pipeline.py: remove local helper; `json` / `re` retained
  (used elsewhere).

Targeted tests 9 passed; full pytest 374 passed (3 pre-existing scripts/
collection errors reproduce on baseline 909bf75, IMP-28 unrelated).
This commit is contained in:
2026-05-20 20:44:19 +09:00
parent 909bf75edc
commit 265d70ed91
6 changed files with 105 additions and 127 deletions

View File

@@ -13,6 +13,7 @@ from typing import Any
import httpx
from src.config import settings
from src.json_utils import parse_json as _parse_json
from src.sse_utils import stream_sse_tokens
logger = logging.getLogger(__name__)
@@ -889,42 +890,6 @@ async def call_kei_overflow_judgment(
return None
def _parse_json(text: str) -> dict[str, Any] | None:
"""텍스트에서 JSON을 추출한다.
Kei API가 마크다운 리스트 접두사(- )를 붙여 응답하는 경우에도 처리.
"""
# 전처리: 각 줄 앞의 마크다운 리스트 접두사(- ) 제거
# Kei API가 JSON을 마크다운 리스트로 감싸서 응답하는 경우 대응
lines = text.split("\n")
cleaned_lines = []
for line in lines:
stripped = line.lstrip()
if stripped.startswith("- "):
cleaned_lines.append(stripped[2:])
elif stripped.startswith("* "):
cleaned_lines.append(stripped[2:])
else:
cleaned_lines.append(stripped)
cleaned = "\n".join(cleaned_lines)
# 원본 + 클린 버전 둘 다 시도
for target in [text, cleaned]:
patterns = [
r"```json\s*(.*?)```",
r"```\s*(.*?)```",
r"(\{.*\})",
]
for pattern in patterns:
match = re.search(pattern, target, re.DOTALL)
if match:
try:
return json.loads(match.group(1).strip())
except json.JSONDecodeError:
continue
return None
async def select_best_candidate(
topic_results: list[dict[str, Any]],
analysis: dict[str, Any],