通过 Ntfy 推送消息
(title: str, content: str)
| 802 | |
| 803 | |
| 804 | def ntfy(title: str, content: str) -> None: |
| 805 | """ |
| 806 | 通过 Ntfy 推送消息 |
| 807 | """ |
| 808 | |
| 809 | def encode_rfc2047(text: str) -> str: |
| 810 | """将文本编码为符合 RFC 2047 标准的格式""" |
| 811 | encoded_bytes = base64.b64encode(text.encode("utf-8")) |
| 812 | encoded_str = encoded_bytes.decode("utf-8") |
| 813 | return f"=?utf-8?B?{encoded_str}?=" |
| 814 | |
| 815 | if not push_config.get("NTFY_TOPIC"): |
| 816 | return |
| 817 | print("ntfy 服务启动") |
| 818 | priority = "3" |
| 819 | if not push_config.get("NTFY_PRIORITY"): |
| 820 | print("ntfy 服务的NTFY_PRIORITY 未设置!!默认设置为3") |
| 821 | else: |
| 822 | priority = push_config.get("NTFY_PRIORITY") |
| 823 | |
| 824 | # 使用 RFC 2047 编码 title |
| 825 | encoded_title = encode_rfc2047(title) |
| 826 | |
| 827 | data = content.encode(encoding="utf-8") |
| 828 | headers = {"Title": encoded_title, "Priority": priority, "Icon": "https://qn.whyour.cn/logo.png"} # 使用编码后的 title |
| 829 | if push_config.get("NTFY_TOKEN"): |
| 830 | headers['Authorization'] = "Bearer " + push_config.get("NTFY_TOKEN") |
| 831 | elif push_config.get("NTFY_USERNAME") and push_config.get("NTFY_PASSWORD"): |
| 832 | authStr = push_config.get("NTFY_USERNAME") + ":" + push_config.get("NTFY_PASSWORD") |
| 833 | headers['Authorization'] = "Basic " + base64.b64encode(authStr.encode('utf-8')).decode('utf-8') |
| 834 | if push_config.get("NTFY_ACTIONS"): |
| 835 | headers['Actions'] = encode_rfc2047(push_config.get("NTFY_ACTIONS")) |
| 836 | |
| 837 | url = push_config.get("NTFY_URL") + "/" + push_config.get("NTFY_TOPIC") |
| 838 | response = requests.post(url, data=data, headers=headers) |
| 839 | if response.status_code == 200: # 使用 response.status_code 进行检查 |
| 840 | print("Ntfy 推送成功!") |
| 841 | else: |
| 842 | print("Ntfy 推送失败!错误信息:", response.text) |
| 843 | |
| 844 | |
| 845 | def wxpusher_bot(title: str, content: str) -> None: |
nothing calls this directly
no test coverage detected