"""Render and send terminal packed-backup notifications to Lark.
The module's public interface is intentionally limited to ``send_notification``.
"""
import json
from datetime import timedelta
from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple
from urllib.request import Request, urlopen
__all__ = ("send_notification",)
_MAX_RESPONSE_BYTES = 65536
_NotificationPart = Tuple[str, str]
_NotificationLine = List[_NotificationPart]
_NotificationField = Tuple[str, Any, str]
_LARK_MARKDOWN_ESCAPE_TABLE = str.maketrans(
{
"&": "&",
">": ">",
"<": "<",
"~": "~",
"-": "-",
"!": "!",
"*": "*",
"/": "/",
"\\": "\",
"[": "[",
"]": "]",
"(": "(",
")": ")",
"#": "#",
":": ":",
"+": "+",
'"': """,
"'": "'",
"`": "`",
"$": "$",
"_": "_",
}
)
class _NotificationError(Exception):
pass
def _format_duration_seconds(value: Any) -> Optional[str]:
if isinstance(value, bool) or not isinstance(value, (int, float)):
return None
try:
total_seconds = round(float(value))
except (OverflowError, ValueError):
return None
if total_seconds < 0:
return None
return str(timedelta(seconds=total_seconds))
def _part(role: str, value: Any) -> _NotificationPart:
return role, str(value)
def _label_line(label: str, value: Any, role: str = "value") -> _NotificationLine:
return [_part("key", f"{label}: "), _part(role, value)]
def _field_line(
label: str,
fields: Sequence[_NotificationField],
) -> _NotificationLine:
line = [_part("key", f"{label}: ")]
for index, (name, value, role) in enumerate(fields):
if index:
line.append(_part("key", " "))
line.extend((_part("key", f"{name}="), _part(role, value)))
return line
def _count_role(value: Any) -> str:
if isinstance(value, bool) or not isinstance(value, (int, float)):
return "value"
return "alert" if value != 0 else "value"
def _notification_lines(result: Mapping[str, Any]) -> List[_NotificationLine]:
succeeded = result.get("status") == "succeeded"
lines = [
[
_part("key", "Packed backup CSV export "),
_part(
"success" if succeeded else "alert",
"succeeded" if succeeded else "failed",
),
_part("key", "."),
],
_label_line("context", result.get("kubernetes_context", "unknown")),
_label_line("run", result.get("run_id", "-")),
_label_line("backup", result.get("backup_name", "-")),
]
if succeeded:
unverified = result.get("unverified_complete_marker_clusters", 0)
mysql_update_failed = result.get("mysql_update_failed_clusters", 0)
lines.extend(
[
_field_line(
"jobs",
[
("pack", result.get("pack_jobs", 0), "value"),
("export", result.get("export_jobs", 0), "value"),
("watermark", result.get("watermark_jobs", 0), "value"),
],
),
_field_line(
"watermark",
[
(
"verified",
result.get("verified_complete_markers", 0),
"value",
),
("unverified", unverified, _count_role(unverified)),
(
"mysql_updated",
result.get("mysql_updated_clusters", 0),
"value",
),
(
"mysql_update_failed",
mysql_update_failed,
_count_role(mysql_update_failed),
),
],
),
]
)
durations = [
_format_duration_seconds(result.get("pack_duration_seconds")),
_format_duration_seconds(result.get("export_duration_seconds")),
_format_duration_seconds(result.get("watermark_duration_seconds")),
]
if all(duration is not None for duration in durations):
lines.append(
_field_line(
"duration",
[
("pack", durations[0], "value"),
("export", durations[1], "value"),
("watermark", durations[2], "value"),
],
)
)
else:
lines.extend(
[
_label_line("stage", result.get("stage", "-")),
_field_line(
"partial",
[
(
"markers",
result.get("verified_complete_markers", 0),
"value",
),
(
"mysql_updated",
result.get("mysql_updated_clusters", 0),
"value",
),
],
),
]
)
error = str(result.get("error", "-")).replace("\n", " ")
if len(error) > 500:
error = error[:500] + "...[truncated]"
lines.append(_label_line("error", error, "alert"))
return lines
def _notification_text(result: Mapping[str, Any]) -> str:
return "\n".join(
"".join(text for _, text in line)
for line in _notification_lines(result)
)
def _escape_lark_markdown(value: str) -> str:
return value.translate(_LARK_MARKDOWN_ESCAPE_TABLE)
def _notification_markdown(result: Mapping[str, Any]) -> str:
rendered_lines = []
for line in _notification_lines(result):
rendered_parts = []
for role, text in line:
escaped = _escape_lark_markdown(text)
if role == "key":
rendered = f"{escaped}"
elif role == "value":
rendered = f"**{escaped}**"
elif role == "success":
rendered = f"**{escaped}**"
elif role == "alert":
rendered = f"**{escaped}**"
else:
raise AssertionError(f"unknown notification role: {role}")
rendered_parts.append(rendered)
rendered_lines.append("".join(rendered_parts))
return "\n".join(rendered_lines)
def _notification_payload(result: Mapping[str, Any]) -> Dict[str, Any]:
return {
"msg_type": "interactive",
"card": {
"elements": [
{
"tag": "div",
"text": {
"tag": "lark_md",
"content": _notification_markdown(result),
},
}
]
},
}
def send_notification(
webhook_url: str,
result: Mapping[str, Any],
) -> Optional[str]:
"""Send one terminal notification and return a sanitized error, if any."""
if not webhook_url:
return None
payload = json.dumps(
_notification_payload(result),
ensure_ascii=False,
separators=(",", ":"),
).encode("utf-8")
request = Request(
webhook_url,
data=payload,
headers={"Content-Type": "application/json"},
method="POST",
)
try:
with urlopen(request, timeout=30) as response:
body = response.read(_MAX_RESPONSE_BYTES + 1)
if len(body) > _MAX_RESPONSE_BYTES:
raise _NotificationError("webhook response is too large")
response_payload = json.loads(body.decode("utf-8"))
if not isinstance(response_payload, dict):
raise _NotificationError("webhook returned a non-object response")
if "StatusCode" in response_payload:
accepted = response_payload.get("StatusCode") == 0
else:
accepted = response_payload.get("code") == 0
if not accepted:
raise _NotificationError("webhook rejected the notification")
return None
except Exception as exc:
if isinstance(exc, _NotificationError):
return str(exc)
return f"{type(exc).__name__} while sending notification"