#!/usr/bin/env python3 """US Equity Kakao Dispatcher (Mac-side, Codex-free). 클로드 예약 작업(셸 없음)이 outputs/pending_kakao_*.md 로 남긴 6줄 요약을 맥의 launchd가 주기적으로 이 스크립트로 실행 → 기존 send_kakao_memo.py 로 카톡 발송, (있으면) 짝꿍 pending_state_*.json 으로 update_dashboard_summary.py 대시보드 갱신, 처리 완료분은 kakao_sent/ 로 이동한다. 설계 원칙: - 기존 전략/스크립트는 건드리지 않는다. 발송/갱신은 정식 스크립트만 호출. - 자격증명은 워크스페이스 루트의 .kakao_env(우선) → macOS Keychain(send_kakao_memo.py 내부) 순. - 중복 발송 방지: 처리분은 즉시 이동, 로그 기록. """ from __future__ import annotations import datetime as dt import glob import json import os import subprocess import sys BASE = os.path.dirname(os.path.abspath(__file__)) # .../outputs ROOT = os.path.dirname(BASE) # 워크스페이스 루트 ENV_FILE = os.path.join(ROOT, ".kakao_env") SENT_DIR = os.path.join(BASE, "kakao_sent") DISPATCH_LOG = os.path.join(BASE, "kakao_dispatch_log.jsonl") SENDER = os.path.join(BASE, "send_kakao_memo.py") DASH = os.path.join(BASE, "update_dashboard_summary.py") def log(rec: dict) -> None: rec["ts"] = dt.datetime.now(dt.timezone.utc).isoformat() try: with open(DISPATCH_LOG, "a", encoding="utf-8") as f: f.write(json.dumps(rec, ensure_ascii=False) + "\n") except Exception: pass def load_env() -> dict: env = dict(os.environ) if os.path.exists(ENV_FILE): with open(ENV_FILE, encoding="utf-8") as f: for line in f: line = line.strip() if not line or line.startswith("#") or "=" not in line: continue k, v = line.split("=", 1) if v: env[k] = v return env def title_from_name(path: str) -> str: name = os.path.basename(path) block = name[len("pending_kakao_"):-len(".md")] if name.startswith("pending_kakao_") else name return f"US Equity - {block}" VALID_MODES = {"RECOVERY", "DEFENSE", "ATTACK", "TACTICAL", "CASH", "AH-URGENT", "WEEKEND-RISK"} def parse_six_line(text: str) -> dict: """6줄 요약([상황]/계좌/시장/행동/금지/다음)에서 대시보드 필드 추출.""" out = {"mode": "TACTICAL", "market_status": "", "verdict": "", "next_check": ""} for raw in text.splitlines(): line = raw.strip() if line.startswith("[상황]"): tok = line.replace("[상황]", "").strip().split() if tok and tok[0].upper() in VALID_MODES: out["mode"] = tok[0].upper() elif line.startswith("시장:"): out["market_status"] = line.split(":", 1)[1].strip() elif line.startswith("행동:"): out["verdict"] = line.split(":", 1)[1].strip() elif line.startswith("다음:"): out["next_check"] = line.split(":", 1)[1].strip() return out def main() -> int: os.makedirs(SENT_DIR, exist_ok=True) env = load_env() # 0a) 분석이 산출한 유동 목표(proposed_targets.json) 반영 applier = os.path.join(BASE, "apply_targets.py") if os.path.exists(applier): try: ar = subprocess.run([sys.executable, applier], capture_output=True, text=True, env=env, timeout=60) if ar.stdout.strip() or ar.returncode != 0: log({"event": "apply_targets", "ok": ar.returncode == 0, "out": ar.stdout.strip()[:200], "err": ar.stderr.strip()[:200]}) except Exception as e: log({"event": "apply_targets_error", "err": str(e)[:200]}) # 0b) 보유종목 현재가 주기 갱신 (매매 변화 없어도 시세 반영) — Yahoo/yfinance 소스 refresher = os.path.join(BASE, "refresh_prices_yf.py") if os.path.exists(refresher): try: pr = subprocess.run([sys.executable, refresher], capture_output=True, text=True, env=env, timeout=90) log({"event": "price_refresh", "ok": pr.returncode == 0, "out": pr.stdout.strip()[:200], "err": pr.stderr.strip()[:200]}) except Exception as e: log({"event": "price_refresh_error", "err": str(e)[:200]}) if not (env.get("KAKAO_REST_API_KEY") and env.get("KAKAO_REFRESH_TOKEN")) and not env.get("KAKAO_ACCESS_TOKEN"): log({"event": "skip", "reason": "no_credentials"}) print("자격증명 없음(.kakao_env 또는 Keychain 확인)") return 0 pending = sorted(glob.glob(os.path.join(BASE, "pending_kakao_*.md"))) if not pending: return 0 for path in pending: try: with open(path, encoding="utf-8") as f: text = f.read().strip() if not text: os.remove(path) continue title = title_from_name(path) # 1) 카톡 발송 (정식 스크립트) res = subprocess.run( [sys.executable, SENDER, "--title", title, "--button-title", "대시보드 열기"], input=text, capture_output=True, text=True, env=env, timeout=60, ) ok = res.returncode == 0 log({"event": "kakao_send", "file": os.path.basename(path), "ok": ok, "out": res.stdout.strip()[:200], "err": res.stderr.strip()[:200]}) # 2) 대시보드 갱신: 짝꿍 state json 우선, 없으면 6줄 요약 파싱 if ok and os.path.exists(DASH): state_path = path.replace("pending_kakao_", "pending_state_").replace(".md", ".json") st = parse_six_line(text) if os.path.exists(state_path): try: st.update({k: v for k, v in json.load(open(state_path, encoding="utf-8")).items() if v}) except Exception: pass try: dargs = [sys.executable, DASH, "--mode", str(st.get("mode") or "TACTICAL"), "--market-status", str(st.get("market_status", "")), "--verdict", str(st.get("verdict", "")), "--next-check", str(st.get("next_check", ""))] dres = subprocess.run(dargs, input=text, capture_output=True, text=True, env=env, timeout=60) log({"event": "dashboard_update", "file": os.path.basename(path), "ok": dres.returncode == 0, "err": dres.stderr.strip()[:200]}) if os.path.exists(state_path): os.replace(state_path, os.path.join(SENT_DIR, os.path.basename(state_path))) except Exception as e: log({"event": "dashboard_error", "file": os.path.basename(path), "err": str(e)[:200]}) # 3) 처리분 이동 (성공 시에만; 실패는 다음 주기 재시도) if ok: stamp = dt.datetime.now().strftime("%Y%m%d_%H%M%S") os.replace(path, os.path.join(SENT_DIR, f"{stamp}_{os.path.basename(path)}")) except Exception as e: log({"event": "error", "file": os.path.basename(path), "err": str(e)[:300]}) return 0 if __name__ == "__main__": raise SystemExit(main())