274 lines
12 KiB
Python
274 lines
12 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
舆情驱动自动化：定期扫描最新新闻 → 重要度评分 → 邮件通知
|
||
- 邮件：SMTP（plain/starttls/ssl 三种模式），配置来自设置区
|
||
- 重要度评分（0-100）：
|
||
类别基础分（公司40/业绩35/机构观点30/行业25/市场15）
|
||
+ 情感绝对值权重（|sentiment| × 权重系数 × 100）
|
||
+ 重要关键词命中（+12/个，上限 2 个）
|
||
+ 关联个股数（+5/只）
|
||
- 去重：记录已处理的最大新闻 id（monitor_last_news_id），只扫描新增
|
||
- 每次扫描结果写入 notification_log
|
||
"""
|
||
import logging
|
||
import smtplib
|
||
import threading
|
||
import time
|
||
from email.mime.text import MIMEText
|
||
from email.utils import formataddr, formatdate
|
||
|
||
from database import query, execute, query_one
|
||
from settings import mail_config, monitor_config, monitor_state, set_monitor_state, \
|
||
quiet_config, in_quiet_period
|
||
|
||
log = logging.getLogger("notifier")
|
||
|
||
_scan_lock = threading.Lock()
|
||
|
||
CATEGORY_BASE = {"公司": 40, "业绩": 35, "机构观点": 30, "行业": 25, "市场": 15}
|
||
STRONG_WORDS = ["回购", "中标", "减持", "问询", "停牌", "重组", "预警",
|
||
"上调", "下调", "超预期", "不及预期", "增持", "定增", "退市", "处罚"]
|
||
|
||
|
||
# ===================================================================== 邮件
|
||
def send_email(subject, html_body, to=None, cfg=None, sender_name=None, attachments=None):
|
||
"""发送 HTML 邮件。cfg 来自设置；attachments: [{filename, content(bytes)}]；失败抛异常"""
|
||
from email.mime.multipart import MIMEMultipart
|
||
from email.mime.base import MIMEBase
|
||
from email import encoders
|
||
cfg = cfg or mail_config()
|
||
to = to or cfg["email_to"]
|
||
if attachments:
|
||
msg = MIMEMultipart()
|
||
msg.attach(MIMEText(html_body, "html", "utf-8"))
|
||
for att in attachments:
|
||
part = MIMEBase("application", "octet-stream")
|
||
part.set_payload(att.get("content") or b"")
|
||
encoders.encode_base64(part)
|
||
part.add_header("Content-Disposition", "attachment",
|
||
filename=("utf-8", "", att.get("filename", "report.html")))
|
||
msg.attach(part)
|
||
else:
|
||
msg = MIMEText(html_body, "html", "utf-8")
|
||
msg["From"] = formataddr((sender_name or cfg["sender_name"], cfg["smtp_user"]))
|
||
msg["To"] = to
|
||
msg["Subject"] = subject
|
||
msg["Date"] = formatdate(localtime=True)
|
||
|
||
mode = cfg.get("smtp_mode", "plain")
|
||
if mode == "ssl":
|
||
server = smtplib.SMTP_SSL(cfg["smtp_host"], cfg["smtp_port"], timeout=30)
|
||
else:
|
||
server = smtplib.SMTP(cfg["smtp_host"], cfg["smtp_port"], timeout=30)
|
||
server.ehlo()
|
||
if mode == "starttls":
|
||
server.starttls()
|
||
server.ehlo()
|
||
server.login(cfg["smtp_user"], cfg["smtp_pass"])
|
||
server.sendmail(cfg["smtp_user"], [to], msg.as_string())
|
||
server.quit()
|
||
return True
|
||
|
||
|
||
# ===================================================================== 重要度
|
||
def importance_score(news, cfg):
|
||
"""返回 (score, [原因...])"""
|
||
score = CATEGORY_BASE.get(news.get("category", "市场"), 15)
|
||
reasons = [f"{news.get('category','市场')}类"]
|
||
# 情感
|
||
sent = abs(float(news.get("sentiment") or 0))
|
||
if sent > 0.05:
|
||
bonus = round(sent * cfg.get("sentiment_weight", 0.2) * 100)
|
||
score += bonus
|
||
reasons.append(f"情感强度{sent:.2f}(+{bonus})")
|
||
# 关键词
|
||
text = (news.get("title") or "") + (news.get("content") or "")
|
||
hit = 0
|
||
for kw in cfg.get("keywords", []):
|
||
if kw and kw in text:
|
||
hit += 1
|
||
if hit > 2:
|
||
break
|
||
if hit:
|
||
score += hit * 12
|
||
reasons.append(f"关键词×{hit}(+{hit*12})")
|
||
# 关联个股
|
||
related = [c for c in (news.get("related_stocks") or "").split(",") if c]
|
||
if related:
|
||
score += min(len(related), 3) * 5
|
||
reasons.append(f"关联{len(related)}只个股(+{min(len(related),3)*5})")
|
||
return min(score, 100), reasons
|
||
|
||
|
||
# ===================================================================== 扫描
|
||
def _fetch_related_names(codes):
|
||
names = {}
|
||
if not codes:
|
||
return names
|
||
for c in codes:
|
||
r = query_one("SELECT name FROM stocks WHERE code=?", (c,))
|
||
if r:
|
||
names[c] = r["name"]
|
||
return names
|
||
|
||
|
||
def build_email_html(items):
|
||
"""把重要新闻渲染成 HTML 邮件正文"""
|
||
rows = []
|
||
for it in items:
|
||
n = it["news"]
|
||
tone = "利好" if n["sentiment"] > 0 else ("利空" if n["sentiment"] < 0 else "中性")
|
||
tone_color = "#e03e3e" if tone == "利好" else ("#17a34a" if tone == "利空" else "#888")
|
||
names = "、".join(f"{k}({v})" for k, v in it["stocks"].items()) or "—"
|
||
rows.append(f"""
|
||
<tr style="border-bottom:1px solid #eee;">
|
||
<td style="padding:10px;">
|
||
<div style="font-size:15px;font-weight:bold;">{n['title']}</div>
|
||
<div style="color:#888;font-size:12px;margin:4px 0;">{n['category']} · {n['source']} · {n['publish_date']}</div>
|
||
<div style="font-size:12px;">重要度 <b>{it['score']:.0f}</b> ｜ 情感 <b style="color:{tone_color};">{tone}（{n['sentiment']:+.2f}）</b> ｜ 关联：{names}</div>
|
||
<div style="color:#555;font-size:13px;margin-top:4px;">{n['content'][:120]}{'…' if len(n['content']) > 120 else ''}</div>
|
||
</td>
|
||
</tr>""")
|
||
return f"""
|
||
<html><body style="font-family:Microsoft YaHei,Arial,sans-serif;background:#f5f6f8;padding:20px;">
|
||
<div style="max-width:680px;margin:auto;background:#fff;border-radius:8px;overflow:hidden;border:1px solid #e5e7eb;">
|
||
<div style="background:#1e293b;color:#fff;padding:16px 20px;">
|
||
<div style="font-size:18px;font-weight:bold;">📰 舆情监控 · 重要资讯通知</div>
|
||
<div style="font-size:12px;opacity:.8;margin-top:4px;">智能荐股系统自动扫描 · {time.strftime('%Y-%m-%d %H:%M:%S')}</div>
|
||
</div>
|
||
<div style="padding:6px 20px 20px;">
|
||
<p style="color:#555;font-size:13px;">本次扫描发现 <b>{len(items)}</b> 条重要资讯（重要度达到阈值）：</p>
|
||
<table style="width:100%;border-collapse:collapse;">{''.join(rows)}</table>
|
||
</div>
|
||
<div style="background:#f8fafc;padding:10px 20px;color:#94a3b8;font-size:11px;text-align:center;">
|
||
本邮件由系统自动生成，内容基于模拟数据，仅供演示，不构成投资建议。可在系统「设置」中关闭通知。
|
||
</div>
|
||
</div></body></html>"""
|
||
|
||
|
||
def scan_news(force=False):
|
||
"""
|
||
扫描一次新闻：找出新增且重要度达标的新闻，发送邮件。
|
||
force=True 时也扫描历史未通知过的（用于手动测试）。
|
||
返回 {"checked": n, "important": n, "sent": n, "failed": n}
|
||
"""
|
||
with _scan_lock:
|
||
cfg = monitor_config()
|
||
if not cfg["enabled"] and not force:
|
||
return {"error": "监控未启用"}
|
||
|
||
news_rows = query("SELECT id, title, content, source, category, publish_date, "
|
||
"related_stocks, sentiment FROM news ORDER BY id ASC")
|
||
if not news_rows:
|
||
return {"checked": 0, "important": 0, "sent": 0, "failed": 0}
|
||
|
||
max_id = max(n["id"] for n in news_rows)
|
||
state = monitor_state()
|
||
last_id = state["last_news_id"]
|
||
|
||
# 首次运行：仅记录水位，不通知历史新闻
|
||
if last_id == 0:
|
||
set_monitor_state(last_news_id=max_id, last_scan=time.strftime("%Y-%m-%d %H:%M:%S"), last_sent=0)
|
||
return {"checked": 0, "important": 0, "sent": 0, "failed": 0, "initialized": True}
|
||
|
||
candidates = [n for n in news_rows if n["id"] > last_id]
|
||
|
||
important = []
|
||
for n in candidates:
|
||
score, _r = importance_score(n, cfg)
|
||
if score >= cfg["importance_threshold"]:
|
||
important.append({"news": n, "score": score, "reasons": _r})
|
||
|
||
sent, failed = 0, 0
|
||
if important:
|
||
# 静默期判断：命中则本次不发送邮件（仍记账推进水位，避免静默结束后补发积压）
|
||
qc = quiet_config("monitor")
|
||
quiet = qc["enabled"] and in_quiet_period(qc["ranges"])
|
||
omitted = 0
|
||
if quiet:
|
||
status, msg = "quiet", "静默期内已抑制发送"
|
||
else:
|
||
# 单封邮件最多带 20 条，避免超大邮件
|
||
shown = important[:20]
|
||
omitted = len(important) - len(shown)
|
||
items = [{"news": it["news"], "score": it["score"],
|
||
"stocks": _fetch_related_names([c for c in (it["news"]["related_stocks"] or "").split(",") if c])}
|
||
for it in shown]
|
||
subject = f"[舆情监控] {len(important)} 条重要资讯：{important[0]['news']['title'][:24]}" + \
|
||
(f" 等{len(important)}条" if len(important) > 1 else "")
|
||
try:
|
||
html = build_email_html(items)
|
||
if omitted:
|
||
html += f"<p style='color:#888;font-size:12px'>……另有 {omitted} 条重要资讯已省略，详见系统。</p>"
|
||
if cfg.get("email_enabled", True):
|
||
send_email(subject, html)
|
||
status, msg = "sent", ""
|
||
else:
|
||
status, msg = "skipped", "邮件功能未启用"
|
||
except Exception as e:
|
||
status, msg = "failed", str(e)
|
||
failed = len(important)
|
||
log.warning("notify email fail: %s", e)
|
||
# 写日志
|
||
for it in important:
|
||
execute(
|
||
"INSERT INTO notification_log(news_id, title, category, sentiment, importance, related, status, message) "
|
||
"VALUES(?,?,?,?,?,?,?,?)",
|
||
(it["news"]["id"], it["news"]["title"], it["news"]["category"],
|
||
it["news"]["sentiment"], round(it["score"], 1),
|
||
it["news"]["related_stocks"], status, msg))
|
||
if status == "sent":
|
||
sent = len(important)
|
||
|
||
# 更新水位：前进到已处理的最后一条
|
||
if candidates:
|
||
set_monitor_state(last_news_id=max(max_id, last_id),
|
||
last_scan=time.strftime("%Y-%m-%d %H:%M:%S"), last_sent=sent)
|
||
return {"checked": len(candidates), "important": len(important),
|
||
"sent": sent, "failed": failed}
|
||
|
||
|
||
def notification_log(limit=50):
|
||
return query("SELECT * FROM notification_log ORDER BY id DESC LIMIT ?", (limit,))
|
||
|
||
|
||
# ===================================================================== 调度器
|
||
class MonitorThread(threading.Thread):
|
||
"""后台调度：每 interval 分钟扫描一次"""
|
||
|
||
def __init__(self):
|
||
super().__init__(daemon=True, name="monitor")
|
||
self._stop = threading.Event()
|
||
|
||
def stop(self):
|
||
self._stop.set()
|
||
|
||
def run(self):
|
||
log.info("舆情监控调度器启动")
|
||
while not self._stop.is_set():
|
||
try:
|
||
cfg = monitor_config()
|
||
if cfg["enabled"]:
|
||
try:
|
||
r = scan_news()
|
||
if r.get("checked"):
|
||
log.info("monitor scan: %s", r)
|
||
except Exception as e:
|
||
log.warning("monitor scan error: %s", e)
|
||
except Exception as e:
|
||
log.warning("monitor loop error: %s", e)
|
||
self._stop.wait(cfg.get("interval_min", 30) * 60)
|
||
log.info("舆情监控调度器停止")
|
||
|
||
|
||
_monitor = None
|
||
|
||
|
||
def start_monitor():
|
||
global _monitor
|
||
if _monitor and _monitor.is_alive():
|
||
return _monitor
|
||
_monitor = MonitorThread()
|
||
_monitor.start()
|
||
return _monitor
|