139 lines
4.7 KiB
Python
139 lines
4.7 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
|
|
|
|
TENANT_TOKEN_URL = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
|
|
MESSAGE_URL = "https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=open_id"
|
|
|
|
|
|
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_bot_message_body(payload: dict[str, Any]) -> dict[str, str]:
|
|
return {
|
|
"receive_id": "",
|
|
"msg_type": "interactive",
|
|
"content": json.dumps(payload["card"], ensure_ascii=False),
|
|
}
|
|
|
|
|
|
def get_tenant_access_token(app_id: str, app_secret: str) -> str:
|
|
if not app_id or not app_secret:
|
|
raise ValueError("FEISHU_APP_ID and FEISHU_APP_SECRET are required")
|
|
|
|
request = urllib.request.Request(
|
|
TENANT_TOKEN_URL,
|
|
data=json.dumps({"app_id": app_id, "app_secret": app_secret}).encode("utf-8"),
|
|
headers={"content-type": "application/json"},
|
|
method="POST",
|
|
)
|
|
with urllib.request.urlopen(request, timeout=10) as response:
|
|
result = json.loads(response.read().decode("utf-8"))
|
|
|
|
token = result.get("tenant_access_token")
|
|
if not token:
|
|
raise RuntimeError(f"Feishu tenant token response missing token: {result}")
|
|
return str(token)
|
|
|
|
|
|
def send_bot_interactive_message(tenant_access_token: str, receive_id: str, payload: dict[str, Any]) -> dict[str, Any]:
|
|
if not receive_id:
|
|
raise ValueError("receive_id is required")
|
|
|
|
body = create_bot_message_body(payload)
|
|
body["receive_id"] = receive_id
|
|
request = urllib.request.Request(
|
|
MESSAGE_URL,
|
|
data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
|
|
headers={
|
|
"authorization": f"Bearer {tenant_access_token}",
|
|
"content-type": "application/json",
|
|
},
|
|
method="POST",
|
|
)
|
|
with urllib.request.urlopen(request, timeout=10) as response:
|
|
result = json.loads(response.read().decode("utf-8"))
|
|
return result
|
|
|
|
|
|
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
|