zyc/kaikai_test
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: send reminders and summaries as bot DMs

Browse Source
This commit is contained in:
Codex 2026-05-07 19:49:48 +08:00
parent ec2c62a529
commit 310a82dc1a
4 changed files with 163 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

50
daily_report/robot_service.py
View File

@ -9,6 +9,9 @@ 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 {
@ -58,6 +61,53 @@ def create_summary_payload(manager_url: str, summary: dict[str, Any]) -> dict[st
}
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()

31
daily_report/scheduled.py
View File

@ -6,7 +6,13 @@ from datetime import date
from .config import read_config
from .db import Database
from .report_service import ReportService
from .robot_service import create_reminder_payload, create_summary_payload, send_webhook
from .robot_service import (
create_reminder_payload,
create_summary_payload,
get_tenant_access_token,
send_bot_interactive_message,
send_webhook,
)
from .workday import is_workday
@ -14,8 +20,21 @@ def send_reminder() -> dict:
config = read_config()
if not is_workday(date.today(), config.workday_calendar_path):
return {"skipped": True, "reason": "not a workday"}
payload = create_reminder_payload(f"{config.base_url}/submit")
return send_webhook(config.feishu_webhook_url, payload, config.feishu_webhook_secret)
database = Database(config.database_path, config.employee_seed_path)
try:
employees = database.list_active_employees()
if config.feishu_app_id and config.feishu_app_secret:
token = get_tenant_access_token(config.feishu_app_id, config.feishu_app_secret)
sent = []
for employee in employees:
send_bot_interactive_message(token, employee["feishu_user_id"], payload)
sent.append(employee["name"])
return {"ok": True, "mode": "bot_private", "sent": sent}
return send_webhook(config.feishu_webhook_url, payload, config.feishu_webhook_secret)
finally:
database.close()
def send_summary(report_date: str | None = None) -> dict:
@ -30,6 +49,14 @@ def send_summary(report_date: str | None = None) -> dict:
selected_date_text = selected_date.isoformat()
summary = service.list_reports_for_date(selected_date_text)
payload = create_summary_payload(f"{config.base_url}/manager?date={selected_date_text}", summary)
if config.feishu_app_id and config.feishu_app_secret:
token = get_tenant_access_token(config.feishu_app_id, config.feishu_app_secret)
admins = [employee for employee in database.list_active_employees() if employee.get("role") == "admin"]
sent = []
for admin in admins:
send_bot_interactive_message(token, admin["feishu_user_id"], payload)
sent.append(admin["name"])
return {"ok": True, "mode": "bot_private", "sent": sent}
return send_webhook(config.feishu_webhook_url, payload, config.feishu_webhook_secret)
finally:
database.close()

10
tests/test_robot_service.py
View File

@ -3,7 +3,7 @@ from __future__ import annotations
import json
import unittest
from daily_report.robot_service import create_reminder_payload, create_summary_payload, create_webhook_sign
from daily_report.robot_service import create_bot_message_body, create_reminder_payload, create_summary_payload, create_webhook_sign
class RobotServiceTest(unittest.TestCase):
@ -39,6 +39,14 @@ class RobotServiceTest(unittest.TestCase):
self.assertIsInstance(signature, str)
self.assertGreater(len(signature), 20)
def test_creates_bot_interactive_message_body(self) -> None:
payload = create_reminder_payload("http://localhost:8787/submit")
body = create_bot_message_body(payload)
self.assertEqual(body["msg_type"], "interactive")
self.assertEqual(body["receive_id"], "")
self.assertIn("请提交今日日报", body["content"])
if __name__ == "__main__":
unittest.main()

75
tests/test_scheduled.py
View File

@ -55,6 +55,81 @@ class ScheduledTest(unittest.TestCase):
self.assertIn("0/2", text)
self.assertIn("Chen、Lin", text)
def test_send_reminder_private_messages_all_active_employees(self) -> None:
with tempfile.TemporaryDirectory(prefix="daily-report-scheduled-") as temp:
root = Path(temp)
seed_path = root / "employees.json"
seed_path.write_text(
json.dumps(
[
{"feishu_user_id": "ou_1", "name": "Lin", "active": True},
{"feishu_user_id": "ou_2", "name": "Chen", "active": True},
],
ensure_ascii=False,
),
encoding="utf-8",
)
fake_file = root / "daily_report" / "config.py"
fake_file.parent.mkdir()
fake_file.write_text("", encoding="utf-8")
env = {
"DATABASE_PATH": "test.sqlite",
"EMPLOYEE_SEED_PATH": "employees.json",
"BASE_URL": "http://localhost:8787",
"FEISHU_APP_ID": "cli_test",
"FEISHU_APP_SECRET": "secret",
}
sent = []
def fake_send(token: str, receive_id: str, payload: dict) -> dict:
sent.append((token, receive_id, payload))
return {"ok": True}
with patch("daily_report.config.__file__", str(fake_file)), patch.dict("os.environ", env, clear=True), patch(
"daily_report.scheduled.get_tenant_access_token", lambda app_id, app_secret: "tenant-token"
), patch("daily_report.scheduled.send_bot_interactive_message", fake_send):
result = scheduled.send_reminder()
self.assertEqual(result["mode"], "bot_private")
self.assertEqual([item[1] for item in sent], ["ou_2", "ou_1"])
def test_send_summary_private_messages_admins_only(self) -> None:
with tempfile.TemporaryDirectory(prefix="daily-report-scheduled-") as temp:
root = Path(temp)
seed_path = root / "employees.json"
seed_path.write_text(
json.dumps(
[
{"feishu_user_id": "ou_admin", "name": "Admin", "active": True, "role": "admin"},
{"feishu_user_id": "ou_staff", "name": "Staff", "active": True, "role": "staff"},
],
ensure_ascii=False,
),
encoding="utf-8",
)
fake_file = root / "daily_report" / "config.py"
fake_file.parent.mkdir()
fake_file.write_text("", encoding="utf-8")
env = {
"DATABASE_PATH": "test.sqlite",
"EMPLOYEE_SEED_PATH": "employees.json",
"BASE_URL": "http://localhost:8787",
"FEISHU_APP_ID": "cli_test",
"FEISHU_APP_SECRET": "secret",
}
sent = []
with patch("daily_report.config.__file__", str(fake_file)), patch.dict("os.environ", env, clear=True), patch(
"daily_report.scheduled.get_tenant_access_token", lambda app_id, app_secret: "tenant-token"
), patch(
"daily_report.scheduled.send_bot_interactive_message",
lambda token, receive_id, payload: sent.append(receive_id) or {"ok": True},
):
result = scheduled.send_summary("2026-05-07")
self.assertEqual(result["mode"], "bot_private")
self.assertEqual(sent, ["ou_admin"])
if __name__ == "__main__":
unittest.main()