CD
752d851f9f
- GeminiCaller: cmd/c 제거, 인자 분리, Semaphore(4) 동시성 제어, GeminiCallError - TaskPipeline: asyncio.gather 병렬 코딩, batch_review 1회, summarize 총평 - FileApplier: Coder 출력 파싱 → 실제 파일 적용 (경로 보안 체크) - Discord Bot: on_message 자동채팅, 의도분류(chat/task/clarify), 대화기억(10메시지) - Prompts: router.md (의도분류), summarizer.md (총평) - Workflows: agent_chat 환경 경로 업데이트
118 lines
3.3 KiB
Python
118 lines
3.3 KiB
Python
"""File Applier — Coder 출력을 실제 파일에 적용.
|
|
|
|
Coder가 출력한 `=== FILE: path === ... === END FILE ===` 블록을
|
|
파싱하여 프로젝트 파일에 실제로 쓰기.
|
|
"""
|
|
|
|
import re
|
|
import logging
|
|
from pathlib import Path
|
|
from dataclasses import dataclass
|
|
|
|
logger = logging.getLogger("variet.applier")
|
|
|
|
|
|
@dataclass
|
|
class FileChange:
|
|
"""파일 변경 단위."""
|
|
path: str # 상대 경로
|
|
content: str # 전체 파일 내용
|
|
is_new: bool # 신규 파일 여부
|
|
|
|
|
|
def parse_code_output(raw: str) -> list[FileChange]:
|
|
"""Coder 출력에서 파일 블록을 추출.
|
|
|
|
지원 형식:
|
|
=== FILE: path/to/file.py ===
|
|
(content)
|
|
=== END FILE ===
|
|
|
|
또는 마크다운 방식:
|
|
```python:path/to/file.py
|
|
(content)
|
|
```
|
|
"""
|
|
changes: list[FileChange] = []
|
|
|
|
# 패턴 1: === FILE: path === ... === END FILE ===
|
|
pattern1 = re.compile(
|
|
r'===\s*FILE:\s*(.+?)\s*===\s*\n(.*?)\n\s*===\s*END\s*FILE\s*===',
|
|
re.DOTALL,
|
|
)
|
|
for match in pattern1.finditer(raw):
|
|
path = match.group(1).strip()
|
|
content = match.group(2)
|
|
changes.append(FileChange(path=path, content=content, is_new=False))
|
|
|
|
# 패턴 2: ```lang:path/to/file.py\n...\n```
|
|
if not changes:
|
|
pattern2 = re.compile(
|
|
r'```\w*:(.+?)\n(.*?)\n```',
|
|
re.DOTALL,
|
|
)
|
|
for match in pattern2.finditer(raw):
|
|
path = match.group(1).strip()
|
|
content = match.group(2)
|
|
changes.append(FileChange(path=path, content=content, is_new=False))
|
|
|
|
return changes
|
|
|
|
|
|
def apply_changes(
|
|
changes: list[FileChange],
|
|
project_path: str | Path,
|
|
dry_run: bool = False,
|
|
) -> list[dict]:
|
|
"""파일 변경사항을 프로젝트에 적용.
|
|
|
|
Args:
|
|
changes: parse_code_output() 결과
|
|
project_path: 프로젝트 루트 경로
|
|
dry_run: True면 실제 파일 쓰기 없이 결과만 반환
|
|
|
|
Returns:
|
|
적용 결과 리스트 [{"path": ..., "action": "created|modified|skipped", "lines": N}]
|
|
"""
|
|
root = Path(project_path).resolve()
|
|
results = []
|
|
|
|
for change in changes:
|
|
# 경로 정규화 + 보안: 프로젝트 밖 경로 차단
|
|
target = (root / change.path).resolve()
|
|
if not str(target).startswith(str(root)):
|
|
logger.warning(f"경로 보안 위반 — 스킵: {change.path}")
|
|
results.append({
|
|
"path": change.path,
|
|
"action": "skipped",
|
|
"reason": "프로젝트 외부 경로",
|
|
})
|
|
continue
|
|
|
|
is_new = not target.exists()
|
|
line_count = len(change.content.splitlines())
|
|
|
|
if dry_run:
|
|
results.append({
|
|
"path": change.path,
|
|
"action": "would_create" if is_new else "would_modify",
|
|
"lines": line_count,
|
|
})
|
|
continue
|
|
|
|
# 디렉토리 생성
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 파일 쓰기
|
|
target.write_text(change.content, encoding="utf-8")
|
|
action = "created" if is_new else "modified"
|
|
logger.info(f"파일 {action}: {change.path} ({line_count}L)")
|
|
|
|
results.append({
|
|
"path": change.path,
|
|
"action": action,
|
|
"lines": line_count,
|
|
})
|
|
|
|
return results
|