83c90f0161
- Consolidated database, file source, and field mapping configurations into dedicated modules under the `config` directory. - Removed hardcoded database connection details from `migration.py` and replaced them with imports from the new configuration structure. - Updated `ntfy_utils.py` and `run_incremental_sync.py` to utilize the new configuration imports for cleaner code and better maintainability. - Deleted `update_config.py` as its contents have been integrated into the new configuration files. - Added a new `settings.local.json` for managing permissions related to script execution. - Enhanced the structure of the migration tasks and Excel configurations for better organization and clarity.
65 lines
1.8 KiB
Python
65 lines
1.8 KiB
Python
# ntfy_utils.py
|
||
import requests
|
||
from config import NTFY_CONFIG
|
||
|
||
def send_ntfy(message, title="数据库同步消息", priority="default", tags=None):
|
||
"""
|
||
向加密的 ntfy 服务器发送消息
|
||
"""
|
||
conf = NTFY_CONFIG
|
||
if not conf.get('enabled', False):
|
||
return
|
||
|
||
# 确保 URL 正确（末尾不要多余斜杠）
|
||
server_url = conf['server_url'].rstrip('/')
|
||
url = f"{server_url}/{conf['topic']}"
|
||
|
||
# 构造请求头
|
||
headers = {
|
||
"Title": title.encode('utf-8'),
|
||
"Priority": priority,
|
||
"Tags": ",".join(tags) if tags else ""
|
||
}
|
||
|
||
# --- 核心：配置秘钥认证 ---
|
||
token = conf.get('token')
|
||
if token:
|
||
# ntfy 使用 Bearer Token 模式
|
||
headers["Authorization"] = f"Bearer {token}"
|
||
|
||
try:
|
||
# 发送请求
|
||
response = requests.post(
|
||
url,
|
||
data=message.encode('utf-8'),
|
||
headers=headers,
|
||
timeout=10
|
||
)
|
||
|
||
# 针对认证失败的处理
|
||
if response.status_code == 401:
|
||
print("ntfy 认证失败：Token 无效")
|
||
elif response.status_code == 403:
|
||
print("ntfy 权限不足：该 Token 无权发布消息")
|
||
|
||
response.raise_for_status()
|
||
except Exception as e:
|
||
print(f"发送 ntfy 通知失败: {e}")
|
||
|
||
def send_error(msg):
|
||
"""便捷方法：发送错误通知"""
|
||
send_ntfy(
|
||
message=str(msg),
|
||
title="❌ 同步任务错误",
|
||
priority=NTFY_CONFIG['priority']['error'],
|
||
tags=["warning", "database"]
|
||
)
|
||
|
||
def send_critical(msg):
|
||
"""便捷方法：发送严重崩溃通知"""
|
||
send_ntfy(
|
||
message=str(msg),
|
||
title="🔥 同步服务崩溃",
|
||
priority=NTFY_CONFIG['priority']['critical'],
|
||
tags=["skull", "critical"]
|
||
) |