#!/usr/bin/env python3 """Send a short text memo to the authenticated user's KakaoTalk My Chatroom. Required environment: - KAKAO_REST_API_KEY - KAKAO_REFRESH_TOKEN Optional environment: - KAKAO_CLIENT_SECRET - KAKAO_ACCESS_TOKEN - KAKAO_DASHBOARD_URL - DASHBOARD_PUBLIC_URL - KAKAO_LINK_URL """ from __future__ import annotations import argparse import datetime as dt import json import os import subprocess import sys import urllib.parse import urllib.request from urllib.parse import urlparse TOKEN_URL = "https://kauth.kakao.com/oauth/token" MEMO_URL = "https://kapi.kakao.com/v2/api/talk/memo/default/send" DEFAULT_LINK = "https://finance.yahoo.com" KEYCHAIN_SERVICE = "codex-us-equity-kakao" LOCAL_HOSTS = {"localhost", "127.0.0.1", "::1"} SEND_LOG = os.path.join(os.path.dirname(__file__), "kakao_send_log.jsonl") DASHBOARD_PUBLIC_URL_FILE = os.path.join(os.path.dirname(__file__), "dashboard_public_url.txt") def get_secret(name: str) -> str | None: value = os.environ.get(name) if value: return value try: result = subprocess.run( [ "security", "find-generic-password", "-s", KEYCHAIN_SERVICE, "-a", name, "-w", ], check=False, capture_output=True, text=True, timeout=5, ) except Exception: return None if result.returncode == 0 and result.stdout.strip(): return result.stdout.strip() return None def post_form(url: str, data: dict[str, str], headers: dict[str, str] | None = None) -> dict: encoded = urllib.parse.urlencode(data).encode("utf-8") req = urllib.request.Request(url, data=encoded, method="POST") req.add_header("Content-Type", "application/x-www-form-urlencoded;charset=utf-8") for key, value in (headers or {}).items(): req.add_header(key, value) with urllib.request.urlopen(req, timeout=20) as response: return json.loads(response.read().decode("utf-8")) def write_send_log(status: str, title: str, text: str, detail: dict | str) -> None: record = { "timestamp": dt.datetime.now(dt.timezone.utc).isoformat(), "status": status, "title": title, "message_chars": len(text), "message_first_line": text.strip().splitlines()[0][:120] if text.strip() else "", "detail": detail, } try: with open(SEND_LOG, "a", encoding="utf-8") as handle: handle.write(json.dumps(record, ensure_ascii=False) + "\n") except Exception: pass def get_access_token() -> str: direct_token = get_secret("KAKAO_ACCESS_TOKEN") rest_api_key = get_secret("KAKAO_REST_API_KEY") refresh_token = get_secret("KAKAO_REFRESH_TOKEN") if rest_api_key and refresh_token: payload = { "grant_type": "refresh_token", "client_id": rest_api_key, "refresh_token": refresh_token, } client_secret = get_secret("KAKAO_CLIENT_SECRET") if client_secret: payload["client_secret"] = client_secret refreshed = post_form(TOKEN_URL, payload) return refreshed["access_token"] if direct_token: return direct_token raise SystemExit( "Missing Kakao credentials. Set KAKAO_REST_API_KEY and KAKAO_REFRESH_TOKEN, " "or provide a temporary KAKAO_ACCESS_TOKEN." ) def normalize_link_url(link_url: str | None, allow_local_link: bool = False) -> str: link_url = (link_url or DEFAULT_LINK).strip() parsed = urlparse(link_url) if not parsed.scheme or not parsed.netloc: return DEFAULT_LINK if not allow_local_link and parsed.hostname in LOCAL_HOSTS: return DEFAULT_LINK return link_url def get_dashboard_link() -> str | None: for name in ("KAKAO_DASHBOARD_URL", "DASHBOARD_PUBLIC_URL", "KAKAO_LINK_URL"): value = get_secret(name) if value: return value try: with open(DASHBOARD_PUBLIC_URL_FILE, encoding="utf-8") as handle: value = handle.read().strip() if value: return value except Exception: pass return None def build_message( text: str, title: str, link_url: str | None = None, button_title: str = "상세 보기", allow_local_link: bool = False, ) -> dict: text = text.strip() if not text: raise SystemExit("No message text provided.") raw_link_url = link_url or get_dashboard_link() or DEFAULT_LINK safe_link_url = normalize_link_url(raw_link_url, allow_local_link=allow_local_link) if safe_link_url != DEFAULT_LINK and safe_link_url not in text: suffix = f"\n\n상세: {safe_link_url}" max_body = 950 - len(suffix) if len(text) > max_body: text = text[: max_body - 3].rstrip() + "..." text = f"{text}{suffix}" elif len(text) > 950: text = text[:947].rstrip() + "..." return { "object_type": "text", "text": f"{title}\n\n{text}", "link": { "web_url": safe_link_url, "mobile_web_url": safe_link_url, }, "button_title": button_title, } def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--title", default="US Equity Analyst 요약") parser.add_argument("--message", help="Message text. If omitted, stdin is used.") parser.add_argument("--link-url", help="Button link URL. Overrides KAKAO_LINK_URL.") parser.add_argument("--button-title", default="상세 보기") parser.add_argument("--allow-local-link", action="store_true") parser.add_argument("--dry-run", action="store_true") args = parser.parse_args() message_text = args.message if args.message is not None else sys.stdin.read() template_object = build_message( message_text, args.title, link_url=args.link_url, button_title=args.button_title, allow_local_link=args.allow_local_link, ) if args.dry_run: print(json.dumps(template_object, ensure_ascii=False, indent=2)) return 0 try: access_token = get_access_token() result = post_form( MEMO_URL, {"template_object": json.dumps(template_object, ensure_ascii=False)}, {"Authorization": f"Bearer {access_token}"}, ) except SystemExit as error: detail = str(error) write_send_log("failure", args.title, message_text, detail) print(detail, file=sys.stderr) return error.code if isinstance(error.code, int) else 1 except Exception as error: write_send_log("failure", args.title, message_text, str(error)) raise write_send_log("success", args.title, message_text, result) print(json.dumps(result, ensure_ascii=False)) return 0 if __name__ == "__main__": raise SystemExit(main())