gahusb
7c5ca15b64
- on_schedule에 autonomous_issue 분기 추가 (eligible 픽만 선별·max_per_day 제한) - _generate_and_preview 메서드: 슬레이트 생성 → 커버 PNG → 인라인 승인 버튼 - messaging.send_photo 신규 추가 (multipart/form-data, reply_markup 지원) - insta_get_preferences 실패를 warning으로 격리해 자율 경로 중단 방지 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
110 lines
3.4 KiB
Python
110 lines
3.4 KiB
Python
"""고수준 메시지 전송 API."""
|
|
import json
|
|
import uuid
|
|
from typing import Optional
|
|
|
|
import httpx
|
|
|
|
from ..config import TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID
|
|
from ..db import save_telegram_callback
|
|
from .client import _enabled, api_call
|
|
from .formatter import MessageKind, format_agent_message
|
|
|
|
|
|
async def send_raw(
|
|
text: str,
|
|
reply_markup: Optional[dict] = None,
|
|
chat_id: Optional[str] = None,
|
|
parse_mode: str = "HTML",
|
|
) -> dict:
|
|
"""가장 저수준. 원문 텍스트 그대로 전송. chat_id 생략 시 기본 TELEGRAM_CHAT_ID로.
|
|
|
|
parse_mode: 기본 'HTML'. MarkdownV2 페이로드(예: 스크리너) 전송 시 명시 지정.
|
|
"""
|
|
if not _enabled():
|
|
return {"ok": False, "message_id": None}
|
|
payload = {
|
|
"chat_id": chat_id or TELEGRAM_CHAT_ID,
|
|
"text": text,
|
|
"parse_mode": parse_mode,
|
|
}
|
|
if reply_markup:
|
|
payload["reply_markup"] = reply_markup
|
|
result = await api_call("sendMessage", payload)
|
|
ok = result.get("ok", False)
|
|
return {
|
|
"ok": ok,
|
|
"message_id": result.get("result", {}).get("message_id") if ok else None,
|
|
"description": result.get("description") if not ok else None,
|
|
"error_code": result.get("error_code") if not ok else None,
|
|
}
|
|
|
|
|
|
async def send_agent_message(
|
|
agent_id: str,
|
|
kind: MessageKind,
|
|
title: str,
|
|
body: str,
|
|
task_id: Optional[str] = None,
|
|
actions: Optional[list] = None,
|
|
metadata: Optional[dict] = None,
|
|
body_is_html: bool = False,
|
|
) -> dict:
|
|
"""통합 에이전트 메시지 API. 모든 에이전트가 이걸 씀.
|
|
|
|
body_is_html=True: 호출자가 이미 HTML-safe 포맷(링크 <a> 등) 구성한 경우.
|
|
"""
|
|
text = format_agent_message(agent_id, kind, title, body, metadata, body_is_html=body_is_html)
|
|
reply_markup = None
|
|
if actions:
|
|
buttons = []
|
|
for action in actions:
|
|
cb_id = f"{action['action']}_{uuid.uuid4().hex[:8]}"
|
|
save_telegram_callback(cb_id, task_id or "", agent_id)
|
|
buttons.append({"text": action["label"], "callback_data": cb_id})
|
|
reply_markup = {"inline_keyboard": [buttons]}
|
|
return await send_raw(text, reply_markup)
|
|
|
|
|
|
async def send_approval_request(
|
|
agent_id: str,
|
|
task_id: str,
|
|
title: str,
|
|
detail: str,
|
|
) -> dict:
|
|
"""승인/거절 단축 헬퍼."""
|
|
return await send_agent_message(
|
|
agent_id=agent_id,
|
|
kind="approval",
|
|
title=title,
|
|
body=detail,
|
|
task_id=task_id,
|
|
actions=[
|
|
{"label": "✅ 승인", "action": "approve"},
|
|
{"label": "❌ 거절", "action": "reject"},
|
|
],
|
|
)
|
|
|
|
|
|
async def send_photo(
|
|
photo_bytes: bytes,
|
|
caption: str = "",
|
|
reply_markup: Optional[dict] = None,
|
|
chat_id: Optional[str] = None,
|
|
) -> dict:
|
|
"""PNG/JPEG 바이트를 sendPhoto로 전송. reply_markup으로 인라인 키보드 첨부 가능."""
|
|
if not TELEGRAM_BOT_TOKEN:
|
|
return {"ok": False, "reason": "no token"}
|
|
url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendPhoto"
|
|
data: dict = {
|
|
"chat_id": chat_id or TELEGRAM_CHAT_ID,
|
|
"caption": caption[:1024],
|
|
"parse_mode": "HTML",
|
|
}
|
|
if reply_markup:
|
|
data["reply_markup"] = json.dumps(reply_markup, ensure_ascii=False)
|
|
files = {"photo": ("cover.png", photo_bytes, "image/png")}
|
|
async with httpx.AsyncClient(timeout=60) as client:
|
|
resp = await client.post(url, data=data, files=files)
|
|
return resp.json()
|