38200a5e93
All checks were successful
CD Pipeline / deploy (push) Successful in 4m51s
- services/openclaw_strategist_service.py：新增 generate_daily_report()（每日09:00業績快報＋競品威脅＋2圖表）和 generate_monthly_report()（每月1日07:00月度全景洞察＋3圖表＋MoM/YoY比較） - services/chart_generator_service.py：新建圖表生成服務（6種深色商業圖表，revenue_trend / category_revenue / monthly_overview / price_gap / price_history_heatmap / price_trend） - services/telegram_templates.py：重建訊息模板系統（5類模板：告警/報告/決策/系統/洞察）、新增 send_photo + send_report_with_charts 圖文推播 - scheduler.py：新增 run_daily_report_task / run_monthly_report_task（含 auto_heal 保護） - run_scheduler.py：每日09:00日報 + 每月1日07:00月報排程（月報用每日gate判斷day==1） - requirements.txt：新增 matplotlib + matplotlib-inline Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
470 lines
22 KiB
Python
470 lines
22 KiB
Python
"""
|
||
services/telegram_templates.py
|
||
Telegram 訊息模板系統 v2
|
||
|
||
═══ 訊息分類 ═══════════════════════════════════════════════
|
||
🚨 告警類 — price_alert_msg / threat_alert_msg / system_alert_msg
|
||
📊 報告類 — daily_report_msg / weekly_report_msg / monthly_report_msg
|
||
💰 決策類 — price_decision / batch_decision_msg
|
||
🤖 系統類 — deploy_msg / heal_msg / meta_analysis_msg
|
||
💡 洞察類 — triaged_alert / insight_summary_msg
|
||
═══════════════════════════════════════════════════════════
|
||
|
||
規範：
|
||
1. 所有模板使用 HTML parse_mode
|
||
2. 一律繁體中文，Agent 名稱保留英文（Hermes/NemoTron/OpenClaw/EA）
|
||
3. 每則訊息須含：標題行 / 核心數據 / 建議行動（三段式）
|
||
4. 圖文訊息使用 send_photo_with_caption（附說明文字）
|
||
"""
|
||
|
||
import io
|
||
import json
|
||
import logging
|
||
import os
|
||
from datetime import datetime
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
sys_log = logging.getLogger("TelegramTpl")
|
||
|
||
TELEGRAM_BOT_TOKEN_ENV = "TELEGRAM_BOT_TOKEN"
|
||
TELEGRAM_CHAT_IDS_ENV = "TELEGRAM_CHAT_IDS"
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
# 基礎工具
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
|
||
def _get_bot_token() -> Optional[str]:
|
||
from dotenv import load_dotenv
|
||
load_dotenv()
|
||
return os.getenv(TELEGRAM_BOT_TOKEN_ENV)
|
||
|
||
|
||
def _get_chat_ids() -> list:
|
||
token = _get_bot_token()
|
||
if not token:
|
||
sys_log.warning("[TelegramTpl] %s 未設定，跳過 Telegram 通知", TELEGRAM_BOT_TOKEN_ENV)
|
||
return []
|
||
raw = os.getenv(TELEGRAM_CHAT_IDS_ENV, "[]")
|
||
try:
|
||
return json.loads(raw)
|
||
except json.JSONDecodeError:
|
||
sys_log.warning("[TelegramTpl] %s 格式錯誤，應為 JSON 陣列", TELEGRAM_CHAT_IDS_ENV)
|
||
return []
|
||
|
||
|
||
def _send_telegram_raw(text: str, chat_ids: Optional[list] = None,
|
||
reply_markup: Optional[Dict[str, Any]] = None,
|
||
parse_mode: str = "HTML") -> bool:
|
||
"""發送純文字訊息"""
|
||
import requests
|
||
token = _get_bot_token()
|
||
if not token:
|
||
return False
|
||
if chat_ids is None:
|
||
chat_ids = _get_chat_ids()
|
||
if not chat_ids:
|
||
chat_ids = [-1003940688311] # fallback
|
||
|
||
url = f"https://api.telegram.org/bot{token}/sendMessage"
|
||
payload = {"chat_id": chat_ids[0], "text": text, "parse_mode": parse_mode}
|
||
if reply_markup:
|
||
payload["reply_markup"] = json.dumps(reply_markup, ensure_ascii=False)
|
||
try:
|
||
r = requests.post(url, json=payload, timeout=10)
|
||
if not r.ok:
|
||
sys_log.warning("[TelegramTpl] sendMessage HTTP %s: %s", r.status_code, r.text[:200])
|
||
return False
|
||
return True
|
||
except Exception as e:
|
||
sys_log.error("[TelegramTpl] send 失敗: %s", e)
|
||
return False
|
||
|
||
|
||
def send_photo(photo_bytes: bytes, caption: str = "",
|
||
chat_ids: Optional[list] = None,
|
||
parse_mode: str = "HTML") -> bool:
|
||
"""發送圖片（含說明文字），供圖表報告使用"""
|
||
import requests
|
||
token = _get_bot_token()
|
||
if not token or not photo_bytes:
|
||
return False
|
||
if chat_ids is None:
|
||
chat_ids = _get_chat_ids()
|
||
if not chat_ids:
|
||
chat_ids = [-1003940688311]
|
||
|
||
url = f"https://api.telegram.org/bot{token}/sendPhoto"
|
||
try:
|
||
r = requests.post(
|
||
url,
|
||
data={"chat_id": chat_ids[0], "caption": caption[:1024],
|
||
"parse_mode": parse_mode},
|
||
files={"photo": ("chart.png", photo_bytes, "image/png")},
|
||
timeout=30,
|
||
)
|
||
if not r.ok:
|
||
sys_log.warning("[TelegramTpl] sendPhoto HTTP %s: %s", r.status_code, r.text[:200])
|
||
return False
|
||
return True
|
||
except Exception as e:
|
||
sys_log.error("[TelegramTpl] sendPhoto 失敗: %s", e)
|
||
return False
|
||
|
||
|
||
def send_report_with_charts(text_msg: str, charts: List[Optional[bytes]],
|
||
chat_ids: Optional[list] = None) -> bool:
|
||
"""先發文字報告，再逐張發送圖表"""
|
||
ok = _send_telegram_raw(text_msg, chat_ids=chat_ids)
|
||
for i, chart in enumerate(charts):
|
||
if chart:
|
||
send_photo(chart, caption=f"圖表 {i+1}/{len(charts)}", chat_ids=chat_ids)
|
||
return ok
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
# 🚨 告警類模板
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
|
||
def price_alert_msg(threats: List[Dict], analysis_period: str = "近48h") -> str:
|
||
"""
|
||
競品削價告警（Hermes + NemoTron 偵測結果）
|
||
┌─────────────────────────────────┐
|
||
│ 🚨 競品削價告警 · N 個 SKU │
|
||
│ ─────────────────────────────── │
|
||
│ ⚠️ SKU名稱 MOMO $X → 競品 $Y │
|
||
│ 價差 +X% 業績跌幅 -Y% │
|
||
└─────────────────────────────────┘
|
||
"""
|
||
n = len(threats)
|
||
high = [t for t in threats if t.get("risk") == "HIGH"]
|
||
lines = [
|
||
f"🚨 <b>競品削價告警</b> · 偵測到 <b>{n}</b> 個 SKU [{analysis_period}]",
|
||
f"🔴 高危：{len(high)} 個 🟡 中危：{n - len(high)} 個",
|
||
"─" * 32,
|
||
]
|
||
for t in threats[:8]:
|
||
risk_icon = "🔴" if t.get("risk") == "HIGH" else "🟡"
|
||
name = str(t.get("name", t.get("sku", "")))[:22]
|
||
gap = float(t.get("gap_pct", 0))
|
||
delta = float(t.get("sales_7d_delta_pct", t.get("sales_delta", 0)))
|
||
momo_p = t.get("momo_price")
|
||
pchome_p = t.get("pchome_price")
|
||
price_str = f" MOMO <b>${momo_p:,.0f}</b> vs 競品 <b>${pchome_p:,.0f}</b>" \
|
||
if momo_p and pchome_p else ""
|
||
lines.append(
|
||
f"{risk_icon} <b>{name}</b>\n"
|
||
f" 價差 <b>{gap:+.1f}%</b> 業績週跌 <b>{delta:+.1f}%</b>{price_str}"
|
||
)
|
||
if n > 8:
|
||
lines.append(f"<i>…另有 {n-8} 個 SKU（查看完整報告）</i>")
|
||
lines += ["─" * 32,
|
||
"💡 <b>建議：</b>優先處理紅色高危品項，確認競品是否促銷或長期低價"]
|
||
return "\n".join(lines)
|
||
|
||
|
||
def threat_alert_msg(sku: str, name: str, momo_price: float,
|
||
comp_price: float, gap_pct: float,
|
||
sales_delta: float, action: str, confidence: float) -> str:
|
||
"""單品威脅告警（NemoTron 派發）"""
|
||
risk = "🔴 高危" if gap_pct > 15 else "🟡 中危" if gap_pct > 5 else "🟢 低危"
|
||
return (
|
||
f"{risk} <b>競品威脅通報</b>\n"
|
||
f"━━━━━━━━━━━━━━━━━━━━\n"
|
||
f"🏷️ <b>{name}</b> <code>{sku}</code>\n\n"
|
||
f"💴 MOMO <b>NT${momo_price:,.0f}</b>\n"
|
||
f"💴 競品（PChome）<b>NT${comp_price:,.0f}</b>\n"
|
||
f"📉 價差 <b>{gap_pct:+.1f}%</b>（我方較貴）\n"
|
||
f"📊 業績週變 <b>{sales_delta:+.1f}%</b>\n\n"
|
||
f"🤖 <b>AI 建議：</b>{action}\n"
|
||
f"📈 信心度：{confidence:.0%}\n"
|
||
f"━━━━━━━━━━━━━━━━━━━━"
|
||
)
|
||
|
||
|
||
def system_alert_msg(level: str, title: str, detail: str, source: str = "") -> str:
|
||
"""系統級告警（AIOps / 部署失敗等）"""
|
||
icons = {"critical": "🚨", "error": "❌", "warning": "⚠️", "info": "ℹ️"}
|
||
icon = icons.get(level, "⚠️")
|
||
src = f" [{source}]" if source else ""
|
||
return (
|
||
f"{icon} <b>{title}</b>{src}\n"
|
||
f"━━━━━━━━━━━━━━━━━━━━\n"
|
||
f"{detail[:600]}\n"
|
||
f"━━━━━━━━━━━━━━━━━━━━\n"
|
||
f"🕐 {datetime.now().strftime('%m/%d %H:%M')}"
|
||
)
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
# 📊 報告類模板（三種週期）
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
|
||
def daily_report_header(date_str: str, revenue: float, wow: float,
|
||
threat_count: int, opportunity_count: int) -> str:
|
||
"""
|
||
日報標題卡（附圖前發送的文字說明）
|
||
"""
|
||
wow_icon = "📈" if wow >= 0 else "📉"
|
||
wow_color = "+" if wow >= 0 else ""
|
||
return (
|
||
f"📊 <b>EwoooC 電商日報</b> · {date_str}\n"
|
||
f"══════════════════════════\n"
|
||
f"💰 今日業績 <b>NT${revenue/10000:.1f} 萬</b>\n"
|
||
f"{wow_icon} 週同比 <b>{wow_color}{wow:.1f}%</b>\n"
|
||
f"🔴 競品威脅 <b>{threat_count}</b> 個 SKU\n"
|
||
f"🟢 市場機會 <b>{opportunity_count}</b> 個 SKU\n"
|
||
f"══════════════════════════\n"
|
||
f"🤖 <i>Hermes + NemoTron + OpenClaw 聯合分析</i>"
|
||
)
|
||
|
||
|
||
def weekly_report_header(period: str, curr_rev: float, prev_rev: float,
|
||
wow: float, top_category: str) -> str:
|
||
"""週報標題卡"""
|
||
wow_icon = "📈" if wow >= 0 else "📉"
|
||
return (
|
||
f"📊 <b>EwoooC 電商週報</b> · {period}\n"
|
||
f"══════════════════════════\n"
|
||
f"💰 本週業績 <b>NT${curr_rev/10000:.1f} 萬</b>\n"
|
||
f"📦 前週業績 <b>NT${prev_rev/10000:.1f} 萬</b>\n"
|
||
f"{wow_icon} 週成長率 <b>{wow:+.1f}%</b>\n"
|
||
f"🏆 最強品類 <b>{top_category}</b>\n"
|
||
f"══════════════════════════\n"
|
||
f"🤖 <i>OpenClaw × MCP 全景策略分析</i>"
|
||
)
|
||
|
||
|
||
def monthly_report_header(month_str: str, revenue: float, mom: float,
|
||
yoy: float, top3_categories: List[str]) -> str:
|
||
"""月報標題卡"""
|
||
mom_icon = "📈" if mom >= 0 else "📉"
|
||
yoy_icon = "🚀" if yoy >= 10 else "📈" if yoy >= 0 else "📉"
|
||
cats = " / ".join(top3_categories[:3])
|
||
return (
|
||
f"📅 <b>EwoooC 電商月報</b> · {month_str}\n"
|
||
f"══════════════════════════\n"
|
||
f"💰 月度業績 <b>NT${revenue/10000:.0f} 萬</b>\n"
|
||
f"{mom_icon} 月成長率 <b>{mom:+.1f}%</b> {yoy_icon} 年成長 <b>{yoy:+.1f}%</b>\n"
|
||
f"🏆 TOP 3 品類 <b>{cats}</b>\n"
|
||
f"══════════════════════════\n"
|
||
f"🤖 <i>OpenClaw × Hermes × MCP 月度全景洞察</i>"
|
||
)
|
||
|
||
|
||
def report_section(icon: str, title: str, lines: List[str]) -> str:
|
||
"""通用報告節段（供日/週/月報各節使用）"""
|
||
body = "\n".join(f" {l}" for l in lines)
|
||
return f"\n{icon} <b>{title}</b>\n{body}"
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
# 💰 決策類模板
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
|
||
def price_decision(product_name: str, product_sku: str,
|
||
current_price: float, suggested_price: float,
|
||
reason: str, insight_id: Optional[int] = None) -> tuple:
|
||
"""降 / 提價決策通知（含 Inline Keyboard）"""
|
||
diff = current_price - suggested_price
|
||
action_text = f"降價 NT${diff:,.0f}" if diff > 0 else \
|
||
f"提價 NT${-diff:,.0f}" if diff < 0 else "維持現價"
|
||
direction = "📉" if diff > 0 else "📈" if diff < 0 else "➡️"
|
||
|
||
message = (
|
||
f"💰 <b>AI 定價決策建議</b>\n"
|
||
f"━━━━━━━━━━━━━━━━━━━━\n"
|
||
f"🏷️ <b>{product_name}</b> <code>{product_sku}</code>\n\n"
|
||
f"現價：<b>NT${current_price:,.0f}</b>\n"
|
||
f"建議：<b>NT${suggested_price:,.0f}</b> {direction} {action_text}\n\n"
|
||
f"💡 <b>依據：</b>{reason}\n"
|
||
)
|
||
if insight_id:
|
||
message += f"🔗 洞察 ID：<code>{insight_id}</code>\n"
|
||
message += f"━━━━━━━━━━━━━━━━━━━━"
|
||
|
||
keyboard = {"inline_keyboard": [[
|
||
{"text": "✅ 確認執行", "callback_data": f"momo:price_decision:approve:{product_sku}"},
|
||
{"text": "❌ 拒絕", "callback_data": f"momo:price_decision:reject:{product_sku}"},
|
||
]]}
|
||
return message, keyboard
|
||
|
||
|
||
def batch_decision_msg(items: List[Dict], batch_id: str) -> tuple:
|
||
"""批次定價決策（多 SKU 一次確認）"""
|
||
lines = [f"💰 <b>批次定價決策</b> · 共 {len(items)} 個 SKU\n━━━━━━━━━━━━━━━━━━━━"]
|
||
for i, item in enumerate(items[:6], 1):
|
||
diff = item.get("current_price", 0) - item.get("suggested_price", 0)
|
||
direction = "📉" if diff > 0 else "📈"
|
||
lines.append(
|
||
f"{i}. {direction} <b>{item.get('name','')[:20]}</b>\n"
|
||
f" ${item.get('current_price',0):,.0f} → <b>${item.get('suggested_price',0):,.0f}</b>"
|
||
)
|
||
if len(items) > 6:
|
||
lines.append(f"<i>…另有 {len(items)-6} 個 SKU</i>")
|
||
lines.append("━━━━━━━━━━━━━━━━━━━━")
|
||
keyboard = {"inline_keyboard": [[
|
||
{"text": f"✅ 全部確認（{len(items)}項）",
|
||
"callback_data": f"momo:batch_decision:approve:{batch_id}"},
|
||
{"text": "❌ 取消",
|
||
"callback_data": f"momo:batch_decision:reject:{batch_id}"},
|
||
]]}
|
||
return "\n".join(lines), keyboard
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
# 🤖 系統類模板
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
|
||
def deploy_msg(status: str, branch: str, commit: str, duration: str = "") -> str:
|
||
"""CI/CD 部署通知"""
|
||
icons = {"success": "✅", "failed": "❌", "started": "🚀"}
|
||
icon = icons.get(status, "ℹ️")
|
||
dur = f" 耗時 {duration}" if duration else ""
|
||
return (
|
||
f"{icon} <b>部署{status.upper()}</b>{dur}\n"
|
||
f"━━━━━━━━━━━━━━━━━━━━\n"
|
||
f"🌿 分支：<code>{branch}</code>\n"
|
||
f"🔖 Commit：<code>{commit[:12]}</code>\n"
|
||
f"🕐 {datetime.now().strftime('%m/%d %H:%M')}"
|
||
)
|
||
|
||
|
||
def heal_msg(status: str, error_type: str, target_file: str,
|
||
commit_sha: str = "", diff_lines: int = 0) -> str:
|
||
"""AiderHeal 自動修復通知"""
|
||
if status == "started":
|
||
icon, title = "🔧", "AiderHeal 啟動"
|
||
elif status == "success":
|
||
icon, title = "✅", "AiderHeal 修復完成"
|
||
elif status == "reverted":
|
||
icon, title = "🔄", "AiderHeal 自動回滾"
|
||
else:
|
||
icon, title = "❌", "AiderHeal 失敗"
|
||
|
||
lines = [f"{icon} <b>{title}</b>", "━━━━━━━━━━━━━━━━━━━━",
|
||
f"🐛 錯誤類型：<code>{error_type}</code>",
|
||
f"📄 目標檔案：<code>{target_file}</code>"]
|
||
if commit_sha:
|
||
lines.append(f"🔖 Commit：<code>{commit_sha[:12]}</code>")
|
||
if diff_lines:
|
||
lines.append(f"📝 修改行數：{diff_lines} 行")
|
||
lines += ["━━━━━━━━━━━━━━━━━━━━",
|
||
f"🕐 {datetime.now().strftime('%m/%d %H:%M')}"]
|
||
return "\n".join(lines)
|
||
|
||
|
||
def meta_analysis_msg(period: str, content: str) -> str:
|
||
"""AI 系統自我審視報告"""
|
||
return (
|
||
f"🤖 <b>AI 系統效能自我審視</b> · {period}\n"
|
||
f"══════════════════════════\n"
|
||
f"{content[:1200]}\n"
|
||
f"══════════════════════════\n"
|
||
f"<i>由 OpenClaw 定期分析 · {datetime.now().strftime('%m/%d %H:%M')}</i>"
|
||
)
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
# 💡 洞察類模板
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
|
||
def triaged_alert(base_event: Dict[str, Any], tier_label: str,
|
||
ai_summary: str, ai_cause: Optional[str] = None,
|
||
ai_actions: Optional[list] = None,
|
||
ai_executed: Optional[list] = None) -> tuple:
|
||
"""EA L1/L2 自主執行通知（保留原有介面，升級排版）"""
|
||
event_type = base_event.get("event_type", "alert")
|
||
title = base_event.get("title", "")
|
||
summary = base_event.get("summary", "")
|
||
event_id = base_event.get("id", "unknown")
|
||
|
||
lines = [
|
||
f"⚡ <b>{tier_label} · {event_type}</b>",
|
||
f"📌 {title}",
|
||
"",
|
||
]
|
||
if summary:
|
||
lines += [f"🔍 <b>概要：</b>{summary}", ""]
|
||
if ai_summary:
|
||
lines += [f"🧠 <b>AI 摘要：</b>{ai_summary[:400]}", ""]
|
||
if ai_cause:
|
||
lines += [f"💡 <b>可能原因：</b>{ai_cause}", ""]
|
||
if ai_actions:
|
||
lines += ["<b>📋 建議行動：</b>"] + [f" • {a}" for a in ai_actions] + [""]
|
||
if ai_executed:
|
||
lines += ["<b>✅ 已執行：</b>"] + [f" • {a}" for a in ai_executed] + [""]
|
||
|
||
trace = base_event.get("trace")
|
||
if trace:
|
||
lines.append(f"<pre>{trace[-400:]}</pre>")
|
||
|
||
keyboard = {"inline_keyboard": [
|
||
[{"text": "🛑 忽略此事件",
|
||
"callback_data": f"momo:event_ignore:{event_id}"}],
|
||
]}
|
||
return "\n".join(lines), keyboard
|
||
|
||
|
||
def insight_summary_msg(insights: List[Dict], period: str = "近24h") -> str:
|
||
"""AI 洞察摘要彙整（供定期推播）"""
|
||
n = len(insights)
|
||
lines = [
|
||
f"💡 <b>AI 洞察摘要</b> · {period} 共 {n} 筆",
|
||
"━━━━━━━━━━━━━━━━━━━━",
|
||
]
|
||
type_icons = {
|
||
"price_alert": "🔴", "recommendation": "💰", "weekly_strategy": "📊",
|
||
"meta_analysis": "🤖", "market_opportunity": "🟢", "mcp_cache": "🌐",
|
||
}
|
||
for ins in insights[:6]:
|
||
icon = type_icons.get(ins.get("insight_type", ""), "💡")
|
||
content_preview = str(ins.get("content", ""))[:80].replace("\n", " ")
|
||
conf = float(ins.get("confidence", 0))
|
||
lines.append(f"{icon} {content_preview}… <i>({conf:.0%})</i>")
|
||
if n > 6:
|
||
lines.append(f"<i>…另有 {n-6} 筆洞察</i>")
|
||
lines.append("━━━━━━━━━━━━━━━━━━━━")
|
||
return "\n".join(lines)
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
# 舊版相容介面（保留供現有程式碼調用）
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
|
||
def alert(title: str, content: str, actions: Optional[list] = None) -> str:
|
||
msg = system_alert_msg("error", title, content)
|
||
if actions:
|
||
msg += "\n" + "\n".join(f" • {a}" for a in actions)
|
||
return msg
|
||
|
||
|
||
def warning(title: str, summary: str, details: Optional[dict] = None) -> str:
|
||
detail_str = "\n".join(f" {k}: {v}" for k, v in (details or {}).items())
|
||
return system_alert_msg("warning", title, summary + ("\n" + detail_str if detail_str else ""))
|
||
|
||
|
||
def info(title: str, module: str, content: str, time: Optional[Any] = None) -> str:
|
||
t_str = f" · {time}" if time else ""
|
||
return f"ℹ️ <b>{title}</b> [{module}]{t_str}\n\n{content}"
|
||
|
||
|
||
def success(title: str, module: str, stats: str = "") -> str:
|
||
return f"✅ <b>{title}</b> [{module}]\n{stats}"
|
||
|
||
|
||
def report(title: str, report_type: str, period: str, content_md: str) -> str:
|
||
icons = {"weekly_strategy": "📊", "daily": "📅", "monthly": "📆", "meta_analysis": "🤖"}
|
||
icon = icons.get(report_type, "📋")
|
||
return (
|
||
f"{icon} <b>{title}</b> ({report_type})\n"
|
||
f"📅 期間：{period}\n"
|
||
f"══════════════════════════\n"
|
||
f"{content_md}"
|
||
)
|
||
|
||
|
||
def _send_telegram(msg: str, chat_ids: Optional[list] = None,
|
||
reply_markup: Optional[Dict[str, Any]] = None) -> bool:
|
||
return _send_telegram_raw(msg, chat_ids=chat_ids, reply_markup=reply_markup)
|