89 lines
2.9 KiB
Python
89 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from typing import Any
|
|
|
|
|
|
def _button(text: str, url: str) -> dict[str, Any]:
|
|
return {
|
|
"tag": "button",
|
|
"text": {"tag": "plain_text", "content": text},
|
|
"type": "primary",
|
|
"url": url,
|
|
}
|
|
|
|
|
|
def create_reminder_payload(submit_url: str, title: str = "每日工作汇报提醒") -> dict[str, Any]:
|
|
return {
|
|
"msg_type": "interactive",
|
|
"card": {
|
|
"header": {
|
|
"title": {"tag": "plain_text", "content": title},
|
|
"template": "blue",
|
|
},
|
|
"elements": [
|
|
{"tag": "div", "text": {"tag": "lark_md", "content": "请提交今日日报。"}},
|
|
{"tag": "action", "actions": [_button("填写日报", submit_url)]},
|
|
],
|
|
},
|
|
}
|
|
|
|
|
|
def create_summary_payload(manager_url: str, summary: dict[str, Any]) -> dict[str, Any]:
|
|
missing_names = "、".join(employee["name"] for employee in summary["missing"]) or "无"
|
|
return {
|
|
"msg_type": "interactive",
|
|
"card": {
|
|
"header": {
|
|
"title": {"tag": "plain_text", "content": f"{summary['date']} 日报提交汇总"},
|
|
"template": "orange" if summary["missingCount"] > 0 else "green",
|
|
},
|
|
"elements": [
|
|
{
|
|
"tag": "div",
|
|
"text": {
|
|
"tag": "lark_md",
|
|
"content": f"已提交：{summary['submittedCount']}/{summary['expectedCount']}\n未提交：{missing_names}",
|
|
},
|
|
},
|
|
{"tag": "action", "actions": [_button("查看全部日报", manager_url)]},
|
|
],
|
|
},
|
|
}
|
|
|
|
|
|
def create_webhook_sign(timestamp: str, secret: str) -> str:
|
|
string_to_sign = f"{timestamp}\n{secret}".encode("utf-8")
|
|
sign = hmac.new(string_to_sign, digestmod=hashlib.sha256).digest()
|
|
return base64.b64encode(sign).decode("utf-8")
|
|
|
|
|
|
def send_webhook(webhook_url: str, payload: dict[str, Any], secret: str = "") -> dict[str, Any]:
|
|
if not webhook_url:
|
|
raise ValueError("FEISHU_WEBHOOK_URL is required")
|
|
|
|
request_payload = dict(payload)
|
|
if secret:
|
|
timestamp = str(int(time.time()))
|
|
request_payload["timestamp"] = timestamp
|
|
request_payload["sign"] = create_webhook_sign(timestamp, secret)
|
|
|
|
request = urllib.request.Request(
|
|
webhook_url,
|
|
data=json.dumps(request_payload).encode("utf-8"),
|
|
headers={"content-type": "application/json"},
|
|
method="POST",
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=10) as response:
|
|
body = response.read().decode("utf-8")
|
|
return json.loads(body) if body else {"ok": True}
|
|
except urllib.error.HTTPError as error:
|
|
raise RuntimeError(f"Feishu webhook failed with status {error.code}") from error
|