Files
hyowons 7e7d759b34 auto: 일일 백업 2026-08-04 02:00
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-08-04 02:00:05 +09:00

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)