# -*- 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 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): """发送 HTML 邮件。cfg 来自设置；失败抛异常（调用方捕获）""" cfg = cfg or mail_config() to = to or cfg["email_to"] 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"""
{n['title']}
{n['category']} · {n['source']} · {n['publish_date']}
重要度 {it['score']:.0f} ｜ 情感 {tone}（{n['sentiment']:+.2f}） ｜ 关联：{names}
{n['content'][:120]}{'…' if len(n['content']) > 120 else ''}
""") return f"""
📰 舆情监控 · 重要资讯通知
智能荐股系统自动扫描 · {time.strftime('%Y-%m-%d %H:%M:%S')}

本次扫描发现 {len(items)} 条重要资讯（重要度达到阈值）：

{''.join(rows)}
本邮件由系统自动生成，内容基于模拟数据，仅供演示，不构成投资建议。可在系统「设置」中关闭通知。
""" 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: # 单封邮件最多带 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"

……另有 {omitted} 条重要资讯已省略，详见系统。

" 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