diff --git a/.env.example b/.env.example index d1ae04b..ef3c54d 100644 --- a/.env.example +++ b/.env.example @@ -2,11 +2,22 @@ ANYROUTER_ACCOUNTS=[{"cookies":{"session":"你的session值"},"api_user":"你的api_user值"}] # 可选：通知配置 -# DINGDING_WEBHOOK=https://oapi.dingtalk.com/robot/send?access_token=xxx -# EMAIL_USER=your_email@example.com -# EMAIL_PASS=your_password -# EMAIL_TO=recipient@example.com -# PUSHPLUS_TOKEN=your_pushplus_token -# SERVERPUSHKEY=your_server_pushkey -# FEISHU_WEBHOOK=https://open.feishu.cn/open-apis/bot/v2/hook/xxx -# WEIXIN_WEBHOOK=https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx + +# 邮件通知配置 +# EMAIL_NOTIF_CONFIG={"user":"your_email@example.com","pass":"your_password","to":"recipient@example.com","template":"可选的自定义模板"} + +# 钉钉通知配置 +# DINGTALK_NOTIF_CONFIG={"webhook":"https://oapi.dingtalk.com/robot/send?access_token=xxx","template":"可选的自定义模板"} + +# 飞书通知配置 +# FEISHU_NOTIF_CONFIG={"webhook":"https://open.feishu.cn/open-apis/bot/v2/hook/xxx","platform_settings":{"use_card":true,"color_theme":"blue"},"template":"可选的自定义模板"} + +# 企业微信通知配置 +# WECOM_NOTIF_CONFIG={"webhook":"https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx","platform_settings":{"markdown_style":"markdown"},"template":"可选的自定义模板"} +# markdown_style 可选值: "markdown" (默认), "markdown_v2", null (使用纯文本), 其他值视为 null + +# PushPlus 通知配置 +# PUSHPLUS_NOTIF_CONFIG={"token":"your_pushplus_token","template":"可选的自定义模板"} + +# Server酱通知配置 +# SERVERPUSH_NOTIF_CONFIG={"send_key":"your_server_pushkey","template":"可选的自定义模板"} diff --git a/.github/workflows/checkin.yml b/.github/workflows/checkin.yml index adf5590..1c54a8c 100644 --- a/.github/workflows/checkin.yml +++ b/.github/workflows/checkin.yml @@ -72,6 +72,7 @@ jobs: - name: 执行签到 env: ANYROUTER_ACCOUNTS: ${{ secrets.ANYROUTER_ACCOUNTS }} + # 旧格式环境变量（向后兼容） DINGDING_WEBHOOK: ${{ secrets.DINGDING_WEBHOOK }} EMAIL_USER: ${{ secrets.EMAIL_USER }} EMAIL_PASS: ${{ secrets.EMAIL_PASS }} @@ -81,6 +82,13 @@ jobs: SERVERPUSHKEY: ${{ secrets.SERVERPUSHKEY }} FEISHU_WEBHOOK: ${{ secrets.FEISHU_WEBHOOK }} WEIXIN_WEBHOOK: ${{ secrets.WEIXIN_WEBHOOK }} + # 新格式环境变量（支持自定义模板） + DINGTALK_NOTIF_CONFIG: ${{ secrets.DINGTALK_NOTIF_CONFIG }} + EMAIL_NOTIF_CONFIG: ${{ secrets.EMAIL_NOTIF_CONFIG }} + PUSHPLUS_NOTIF_CONFIG: ${{ secrets.PUSHPLUS_NOTIF_CONFIG }} + SERVERPUSH_NOTIF_CONFIG: ${{ secrets.SERVERPUSH_NOTIF_CONFIG }} + FEISHU_NOTIF_CONFIG: ${{ secrets.FEISHU_NOTIF_CONFIG }} + WECOM_NOTIF_CONFIG: ${{ secrets.WECOM_NOTIF_CONFIG }} run: | uv run checkin.py diff --git a/README.md b/README.md index ec12794..e55e8bf 100644 --- a/README.md +++ b/README.md @@ -11,6 +11,7 @@ - ✅ 单个/多账号自动签到 - ✅ 多种机器人通知（可选） - ✅ 绕过 WAF 限制 +- ✅ 支持 Stencil 模板自定义通知内容 ## 使用方法 @@ -140,27 +141,170 @@ 脚本支持多种通知方式，可以通过配置以下环境变量开启，如果 `webhook` 有要求安全设置，例如钉钉，可以在新建机器人时选择自定义关键词，填写 `AnyRouter`。 -### 邮箱通知 +### 传统配置方式（向后兼容） + +> 本配置方式有可能将在未来的某个版本中被彻底移除，请尽快升级至[新版本的配置方式](#新的配置方式推荐)。 + +#### 邮箱通知 - `EMAIL_USER`: 发件人邮箱地址 - `EMAIL_PASS`: 发件人邮箱密码/授权码 - `CUSTOM_SMTP_SERVER`: 自定义发件人SMTP服务器(可选) - `EMAIL_TO`: 收件人邮箱地址 -### 钉钉机器人 + +#### 钉钉机器人 - `DINGDING_WEBHOOK`: 钉钉机器人的 Webhook 地址 -### 飞书机器人 +#### 飞书机器人 - `FEISHU_WEBHOOK`: 飞书机器人的 Webhook 地址 -### 企业微信机器人 +#### 企业微信机器人 - `WEIXIN_WEBHOOK`: 企业微信机器人的 Webhook 地址 -### PushPlus 推送 +#### PushPlus 推送 - `PUSHPLUS_TOKEN`: PushPlus 的 Token -### Server酱 +#### Server酱 - `SERVERPUSHKEY`: Server酱的 SendKey -配置步骤： +### 新的配置方式（推荐） + +**支持 Stencil 模板自定义通知内容** + +新的配置方式支持： +- 自定义通知模板（使用 Stencil 模板语法） +- 平台特定设置（如飞书的卡片模式、颜色主题等） +- 向后兼容旧的配置方式 + +#### 邮箱通知 +```bash +# JSON 配置（支持自定义模板） +EMAIL_NOTIF_CONFIG='{"user":"your_email@example.com","pass":"your_password","to":"recipient@example.com","template":"自定义模板内容"}' +``` + +#### 钉钉机器人 +```bash +# 方式一：JSON 配置 +DINGTALK_NOTIF_CONFIG='{"webhook":"https://oapi.dingtalk.com/robot/send?access_token=xxx","template":"自定义模板"}' + +# 方式二：纯 Webhook URL +DINGTALK_NOTIF_CONFIG="https://oapi.dingtalk.com/robot/send?access_token=xxx" +``` + +#### 飞书机器人 +```bash +# 方式一：JSON 配置（支持卡片模式和颜色主题，以及自定义模板） +FEISHU_NOTIF_CONFIG='{"webhook":"https://open.feishu.cn/open-apis/bot/v2/hook/xxx","platform_settings":{"use_card":true,"color_theme":"blue"},"template":"自定义模板"}' + +# 方式二：纯 Webhook URL（使用默认模板） +FEISHU_NOTIF_CONFIG="https://open.feishu.cn/open-apis/bot/v2/hook/xxx" +``` + +#### 企业微信机器人 +```bash +# 方式一：JSON 配置（支持 markdown 样式和自定义模板） +WECOM_NOTIF_CONFIG='{"webhook":"https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx","platform_settings":{"markdown_style":"markdown"},"template":"自定义模板"}' + +# 方式二：纯 Webhook URL（使用默认模板） +WECOM_NOTIF_CONFIG="https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx" +``` + +**`markdown_style` 配置说明**： +- `"markdown"`：使用 markdown 格式（默认值） +- `"markdown_v2"`：使用 markdown_v2 格式 +- `null` 或其他值：使用纯文本格式 + +#### PushPlus 推送 +```bash +# 方式一：JSON 配置（支持自定义模板） +PUSHPLUS_NOTIF_CONFIG='{"token":"your_pushplus_token","template":"自定义模板"}' + +# 方式二：纯 Token（使用默认模板） +PUSHPLUS_NOTIF_CONFIG="your_pushplus_token" +``` + +#### Server酱 +```bash +# 方式一：JSON 配置（支持自定义模板） +SERVERPUSH_NOTIF_CONFIG='{"send_key":"your_server_pushkey","template":"自定义模板"}' + +# 方式二：纯 SendKey（使用默认模板） +SERVERPUSH_NOTIF_CONFIG="your_server_pushkey" +``` + +### 模板语法说明 + +使用 Stencil 模板语法，支持以下变量： + +#### 可用变量 + +**基础变量**： +- `timestamp`: 执行时间戳字符串（格式：`YYYY-MM-DD HH:MM:SS`） + +**统计数据**： +- `stats`: 统计数据对象，包含以下属性： + - `stats.success_count`: 成功签到的账号数量 + - `stats.failed_count`: 失败的账号数量 + - `stats.total_count`: 总账号数量 + +**账号列表**： +- `accounts`: 完整的账号列表数组（包含所有账号），每个账号对象包含以下属性： + - `name`: 账号名称 + - `status`: 状态（ `"success"` 或 `"failed"`） + - `quota`: 当前余额（仅成功时有值，类型为 float） + - `used`: 已使用余额（仅成功时有值，类型为 float） + - `error`: 错误信息（仅失败时有值，类型为 string） + +**便利变量**： +- `success_accounts`: 成功的账号列表（已按状态过滤） +- `failed_accounts`: 失败的账号列表（已按状态过滤） +- `has_success`: 布尔值，是否有成功的账号 +- `has_failed`: 布尔值，是否有失败的账号 +- `all_success`: 布尔值，是否所有账号都成功 +- `all_failed`: 布尔值，是否所有账号都失败 +- `partial_success`: 布尔值，是否部分成功部分失败 + +#### 重要说明 + +**Stencil 模板引擎限制**： +- ❌ 不支持比较操作符（`==`、`!=`、`<`、`>` 等） +- ❌ 不支持在循环中使用 `{% if account.status == "success" %}` 来判断状态 + +#### 模板示例 + +**推荐写法**（使用分组列表）： +``` +{% if timestamp %}[TIME] 执行时间: {{ timestamp }}\n\n{% endif %}{% if success_accounts %}[SUCCESS] 成功账号: +{% for account in success_accounts %}• {{ account.name }} +💰 余额: ${{ account.quota }}, 已用: ${{ account.used }} +{% endfor %} +{% endif %}{% if failed_accounts %}[FAIL] 失败账号: +{% for account in failed_accounts %}• {{ account.name }} +⚠️ {{ account.error }} +{% endfor %} +{% endif %}[STATS] 签到统计: +✅ 成功: {{ stats.success_count }}/{{ stats.total_count }} +❌ 失败: {{ stats.failed_count }}/{{ stats.total_count }} +``` + +**简单模板示例**： +``` +{{ timestamp }} - 成功: {{ success_count }}个，失败: {{ failed_count }}个 +``` + +**条件渲染示例**： +``` +{% if has_success %}有成功的账号{% endif %} +{% if has_failed %}有失败的账号{% endif %} +``` + +### 配置优先级 + +配置按以下优先级加载： +1. 新环境变量配置（如 `EMAIL_NOTIF_CONFIG`） +2. 默认模板配置文件（`notif_config/*.json`） +3. 旧环境变量配置（如 `EMAIL_USER`、`EMAIL_PASS` 等） + +## 配置步骤： 1. 在仓库的 Settings -> Environments -> production -> Environment secrets 中添加上述环境变量 2. 每个通知方式都是独立的，可以只配置你需要的推送方式 3. 如果某个通知方式配置不正确或未配置，脚本会自动跳过该通知方式 @@ -192,16 +336,32 @@ uv run checkin.py ## 测试 +项目包含完整的测试套件，分为单元测试和集成测试： + ```bash +# 安装所有依赖 uv sync --dev # 安装 Playwright 浏览器 playwright install chromium -# 运行测试 +# 运行所有测试 uv run pytest tests/ + +# 仅运行单元测试（快速） +uv run pytest tests/unit/ + +# 仅运行集成测试（需要真实接口） +uv run pytest tests/integration/ + +# 运行特定测试文件 +uv run pytest tests/unit/test_template_rendering.py -v ``` +**测试目录说明**： +- `tests/unit/` - 单元测试：配置解析、数据模型、发送功能、模板渲染 +- `tests/integration/` - 集成测试：真实通知接口测试 + ## 免责声明 本脚本仅用于学习和研究目的，使用前请确保遵守相关网站的使用条款. diff --git a/checkin.py b/checkin.py index f9dd7e3..6955598 100644 --- a/checkin.py +++ b/checkin.py @@ -9,14 +9,17 @@ import os import sys from datetime import datetime +from typing import List import httpx from dotenv import load_dotenv from playwright.async_api import async_playwright from notify import notify +from models import NotificationData, AccountResult, NotificationStats -load_dotenv() +# 禁用变量插值以保留模板中的 $ 符号 +load_dotenv(interpolate=False) BALANCE_HASH_FILE = 'balance_hash.txt' @@ -300,7 +303,7 @@ async def main(): # 为每个账号执行签到 success_count = 0 total_count = len(accounts) - notification_content = [] + account_results: List[AccountResult] = [] # 使用结构化数据存储结果 current_balances = {} need_notify = False # 是否需要发送通知 balance_changed = False # 余额是否有变化 @@ -309,6 +312,14 @@ async def main(): account_key = f'account_{i + 1}' try: success, user_info = await check_in_account(account, i) + account_name = get_account_display_name(account, i) + + # 创建账号结果 + account_result = AccountResult( + name=account_name, + status="success" if success else "failed" + ) + if success: success_count += 1 @@ -319,31 +330,37 @@ async def main(): if not success: should_notify_this_account = True need_notify = True - account_name = get_account_display_name(account, i) print(f'[NOTIFY] {account_name} failed, will send notification') - # 收集余额数据 + # 收集余额数据和处理结果 if user_info and user_info.get('success'): current_quota = user_info['quota'] current_used = user_info['used_quota'] current_balances[account_key] = {'quota': current_quota, 'used': current_used} - # 只有需要通知的账号才收集内容 + # 设置账号结果的余额信息 + account_result.quota = current_quota + account_result.used = current_used + elif user_info: + # 设置错误信息 + account_result.error = user_info.get('error', '未知错误') + + # 只有需要通知的账号才添加到结果列表 if should_notify_this_account: - account_name = get_account_display_name(account, i) - status = '[SUCCESS]' if success else '[FAIL]' - account_result = f'{status} {account_name}' - if user_info and user_info.get('success'): - account_result += f'\n{user_info["display"]}' - elif user_info: - account_result += f'\n{user_info.get("error", "Unknown error")}' - notification_content.append(account_result) + account_results.append(account_result) except Exception as e: account_name = get_account_display_name(account, i) print(f'[FAILED] {account_name} processing exception: {e}') need_notify = True # 异常也需要通知 - notification_content.append(f'[FAIL] {account_name} exception: {str(e)[:50]}...') + + # 创建失败的账号结果 + account_result = AccountResult( + name=account_name, + status="failed", + error=f'异常: {str(e)[:50]}...' + ) + account_results.append(account_result) # 检查余额变化 current_balance_hash = generate_balance_hash(current_balances) if current_balances else None @@ -367,42 +384,43 @@ async def main(): account_key = f'account_{i + 1}' if account_key in current_balances: account_name = get_account_display_name(account, i) - # 只添加成功获取余额的账号，且避免重复添加 - account_result = f'[BALANCE] {account_name}' - account_result += f'\n:money: Current balance: ${current_balances[account_key]["quota"]}, Used: ${current_balances[account_key]["used"]}' - # 检查是否已经在通知内容中（避免重复） - if not any(account_name in item for item in notification_content): - notification_content.append(account_result) + # 检查是否已经在结果列表中（避免重复） + if not any(result.name == account_name for result in account_results): + account_result = AccountResult( + name=account_name, + status="success", + quota=current_balances[account_key]["quota"], + used=current_balances[account_key]["used"] + ) + account_results.append(account_result) # 保存当前余额hash if current_balance_hash: save_balance_hash(current_balance_hash) - if need_notify and notification_content: - # 构建通知内容 - summary = [ - '[STATS] Check-in result statistics:', - f'[SUCCESS] Success: {success_count}/{total_count}', - f'[FAIL] Failed: {total_count - success_count}/{total_count}', - ] - - if success_count == total_count: - summary.append('[SUCCESS] All accounts check-in successful!') - elif success_count > 0: - summary.append('[WARN] Some accounts check-in successful') - else: - summary.append('[ERROR] All accounts check-in failed') - - time_info = f'[TIME] Execution time: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}' - - notify_content = '\n\n'.join([time_info, '\n'.join(notification_content), '\n'.join(summary)]) - - print(notify_content) - notify.push_message('AnyRouter Check-in Alert', notify_content, msg_type='text') + if need_notify and account_results: + # 构建结构化通知数据 + stats = NotificationStats( + success_count=success_count, + failed_count=total_count - success_count, + total_count=total_count + ) + + notification_data = NotificationData( + accounts=account_results, + stats=stats, + timestamp=datetime.now().strftime('%Y-%m-%d %H:%M:%S') + ) + + # 发送通知 + notify.push_message('AnyRouter 签到提醒', notification_data, msg_type='text') print('[NOTIFY] Notification sent due to failures or balance changes') else: print('[INFO] All accounts successful and no balance changes detected, notification skipped') + # 日志总结 + print(f'[RESULT] Final result: Success {success_count}/{total_count}, Failed {total_count - success_count}/{total_count}') + # 设置退出码 sys.exit(0 if success_count > 0 else 1) diff --git a/models/__init__.py b/models/__init__.py new file mode 100644 index 0000000..346b8fe --- /dev/null +++ b/models/__init__.py @@ -0,0 +1,15 @@ +from models.email_config import EmailConfig +from models.webhook_config import WebhookConfig +from models.pushplus_config import PushPlusConfig +from models.serverpush_config import ServerPushConfig +from models.notification_data import NotificationData, AccountResult, NotificationStats + +__all__ = [ + 'EmailConfig', + 'WebhookConfig', + 'PushPlusConfig', + 'ServerPushConfig', + 'NotificationData', + 'AccountResult', + 'NotificationStats', +] diff --git a/models/email_config.py b/models/email_config.py new file mode 100644 index 0000000..b7b3760 --- /dev/null +++ b/models/email_config.py @@ -0,0 +1,25 @@ +from typing import Dict, Any, Optional +from dataclasses import dataclass + + +@dataclass +class EmailConfig: + """邮件配置参数类""" + + # 发件人邮箱地址 + user: str + + # 发件人邮箱密码或授权码 + password: str + + # 收件人邮箱地址 + to: str + + # 自定义 SMTP 服务器地址（可选），如果不指定则自动从邮箱地址推断 + smtp_server: Optional[str] = None + + # 平台设置 + platform_settings: Optional[Dict[str, Any]] = None + + # 模板内容，如果为空则使用默认模板 + template: Optional[str] = None diff --git a/models/notification_data.py b/models/notification_data.py new file mode 100644 index 0000000..a13c3cc --- /dev/null +++ b/models/notification_data.py @@ -0,0 +1,65 @@ +from typing import List, Optional +from dataclasses import dataclass + + +@dataclass +class AccountResult: + """单个账号的处理结果""" + + # 账号名称 + name: str + + # 处理状态：success 或 failed + status: str + + # 当前余额，成功时才有 + quota: Optional[float] = None + + # 已使用余额，成功时才有 + used: Optional[float] = None + + # 错误信息，失败时才有 + error: Optional[str] = None + + +@dataclass +class NotificationStats: + """通知统计信息""" + + # 成功数量 + success_count: int + + # 失败数量 + failed_count: int + + # 总数量 + total_count: int + + +@dataclass +class NotificationData: + """通知数据结构""" + + # 账号列表和处理结果 + accounts: List[AccountResult] + + # 统计信息 + stats: NotificationStats + + # 执行时间戳 + timestamp: Optional[str] = None + + @property + def all_success(self) -> bool: + """是否全部成功""" + return self.stats.failed_count == 0 + + @property + def all_failed(self) -> bool: + """是否全部失败""" + return self.stats.success_count == 0 + + @property + def partial_success(self) -> bool: + """是否部分成功""" + return self.stats.success_count > 0 and self.stats.failed_count > 0 diff --git a/models/pushplus_config.py b/models/pushplus_config.py new file mode 100644 index 0000000..fb03a29 --- /dev/null +++ b/models/pushplus_config.py @@ -0,0 +1,16 @@ +from typing import Dict, Any, Optional +from dataclasses import dataclass + + +@dataclass +class PushPlusConfig: + """PushPlus 配置参数类""" + + # PushPlus Token + token: str + + # 平台设置 + platform_settings: Optional[Dict[str, Any]] = None + + # 模板内容，如果为空则使用默认模板 + template: Optional[str] = None diff --git a/models/serverpush_config.py b/models/serverpush_config.py new file mode 100644 index 0000000..2c71db8 --- /dev/null +++ b/models/serverpush_config.py @@ -0,0 +1,16 @@ +from typing import Dict, Any, Optional +from dataclasses import dataclass + + +@dataclass +class ServerPushConfig: + """Server 酱配置参数类""" + + # Server 酱 SendKey + send_key: str + + # 平台设置 + platform_settings: Optional[Dict[str, Any]] = None + + # 模板内容，如果为空则使用默认模板 + template: Optional[str] = None diff --git a/models/webhook_config.py b/models/webhook_config.py new file mode 100644 index 0000000..ba3f26f --- /dev/null +++ b/models/webhook_config.py @@ -0,0 +1,16 @@ +from typing import Dict, Any, Optional +from dataclasses import dataclass + + +@dataclass +class WebhookConfig: + """Webhook 配置参数类""" + + # Webhook URL 地址 + webhook: str + + # 平台设置，比如 use_card、color_theme 等 + platform_settings: Optional[Dict[str, Any]] = None + + # 模板内容，如果为空则使用默认模板 + template: Optional[str] = None diff --git a/notif_config/dingtalk.json b/notif_config/dingtalk.json new file mode 100644 index 0000000..615ae16 --- /dev/null +++ b/notif_config/dingtalk.json @@ -0,0 +1,4 @@ +{ + "platform_settings": {}, + "template": "{# 钉钉通知模板 #}{% if timestamp %}[TIME] Execution time: {{ timestamp }}\n\n{% endif %}{% if success_accounts %}{% for account in success_accounts %}[BALANCE] {{ account.name }}\n:money: Current balance: ${{ account.quota }}, Used: ${{ account.used }}\n{% endfor %}{% endif %}{% if failed_accounts %}{% for account in failed_accounts %}[FAIL] {{ account.name }} exception: {{ account.error }}\n{% endfor %}{% endif %}\n[STATS] Check-in result statistics:\n[SUCCESS] Success: {{ stats.success_count }}/{{ stats.total_count }}\n[FAIL] Failed: {{ stats.failed_count }}/{{ stats.total_count }}\n{% if all_success %}[SUCCESS] All accounts check-in successful!{% else %}{% if partial_success %}[WARN] Some accounts check-in successful{% else %}[ERROR] All accounts check-in failed{% endif %}{% endif %}" +} \ No newline at end of file diff --git a/notif_config/email.json b/notif_config/email.json new file mode 100644 index 0000000..ab90cdd --- /dev/null +++ b/notif_config/email.json @@ -0,0 +1,4 @@ +{ + "platform_settings": {}, + "template": "{# 邮箱通知模板 #}{% if timestamp %}[TIME] Execution time: {{ timestamp }}\n\n{% endif %}{% if success_accounts %}{% for account in success_accounts %}[BALANCE] {{ account.name }}\n:money: Current balance: ${{ account.quota }}, Used: ${{ account.used }}\n{% endfor %}{% endif %}{% if failed_accounts %}{% for account in failed_accounts %}[FAIL] {{ account.name }} exception: {{ account.error }}\n{% endfor %}{% endif %}\n[STATS] Check-in result statistics:\n[SUCCESS] Success: {{ stats.success_count }}/{{ stats.total_count }}\n[FAIL] Failed: {{ stats.failed_count }}/{{ stats.total_count }}\n{% if all_success %}[SUCCESS] All accounts check-in successful!{% else %}{% if partial_success %}[WARN] Some accounts check-in successful{% else %}[ERROR] All accounts check-in failed{% endif %}{% endif %}" +} \ No newline at end of file diff --git a/notif_config/feishu.json b/notif_config/feishu.json new file mode 100644 index 0000000..0def9cc --- /dev/null +++ b/notif_config/feishu.json @@ -0,0 +1,7 @@ +{ + "platform_settings": { + "use_card": true, + "color_theme": "blue" + }, + "template": "{# 飞书通知模板 #}{% if timestamp %}**[TIME] Execution time:** {{ timestamp }}\n\n{% endif %}{% if success_accounts %}{% for account in success_accounts %}**[BALANCE] {{ account.name }}**\n:money: Current balance: ${{ account.quota }}, Used: ${{ account.used }}\n{% endfor %}{% endif %}{% if failed_accounts %}{% for account in failed_accounts %}**[FAIL] {{ account.name }} exception:** {{ account.error }}\n{% endfor %}{% endif %}\n**[STATS] Check-in result statistics:**\n**[SUCCESS] Success:** {{ stats.success_count }}/{{ stats.total_count }}\n**[FAIL] Failed:** {{ stats.failed_count }}/{{ stats.total_count }}\n{% if all_success %}**[SUCCESS] All accounts check-in successful!**{% else %}{% if partial_success %}**[WARN] Some accounts check-in successful**{% else %}**[ERROR] All accounts check-in failed**{% endif %}{% endif %}" +} \ No newline at end of file diff --git a/notif_config/pushplus.json b/notif_config/pushplus.json new file mode 100644 index 0000000..c405124 --- /dev/null +++ b/notif_config/pushplus.json @@ -0,0 +1,4 @@ +{ + "platform_settings": {}, + "template": "{# PushPlus 通知模板 #}{% if timestamp %}[TIME] Execution time: {{ timestamp }}\n\n{% endif %}{% if success_accounts %}{% for account in success_accounts %}[BALANCE] {{ account.name }}\n:money: Current balance: ${{ account.quota }}, Used: ${{ account.used }}\n{% endfor %}{% endif %}{% if failed_accounts %}{% for account in failed_accounts %}[FAIL] {{ account.name }} exception: {{ account.error }}\n{% endfor %}{% endif %}\n[STATS] Check-in result statistics:\n[SUCCESS] Success: {{ stats.success_count }}/{{ stats.total_count }}\n[FAIL] Failed: {{ stats.failed_count }}/{{ stats.total_count }}\n{% if all_success %}[SUCCESS] All accounts check-in successful!{% else %}{% if partial_success %}[WARN] Some accounts check-in successful{% else %}[ERROR] All accounts check-in failed{% endif %}{% endif %}" +} \ No newline at end of file diff --git a/notif_config/serverpush.json b/notif_config/serverpush.json new file mode 100644 index 0000000..2695659 --- /dev/null +++ b/notif_config/serverpush.json @@ -0,0 +1,4 @@ +{ + "platform_settings": {}, + "template": "{# Server酱通知模板 #}{% if timestamp %}[TIME] Execution time: {{ timestamp }}\n\n{% endif %}{% if success_accounts %}{% for account in success_accounts %}[BALANCE] {{ account.name }}\n:money: Current balance: ${{ account.quota }}, Used: ${{ account.used }}\n{% endfor %}{% endif %}{% if failed_accounts %}{% for account in failed_accounts %}[FAIL] {{ account.name }} exception: {{ account.error }}\n{% endfor %}{% endif %}\n[STATS] Check-in result statistics:\n[SUCCESS] Success: {{ stats.success_count }}/{{ stats.total_count }}\n[FAIL] Failed: {{ stats.failed_count }}/{{ stats.total_count }}\n{% if all_success %}[SUCCESS] All accounts check-in successful!{% else %}{% if partial_success %}[WARN] Some accounts check-in successful{% else %}[ERROR] All accounts check-in failed{% endif %}{% endif %}" +} \ No newline at end of file diff --git a/notif_config/wecom.json b/notif_config/wecom.json new file mode 100644 index 0000000..2833f80 --- /dev/null +++ b/notif_config/wecom.json @@ -0,0 +1,6 @@ +{ + "platform_settings": { + "markdown_style": "markdown" + }, + "template": "{# 企业微信通知模板 #}{% if timestamp %}**[TIME] Execution time:** {{ timestamp }}\\n\\n{% endif %}{% if success_accounts %}{% for account in success_accounts %}**[BALANCE] {{ account.name }}**\\n:money: Current balance: ${{ account.quota }}, Used: ${{ account.used }}\\n{% endfor %}{% endif %}{% if failed_accounts %}{% for account in failed_accounts %}**[FAIL] {{ account.name }} exception:** {{ account.error }}\\n{% endfor %}{% endif %}\\n**[STATS] Check-in result statistics:**\\n**[SUCCESS] Success:** {{ stats.success_count }}/{{ stats.total_count }}\\n**[FAIL] Failed:** {{ stats.failed_count }}/{{ stats.total_count }}\\n{% if all_success %}**[SUCCESS] All accounts check-in successful!**{% else %}{% if partial_success %}**[WARN] Some accounts check-in successful**{% else %}**[ERROR] All accounts check-in failed**{% endif %}{% endif %}" +} \ No newline at end of file diff --git a/notify.py b/notify.py index d0beb5d..97d76a2 100644 --- a/notify.py +++ b/notify.py @@ -1,103 +1,453 @@ +import json import os import smtplib from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText -from typing import Literal +from pathlib import Path +from typing import Dict, Any, Optional, Literal, Union import httpx +import stencil + +from models import ( + EmailConfig, + WebhookConfig, + PushPlusConfig, + ServerPushConfig, + NotificationData, +) class NotificationKit: def __init__(self): - self.email_user: str = os.getenv('EMAIL_USER', '') - self.email_pass: str = os.getenv('EMAIL_PASS', '') - self.email_to: str = os.getenv('EMAIL_TO', '') - self.smtp_server: str = os.getenv('CUSTOM_SMTP_SERVER', '') - self.pushplus_token = os.getenv('PUSHPLUS_TOKEN') - self.server_push_key = os.getenv('SERVERPUSHKEY') - self.dingding_webhook = os.getenv('DINGDING_WEBHOOK') - self.feishu_webhook = os.getenv('FEISHU_WEBHOOK') - self.weixin_webhook = os.getenv('WEIXIN_WEBHOOK') + # 配置文件路径 + self.config_dir = Path(__file__).parent / 'notif_config' + + # 加载各平台配置 + self.email_config = self._load_email_config() + self.dingtalk_config = self._load_dingtalk_config() + self.feishu_config = self._load_feishu_config() + self.wecom_config = self._load_wecom_config() + self.pushplus_config = self._load_pushplus_config() + self.serverpush_config = self._load_serverpush_config() def send_email(self, title: str, content: str, msg_type: Literal['text', 'html'] = 'text'): - if not self.email_user or not self.email_pass or not self.email_to: - raise ValueError('Email configuration not set') + if not self.email_config: + raise ValueError('未配置邮箱信息') msg = MIMEMultipart() - msg['From'] = f'AnyRouter Assistant <{self.email_user}>' - msg['To'] = self.email_to + msg['From'] = f'AnyRouter Assistant <{self.email_config.user}>' + msg['To'] = self.email_config.to msg['Subject'] = title body = MIMEText(content, msg_type, 'utf-8') msg.attach(body) - smtp_server = self.smtp_server if self.smtp_server else f'smtp.{self.email_user.split("@")[1]}' + smtp_server = self.email_config.smtp_server if self.email_config.smtp_server else f'smtp.{self.email_config.user.split("@")[1]}' with smtplib.SMTP_SSL(smtp_server, 465) as server: - server.login(self.email_user, self.email_pass) + server.login(self.email_config.user, self.email_config.password) server.send_message(msg) def send_pushplus(self, title: str, content: str): - if not self.pushplus_token: - raise ValueError('PushPlus Token not configured') + if not self.pushplus_config: + raise ValueError('未配置PushPlus Token') - data = {'token': self.pushplus_token, 'title': title, 'content': content, 'template': 'html'} + data = { + 'token': self.pushplus_config.token, + 'title': title, + 'content': content, + 'template': 'html' + } with httpx.Client(timeout=30.0) as client: client.post('http://www.pushplus.plus/send', json=data) - def send_serverPush(self, title: str, content: str): - if not self.server_push_key: - raise ValueError('Server Push key not configured') + def send_serverpush(self, title: str, content: str): + if not self.serverpush_config: + raise ValueError('未配置Server Push key') data = {'title': title, 'desp': content} with httpx.Client(timeout=30.0) as client: - client.post(f'https://sctapi.ftqq.com/{self.server_push_key}.send', json=data) + client.post(f'https://sctapi.ftqq.com/{self.serverpush_config.send_key}.send', json=data) def send_dingtalk(self, title: str, content: str): - if not self.dingding_webhook: - raise ValueError('DingTalk Webhook not configured') + if not self.dingtalk_config: + raise ValueError('未配置钉钉 Webhook') data = {'msgtype': 'text', 'text': {'content': f'{title}\n{content}'}} with httpx.Client(timeout=30.0) as client: - client.post(self.dingding_webhook, json=data) + client.post(self.dingtalk_config.webhook, json=data) def send_feishu(self, title: str, content: str): - if not self.feishu_webhook: - raise ValueError('Feishu Webhook not configured') + if not self.feishu_config: + raise ValueError('未配置飞书 Webhook') + + # 检查是否使用卡片模式（确保 platform_settings 不为 None） + platform_settings = self.feishu_config.platform_settings or {} + use_card = platform_settings.get('use_card', True) + color_theme = platform_settings.get('color_theme', 'blue') + + if use_card: + data = { + 'msg_type': 'interactive', + 'card': { + 'elements': [{'tag': 'markdown', 'content': content, 'text_align': 'left'}], + 'header': {'template': color_theme, 'title': {'content': title, 'tag': 'plain_text'}}, + }, + } + else: + data = {'msg_type': 'text', 'text': {'content': f'{title}\n{content}'}} - data = { - 'msg_type': 'interactive', - 'card': { - 'elements': [{'tag': 'markdown', 'content': content, 'text_align': 'left'}], - 'header': {'template': 'blue', 'title': {'content': title, 'tag': 'plain_text'}}, - }, - } with httpx.Client(timeout=30.0) as client: - client.post(self.feishu_webhook, json=data) + client.post(self.feishu_config.webhook, json=data) def send_wecom(self, title: str, content: str): - if not self.weixin_webhook: - raise ValueError('WeChat Work Webhook not configured') + if not self.wecom_config: + raise ValueError('未配置企业微信 Webhook') + + # 检查 markdown_style 配置（确保 platform_settings 不为 None） + platform_settings = self.wecom_config.platform_settings or {} + markdown_style = platform_settings.get('markdown_style', 'markdown') + + # 根据 markdown_style 选择消息格式 + # 如果 markdown_style 包含 'markdown'，则直接作为消息类型；否则使用 text 模式 + if markdown_style and 'markdown' in str(markdown_style): + data = {'msgtype': markdown_style, markdown_style: {'content': f'**{title}**\n{content}'}} + else: + data = {'msgtype': 'text', 'text': {'content': f'{title}\n{content}'}} - data = {'msgtype': 'text', 'text': {'content': f'{title}\n{content}'}} with httpx.Client(timeout=30.0) as client: - client.post(self.weixin_webhook, json=data) + client.post(self.wecom_config.webhook, json=data) + + def push_message(self, title: str, content: Union[str, NotificationData], msg_type: Literal['text', 'html'] = 'text'): + """发送通知消息 - def push_message(self, title: str, content: str, msg_type: Literal['text', 'html'] = 'text'): + Args: + title: 消息标题 + content: 消息内容，可以是字符串（向后兼容）或 NotificationData 结构 + msg_type: 消息类型 + """ notifications = [ - ('Email', lambda: self.send_email(title, content, msg_type)), - ('PushPlus', lambda: self.send_pushplus(title, content)), - ('Server Push', lambda: self.send_serverPush(title, content)), - ('DingTalk', lambda: self.send_dingtalk(title, content)), - ('Feishu', lambda: self.send_feishu(title, content)), - ('WeChat Work', lambda: self.send_wecom(title, content)), + ('邮箱', self._send_email_with_template), + ('PushPlus', self._send_pushplus_with_template), + ('Server 酱', self._send_serverpush_with_template), + ('钉钉', self._send_dingtalk_with_template), + ('飞书', self._send_feishu_with_template), + ('企业微信', self._send_wecom_with_template), ] for name, func in notifications: try: - func() + func(title, content, msg_type) print(f'[{name}]: Message push successful!') except Exception as e: print(f'[{name}]: Message push failed! Reason: {str(e)}') + # 配置加载方法 + def _load_default_config(self, platform: str) -> Optional[Dict[str, Any]]: + """加载默认配置文件""" + config_file = self.config_dir / f'{platform}.json' + if config_file.exists(): + try: + with open(config_file, 'r', encoding='utf-8') as f: + return json.load(f) + except Exception as e: + print(f'Warning: Failed to load default config file {config_file}: {e}') + return None + + def _parse_env_config(self, env_value: str) -> Union[str, Dict[str, Any]]: + """解析环境变量配置""" + try: + # 尝试解析为 JSON + return json.loads(env_value) + except json.JSONDecodeError: + # 如果不是 JSON，就当做纯字符串（Token 或 Webhook URL） + return env_value + + def _load_email_config(self) -> Optional[EmailConfig]: + """加载邮箱配置""" + # 检查新的配置方式 + email_notif_config = os.getenv('EMAIL_NOTIF_CONFIG') + if email_notif_config: + parsed = self._parse_env_config(email_notif_config) + if isinstance(parsed, dict): + # 如果 template 为 None，则使用默认模板 + template = parsed.get('template') + if template is None: + default_config = self._load_default_config('email') + template = default_config.get('template') if default_config else None + + return EmailConfig( + user=parsed['user'], + password=parsed['pass'], + to=parsed['to'], + smtp_server=parsed.get('smtp_server'), + platform_settings=parsed.get('platform_settings'), + template=template + ) + + # 检查旧的配置方式 + email_user = os.getenv('EMAIL_USER') + email_pass = os.getenv('EMAIL_PASS') + email_to = os.getenv('EMAIL_TO') + custom_smtp = os.getenv('CUSTOM_SMTP_SERVER') + + if email_user and email_pass and email_to: + # 加载默认模板 + default_config = self._load_default_config('email') + return EmailConfig( + user=email_user, + password=email_pass, + to=email_to, + smtp_server=custom_smtp, + platform_settings=default_config.get('platform_settings') if default_config else None, + template=default_config.get('template') if default_config else None + ) + + return None + + def _load_webhook_config(self, platform: str, env_key: str, notif_config_key: str) -> Optional[WebhookConfig]: + """加载 Webhook 配置的通用方法""" + # 检查新的配置方式 + notif_config = os.getenv(notif_config_key) + if notif_config: + parsed = self._parse_env_config(notif_config) + if isinstance(parsed, dict): + # 如果 template 为 None，则使用默认模板 + template = parsed.get('template') + if template is None: + default_config = self._load_default_config(platform) + template = default_config.get('template') if default_config else None + + return WebhookConfig( + webhook=parsed['webhook'], + platform_settings=parsed.get('platform_settings'), + template=template + ) + else: + # 纯字符串，当做 webhook URL，使用默认模板 + default_config = self._load_default_config(platform) + return WebhookConfig( + webhook=parsed, + platform_settings=default_config.get('platform_settings') if default_config else None, + template=default_config.get('template') if default_config else None + ) + + # 检查旧的配置方式 + old_webhook = os.getenv(env_key) + if old_webhook: + default_config = self._load_default_config(platform) + return WebhookConfig( + webhook=old_webhook, + platform_settings=default_config.get('platform_settings') if default_config else None, + template=default_config.get('template') if default_config else None + ) + + return None + + def _load_dingtalk_config(self) -> Optional[WebhookConfig]: + return self._load_webhook_config('dingtalk', 'DINGDING_WEBHOOK', 'DINGTALK_NOTIF_CONFIG') + + def _load_feishu_config(self) -> Optional[WebhookConfig]: + return self._load_webhook_config('feishu', 'FEISHU_WEBHOOK', 'FEISHU_NOTIF_CONFIG') + + def _load_wecom_config(self) -> Optional[WebhookConfig]: + return self._load_webhook_config('wecom', 'WEIXIN_WEBHOOK', 'WECOM_NOTIF_CONFIG') + + def _load_pushplus_config(self) -> Optional[PushPlusConfig]: + """加载 PushPlus 配置""" + # 检查新的配置方式 + pushplus_notif_config = os.getenv('PUSHPLUS_NOTIF_CONFIG') + if pushplus_notif_config: + parsed = self._parse_env_config(pushplus_notif_config) + if isinstance(parsed, dict): + # 如果 template 为 None，则使用默认模板 + template = parsed.get('template') + if template is None: + default_config = self._load_default_config('pushplus') + template = default_config.get('template') if default_config else None + + return PushPlusConfig( + token=parsed['token'], + platform_settings=parsed.get('platform_settings'), + template=template + ) + else: + # 纯字符串，当做 token，使用默认模板 + default_config = self._load_default_config('pushplus') + return PushPlusConfig( + token=parsed, + platform_settings=default_config.get('platform_settings') if default_config else None, + template=default_config.get('template') if default_config else None + ) + + # 检查旧的配置方式 + pushplus_token = os.getenv('PUSHPLUS_TOKEN') + if pushplus_token: + default_config = self._load_default_config('pushplus') + return PushPlusConfig( + token=pushplus_token, + platform_settings=default_config.get('platform_settings') if default_config else None, + template=default_config.get('template') if default_config else None + ) + + return None + + def _load_serverpush_config(self) -> Optional[ServerPushConfig]: + """加载 Server 酱配置""" + # 检查新的配置方式 + serverpush_notif_config = os.getenv('SERVERPUSH_NOTIF_CONFIG') + if serverpush_notif_config: + parsed = self._parse_env_config(serverpush_notif_config) + if isinstance(parsed, dict): + # 如果 template 为 None，则使用默认模板 + template = parsed.get('template') + if template is None: + default_config = self._load_default_config('serverpush') + template = default_config.get('template') if default_config else None + + return ServerPushConfig( + send_key=parsed['send_key'], + platform_settings=parsed.get('platform_settings'), + template=template + ) + else: + # 纯字符串，当做 send_key，使用默认模板 + default_config = self._load_default_config('serverpush') + return ServerPushConfig( + send_key=parsed, + platform_settings=default_config.get('platform_settings') if default_config else None, + template=default_config.get('template') if default_config else None + ) + + # 检查旧的配置方式 + server_push_key = os.getenv('SERVERPUSHKEY') + if server_push_key: + default_config = self._load_default_config('serverpush') + return ServerPushConfig( + send_key=server_push_key, + platform_settings=default_config.get('platform_settings') if default_config else None, + template=default_config.get('template') if default_config else None + ) + + return None + + # 模板渲染方法 + def _render_template(self, template: str, data: NotificationData) -> str: + """渲染模板 + + 注意: Stencil 模板引擎有以下限制: + 1. 不支持比较操作符 (==, !=, <, > 等) + 2. 不支持字典的点访问，只能访问对象属性 + 因此我们提供分组的账号列表和对象形式的数据 + """ + try: + # 分组账号（因为 Stencil 不支持 == 比较） + success_accounts = [acc for acc in data.accounts if acc.status == 'success'] + failed_accounts = [acc for acc in data.accounts if acc.status != 'success'] + + context_data = { + 'timestamp': data.timestamp, + 'stats': data.stats, # dataclass 对象，支持 {{ stats.success_count }} + + # 提供分组的账号列表（AccountResult 对象） + 'success_accounts': success_accounts, + 'failed_accounts': failed_accounts, + + # 保留完整列表供需要的模板使用 + 'accounts': data.accounts, # AccountResult 对象列表 + + # 便利变量：布尔标志 + 'has_success': len(success_accounts) > 0, + 'has_failed': len(failed_accounts) > 0, + 'all_success': len(failed_accounts) == 0, + 'all_failed': len(success_accounts) == 0, + 'partial_success': len(success_accounts) > 0 and len(failed_accounts) > 0, + } + + # 解析并渲染模板 + template_obj = stencil.Template(template) + context = stencil.Context(context_data) + rendered_result = template_obj.render(context) + + # 处理换行符：将 \\n 转换为真正的换行符 + rendered_result = rendered_result.replace('\\n', '\n') + + return rendered_result + except Exception as e: + print(f'ERROR: Template rendering failed: {e}') + # 如果模板渲染失败，返回简单格式 + return f'{data.timestamp}\n\n' + '\n\n'.join([ + f'[{"SUCCESS" if account.status == "success" else "FAIL"}] {account.name}' + for account in data.accounts + ]) + + # 带模板的发送方法 + def _send_email_with_template(self, title: str, content: Union[str, NotificationData], msg_type: Literal['text', 'html'] = 'text'): + if not self.email_config: + return + + if isinstance(content, NotificationData) and self.email_config.template: + rendered_content = self._render_template(self.email_config.template, content) + else: + rendered_content = str(content) + + self.send_email(title, rendered_content, msg_type) + + def _send_pushplus_with_template(self, title: str, content: Union[str, NotificationData], msg_type: Literal['text', 'html'] = 'text'): + if not self.pushplus_config: + return + + if isinstance(content, NotificationData) and self.pushplus_config.template: + rendered_content = self._render_template(self.pushplus_config.template, content) + else: + rendered_content = str(content) + + self.send_pushplus(title, rendered_content) + + def _send_serverpush_with_template(self, title: str, content: Union[str, NotificationData], msg_type: Literal['text', 'html'] = 'text'): + if not self.serverpush_config: + return + + if isinstance(content, NotificationData) and self.serverpush_config.template: + rendered_content = self._render_template(self.serverpush_config.template, content) + else: + rendered_content = str(content) + + self.send_serverpush(title, rendered_content) + + def _send_dingtalk_with_template(self, title: str, content: Union[str, NotificationData], msg_type: Literal['text', 'html'] = 'text'): + if not self.dingtalk_config: + return + + if isinstance(content, NotificationData) and self.dingtalk_config.template: + rendered_content = self._render_template(self.dingtalk_config.template, content) + else: + rendered_content = str(content) + + self.send_dingtalk(title, rendered_content) + + def _send_feishu_with_template(self, title: str, content: Union[str, NotificationData], msg_type: Literal['text', 'html'] = 'text'): + if not self.feishu_config: + return + + if isinstance(content, NotificationData) and self.feishu_config.template: + rendered_content = self._render_template(self.feishu_config.template, content) + else: + rendered_content = str(content) + + self.send_feishu(title, rendered_content) + + def _send_wecom_with_template(self, title: str, content: Union[str, NotificationData], msg_type: Literal['text', 'html'] = 'text'): + if not self.wecom_config: + return + + if isinstance(content, NotificationData) and self.wecom_config.template: + rendered_content = self._render_template(self.wecom_config.template, content) + else: + rendered_content = str(content) + + self.send_wecom(title, rendered_content) + + notify = NotificationKit() diff --git a/pyproject.toml b/pyproject.toml index abcd3c8..fa8548f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -6,7 +6,8 @@ requires-python = ">=3.11" dependencies = [ "httpx[http2]>=0.24.0", "playwright>=1.40.0", - "python-dotenv>=1.0.0" + "python-dotenv>=1.0.0", + "stencil-template>=4.0.0" ] [dependency-groups] diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..f0c55e0 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +"""Tests 包初始化文件""" \ No newline at end of file diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..9cb3065 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,12 @@ +import sys +from pathlib import Path + +from dotenv import load_dotenv + +# 添加项目根目录到 PATH +project_root = Path(__file__).parent.parent +sys.path.insert(0, str(project_root)) + +# 加载环境变量 +# 禁用变量插值以保留模板中的 $ 符号 +load_dotenv(project_root / '.env', interpolate=False) diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/integration/test_real_notifications.py b/tests/integration/test_real_notifications.py new file mode 100644 index 0000000..8da8d9f --- /dev/null +++ b/tests/integration/test_real_notifications.py @@ -0,0 +1,103 @@ +import os +from unittest.mock import patch + +import pytest + + +class TestIntegration: + """集成测试 - 测试端到端流程""" + + @pytest.fixture + def notification_kit_from_env(self): + """从环境变量创建 NotificationKit 实例（用于集成测试）""" + from notify import NotificationKit + return NotificationKit() + + @pytest.fixture + def test_notification_data(self): + """创建测试用的通知数据""" + from models import NotificationData, AccountResult, NotificationStats + + return NotificationData( + accounts=[ + AccountResult( + name='测试账号', + status='success', + quota=25.0, + used=5.0, + error=None + ) + ], + stats=NotificationStats(success_count=1, failed_count=0, total_count=1), + timestamp='2024-01-01 12:00:00' + ) + + def test_real_notification_with_env_config( + self, + notification_kit_from_env, + test_notification_data + ): + """ + 真实接口测试 - 需要在 .env.local 文件中配置 ENABLE_REAL_TEST=true + + 此测试会实际发送通知到配置的平台，用于验证端到端流程。 + """ + if os.getenv('ENABLE_REAL_TEST') != 'true': + pytest.skip('未启用真实接口测试。请在 .env.local 中设置 ENABLE_REAL_TEST=true') + + # 尝试发送通知 + notification_kit_from_env.push_message('集成测试消息', test_notification_data) + + # 如果没有抛出异常，则测试通过 + # 注意：这个测试不验证通知是否真的成功发送，只验证代码执行流程正确 + + @patch('notify.NotificationKit._send_email_with_template') + @patch('notify.NotificationKit._send_dingtalk_with_template') + @patch('notify.NotificationKit._send_wecom_with_template') + @patch('notify.NotificationKit._send_pushplus_with_template') + @patch('notify.NotificationKit._send_feishu_with_template') + @patch('notify.NotificationKit._send_serverpush_with_template') + def test_push_message_routing_logic( + self, + mock_serverpush, + mock_feishu, + mock_pushplus, + mock_wecom, + mock_dingtalk, + mock_email, + notification_kit_from_env, + test_notification_data + ): + """测试 push_message 的路由逻辑 - 验证根据配置调用相应平台""" + notification_kit_from_env.push_message('测试标题', test_notification_data) + + # 验证有配置的平台被调用 + if notification_kit_from_env.email_config: + assert mock_email.called + else: + assert not mock_email.called + + if notification_kit_from_env.dingtalk_config: + assert mock_dingtalk.called + else: + assert not mock_dingtalk.called + + if notification_kit_from_env.wecom_config: + assert mock_wecom.called + else: + assert not mock_wecom.called + + if notification_kit_from_env.pushplus_config: + assert mock_pushplus.called + else: + assert not mock_pushplus.called + + if notification_kit_from_env.feishu_config: + assert mock_feishu.called + else: + assert not mock_feishu.called + + if notification_kit_from_env.serverpush_config: + assert mock_serverpush.called + else: + assert not mock_serverpush.called diff --git a/tests/test_notify.py b/tests/test_notify.py deleted file mode 100644 index fe5ad32..0000000 --- a/tests/test_notify.py +++ /dev/null @@ -1,105 +0,0 @@ -import os -import sys -from datetime import datetime -from pathlib import Path -from unittest.mock import MagicMock, patch - -import pytest -from dotenv import load_dotenv - -# 添加项目根目录到 PATH -project_root = Path(__file__).parent.parent -sys.path.insert(0, str(project_root)) - -load_dotenv(project_root / '.env') - -from notify import NotificationKit - - -@pytest.fixture -def notification_kit(): - return NotificationKit() - - -def test_real_notification(notification_kit): - """真实接口测试，需要配置.env.local文件""" - if os.getenv('ENABLE_REAL_TEST') != 'true': - pytest.skip('未启用真实接口测试') - - notification_kit.push_message( - '测试消息', f'这是一条测试消息\n发送时间: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}' - ) - - -@patch('smtplib.SMTP_SSL') -def test_send_email(mock_smtp, notification_kit): - mock_server = MagicMock() - mock_smtp.return_value.__enter__.return_value = mock_server - - notification_kit.send_email('测试标题', '测试内容') - - assert mock_server.login.called - assert mock_server.send_message.called - - -@patch('requests.post') -def test_send_pushplus(mock_post, notification_kit): - notification_kit.send_pushplus('测试标题', '测试内容') - - mock_post.assert_called_once() - args = mock_post.call_args[1] - assert 'test_token' in str(args) - - -@patch('requests.post') -def test_send_dingtalk(mock_post, notification_kit): - notification_kit.send_dingtalk('测试标题', '测试内容') - - expected_webhook = 'https://oapi.dingtalk.com/robot/send?access_token=fbcd45f32f17dea5c762e82644c7f28945075e0b4d22953c8eebe064b106a96f' - expected_data = {'msgtype': 'text', 'text': {'content': '测试标题\n测试内容'}} - - mock_post.assert_called_once_with(expected_webhook, json=expected_data) - - -@patch('requests.post') -def test_send_feishu(mock_post, notification_kit): - notification_kit.send_feishu('测试标题', '测试内容') - - mock_post.assert_called_once() - args = mock_post.call_args[1] - assert 'card' in args['json'] - - -@patch('requests.post') -def test_send_wecom(mock_post, notification_kit): - notification_kit.send_wecom('测试标题', '测试内容') - - mock_post.assert_called_once_with( - 'http://weixin.example.com', json={'msgtype': 'text', 'text': {'content': '测试标题\n测试内容'}} - ) - - -def test_missing_config(): - os.environ.clear() - kit = NotificationKit() - - with pytest.raises(ValueError, match='未配置邮箱信息'): - kit.send_email('测试', '测试') - - with pytest.raises(ValueError, match='未配置PushPlus Token'): - kit.send_pushplus('测试', '测试') - - -@patch('anyrouter.notify.NotificationKit.send_email') -@patch('anyrouter.notify.NotificationKit.send_dingtalk') -@patch('anyrouter.notify.NotificationKit.send_wecom') -@patch('anyrouter.notify.NotificationKit.send_pushplus') -@patch('anyrouter.notify.NotificationKit.send_feishu') -def test_push_message(mock_feishu, mock_pushplus, mock_wecom, mock_dingtalk, mock_email, notification_kit): - notification_kit.push_message('测试标题', '测试内容') - - assert mock_email.called - assert mock_dingtalk.called - assert mock_wecom.called - assert mock_pushplus.called - assert mock_feishu.called diff --git a/tests/unit/__init__.py b/tests/unit/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/test_config_parsing.py b/tests/unit/test_config_parsing.py new file mode 100644 index 0000000..2e1fadb --- /dev/null +++ b/tests/unit/test_config_parsing.py @@ -0,0 +1,303 @@ +import json +import os + +import pytest + + +class TestConfigParsing: + """测试配置解析功能""" + + def test_no_config_returns_none(self, monkeypatch): + """测试无配置时所有配置为 None""" + # 清空所有相关环境变量 + for key in list(os.environ.keys()): + if 'NOTIF_CONFIG' in key or key in ['EMAIL_USER', 'WEIXIN_WEBHOOK', 'DINGDING_WEBHOOK', 'FEISHU_WEBHOOK', 'PUSHPLUS_TOKEN', 'SERVERPUSHKEY']: + monkeypatch.delenv(key, raising=False) + + from notify import NotificationKit + kit = NotificationKit() + + assert kit.email_config is None + assert kit.dingtalk_config is None + assert kit.feishu_config is None + assert kit.wecom_config is None + assert kit.pushplus_config is None + assert kit.serverpush_config is None + + @pytest.mark.parametrize('config_format,has_template,template_value', [ + ('json_with_custom_template', True, 'custom template content'), + ('json_with_null_template', False, None), + ('json_without_template_key', False, None), + ]) + def test_wecom_config_formats(self, monkeypatch, config_format, has_template, template_value): + """测试企业微信配置的各种格式""" + # 清空环境 + for key in list(os.environ.keys()): + if 'NOTIF_CONFIG' in key or 'WEBHOOK' in key: + monkeypatch.delenv(key, raising=False) + + # 构造配置 + if config_format == 'json_with_custom_template': + config = {'webhook': 'https://example.com/webhook', 'template': template_value} + elif config_format == 'json_with_null_template': + config = {'webhook': 'https://example.com/webhook', 'template': None} + else: # json_without_template_key + config = {'webhook': 'https://example.com/webhook'} + + monkeypatch.setenv('WECOM_NOTIF_CONFIG', json.dumps(config)) + + from notify import NotificationKit + kit = NotificationKit() + + assert kit.wecom_config is not None + assert kit.wecom_config.webhook == 'https://example.com/webhook' + + # 验证模板处理 + if has_template: + assert kit.wecom_config.template == template_value + else: + # template 为 null 或不存在时，应该加载默认模板 + assert kit.wecom_config.template is not None + assert len(kit.wecom_config.template) > 0 + + @pytest.mark.parametrize('config_format,webhook_value', [ + ('old_format_string', 'https://example.com/webhook'), + ]) + def test_wecom_old_format_loads_default_template(self, monkeypatch, config_format, webhook_value): + """测试企业微信旧格式配置（字符串形式）加载默认模板""" + # 清空环境 + for key in list(os.environ.keys()): + if 'NOTIF_CONFIG' in key or 'WEBHOOK' in key: + monkeypatch.delenv(key, raising=False) + + monkeypatch.setenv('WEIXIN_WEBHOOK', webhook_value) + + from notify import NotificationKit + kit = NotificationKit() + + assert kit.wecom_config is not None + assert kit.wecom_config.webhook == webhook_value + assert kit.wecom_config.template is not None + assert len(kit.wecom_config.template) > 0 + + @pytest.mark.parametrize('platform,env_key,config_value,expected_webhook', [ + ('dingtalk', 'DINGTALK_NOTIF_CONFIG', {'webhook': 'https://dingtalk.com/webhook', 'template': 'custom'}, 'https://dingtalk.com/webhook'), + ('dingtalk', 'DINGTALK_NOTIF_CONFIG', {'webhook': 'https://dingtalk.com/webhook', 'template': None}, 'https://dingtalk.com/webhook'), + ('feishu', 'FEISHU_NOTIF_CONFIG', {'webhook': 'https://feishu.com/webhook', 'template': 'custom'}, 'https://feishu.com/webhook'), + ('feishu', 'FEISHU_NOTIF_CONFIG', {'webhook': 'https://feishu.com/webhook', 'template': None}, 'https://feishu.com/webhook'), + ]) + def test_webhook_platforms_json_config( + self, + monkeypatch, + platform, + env_key, + config_value, + expected_webhook + ): + """测试钉钉、飞书等 webhook 平台的 JSON 配置""" + # 清空环境 + for key in list(os.environ.keys()): + if 'NOTIF_CONFIG' in key or 'WEBHOOK' in key: + monkeypatch.delenv(key, raising=False) + + monkeypatch.setenv(env_key, json.dumps(config_value)) + + from notify import NotificationKit + kit = NotificationKit() + + config_attr = f'{platform}_config' + config = getattr(kit, config_attr) + + assert config is not None + assert config.webhook == expected_webhook + + # 验证模板 + if config_value.get('template') is None: + assert config.template is not None + assert len(config.template) > 0 + else: + assert config.template == config_value['template'] + + @pytest.mark.parametrize('platform,env_key,webhook_value', [ + ('dingtalk', 'DINGDING_WEBHOOK', 'https://dingtalk.com/webhook'), + ('feishu', 'FEISHU_WEBHOOK', 'https://feishu.com/webhook'), + ]) + def test_webhook_platforms_old_format(self, monkeypatch, platform, env_key, webhook_value): + """测试钉钉、飞书旧格式配置（字符串）""" + # 清空环境 + for key in list(os.environ.keys()): + if 'NOTIF_CONFIG' in key or 'WEBHOOK' in key: + monkeypatch.delenv(key, raising=False) + + monkeypatch.setenv(env_key, webhook_value) + + from notify import NotificationKit + kit = NotificationKit() + + config_attr = f'{platform}_config' + config = getattr(kit, config_attr) + + assert config is not None + assert config.webhook == webhook_value + assert config.template is not None + assert len(config.template) > 0 + + @pytest.mark.parametrize('config_format,has_template', [ + ('json_with_custom_template', True), + ('json_with_null_template', False), + ]) + def test_pushplus_config_formats(self, monkeypatch, config_format, has_template): + """测试 PushPlus 配置的各种格式""" + # 清空环境 + for key in list(os.environ.keys()): + if 'NOTIF_CONFIG' in key or 'PUSHPLUS' in key: + monkeypatch.delenv(key, raising=False) + + # 构造配置 + if config_format == 'json_with_custom_template': + config = {'token': 'test_token_123', 'template': 'custom pushplus template'} + else: # json_with_null_template + config = {'token': 'test_token_123', 'template': None} + + monkeypatch.setenv('PUSHPLUS_NOTIF_CONFIG', json.dumps(config)) + + from notify import NotificationKit + kit = NotificationKit() + + assert kit.pushplus_config is not None + assert kit.pushplus_config.token == 'test_token_123' + + if has_template: + assert kit.pushplus_config.template == 'custom pushplus template' + else: + assert kit.pushplus_config.template is not None + assert len(kit.pushplus_config.template) > 0 + + def test_pushplus_old_format(self, monkeypatch): + """测试 PushPlus 旧格式配置""" + # 清空环境 + for key in list(os.environ.keys()): + if 'NOTIF_CONFIG' in key or 'PUSHPLUS' in key: + monkeypatch.delenv(key, raising=False) + + monkeypatch.setenv('PUSHPLUS_TOKEN', 'old_format_token') + + from notify import NotificationKit + kit = NotificationKit() + + assert kit.pushplus_config is not None + assert kit.pushplus_config.token == 'old_format_token' + assert kit.pushplus_config.template is not None + assert len(kit.pushplus_config.template) > 0 + + @pytest.mark.parametrize('config_format,has_template', [ + ('json_with_custom_template', True), + ('json_with_null_template', False), + ]) + def test_serverpush_config_formats(self, monkeypatch, config_format, has_template): + """测试 Server 酱配置的各种格式""" + # 清空环境 + for key in list(os.environ.keys()): + if 'NOTIF_CONFIG' in key or 'SERVERPUSH' in key: + monkeypatch.delenv(key, raising=False) + + # 构造配置 + if config_format == 'json_with_custom_template': + config = {'send_key': 'test_send_key', 'template': 'custom serverpush template'} + else: # json_with_null_template + config = {'send_key': 'test_send_key', 'template': None} + + monkeypatch.setenv('SERVERPUSH_NOTIF_CONFIG', json.dumps(config)) + + from notify import NotificationKit + kit = NotificationKit() + + assert kit.serverpush_config is not None + assert kit.serverpush_config.send_key == 'test_send_key' + + if has_template: + assert kit.serverpush_config.template == 'custom serverpush template' + else: + assert kit.serverpush_config.template is not None + assert len(kit.serverpush_config.template) > 0 + + def test_serverpush_old_format(self, monkeypatch): + """测试 Server 酱旧格式配置""" + # 清空环境 + for key in list(os.environ.keys()): + if 'NOTIF_CONFIG' in key or 'SERVERPUSH' in key: + monkeypatch.delenv(key, raising=False) + + monkeypatch.setenv('SERVERPUSHKEY', 'old_format_key') + + from notify import NotificationKit + kit = NotificationKit() + + assert kit.serverpush_config is not None + assert kit.serverpush_config.send_key == 'old_format_key' + assert kit.serverpush_config.template is not None + assert len(kit.serverpush_config.template) > 0 + + @pytest.mark.parametrize('config_format,has_template', [ + ('json_with_custom_template', True), + ('json_with_null_template', False), + ]) + def test_email_config_formats(self, monkeypatch, config_format, has_template): + """测试邮箱配置的各种格式""" + # 清空环境 + for key in list(os.environ.keys()): + if 'NOTIF_CONFIG' in key or 'EMAIL' in key: + monkeypatch.delenv(key, raising=False) + + # 构造配置 + if config_format == 'json_with_custom_template': + config = { + 'user': 'test@example.com', + 'pass': 'test_password', + 'to': 'recipient@example.com', + 'template': 'custom email template' + } + else: # json_with_null_template + config = { + 'user': 'test@example.com', + 'pass': 'test_password', + 'to': 'recipient@example.com', + 'template': None + } + + monkeypatch.setenv('EMAIL_NOTIF_CONFIG', json.dumps(config)) + + from notify import NotificationKit + kit = NotificationKit() + + assert kit.email_config is not None + assert kit.email_config.user == 'test@example.com' + assert kit.email_config.password == 'test_password' + assert kit.email_config.to == 'recipient@example.com' + + if has_template: + assert kit.email_config.template == 'custom email template' + else: + assert kit.email_config.template is not None + assert len(kit.email_config.template) > 0 + + def test_email_old_format(self, monkeypatch): + """测试邮箱旧格式配置""" + # 清空环境 + for key in list(os.environ.keys()): + if 'NOTIF_CONFIG' in key or 'EMAIL' in key: + monkeypatch.delenv(key, raising=False) + + monkeypatch.setenv('EMAIL_USER', 'old@example.com') + monkeypatch.setenv('EMAIL_PASS', 'old_password') + monkeypatch.setenv('EMAIL_TO', 'old_recipient@example.com') + + from notify import NotificationKit + kit = NotificationKit() + + assert kit.email_config is not None + assert kit.email_config.user == 'old@example.com' + assert kit.email_config.password == 'old_password' + assert kit.email_config.to == 'old_recipient@example.com' + assert kit.email_config.template is not None + assert len(kit.email_config.template) > 0 diff --git a/tests/unit/test_data_models.py b/tests/unit/test_data_models.py new file mode 100644 index 0000000..e94617c --- /dev/null +++ b/tests/unit/test_data_models.py @@ -0,0 +1,84 @@ +import pytest + + +class TestNotificationDataModel: + """测试 NotificationData 数据模型""" + + @pytest.fixture + def create_account_result(self): + """创建账号结果的工厂函数""" + from models import AccountResult + + def _create( + name: str = '测试账号', + status: str = 'success', + quota: float = 25.0, + used: float = 5.0, + error: str = None + ) -> AccountResult: + return AccountResult( + name=name, + status=status, + quota=quota if status == 'success' else None, + used=used if status == 'success' else None, + error=error if status != 'success' else None + ) + + return _create + + @pytest.fixture + def create_notification_data(self, create_account_result): + """创建通知数据的工厂函数""" + from typing import List + from models import NotificationData, AccountResult, NotificationStats + + def _create( + accounts: List[AccountResult], + timestamp: str = '2024-01-01 12:00:00' + ) -> NotificationData: + success_count = sum(1 for acc in accounts if acc.status == 'success') + failed_count = len(accounts) - success_count + + stats = NotificationStats( + success_count=success_count, + failed_count=failed_count, + total_count=len(accounts) + ) + + return NotificationData( + accounts=accounts, + stats=stats, + timestamp=timestamp + ) + + return _create + + def test_all_success_property(self, create_account_result, create_notification_data): + """测试 all_success 属性""" + data = create_notification_data([ + create_account_result(name='Account-1'), + create_account_result(name='Account-2') + ]) + assert data.all_success is True + assert data.all_failed is False + assert data.partial_success is False + + def test_all_failed_property(self, create_account_result, create_notification_data): + """测试 all_failed 属性""" + data = create_notification_data([ + create_account_result(name='Account-1', status='failed', error='Error 1'), + create_account_result(name='Account-2', status='failed', error='Error 2') + ]) + assert data.all_success is False + assert data.all_failed is True + assert data.partial_success is False + + def test_partial_success_property(self, create_account_result, create_notification_data): + """测试 partial_success 属性""" + data = create_notification_data([ + create_account_result(name='Account-1'), + create_account_result(name='Account-2', status='failed', error='Error') + ]) + assert data.all_success is False + assert data.all_failed is False + assert data.partial_success is True diff --git a/tests/unit/test_send_functions.py b/tests/unit/test_send_functions.py new file mode 100644 index 0000000..a2c258b --- /dev/null +++ b/tests/unit/test_send_functions.py @@ -0,0 +1,189 @@ +import json +import os +from unittest.mock import MagicMock, patch + +import pytest + + +class TestNotificationSending: + """测试通知发送功能""" + + @pytest.mark.parametrize('platform,method_name,env_config,error_message', [ + ('email', 'send_email', {}, '未配置邮箱信息'), + ('pushplus', 'send_pushplus', {}, '未配置PushPlus Token'), + ('serverpush', 'send_serverpush', {}, '未配置Server Push key'), + ('dingtalk', 'send_dingtalk', {}, '未配置钉钉 Webhook'), + ('feishu', 'send_feishu', {}, '未配置飞书 Webhook'), + ('wecom', 'send_wecom', {}, '未配置企业微信 Webhook'), + ]) + def test_send_without_config_raises_error( + self, + monkeypatch, + platform, + method_name, + env_config, + error_message + ): + """测试无配置时发送抛出异常""" + # 清空所有配置相关环境变量 + for key in list(os.environ.keys()): + if 'NOTIF_CONFIG' in key or any(x in key for x in ['EMAIL', 'WEBHOOK', 'TOKEN', 'KEY']): + monkeypatch.delenv(key, raising=False) + + from notify import NotificationKit + kit = NotificationKit() + + with pytest.raises(ValueError, match=error_message): + getattr(kit, method_name)('测试标题', '测试内容') + + @patch('smtplib.SMTP_SSL') + def test_email_sending_with_mock(self, mock_smtp, monkeypatch): + """测试邮箱发送逻辑（使用 mock）""" + # 清空环境 + for key in list(os.environ.keys()): + if 'EMAIL' in key or 'NOTIF_CONFIG' in key: + monkeypatch.delenv(key, raising=False) + + # 设置邮箱配置 + monkeypatch.setenv('EMAIL_USER', 'test@example.com') + monkeypatch.setenv('EMAIL_PASS', 'testpass') + monkeypatch.setenv('EMAIL_TO', 'recipient@example.com') + + mock_server = MagicMock() + mock_smtp.return_value.__enter__.return_value = mock_server + + from notify import NotificationKit + kit = NotificationKit() + kit.send_email('测试标题', '测试内容') + + assert mock_server.login.called + assert mock_server.send_message.called + + @pytest.mark.parametrize('platform,method_name,env_key,env_value', [ + ('pushplus', 'send_pushplus', 'PUSHPLUS_TOKEN', 'test_token'), + ('serverpush', 'send_serverpush', 'SERVERPUSHKEY', 'test_key'), + ('dingtalk', 'send_dingtalk', 'DINGDING_WEBHOOK', 'https://example.com/webhook'), + ('feishu', 'send_feishu', 'FEISHU_WEBHOOK', 'https://example.com/webhook'), + ('wecom', 'send_wecom', 'WEIXIN_WEBHOOK', 'https://example.com/webhook'), + ]) + @patch('httpx.Client') + def test_http_platforms_sending_with_mock( + self, + mock_client, + monkeypatch, + platform, + method_name, + env_key, + env_value + ): + """测试各 HTTP 平台的发送逻辑（使用 mock）""" + # 清空环境 + for key in list(os.environ.keys()): + if 'NOTIF_CONFIG' in key or any(x in key for x in ['WEBHOOK', 'TOKEN', 'KEY']): + monkeypatch.delenv(key, raising=False) + + monkeypatch.setenv(env_key, env_value) + + mock_client_instance = MagicMock() + mock_client.return_value.__enter__.return_value = mock_client_instance + + from notify import NotificationKit + kit = NotificationKit() + getattr(kit, method_name)('测试标题', '测试内容') + + mock_client_instance.post.assert_called_once() + + @pytest.mark.parametrize('use_card,color_theme', [ + (True, 'red'), + (True, 'blue'), + (False, None), + ]) + @patch('httpx.Client') + def test_feishu_card_modes(self, mock_client, monkeypatch, use_card, color_theme): + """测试飞书的卡片模式和普通模式""" + # 清空环境 + for key in list(os.environ.keys()): + if 'FEISHU' in key or 'NOTIF_CONFIG' in key: + monkeypatch.delenv(key, raising=False) + + # 构造配置 + config = { + 'webhook': 'https://example.com/webhook', + 'platform_settings': { + 'use_card': use_card + }, + 'template': None + } + + if use_card and color_theme: + config['platform_settings']['color_theme'] = color_theme + + monkeypatch.setenv('FEISHU_NOTIF_CONFIG', json.dumps(config)) + + mock_client_instance = MagicMock() + mock_client.return_value.__enter__.return_value = mock_client_instance + + from notify import NotificationKit + kit = NotificationKit() + kit.send_feishu('测试标题', '测试内容') + + # 验证调用了 post 方法 + assert mock_client_instance.post.called + call_args = mock_client_instance.post.call_args + data = call_args[1]['json'] + + if use_card: + # 验证是卡片模式 + assert data['msg_type'] == 'interactive' + assert 'card' in data + if color_theme: + assert data['card']['header']['template'] == color_theme + else: + # 验证是文本模式 + assert data['msg_type'] == 'text' + assert 'text' in data + + @pytest.mark.parametrize('markdown_style,expected_msgtype', [ + ('markdown', 'markdown'), + ('markdown_v2', 'markdown_v2'), + (None, 'text'), + ('invalid', 'text'), + ]) + @patch('httpx.Client') + def test_wecom_markdown_style_modes(self, mock_client, monkeypatch, markdown_style, expected_msgtype): + """测试企业微信的 markdown_style 配置""" + # 清空环境 + for key in list(os.environ.keys()): + if 'WECOM' in key or 'WEIXIN' in key or 'NOTIF_CONFIG' in key: + monkeypatch.delenv(key, raising=False) + + # 构造配置 + config = { + 'webhook': 'https://example.com/webhook', + 'platform_settings': { + 'markdown_style': markdown_style + }, + 'template': None + } + + monkeypatch.setenv('WECOM_NOTIF_CONFIG', json.dumps(config)) + + mock_client_instance = MagicMock() + mock_client.return_value.__enter__.return_value = mock_client_instance + + from notify import NotificationKit + kit = NotificationKit() + kit.send_wecom('测试标题', '测试内容') + + # 验证调用了 post 方法 + assert mock_client_instance.post.called + call_args = mock_client_instance.post.call_args + data = call_args[1]['json'] + + # 验证消息格式 + assert data['msgtype'] == expected_msgtype + if 'markdown' in expected_msgtype: + assert 'markdown' in data + assert '**测试标题**' in data['markdown']['content'] + else: + assert 'text' in data diff --git a/tests/unit/test_template_rendering.py b/tests/unit/test_template_rendering.py new file mode 100644 index 0000000..af1e2b4 --- /dev/null +++ b/tests/unit/test_template_rendering.py @@ -0,0 +1,272 @@ +import json +from pathlib import Path + +import pytest + + +project_root = Path(__file__).parent.parent.parent + + +class TestTemplateRendering: + """测试模板渲染功能""" + + @pytest.fixture + def notification_kit(self): + """创建不依赖环境变量的 NotificationKit 实例""" + from notify import NotificationKit + return NotificationKit() + + @pytest.fixture + def single_success_data(self): + """单账号成功的测试数据""" + from models import NotificationData, AccountResult, NotificationStats + + return NotificationData( + accounts=[ + AccountResult( + name='Account-1', + status='success', + quota=25.0, + used=5.0, + error=None + ) + ], + stats=NotificationStats(success_count=1, failed_count=0, total_count=1), + timestamp='2024-01-01 12:00:00' + ) + + @pytest.fixture + def single_failure_data(self): + """单账号失败的测试数据""" + from models import NotificationData, AccountResult, NotificationStats + + return NotificationData( + accounts=[ + AccountResult( + name='Account-1', + status='failed', + quota=None, + used=None, + error='Connection timeout' + ) + ], + stats=NotificationStats(success_count=0, failed_count=1, total_count=1), + timestamp='2024-01-01 12:00:00' + ) + + @pytest.fixture + def multiple_mixed_data(self): + """多账号混合的测试数据""" + from models import NotificationData, AccountResult, NotificationStats + + return NotificationData( + accounts=[ + AccountResult(name='Account-1', status='success', quota=25.0, used=5.0, error=None), + AccountResult(name='Account-2', status='success', quota=30.0, used=10.0, error=None), + AccountResult(name='Account-3', status='failed', quota=None, used=None, error='Authentication failed') + ], + stats=NotificationStats(success_count=2, failed_count=1, total_count=3), + timestamp='2024-01-01 12:00:00' + ) + + @pytest.mark.parametrize('platform,template_file', [ + ('dingtalk', 'dingtalk.json'), + ('email', 'email.json'), + ('pushplus', 'pushplus.json'), + ('serverpush', 'serverpush.json'), + ]) + def test_default_template_single_success( + self, + notification_kit, + single_success_data, + platform, + template_file + ): + """测试默认模板渲染单账号成功场景""" + config_path = project_root / 'notif_config' / template_file + with open(config_path) as f: + config = json.load(f) + + result = notification_kit._render_template(config['template'], single_success_data) + + assert '[TIME] Execution time: 2024-01-01 12:00:00' in result + assert '[BALANCE] Account-1' in result + assert ':money: Current balance: $25.0, Used: $5.0' in result + assert '[STATS] Check-in result statistics:' in result + assert '[SUCCESS] Success: 1/1' in result + assert '[FAIL] Failed: 0/1' in result + assert '[SUCCESS] All accounts check-in successful!' in result + + def test_wecom_default_template_single_success(self, notification_kit, single_success_data): + """测试企业微信默认模板渲染单账号成功场景（Markdown 格式）""" + config_path = project_root / 'notif_config' / 'wecom.json' + with open(config_path) as f: + config = json.load(f) + + result = notification_kit._render_template(config['template'], single_success_data) + + assert '**[TIME] Execution time:** 2024-01-01 12:00:00' in result + assert '**[BALANCE] Account-1**' in result + assert ':money: Current balance: $25.0, Used: $5.0' in result + assert '**[STATS] Check-in result statistics:**' in result + assert '**[SUCCESS] Success:** 1/1' in result + assert '**[FAIL] Failed:** 0/1' in result + assert '**[SUCCESS] All accounts check-in successful!**' in result + + def test_feishu_default_template_single_success(self, notification_kit, single_success_data): + """测试飞书默认模板渲染单账号成功场景（Markdown 格式）""" + config_path = project_root / 'notif_config' / 'feishu.json' + with open(config_path) as f: + config = json.load(f) + + result = notification_kit._render_template(config['template'], single_success_data) + + assert '**[TIME] Execution time:** 2024-01-01 12:00:00' in result + assert '**[BALANCE] Account-1**' in result + assert ':money: Current balance: $25.0, Used: $5.0' in result + assert '**[STATS] Check-in result statistics:**' in result + assert '**[SUCCESS] Success:** 1/1' in result + assert '**[FAIL] Failed:** 0/1' in result + assert '**[SUCCESS] All accounts check-in successful!**' in result + + @pytest.mark.parametrize('platform,template_file', [ + ('dingtalk', 'dingtalk.json'), + ]) + def test_default_template_single_failure( + self, + notification_kit, + single_failure_data, + platform, + template_file + ): + """测试默认模板渲染单账号失败场景""" + config_path = project_root / 'notif_config' / template_file + with open(config_path) as f: + config = json.load(f) + + result = notification_kit._render_template(config['template'], single_failure_data) + + assert '[TIME] Execution time: 2024-01-01 12:00:00' in result + assert '[FAIL] Account-1 exception: Connection timeout' in result + assert '[STATS] Check-in result statistics:' in result + assert '[SUCCESS] Success: 0/1' in result + assert '[FAIL] Failed: 1/1' in result + assert '[ERROR] All accounts check-in failed' in result + + def test_wecom_default_template_single_failure(self, notification_kit, single_failure_data): + """测试企业微信默认模板渲染单账号失败场景（Markdown 格式）""" + config_path = project_root / 'notif_config' / 'wecom.json' + with open(config_path) as f: + config = json.load(f) + + result = notification_kit._render_template(config['template'], single_failure_data) + + assert '**[TIME] Execution time:** 2024-01-01 12:00:00' in result + assert '**[FAIL] Account-1 exception:** Connection timeout' in result + assert '**[STATS] Check-in result statistics:**' in result + assert '**[SUCCESS] Success:** 0/1' in result + assert '**[FAIL] Failed:** 1/1' in result + assert '**[ERROR] All accounts check-in failed**' in result + + @pytest.mark.parametrize('platform,template_file', [ + ('dingtalk', 'dingtalk.json'), + ]) + def test_default_template_multiple_mixed( + self, + notification_kit, + multiple_mixed_data, + platform, + template_file + ): + """测试默认模板渲染多账号混合场景""" + config_path = project_root / 'notif_config' / template_file + with open(config_path) as f: + config = json.load(f) + + result = notification_kit._render_template(config['template'], multiple_mixed_data) + + assert '[TIME] Execution time: 2024-01-01 12:00:00' in result + assert '[BALANCE] Account-1' in result + assert ':money: Current balance: $25.0, Used: $5.0' in result + assert '[BALANCE] Account-2' in result + assert ':money: Current balance: $30.0, Used: $10.0' in result + assert '[FAIL] Account-3 exception: Authentication failed' in result + assert '[STATS] Check-in result statistics:' in result + assert '[SUCCESS] Success: 2/3' in result + assert '[FAIL] Failed: 1/3' in result + assert '[WARN] Some accounts check-in successful' in result + + def test_wecom_default_template_multiple_mixed(self, notification_kit, multiple_mixed_data): + """测试企业微信默认模板渲染多账号混合场景（Markdown 格式）""" + config_path = project_root / 'notif_config' / 'wecom.json' + with open(config_path) as f: + config = json.load(f) + + result = notification_kit._render_template(config['template'], multiple_mixed_data) + + assert '**[TIME] Execution time:** 2024-01-01 12:00:00' in result + assert '**[BALANCE] Account-1**' in result + assert ':money: Current balance: $25.0, Used: $5.0' in result + assert '**[BALANCE] Account-2**' in result + assert ':money: Current balance: $30.0, Used: $10.0' in result + assert '**[FAIL] Account-3 exception:** Authentication failed' in result + assert '**[STATS] Check-in result statistics:**' in result + assert '**[SUCCESS] Success:** 2/3' in result + assert '**[FAIL] Failed:** 1/3' in result + assert '**[WARN] Some accounts check-in successful**' in result + + def test_custom_template_with_variables(self, notification_kit, single_success_data): + """测试自定义模板的变量访问""" + template = '{{ timestamp }} - {% for account in accounts %}{{ account.name }}: {{ account.status }}{% endfor %} - 成功: {{ stats.success_count }}/{{ stats.total_count }}' + result = notification_kit._render_template(template, single_success_data) + + assert '2024-01-01 12:00:00' in result + assert 'Account-1' in result + assert 'success' in result + assert '1/1' in result + + def test_custom_template_with_convenience_flags(self, notification_kit): + """测试自定义模板使用便利标志（all_success, all_failed, partial_success）""" + from models import NotificationData, AccountResult, NotificationStats + + template = '{% if all_success %}ALL SUCCESS{% endif %}{% if all_failed %}ALL FAILED{% endif %}{% if partial_success %}PARTIAL{% endif %}' + + # 测试 all_success + data_all_success = NotificationData( + accounts=[AccountResult(name='A1', status='success', quota=25.0, used=5.0, error=None)], + stats=NotificationStats(success_count=1, failed_count=0, total_count=1), + timestamp='2024-01-01 12:00:00' + ) + result = notification_kit._render_template(template, data_all_success) + assert 'ALL SUCCESS' in result + + # 测试 all_failed + data_all_failed = NotificationData( + accounts=[AccountResult(name='A1', status='failed', quota=None, used=None, error='Error')], + stats=NotificationStats(success_count=0, failed_count=1, total_count=1), + timestamp='2024-01-01 12:00:00' + ) + result = notification_kit._render_template(template, data_all_failed) + assert 'ALL FAILED' in result + + # 测试 partial_success + data_partial = NotificationData( + accounts=[ + AccountResult(name='A1', status='success', quota=25.0, used=5.0, error=None), + AccountResult(name='A2', status='failed', quota=None, used=None, error='Error') + ], + stats=NotificationStats(success_count=1, failed_count=1, total_count=2), + timestamp='2024-01-01 12:00:00' + ) + result = notification_kit._render_template(template, data_partial) + assert 'PARTIAL' in result + + def test_invalid_template_fallback(self, notification_kit, single_success_data): + """测试无效模板语法时的回退处理""" + # 使用会触发解析错误的模板语法 + invalid_template = '{% if unclosed_block %}' + result = notification_kit._render_template(invalid_template, single_success_data) + + # 应该返回回退格式 + assert '2024-01-01 12:00:00' in result + assert 'Account-1' in result