fix: harden alerts and backup deployment guard
All checks were successful
CD Pipeline / deploy (push) Successful in 1m4s
All checks were successful
CD Pipeline / deploy (push) Successful in 1m4s
This commit is contained in:
@@ -31,6 +31,11 @@ sys_log = logging.getLogger("TelegramTpl")
|
||||
TELEGRAM_BOT_TOKEN_ENV = "TELEGRAM_BOT_TOKEN"
|
||||
TELEGRAM_CHAT_IDS_ENV = "TELEGRAM_CHAT_IDS"
|
||||
_TELEGRAM_HTML_BR_RE = re.compile(r"<\s*br\s*/?\s*>", re.IGNORECASE)
|
||||
_TELEGRAM_HTML_TAG_RE = re.compile(r"<[^<>\n]{1,500}>")
|
||||
_TELEGRAM_ALLOWED_HTML_TAG_RE = re.compile(
|
||||
r"(?:</?(?:b|strong|i|em|u|s|strike|del|code|pre)>|<a\s+href=\"[^\"]+\">|</a>)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
|
||||
# ══════════════════════════════════════════════════════════════════════════════
|
||||
@@ -57,13 +62,25 @@ def _get_chat_ids() -> list:
|
||||
|
||||
|
||||
def _sanitize_telegram_html(text: str, parse_mode: Optional[str] = "HTML") -> str:
|
||||
"""Telegram HTML 不支援 <br>,統一轉為換行避免 sendMessage 400。"""
|
||||
"""Telegram HTML 只保留白名單標籤,其餘轉成文字避免 sendMessage 400。"""
|
||||
value = str(text or "")
|
||||
if parse_mode and str(parse_mode).upper() == "HTML":
|
||||
return _TELEGRAM_HTML_BR_RE.sub("\n", value)
|
||||
value = _normalize_telegram_html_linebreaks(value)
|
||||
return _TELEGRAM_HTML_TAG_RE.sub(_escape_unsupported_telegram_html_tag, value)
|
||||
return value
|
||||
|
||||
|
||||
def _normalize_telegram_html_linebreaks(text: str) -> str:
|
||||
return _TELEGRAM_HTML_BR_RE.sub("\n", str(text or ""))
|
||||
|
||||
|
||||
def _escape_unsupported_telegram_html_tag(match: re.Match) -> str:
|
||||
tag = match.group(0)
|
||||
if _TELEGRAM_ALLOWED_HTML_TAG_RE.fullmatch(tag):
|
||||
return tag
|
||||
return escape(tag)
|
||||
|
||||
|
||||
def _callback_payload_utf8(value: Any, max_bytes: int = 52) -> str:
|
||||
"""Clamp callback payload by UTF-8 bytes without splitting multibyte chars."""
|
||||
text = str(value or "unknown").strip() or "unknown"
|
||||
@@ -352,7 +369,7 @@ def price_decision(product_name: str, product_sku: str,
|
||||
direction = "📉" if diff > 0 else "📈" if diff < 0 else "➡️"
|
||||
safe_name = escape(str(product_name or ""))
|
||||
safe_sku = escape(str(product_sku or ""))
|
||||
safe_reason = escape(_sanitize_telegram_html(str(reason or ""), "HTML"))
|
||||
safe_reason = escape(_normalize_telegram_html_linebreaks(str(reason or "")))
|
||||
|
||||
message = (
|
||||
f"💰 <b>AI 定價決策建議</b>\n"
|
||||
@@ -854,7 +871,7 @@ def _format_price_decision_envelope(envelope: Dict[str, Any]) -> List[str]:
|
||||
|
||||
lines = [
|
||||
"🧭 <b>決策信封</b>",
|
||||
f"• 狀態:<code>{decision_type}</code> 等級:<b>{severity}</b>{confidence_text}",
|
||||
f"• 類型:<code>{decision_type}</code> 嚴重度:<b>{severity}</b>{confidence_text}",
|
||||
f"• 資料品質:<code>{data_quality}</code> 自動執行:<b>{'允許' if can_auto_execute else '不允許'}</b>",
|
||||
]
|
||||
if blocked_reason:
|
||||
|
||||
Reference in New Issue
Block a user