aszerW
844e609ff7
- 将 notify.py 和 oss_utils.py 从 archive/legacy_core 移至 core/common/ - 内联钉钉配置读取函数，移除对 config.settings 的依赖 - 删除 config/ 目录（settings.py 不再需要） - daily_scheduler.py 移除归档路径的 sys.path hack - 新增 --no-detail 和 --no-report 命令行参数控制导出 - 全标的排名表新增退场日期和退场价格列
607 lines
19 KiB
Python
607 lines
19 KiB
Python
"""
|
||
通知模块 - 支持钉钉、日志等多种通知方式
|
||
"""
|
||
|
||
import requests
|
||
import time
|
||
import hmac
|
||
import hashlib
|
||
import base64
|
||
import urllib.parse
|
||
import os
|
||
from loguru import logger
|
||
from typing import Optional
|
||
|
||
from core.common.oss_utils import upload_image_to_oss
|
||
|
||
|
||
def _get_dingtalk_config() -> dict:
|
||
"""获取钉钉机器人配置（第一群）"""
|
||
return {
|
||
"webhook": os.getenv("DINGTALK_WEBHOOK", ""),
|
||
"secret": os.getenv("DINGTALK_SECRET", ""),
|
||
}
|
||
|
||
|
||
def _get_all_dingtalk_configs() -> list:
|
||
"""获取所有钉钉机器人配置（支持多群）
|
||
环境变量格式: DINGTALK_WEBHOOK_1 + DINGTALK_SECRET_1, ...
|
||
"""
|
||
configs = []
|
||
i = 1
|
||
while True:
|
||
webhook = os.getenv(f"DINGTALK_WEBHOOK_{i}", "")
|
||
secret = os.getenv(f"DINGTALK_SECRET_{i}", "")
|
||
if not webhook:
|
||
break
|
||
configs.append({"webhook": webhook, "secret": secret})
|
||
i += 1
|
||
return configs
|
||
|
||
|
||
class DingTalkBot:
|
||
"""钉钉机器人类"""
|
||
|
||
def __init__(self, webhook: str = None, secret: str = None):
|
||
"""
|
||
初始化钉钉机器人
|
||
|
||
Args:
|
||
webhook: 钉钉自定义机器人webhook地址
|
||
secret: 加签密钥（可选）
|
||
"""
|
||
config = _get_dingtalk_config()
|
||
self.webhook = webhook or config.get("webhook", "")
|
||
self.secret = secret or config.get("secret", "")
|
||
|
||
if not self.webhook:
|
||
logger.warning("钉钉webhook未配置，消息将不会被发送")
|
||
|
||
def _gen_signed_url(self) -> str:
|
||
"""生成带签名的URL"""
|
||
if not self.secret:
|
||
return self.webhook
|
||
|
||
timestamp = str(round(time.time() * 1000))
|
||
secret_enc = self.secret.encode("utf-8")
|
||
string_to_sign = f"{timestamp}\n{self.secret}"
|
||
string_to_sign_enc = string_to_sign.encode("utf-8")
|
||
hmac_code = hmac.new(
|
||
secret_enc, string_to_sign_enc, digestmod=hashlib.sha256
|
||
).digest()
|
||
sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
|
||
return f"{self.webhook}×tamp={timestamp}&sign={sign}"
|
||
|
||
def send_text(
|
||
self, content: str, at_mobiles: list = None, is_at_all: bool = False
|
||
) -> bool:
|
||
"""
|
||
发送文本消息
|
||
|
||
Args:
|
||
content: 消息内容
|
||
at_mobiles: 需要@的手机号列表
|
||
is_at_all: 是否@所有人
|
||
|
||
Returns:
|
||
bool: 是否发送成功
|
||
"""
|
||
if not self.webhook:
|
||
logger.warning(f"[钉钉消息未发送] {content[:100]}...")
|
||
return False
|
||
|
||
at_mobiles = at_mobiles or []
|
||
data = {
|
||
"msgtype": "text",
|
||
"text": {"content": content},
|
||
"at": {"atMobiles": at_mobiles, "isAtAll": is_at_all},
|
||
}
|
||
|
||
url = self._gen_signed_url()
|
||
|
||
try:
|
||
response = requests.post(url, json=data, timeout=5)
|
||
response.raise_for_status()
|
||
result = response.json()
|
||
if result.get("errcode", -1) != 0:
|
||
logger.error(f"钉钉消息发送失败: {result}")
|
||
return False
|
||
logger.info("钉钉消息发送成功")
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"钉钉消息发送异常: {e}")
|
||
return False
|
||
|
||
def send_markdown(
|
||
self,
|
||
title: str,
|
||
text: str,
|
||
at_mobiles: list = None,
|
||
is_at_all: bool = False,
|
||
) -> bool:
|
||
"""
|
||
发送markdown消息
|
||
|
||
Args:
|
||
title: 消息标题
|
||
text: markdown格式的消息内容
|
||
at_mobiles: 需要@的手机号列表
|
||
is_at_all: 是否@所有人
|
||
|
||
Returns:
|
||
bool: 是否发送成功
|
||
"""
|
||
if not self.webhook:
|
||
logger.warning(f"[钉钉Markdown未发送] {title}")
|
||
return False
|
||
|
||
at_mobiles = at_mobiles or []
|
||
data = {
|
||
"msgtype": "markdown",
|
||
"markdown": {"title": title, "text": text},
|
||
"at": {"atMobiles": at_mobiles, "isAtAll": is_at_all},
|
||
}
|
||
|
||
url = self._gen_signed_url()
|
||
|
||
try:
|
||
response = requests.post(url, json=data, timeout=5)
|
||
response.raise_for_status()
|
||
result = response.json()
|
||
if result.get("errcode", -1) != 0:
|
||
logger.error(f"钉钉markdown消息发送失败: {result}")
|
||
return False
|
||
logger.info("钉钉markdown消息发送成功")
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"钉钉markdown消息发送异常: {e}")
|
||
return False
|
||
|
||
def send_image(self, image_path: str, title: str = "图片", max_size_kb: int = 6) -> bool:
|
||
"""
|
||
发送图片消息（自动压缩以适应钉钉20KB限制）
|
||
注意：钉钉限制的是整个请求body大小，base64编码会增加约33%体积，所以图片本身需要更小
|
||
|
||
Args:
|
||
image_path: 图片文件路径（本地路径）
|
||
title: 消息标题
|
||
max_size_kb: 最大图片大小（KB），默认6KB（base64后约8KB，加上其他字段约10KB）
|
||
|
||
Returns:
|
||
bool: 是否发送成功
|
||
"""
|
||
if not self.webhook:
|
||
logger.warning(f"[钉钉图片未发送] {title}")
|
||
return False
|
||
|
||
if not os.path.exists(image_path):
|
||
logger.error(f"图片文件不存在: {image_path}")
|
||
return False
|
||
|
||
try:
|
||
# 读取并压缩图片
|
||
image_data = self._compress_image(image_path, max_size_kb)
|
||
|
||
if not image_data:
|
||
logger.error(f"图片压缩失败: {image_path}")
|
||
return False
|
||
|
||
# 转为base64
|
||
image_base64 = base64.b64encode(image_data).decode("utf-8")
|
||
# 计算图片md5（钉钉需要）
|
||
image_md5 = hashlib.md5(image_data).hexdigest()
|
||
|
||
data = {
|
||
"msgtype": "image",
|
||
"image": {
|
||
"base64": image_base64,
|
||
"md5": image_md5
|
||
}
|
||
}
|
||
|
||
url = self._gen_signed_url()
|
||
response = requests.post(url, json=data, timeout=10)
|
||
response.raise_for_status()
|
||
result = response.json()
|
||
|
||
if result.get("errcode", -1) != 0:
|
||
logger.error(f"钉钉图片发送失败: {result}")
|
||
return False
|
||
|
||
logger.info(f"钉钉图片发送成功: {image_path}")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"钉钉图片发送异常: {e}")
|
||
return False
|
||
|
||
def _compress_image(self, image_path: str, max_size_kb: int) -> bytes:
|
||
"""
|
||
压缩图片到指定大小以下
|
||
|
||
Args:
|
||
image_path: 图片路径
|
||
max_size_kb: 最大大小（KB）
|
||
|
||
Returns:
|
||
bytes: 压缩后的图片数据
|
||
"""
|
||
from PIL import Image
|
||
import io
|
||
|
||
max_size_bytes = max_size_kb * 1024
|
||
|
||
# 先尝试读取原图
|
||
with open(image_path, "rb") as f:
|
||
image_data = f.read()
|
||
|
||
# 如果已经小于限制，直接返回
|
||
if len(image_data) <= max_size_bytes:
|
||
return image_data
|
||
|
||
# 需要压缩，使用PIL重新保存
|
||
img = Image.open(image_path)
|
||
|
||
# 转换为RGB（去除alpha通道，减小大小）
|
||
if img.mode in ('RGBA', 'P'):
|
||
img = img.convert('RGB')
|
||
|
||
# 逐步降低质量直到满足大小要求
|
||
quality = 85
|
||
min_quality = 30
|
||
|
||
while quality >= min_quality:
|
||
buffer = io.BytesIO()
|
||
img.save(buffer, format='JPEG', quality=quality, optimize=True)
|
||
compressed_data = buffer.getvalue()
|
||
|
||
if len(compressed_data) <= max_size_bytes:
|
||
logger.info(f"图片压缩成功: {len(image_data)/1024:.1f}KB -> {len(compressed_data)/1024:.1f}KB (quality={quality})")
|
||
return compressed_data
|
||
|
||
quality -= 10
|
||
|
||
# 如果质量降到最小还不行，尝试缩小尺寸
|
||
logger.warning(f"降低质量无法满足要求，尝试缩小尺寸")
|
||
width, height = img.size
|
||
|
||
while width > 200 and height > 200:
|
||
width = int(width * 0.8)
|
||
height = int(height * 0.8)
|
||
resized_img = img.resize((width, height), Image.Resampling.LANCZOS)
|
||
|
||
buffer = io.BytesIO()
|
||
resized_img.save(buffer, format='JPEG', quality=min_quality, optimize=True)
|
||
compressed_data = buffer.getvalue()
|
||
|
||
if len(compressed_data) <= max_size_bytes:
|
||
logger.info(f"图片压缩成功: {len(image_data)/1024:.1f}KB -> {len(compressed_data)/1024:.1f}KB ({width}x{height})")
|
||
return compressed_data
|
||
|
||
logger.error(f"无法将图片压缩到 {max_size_kb}KB 以下")
|
||
return None
|
||
|
||
def send_image_with_text(self, image_path: str, title: str = "图片", text: str = "") -> bool:
|
||
"""
|
||
发送图文消息（markdown格式嵌入图片链接）
|
||
注意：需要使用钉钉的媒体文件上传接口获取URL，这里使用简化的markdown图片语法
|
||
|
||
Args:
|
||
image_path: 图片文件路径
|
||
title: 消息标题
|
||
text: accompanying text
|
||
|
||
Returns:
|
||
bool: 是否发送成功
|
||
"""
|
||
if not self.webhook:
|
||
logger.warning(f"[钉钉图文未发送] {title}")
|
||
return False
|
||
|
||
if not os.path.exists(image_path):
|
||
logger.error(f"图片文件不存在: {image_path}")
|
||
return False
|
||
|
||
# 先尝试直接发送图片
|
||
success = self.send_image(image_path, title)
|
||
|
||
# 如果图片发送成功且有文字，再发送文字
|
||
if success and text:
|
||
time.sleep(0.5) # 避免发送过快
|
||
return self.send_text(f"{title}\n{text}")
|
||
|
||
return success
|
||
|
||
def send_file(self, file_path: str, title: str = "文件") -> bool:
|
||
"""
|
||
发送文件（通过钉钉文件上传接口）
|
||
注意：需要企业版钉钉机器人，个人版可能不支持
|
||
|
||
Args:
|
||
file_path: 文件路径
|
||
title: 文件标题
|
||
|
||
Returns:
|
||
bool: 是否发送成功
|
||
"""
|
||
if not self.webhook:
|
||
logger.warning(f"[钉钉文件未发送] {title}")
|
||
return False
|
||
|
||
if not os.path.exists(file_path):
|
||
logger.error(f"文件不存在: {file_path}")
|
||
return False
|
||
|
||
try:
|
||
# 获取文件大小
|
||
file_size = os.path.getsize(file_path)
|
||
file_name = os.path.basename(file_path)
|
||
|
||
# 钉钉文件大小限制：20MB
|
||
if file_size > 20 * 1024 * 1024:
|
||
logger.error(f"文件过大: {file_size/1024/1024:.1f}MB > 20MB")
|
||
return False
|
||
|
||
# 读取文件并转为base64
|
||
with open(file_path, "rb") as f:
|
||
file_data = f.read()
|
||
|
||
file_base64 = base64.b64encode(file_data).decode("utf-8")
|
||
|
||
# 构建消息
|
||
data = {
|
||
"msgtype": "file",
|
||
"file": {
|
||
"base64": file_base64,
|
||
"name": file_name
|
||
}
|
||
}
|
||
|
||
url = self._gen_signed_url()
|
||
response = requests.post(url, json=data, timeout=30)
|
||
response.raise_for_status()
|
||
result = response.json()
|
||
|
||
if result.get("errcode", -1) != 0:
|
||
# 如果文件发送失败，尝试作为图片发送
|
||
if file_path.lower().endswith(('.png', '.jpg', '.jpeg', '.gif', '.bmp')):
|
||
logger.warning(f"文件发送失败，尝试作为图片发送: {result}")
|
||
return self.send_image(file_path, title)
|
||
logger.error(f"钉钉文件发送失败: {result}")
|
||
return False
|
||
|
||
logger.info(f"钉钉文件发送成功: {file_name}")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"钉钉文件发送异常: {e}")
|
||
return False
|
||
|
||
def send_local_image_as_link(self, image_path: str, title: str = "图片", text: str = "") -> bool:
|
||
"""
|
||
发送本地图片（转换为base64嵌入markdown）
|
||
注意：钉钉不支持直接显示base64图片，此方法会发送图片链接文本
|
||
|
||
Args:
|
||
image_path: 图片文件路径
|
||
title: 消息标题
|
||
text: 附加文本
|
||
|
||
Returns:
|
||
bool: 是否发送成功
|
||
"""
|
||
if not self.webhook:
|
||
logger.warning(f"[钉钉图片链接未发送] {title}")
|
||
return False
|
||
|
||
if not os.path.exists(image_path):
|
||
logger.error(f"图片文件不存在: {image_path}")
|
||
return False
|
||
|
||
try:
|
||
# 读取图片
|
||
with open(image_path, "rb") as f:
|
||
image_data = f.read()
|
||
|
||
# 转为base64
|
||
image_base64 = base64.b64encode(image_data).decode("utf-8")
|
||
|
||
# 获取图片格式
|
||
ext = os.path.splitext(image_path)[1].lower().replace('.', '')
|
||
if ext == 'jpg':
|
||
ext = 'jpeg'
|
||
if ext not in ['png', 'jpeg', 'gif', 'bmp']:
|
||
ext = 'png'
|
||
|
||
# 构建data URL（钉钉可能不支持直接显示，但可以作为链接）
|
||
data_url = f"data:image/{ext};base64,{image_base64[:100]}..."
|
||
|
||
# 发送markdown消息
|
||
markdown = f"## {title}\n\n"
|
||
if text:
|
||
markdown += f"{text}\n\n"
|
||
markdown += f"**图片**: {os.path.basename(image_path)}\n\n"
|
||
markdown += f"大小: {len(image_data)/1024:.1f}KB\n"
|
||
|
||
return self.send_markdown(title, markdown)
|
||
|
||
except Exception as e:
|
||
logger.error(f"发送图片链接异常: {e}")
|
||
return False
|
||
|
||
def send_image_via_oss(
|
||
self,
|
||
image_path: str,
|
||
title: str = "策略图表",
|
||
text: str = "",
|
||
expire_days: int = 7,
|
||
) -> bool:
|
||
"""
|
||
上传图片到 OSS 并通过 Markdown 发送到钉钉
|
||
这是发送图片的推荐方式，不受 20KB 限制
|
||
|
||
Args:
|
||
image_path: 本地图片路径
|
||
title: 消息标题
|
||
text: 附加文本
|
||
expire_days: OSS 链接有效期（天）
|
||
|
||
Returns:
|
||
bool: 是否发送成功
|
||
"""
|
||
if not self.webhook:
|
||
logger.warning(f"[钉钉OSS图片未发送] {title}")
|
||
return False
|
||
|
||
if not os.path.exists(image_path):
|
||
logger.error(f"图片文件不存在: {image_path}")
|
||
return False
|
||
|
||
try:
|
||
# 上传图片到 OSS
|
||
image_url = upload_image_to_oss(image_path, expire_days)
|
||
|
||
if not image_url:
|
||
logger.error("图片上传到 OSS 失败")
|
||
# 尝试直接发送压缩后的图片
|
||
return self.send_image(image_path, title)
|
||
|
||
# 构建 Markdown 消息
|
||
markdown = f"## {title}\n\n"
|
||
|
||
if text:
|
||
markdown += f"{text}\n\n"
|
||
|
||
# 添加图片（钉钉 Markdown 支持图片语法）
|
||
markdown += f"\n\n"
|
||
|
||
# 添加图片信息
|
||
file_size = os.path.getsize(image_path) / 1024
|
||
markdown += f"---\n"
|
||
markdown += f"**图片**: {os.path.basename(image_path)} ({file_size:.1f}KB)\n"
|
||
markdown += f"**有效期**: {expire_days}天\n"
|
||
|
||
# 发送 Markdown 消息
|
||
return self.send_markdown(title, markdown)
|
||
|
||
except Exception as e:
|
||
logger.error(f"发送 OSS 图片异常: {e}")
|
||
# 失败时尝试直接发送
|
||
return self.send_image(image_path, title)
|
||
|
||
|
||
def send_to_all_groups(
|
||
send_func_name: str,
|
||
**kwargs,
|
||
) -> bool:
|
||
"""
|
||
向所有已配置的钉钉群发送消息
|
||
|
||
Args:
|
||
send_func_name: DingTalkBot 的发送方法名，如 'send_text', 'send_markdown', 'send_image_via_oss'
|
||
**kwargs: 传递给发送方法的参数
|
||
|
||
Returns:
|
||
bool: 是否全部发送成功
|
||
"""
|
||
configs = _get_all_dingtalk_configs()
|
||
if not configs:
|
||
logger.warning("没有配置任何钉钉群，消息未发送")
|
||
return False
|
||
|
||
all_success = True
|
||
for i, cfg in enumerate(configs, 1):
|
||
bot = DingTalkBot(webhook=cfg["webhook"], secret=cfg["secret"])
|
||
method = getattr(bot, send_func_name, None)
|
||
if method is None:
|
||
logger.error(f"DingTalkBot 没有方法: {send_func_name}")
|
||
return False
|
||
try:
|
||
success = method(**kwargs)
|
||
if success:
|
||
logger.info(f"群{i} 发送成功")
|
||
else:
|
||
logger.error(f"群{i} 发送失败")
|
||
all_success = False
|
||
except Exception as e:
|
||
logger.error(f"群{i} 发送异常: {e}")
|
||
all_success = False
|
||
return all_success
|
||
|
||
|
||
class NotificationManager:
|
||
"""通知管理器 - 统一管理多种通知渠道"""
|
||
|
||
def __init__(self):
|
||
self.dingtalk = DingTalkBot()
|
||
|
||
def notify(self, message: str, title: str = "系统通知", use_markdown: bool = False):
|
||
"""
|
||
发送通知（优先使用钉钉，失败则记录日志）
|
||
|
||
Args:
|
||
message: 消息内容
|
||
title: 消息标题（markdown模式使用）
|
||
use_markdown: 是否使用markdown格式
|
||
"""
|
||
if use_markdown:
|
||
success = self.dingtalk.send_markdown(title, message)
|
||
else:
|
||
success = self.dingtalk.send_text(message)
|
||
|
||
if not success:
|
||
# 钉钉发送失败，记录到日志
|
||
logger.info(f"[通知] {title}: {message}")
|
||
|
||
def notify_error(self, error_msg: str):
|
||
"""发送错误通知"""
|
||
markdown = f"""## 错误告警
|
||
|
||
**时间**: {time.strftime('%Y-%m-%d %H:%M:%S')}
|
||
|
||
**错误信息**:
|
||
```
|
||
{error_msg}
|
||
```
|
||
"""
|
||
self.notify(markdown, title="系统错误", use_markdown=True)
|
||
|
||
def notify_signal(self, signals: list, signal_type: str = "CCI超卖"):
|
||
"""
|
||
发送交易信号通知
|
||
|
||
Args:
|
||
signals: 信号列表，每项为dict包含code, name等指标
|
||
signal_type: 信号类型名称
|
||
"""
|
||
if not signals:
|
||
logger.info(f"[{signal_type}] 无信号")
|
||
return
|
||
|
||
# 构建markdown表格
|
||
if signals:
|
||
headers = signals[0].keys()
|
||
header_line = " | ".join(headers)
|
||
separator = " | ".join(["---"] * len(headers))
|
||
|
||
rows = []
|
||
for s in signals:
|
||
row = " | ".join(str(v) for v in s.values())
|
||
rows.append(row)
|
||
|
||
table = f"{header_line}\n{separator}\n" + "\n".join(rows)
|
||
else:
|
||
table = "无"
|
||
|
||
markdown = f"""## {signal_type}信号
|
||
|
||
**时间**: {time.strftime('%Y-%m-%d %H:%M:%S')}
|
||
|
||
**筛选结果**:
|
||
|
||
{table}
|
||
|
||
共 {len(signals)} 个标的符合筛选条件。
|
||
"""
|
||
self.notify(markdown, title=f"{signal_type}信号", use_markdown=True)
|