7e7d759b34
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
82 lines
3.2 KiB
Python
82 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
"""imsg CLI 호출 공용 래퍼.
|
|
|
|
호출부가 stdout 만 JSON 파싱하고 stderr·returncode 를 버리면, FDA(전체 디스크 접근 권한)
|
|
누락 같은 권한 오류가 `Expecting value: line 1 column 1 (char 0)` 파싱 에러로 둔갑한다.
|
|
2026-08-03 골디 오진 사고의 원인 — 실제 원인은 homebrew node 업그레이드로 게이트웨이
|
|
바이너리 경로가 바뀌어 FDA 승인이 무효화된 것이었다. 판정을 한곳에 모아 사유를 그대로 노출한다.
|
|
"""
|
|
import json
|
|
import subprocess
|
|
|
|
# imsg 가 권한 거부 시 내는 문구 (imsg 0.5.0 기준).
|
|
FDA_MARKERS = ("Full Disk Access", "Cannot access Messages database")
|
|
FDA_HINT = (
|
|
"Messages DB 접근 거부 — 전체 디스크 접근 권한(FDA) 누락. "
|
|
"skills/whooing-sync/SKILL.md 의 'Full Disk Access' 섹션 참고"
|
|
)
|
|
|
|
|
|
# `imsg chats` 기본값은 최근 20건이라, 오래 조용한 발신번호(현대카드 SMS·신한은행 등)가
|
|
# 목록에서 빠진다. 그 상태로 chat id 캐시를 다시 만들면 해당 carrier 수집이 조용히 끊긴다.
|
|
# chat.db 전체 대화가 130개 수준이라 넉넉히 잡는다. (2026-08-03 실측으로 발견)
|
|
CHATS_LIMIT = 200
|
|
|
|
|
|
def _reason(stderr: str, stdout: str, fallback: str) -> str:
|
|
"""실패 사유 한 줄. FDA 마커는 stdout/stderr 어느 쪽에 있어도 잡는다."""
|
|
err = (stderr or "").strip()
|
|
out = (stdout or "").strip()
|
|
if any(m in err or m in out for m in FDA_MARKERS):
|
|
return FDA_HINT
|
|
detail = err or out
|
|
if detail:
|
|
return " ".join(detail.split())[:200]
|
|
return fallback
|
|
|
|
|
|
def run_json(args: list, timeout: int) -> "tuple[list, str | None]":
|
|
"""imsg 실행 → (rows, error). 성공 시 (list[dict], None), 실패 시 ([], 사유).
|
|
|
|
결과 0건과 실패를 구분한다 — 둘 다 빈 리스트지만 실패에만 error 가 붙는다.
|
|
"""
|
|
try:
|
|
p = subprocess.run(
|
|
["imsg"] + list(args),
|
|
capture_output=True, text=True, timeout=timeout,
|
|
)
|
|
except FileNotFoundError:
|
|
return [], "imsg CLI가 설치되어 있지 않습니다"
|
|
except subprocess.TimeoutExpired:
|
|
return [], f"타임아웃 ({timeout}s)"
|
|
except Exception as e:
|
|
return [], f"{type(e).__name__}: {e}"
|
|
|
|
out = p.stdout or ""
|
|
if p.returncode != 0:
|
|
return [], _reason(p.stderr, out, f"rc={p.returncode}")
|
|
|
|
if not out.strip():
|
|
# stderr 만 있고 stdout 이 비면 실패. 둘 다 비면 정상적으로 결과 0건.
|
|
if (p.stderr or "").strip():
|
|
return [], _reason(p.stderr, "", "빈 출력")
|
|
return [], None
|
|
|
|
rows = []
|
|
for line in out.splitlines():
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
try:
|
|
rows.append(json.loads(line))
|
|
except json.JSONDecodeError:
|
|
return [], _reason(p.stderr, line, "JSON 파싱 실패")
|
|
return rows, None
|
|
|
|
|
|
def chats(timeout: int) -> "tuple[list, str | None]":
|
|
"""대화 목록 조회 → (rows, error). CHATS_LIMIT 만큼 넓게 훑는다."""
|
|
return run_json(["chats", "--limit", str(CHATS_LIMIT), "--json"], timeout)
|