970 lines
32 KiB
Python
970 lines
32 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
新闻智能跟踪系统 - SQLite 数据访问层
|
||
原生 sqlite3，无 ORM，结构与其它项目保持一致。
|
||
"""
|
||
import sqlite3
|
||
import json
|
||
import os
|
||
from datetime import datetime
|
||
|
||
import config
|
||
|
||
_SCHEMA = """
|
||
CREATE TABLE IF NOT EXISTS sources (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
name TEXT NOT NULL,
|
||
type TEXT DEFAULT '',
|
||
url TEXT DEFAULT '',
|
||
description TEXT DEFAULT '',
|
||
weight REAL DEFAULT 1.0,
|
||
kind TEXT DEFAULT 'normal', -- normal 普通(按权重打分) / custom 定制监控(按推送标准LLM判断)
|
||
monitor_standard TEXT DEFAULT '', -- 定制监控：推送标准说明
|
||
scan_interval_min INTEGER DEFAULT 0, -- 本源采集间隔(分钟)，0=跟随所属机制的全局值
|
||
fetch_method TEXT DEFAULT 'auto', -- 获取网页方式: auto(优先web-capture-api,失败回退直接抓取)/webcapture/direct
|
||
capture_params TEXT DEFAULT '{}', -- web-capture-api 抓取参数(JSON: action/wait_time/scroll_times/scroll_delay/backend/viewport...)
|
||
enabled INTEGER DEFAULT 1,
|
||
status TEXT DEFAULT 'ok', -- ok / error
|
||
last_fetch TEXT DEFAULT '',
|
||
last_count INTEGER DEFAULT 0,
|
||
created_at TEXT DEFAULT ''
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS system_error_log (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
source TEXT DEFAULT '', -- 错误来源(采集/分析/通知/汇总/系统...)
|
||
message TEXT DEFAULT '', -- 错误信息
|
||
detail TEXT DEFAULT '', -- 详情
|
||
count INTEGER DEFAULT 1, -- 同类错误累计次数
|
||
pending INTEGER DEFAULT 1, -- 1=待通知(尚未发送过邮件) 0=已通知
|
||
first_seen TEXT DEFAULT '',
|
||
last_seen TEXT DEFAULT '',
|
||
last_notified TEXT DEFAULT '',
|
||
notified_count INTEGER DEFAULT 0,
|
||
created_at TEXT DEFAULT ''
|
||
);
|
||
CREATE INDEX IF NOT EXISTS idx_errlog_pending ON system_error_log(pending);
|
||
|
||
CREATE TABLE IF NOT EXISTS source_snapshots (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
source_id INTEGER NOT NULL,
|
||
fetched_at TEXT DEFAULT '', -- 本次采样时间
|
||
count INTEGER DEFAULT 0, -- 本次采到条数
|
||
status TEXT DEFAULT 'ok', -- ok / error
|
||
detail TEXT DEFAULT '', -- 错误信息 / 备注
|
||
created_at TEXT DEFAULT ''
|
||
);
|
||
CREATE INDEX IF NOT EXISTS idx_snapshots_source ON source_snapshots(source_id, id);
|
||
|
||
CREATE TABLE IF NOT EXISTS articles (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
source_id INTEGER DEFAULT 0,
|
||
title TEXT NOT NULL,
|
||
url TEXT DEFAULT '',
|
||
author TEXT DEFAULT '',
|
||
content TEXT DEFAULT '',
|
||
summary TEXT DEFAULT '',
|
||
domain TEXT DEFAULT '', -- 分类：AI模型与算法 / 芯片与硬件 ...
|
||
entities TEXT DEFAULT '[]', -- JSON 数组：涉及的公司/实体
|
||
importance INTEGER DEFAULT 0, -- LLM 深度评分 1-10
|
||
relevance INTEGER DEFAULT 0, -- 兴趣相关度 0-100
|
||
total_score INTEGER DEFAULT 0, -- 综合分 0-100
|
||
is_important INTEGER DEFAULT 0,
|
||
analysis TEXT DEFAULT '', -- LLM 分析结论(为什么重要)
|
||
llm_status TEXT DEFAULT 'pending', -- pending / done / skipped / error
|
||
notified INTEGER DEFAULT 0, -- 是否已实时邮件通知
|
||
status TEXT DEFAULT 'new', -- new / summarized
|
||
published_at TEXT DEFAULT '',
|
||
collected_at TEXT DEFAULT ''
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS keywords (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
keyword TEXT NOT NULL,
|
||
weight INTEGER DEFAULT 5,
|
||
enabled INTEGER DEFAULT 1
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS domains (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
name TEXT NOT NULL,
|
||
weight INTEGER DEFAULT 5,
|
||
enabled INTEGER DEFAULT 1
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS companies (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
name TEXT NOT NULL,
|
||
enabled INTEGER DEFAULT 1
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS notification_log (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
type TEXT DEFAULT 'realtime', -- realtime / summary
|
||
subject TEXT DEFAULT '',
|
||
count INTEGER DEFAULT 0,
|
||
article_ids TEXT DEFAULT '[]',
|
||
status TEXT DEFAULT 'ok',
|
||
detail TEXT DEFAULT '',
|
||
sent_at TEXT DEFAULT ''
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS settings (
|
||
key TEXT PRIMARY KEY,
|
||
value TEXT DEFAULT ''
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS llm_providers (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
name TEXT NOT NULL,
|
||
base_url TEXT DEFAULT '',
|
||
api_key TEXT DEFAULT '',
|
||
model TEXT DEFAULT '',
|
||
active INTEGER DEFAULT 0, -- 1=当前激活（网页一键切换）
|
||
enabled INTEGER DEFAULT 1,
|
||
created_at TEXT DEFAULT ''
|
||
);
|
||
|
||
CREATE INDEX IF NOT EXISTS idx_articles_collected ON articles(collected_at);
|
||
CREATE INDEX IF NOT EXISTS idx_articles_score ON articles(total_score);
|
||
CREATE INDEX IF NOT EXISTS idx_articles_status ON articles(llm_status);
|
||
"""
|
||
|
||
|
||
def get_conn():
|
||
conn = sqlite3.connect(config.DB_PATH)
|
||
conn.row_factory = sqlite3.Row
|
||
conn.execute("PRAGMA journal_mode=WAL")
|
||
return conn
|
||
|
||
|
||
def init_db():
|
||
os.makedirs(config.DATA_DIR, exist_ok=True)
|
||
conn = get_conn()
|
||
conn.executescript(_SCHEMA)
|
||
# 迁移：旧库补充 sources.kind / monitor_standard / scan_interval_min
|
||
cols = [r["name"] for r in conn.execute("PRAGMA table_info(sources)").fetchall()]
|
||
if "kind" not in cols:
|
||
conn.execute("ALTER TABLE sources ADD COLUMN kind TEXT DEFAULT 'normal'")
|
||
if "monitor_standard" not in cols:
|
||
conn.execute("ALTER TABLE sources ADD COLUMN monitor_standard TEXT DEFAULT ''")
|
||
if "scan_interval_min" not in cols:
|
||
conn.execute("ALTER TABLE sources ADD COLUMN scan_interval_min INTEGER DEFAULT 0")
|
||
# 迁移：历史采样表（每次采集留档）
|
||
conn.execute("""CREATE TABLE IF NOT EXISTS source_snapshots (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
source_id INTEGER NOT NULL,
|
||
fetched_at TEXT DEFAULT '',
|
||
count INTEGER DEFAULT 0,
|
||
status TEXT DEFAULT 'ok',
|
||
detail TEXT DEFAULT '',
|
||
created_at TEXT DEFAULT ''
|
||
)""")
|
||
conn.execute("CREATE INDEX IF NOT EXISTS idx_snapshots_source ON source_snapshots(source_id, id)")
|
||
# 迁移：数据源补充获取方式 / 抓取参数（web-capture-api 集成）
|
||
cols = [r["name"] for r in conn.execute("PRAGMA table_info(sources)").fetchall()]
|
||
if "fetch_method" not in cols:
|
||
conn.execute("ALTER TABLE sources ADD COLUMN fetch_method TEXT DEFAULT 'auto'")
|
||
if "capture_params" not in cols:
|
||
conn.execute("ALTER TABLE sources ADD COLUMN capture_params TEXT DEFAULT '{}'")
|
||
# 迁移：系统错误日志表
|
||
conn.execute("""CREATE TABLE IF NOT EXISTS system_error_log (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
source TEXT DEFAULT '',
|
||
message TEXT DEFAULT '',
|
||
detail TEXT DEFAULT '',
|
||
count INTEGER DEFAULT 1,
|
||
pending INTEGER DEFAULT 1,
|
||
first_seen TEXT DEFAULT '',
|
||
last_seen TEXT DEFAULT '',
|
||
last_notified TEXT DEFAULT '',
|
||
notified_count INTEGER DEFAULT 0,
|
||
created_at TEXT DEFAULT ''
|
||
)""")
|
||
conn.execute("CREATE INDEX IF NOT EXISTS idx_errlog_pending ON system_error_log(pending)")
|
||
conn.commit()
|
||
# 迁移：旧库补充 full_text 列（存页面可读正文）
|
||
cols = [r["name"] for r in conn.execute("PRAGMA table_info(articles)").fetchall()]
|
||
if "full_text" not in cols:
|
||
conn.execute("ALTER TABLE articles ADD COLUMN full_text TEXT DEFAULT ''")
|
||
conn.commit()
|
||
# 迁移：大模型接口增加优先级字段（小优先）+ 通知日志增加 body 列 + LLM 统计表
|
||
cols = [r["name"] for r in conn.execute("PRAGMA table_info(llm_providers)").fetchall()]
|
||
if "priority" not in cols:
|
||
conn.execute("ALTER TABLE llm_providers ADD COLUMN priority INTEGER DEFAULT 0")
|
||
cols = [r["name"] for r in conn.execute("PRAGMA table_info(notification_log)").fetchall()]
|
||
if "body" not in cols:
|
||
conn.execute("ALTER TABLE notification_log ADD COLUMN body TEXT DEFAULT ''")
|
||
conn.execute("""CREATE TABLE IF NOT EXISTS llm_stats (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
provider TEXT DEFAULT '',
|
||
provider_id INTEGER DEFAULT 0,
|
||
day TEXT DEFAULT '',
|
||
calls INTEGER DEFAULT 0,
|
||
prompt_tokens INTEGER DEFAULT 0,
|
||
completion_tokens INTEGER DEFAULT 0,
|
||
total_tokens INTEGER DEFAULT 0,
|
||
failures INTEGER DEFAULT 0
|
||
)""")
|
||
conn.execute("CREATE INDEX IF NOT EXISTS idx_llmstats_prov ON llm_stats(provider, day)")
|
||
conn.execute("CREATE INDEX IF NOT EXISTS idx_llmstats_day ON llm_stats(day)")
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def now_str():
|
||
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
|
||
# ---------------- settings ----------------
|
||
def get_setting(key, default=None):
|
||
conn = get_conn()
|
||
row = conn.execute("SELECT value FROM settings WHERE key=?", (key,)).fetchone()
|
||
conn.close()
|
||
if row is None:
|
||
return default
|
||
try:
|
||
return json.loads(row["value"])
|
||
except Exception:
|
||
return row["value"]
|
||
|
||
|
||
def set_setting(key, value):
|
||
conn = get_conn()
|
||
conn.execute(
|
||
"INSERT INTO settings(key,value) VALUES(?,?) "
|
||
"ON CONFLICT(key) DO UPDATE SET value=excluded.value",
|
||
(key, json.dumps(value, ensure_ascii=False)),
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def get_all_settings():
|
||
conn = get_conn()
|
||
rows = conn.execute("SELECT key,value FROM settings").fetchall()
|
||
conn.close()
|
||
out = {}
|
||
for r in rows:
|
||
try:
|
||
out[r["key"]] = json.loads(r["value"])
|
||
except Exception:
|
||
out[r["key"]] = r["value"]
|
||
return out
|
||
|
||
|
||
# ---------------- sources ----------------
|
||
def list_sources(only_enabled=False):
|
||
conn = get_conn()
|
||
sql = "SELECT * FROM sources"
|
||
if only_enabled:
|
||
sql += " WHERE enabled=1"
|
||
sql += " ORDER BY id"
|
||
rows = conn.execute(sql).fetchall()
|
||
conn.close()
|
||
return [dict(r) for r in rows]
|
||
|
||
|
||
def get_source(sid):
|
||
conn = get_conn()
|
||
row = conn.execute("SELECT * FROM sources WHERE id=?", (sid,)).fetchone()
|
||
conn.close()
|
||
return dict(row) if row else None
|
||
|
||
|
||
def add_source(name, type_, url, desc, weight=1.0, kind="normal", monitor_standard="",
|
||
scan_interval_min=0, fetch_method="auto", capture_params="{}"):
|
||
conn = get_conn()
|
||
cur = conn.execute(
|
||
"INSERT INTO sources(name,type,url,description,weight,kind,monitor_standard,"
|
||
"scan_interval_min,fetch_method,capture_params,created_at) "
|
||
"VALUES(?,?,?,?,?,?,?,?,?,?,?)",
|
||
(name, type_, url, desc, weight, kind, monitor_standard, scan_interval_min,
|
||
fetch_method, capture_params, now_str()),
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
return cur.lastrowid
|
||
|
||
|
||
def update_source(sid, **fields):
|
||
allowed = {"name", "type", "url", "description", "weight", "kind", "monitor_standard",
|
||
"scan_interval_min", "fetch_method", "capture_params", "enabled"}
|
||
sets, vals = [], []
|
||
for k, v in fields.items():
|
||
if k in allowed:
|
||
sets.append(f"{k}=?")
|
||
vals.append(v)
|
||
if not sets:
|
||
return
|
||
vals.append(sid)
|
||
conn = get_conn()
|
||
conn.execute(f"UPDATE sources SET {','.join(sets)} WHERE id=?", vals)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def delete_source(sid):
|
||
conn = get_conn()
|
||
conn.execute("DELETE FROM sources WHERE id=?", (sid,))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def set_all_sources_enabled(val):
|
||
"""一键全部启用/停用。返回受影响的数据源数。"""
|
||
conn = get_conn()
|
||
cur = conn.execute("UPDATE sources SET enabled=?", (1 if val else 0,))
|
||
conn.commit()
|
||
n = cur.rowcount
|
||
conn.close()
|
||
return n
|
||
|
||
|
||
def update_source_fetch(sid, status="ok", count=0):
|
||
conn = get_conn()
|
||
conn.execute(
|
||
"UPDATE sources SET status=?, last_fetch=?, last_count=? WHERE id=?",
|
||
(status, now_str(), count, sid),
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
# ---------------- 历史采样（每次采集留档，可查看/自动流程提取） ----------------
|
||
def add_source_snapshot(source_id, count=0, status="ok", detail=""):
|
||
conn = get_conn()
|
||
conn.execute(
|
||
"INSERT INTO source_snapshots(source_id,fetched_at,count,status,detail,created_at) "
|
||
"VALUES(?,?,?,?,?,?)",
|
||
(source_id, now_str(), count, status, detail, now_str()),
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def list_source_snapshots(source_id, limit=50):
|
||
conn = get_conn()
|
||
rows = conn.execute(
|
||
"SELECT * FROM source_snapshots WHERE source_id=? ORDER BY id DESC LIMIT ?",
|
||
(source_id, limit),
|
||
).fetchall()
|
||
conn.close()
|
||
return [dict(r) for r in rows]
|
||
|
||
|
||
# ---------------- articles ----------------
|
||
def article_exists(url):
|
||
conn = get_conn()
|
||
row = conn.execute("SELECT id FROM articles WHERE url=?", (url,)).fetchone()
|
||
conn.close()
|
||
return row is not None
|
||
|
||
|
||
def add_article(a):
|
||
conn = get_conn()
|
||
cur = conn.execute(
|
||
"""INSERT INTO articles(source_id,title,url,author,content,summary,domain,entities,
|
||
importance,relevance,total_score,is_important,analysis,llm_status,status,
|
||
published_at,collected_at,full_text)
|
||
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) """,
|
||
(a.get("source_id", 0), a["title"], a.get("url", ""), a.get("author", ""),
|
||
a.get("content", ""), a.get("summary", ""), a.get("domain", ""),
|
||
json.dumps(a.get("entities", []), ensure_ascii=False),
|
||
a.get("importance", 0), a.get("relevance", 0), a.get("total_score", 0),
|
||
a.get("is_important", 0), a.get("analysis", ""), a.get("llm_status", "pending"),
|
||
a.get("status", "new"), a.get("published_at", ""), now_str(),
|
||
a.get("full_text", "")),
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
return cur.lastrowid
|
||
|
||
|
||
def update_article(aid, **fields):
|
||
allowed = {"summary", "domain", "entities", "importance", "relevance",
|
||
"total_score", "is_important", "analysis", "llm_status", "notified", "status",
|
||
"full_text", "content", "title", "url"}
|
||
sets, vals = [], []
|
||
for k, v in fields.items():
|
||
if k in allowed:
|
||
if k == "entities":
|
||
v = json.dumps(v, ensure_ascii=False)
|
||
sets.append(f"{k}=?")
|
||
vals.append(v)
|
||
if not sets:
|
||
return
|
||
vals.append(aid)
|
||
conn = get_conn()
|
||
conn.execute(f"UPDATE articles SET {','.join(sets)} WHERE id=?", vals)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def get_article(aid):
|
||
conn = get_conn()
|
||
row = conn.execute(
|
||
"SELECT a.*, s.name AS source_name FROM articles a "
|
||
"LEFT JOIN sources s ON a.source_id=s.id WHERE a.id=?", (aid,)).fetchone()
|
||
conn.close()
|
||
if not row:
|
||
return None
|
||
d = dict(row)
|
||
try:
|
||
d["entities"] = json.loads(d["entities"] or "[]")
|
||
except Exception:
|
||
d["entities"] = []
|
||
return d
|
||
|
||
|
||
def list_articles(**filters):
|
||
"""支持 source_id/domain/is_important/llm_status/min_score/q/date_from/order/limit/offset"""
|
||
sql = ("SELECT a.*, s.name AS source_name FROM articles a "
|
||
"LEFT JOIN sources s ON a.source_id=s.id WHERE 1=1")
|
||
args = []
|
||
if filters.get("source_id"):
|
||
sql += " AND a.source_id=?"
|
||
args.append(filters["source_id"])
|
||
if filters.get("domain"):
|
||
sql += " AND a.domain=?"
|
||
args.append(filters["domain"])
|
||
if filters.get("is_important") is not None:
|
||
sql += " AND a.is_important=?"
|
||
args.append(1 if filters["is_important"] else 0)
|
||
if filters.get("llm_status"):
|
||
sql += " AND a.llm_status=?"
|
||
args.append(filters["llm_status"])
|
||
if filters.get("min_score"):
|
||
sql += " AND a.total_score>=?"
|
||
args.append(filters["min_score"])
|
||
if filters.get("q"):
|
||
sql += " AND (a.title LIKE ? OR a.content LIKE ?)"
|
||
args += [f"%{filters['q']}%", f"%{filters['q']}%"]
|
||
if filters.get("date_from"):
|
||
sql += " AND a.published_at>=?"
|
||
args.append(filters["date_from"])
|
||
order = filters.get("order", "a.collected_at DESC")
|
||
sql += f" ORDER BY {order}"
|
||
if filters.get("limit"):
|
||
sql += " LIMIT ?"
|
||
args.append(filters["limit"])
|
||
if filters.get("offset"):
|
||
sql += " OFFSET ?"
|
||
args.append(filters["offset"])
|
||
conn = get_conn()
|
||
rows = conn.execute(sql, args).fetchall()
|
||
conn.close()
|
||
out = []
|
||
for r in rows:
|
||
d = dict(r)
|
||
try:
|
||
d["entities"] = json.loads(d["entities"] or "[]")
|
||
except Exception:
|
||
d["entities"] = []
|
||
out.append(d)
|
||
return out
|
||
|
||
|
||
def count_articles(**filters):
|
||
sql = "SELECT COUNT(*) c FROM articles a WHERE 1=1"
|
||
args = []
|
||
if filters.get("domain"):
|
||
sql += " AND a.domain=?"
|
||
args.append(filters["domain"])
|
||
if filters.get("is_important") is not None:
|
||
sql += " AND a.is_important=?"
|
||
args.append(1 if filters["is_important"] else 0)
|
||
conn = get_conn()
|
||
c = conn.execute(sql, args).fetchone()["c"]
|
||
conn.close()
|
||
return c
|
||
|
||
|
||
def article_stats():
|
||
conn = get_conn()
|
||
total = conn.execute("SELECT COUNT(*) c FROM articles").fetchone()["c"]
|
||
important = conn.execute("SELECT COUNT(*) c FROM articles WHERE is_important=1").fetchone()["c"]
|
||
pending_llm = conn.execute("SELECT COUNT(*) c FROM articles WHERE llm_status='pending'").fetchone()["c"]
|
||
notified = conn.execute("SELECT COUNT(*) c FROM articles WHERE notified=1").fetchone()["c"]
|
||
conn.close()
|
||
return {"total": total, "important": important, "pending_llm": pending_llm, "notified": notified}
|
||
|
||
|
||
def latest_articles_for_summary(window_hours=24):
|
||
"""新闻机制汇总：普通源(normal)，窗口内 total_score>=50"""
|
||
conn = get_conn()
|
||
rows = conn.execute(
|
||
"SELECT a.*, s.name AS source_name, s.kind AS source_kind FROM articles a "
|
||
"LEFT JOIN sources s ON a.source_id=s.id "
|
||
"WHERE a.published_at >= datetime('now','localtime','-%d hours') "
|
||
"AND (s.kind IS NULL OR s.kind != 'custom') "
|
||
"AND a.total_score>=50 ORDER BY a.total_score DESC LIMIT 200" % int(window_hours)
|
||
).fetchall()
|
||
conn.close()
|
||
out = []
|
||
for r in rows:
|
||
d = dict(r)
|
||
try:
|
||
d["entities"] = json.loads(d["entities"] or "[]")
|
||
except Exception:
|
||
d["entities"] = []
|
||
out.append(d)
|
||
return out
|
||
|
||
|
||
def custom_articles_for_summary(window_hours=24):
|
||
"""定制监控汇总：定制源(custom)，窗口内命中推送标准(is_important=1)的资讯"""
|
||
conn = get_conn()
|
||
rows = conn.execute(
|
||
"SELECT a.*, s.name AS source_name, s.kind AS source_kind FROM articles a "
|
||
"LEFT JOIN sources s ON a.source_id=s.id "
|
||
"WHERE a.published_at >= datetime('now','localtime','-%d hours') "
|
||
"AND s.kind='custom' AND a.is_important=1 "
|
||
"ORDER BY a.published_at DESC, a.id DESC LIMIT 200" % int(window_hours)
|
||
).fetchall()
|
||
conn.close()
|
||
out = []
|
||
for r in rows:
|
||
d = dict(r)
|
||
try:
|
||
d["entities"] = json.loads(d["entities"] or "[]")
|
||
except Exception:
|
||
d["entities"] = []
|
||
out.append(d)
|
||
return out
|
||
|
||
|
||
def pending_llm_articles(limit=10):
|
||
conn = get_conn()
|
||
rows = conn.execute(
|
||
"SELECT a.*, s.name AS source_name FROM articles a "
|
||
"LEFT JOIN sources s ON a.source_id=s.id "
|
||
"WHERE a.llm_status='pending' ORDER BY a.total_score DESC LIMIT ?", (limit,)
|
||
).fetchall()
|
||
conn.close()
|
||
out = []
|
||
for r in rows:
|
||
d = dict(r)
|
||
try:
|
||
d["entities"] = json.loads(d["entities"] or "[]")
|
||
except Exception:
|
||
d["entities"] = []
|
||
out.append(d)
|
||
return out
|
||
|
||
|
||
# ---------------- 兴趣画像 ----------------
|
||
def list_keywords():
|
||
conn = get_conn()
|
||
rows = conn.execute("SELECT * FROM keywords ORDER BY weight DESC").fetchall()
|
||
conn.close()
|
||
return [dict(r) for r in rows]
|
||
|
||
|
||
def add_keyword(kw, weight=5):
|
||
conn = get_conn()
|
||
conn.execute("INSERT OR IGNORE INTO keywords(keyword,weight) VALUES(?,?)", (kw, weight))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def delete_keyword(kid):
|
||
conn = get_conn()
|
||
conn.execute("DELETE FROM keywords WHERE id=?", (kid,))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def list_domains():
|
||
conn = get_conn()
|
||
rows = conn.execute("SELECT * FROM domains ORDER BY weight DESC").fetchall()
|
||
conn.close()
|
||
return [dict(r) for r in rows]
|
||
|
||
|
||
def add_domain(name, weight=5):
|
||
conn = get_conn()
|
||
conn.execute("INSERT OR IGNORE INTO domains(name,weight) VALUES(?,?)", (name, weight))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def delete_domain(did):
|
||
conn = get_conn()
|
||
conn.execute("DELETE FROM domains WHERE id=?", (did,))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def list_companies():
|
||
conn = get_conn()
|
||
rows = conn.execute("SELECT * FROM companies ORDER BY id").fetchall()
|
||
conn.close()
|
||
return [dict(r) for r in rows]
|
||
|
||
|
||
def add_company(name):
|
||
conn = get_conn()
|
||
conn.execute("INSERT OR IGNORE INTO companies(name) VALUES(?)", (name,))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def delete_company(cid):
|
||
conn = get_conn()
|
||
conn.execute("DELETE FROM companies WHERE id=?", (cid,))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
# ---------------- 大模型接口（llm_providers） ----------------
|
||
def list_providers():
|
||
conn = get_conn()
|
||
rows = conn.execute("SELECT * FROM llm_providers ORDER BY active DESC, id").fetchall()
|
||
conn.close()
|
||
return [dict(r) for r in rows]
|
||
|
||
|
||
def get_provider(pid):
|
||
conn = get_conn()
|
||
row = conn.execute("SELECT * FROM llm_providers WHERE id=?", (pid,)).fetchone()
|
||
conn.close()
|
||
return dict(row) if row else None
|
||
|
||
|
||
def add_provider(name, base_url, api_key, model, active=0, enabled=1):
|
||
conn = get_conn()
|
||
cur = conn.execute(
|
||
"INSERT INTO llm_providers(name,base_url,api_key,model,active,enabled,created_at) "
|
||
"VALUES(?,?,?,?,?,?,?)",
|
||
(name, base_url, api_key, model, active, enabled, now_str()),
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
return cur.lastrowid
|
||
|
||
|
||
def update_provider(pid, **fields):
|
||
allowed = {"name", "base_url", "api_key", "model", "active", "enabled"}
|
||
sets, vals = [], []
|
||
for k, v in fields.items():
|
||
if k in allowed:
|
||
sets.append(f"{k}=?")
|
||
vals.append(v)
|
||
if not sets:
|
||
return
|
||
vals.append(pid)
|
||
conn = get_conn()
|
||
conn.execute(f"UPDATE llm_providers SET {','.join(sets)} WHERE id=?", vals)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def delete_provider(pid):
|
||
conn = get_conn()
|
||
conn.execute("DELETE FROM llm_providers WHERE id=?", (pid,))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def get_active_provider():
|
||
conn = get_conn()
|
||
row = conn.execute("SELECT * FROM llm_providers WHERE active=1 AND enabled=1 LIMIT 1").fetchone()
|
||
conn.close()
|
||
return dict(row) if row else None
|
||
|
||
|
||
def set_active_provider(pid):
|
||
conn = get_conn()
|
||
conn.execute("UPDATE llm_providers SET active=0")
|
||
conn.execute("UPDATE llm_providers SET active=1, enabled=1 WHERE id=?", (pid,))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def enabled_providers_except(pid):
|
||
conn = get_conn()
|
||
rows = conn.execute(
|
||
"SELECT * FROM llm_providers WHERE enabled=1 AND id!=? ORDER BY id", (pid,)).fetchall()
|
||
conn.close()
|
||
return [dict(r) for r in rows]
|
||
|
||
|
||
def list_providers_ordered():
|
||
"""按优先级排序的已启用大模型接口：priority 小优先，其次 id（用于调用顺序）。"""
|
||
conn = get_conn()
|
||
rows = conn.execute(
|
||
"SELECT * FROM llm_providers WHERE enabled=1 ORDER BY priority, id").fetchall()
|
||
conn.close()
|
||
return [dict(r) for r in rows]
|
||
|
||
|
||
def reorder_provider(pid, direction):
|
||
"""把某接口在优先级列表中上移(-1)/下移(+1)。
|
||
优先级列表包含全部接口（含停用的），但调用链只取启用的。返回是否成功。"""
|
||
conn = get_conn()
|
||
rows = conn.execute(
|
||
"SELECT id, priority FROM llm_providers ORDER BY priority, id").fetchall()
|
||
idx = next((i for i, r in enumerate(rows) if r["id"] == pid), None)
|
||
if idx is None:
|
||
conn.close()
|
||
return False
|
||
j = idx + direction
|
||
if j < 0 or j >= len(rows):
|
||
conn.close()
|
||
return False
|
||
a, b = rows[idx], rows[j]
|
||
if a["priority"] == b["priority"]:
|
||
# 优先级相等时先重排为 0..n-1，再交换目标两个位置
|
||
for k, r in enumerate(rows):
|
||
conn.execute("UPDATE llm_providers SET priority=? WHERE id=?", (k, r["id"]))
|
||
conn.execute("UPDATE llm_providers SET priority=? WHERE id=?", (j, a["id"]))
|
||
conn.execute("UPDATE llm_providers SET priority=? WHERE id=?", (idx, b["id"]))
|
||
else:
|
||
conn.execute("UPDATE llm_providers SET priority=? WHERE id=?", (b["priority"], a["id"]))
|
||
conn.execute("UPDATE llm_providers SET priority=? WHERE id=?", (a["priority"], b["id"]))
|
||
conn.commit()
|
||
conn.close()
|
||
return True
|
||
|
||
|
||
# ---------------- LLM 调用统计（次数 + Token） ----------------
|
||
def add_llm_stat(provider, provider_id, prompt_tokens, completion_tokens, success=True):
|
||
"""记录一次大模型调用（按天/接口聚合）。success=False 时只累加 calls 与 failures。"""
|
||
day = datetime.now().strftime("%Y-%m-%d")
|
||
conn = get_conn()
|
||
row = conn.execute(
|
||
"SELECT id FROM llm_stats WHERE provider=? AND day=?", (provider, day)).fetchone()
|
||
if row:
|
||
if success:
|
||
conn.execute(
|
||
"UPDATE llm_stats SET calls=calls+1, prompt_tokens=prompt_tokens+?, "
|
||
"completion_tokens=completion_tokens+?, total_tokens=total_tokens+? WHERE id=?",
|
||
(prompt_tokens, completion_tokens, prompt_tokens + completion_tokens, row["id"]))
|
||
else:
|
||
conn.execute("UPDATE llm_stats SET calls=calls+1, failures=failures+1 WHERE id=?",
|
||
(row["id"],))
|
||
else:
|
||
if success:
|
||
conn.execute(
|
||
"INSERT INTO llm_stats(provider,provider_id,day,calls,prompt_tokens,"
|
||
"completion_tokens,total_tokens,failures) VALUES(?,?,?,1,?,?,?,0)",
|
||
(provider, provider_id, day, prompt_tokens, completion_tokens,
|
||
prompt_tokens + completion_tokens))
|
||
else:
|
||
conn.execute(
|
||
"INSERT INTO llm_stats(provider,provider_id,day,calls,prompt_tokens,"
|
||
"completion_tokens,total_tokens,failures) VALUES(?,?,?,1,0,0,0,1)",
|
||
(provider, provider_id, day))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def llm_stats_summary(day=None):
|
||
"""各接口累计统计。day='YYYY-MM-DD' 只看当天；None 看全部。"""
|
||
if day:
|
||
where, args = " WHERE day=?", [day]
|
||
else:
|
||
where, args = "", []
|
||
conn = get_conn()
|
||
rows = conn.execute(
|
||
f"SELECT provider, provider_id, SUM(calls) calls, SUM(prompt_tokens) prompt_tokens, "
|
||
f"SUM(completion_tokens) completion_tokens, SUM(total_tokens) total_tokens, "
|
||
f"SUM(failures) failures FROM llm_stats{where} "
|
||
"GROUP BY provider, provider_id ORDER BY calls DESC", args).fetchall()
|
||
conn.close()
|
||
return [dict(r) for r in rows]
|
||
|
||
|
||
def llm_stats_daily(limit=14):
|
||
"""最近 N 天每日汇总（全部接口合计）。"""
|
||
conn = get_conn()
|
||
rows = conn.execute(
|
||
"SELECT day, SUM(calls) calls, SUM(prompt_tokens) prompt_tokens, "
|
||
"SUM(completion_tokens) completion_tokens, SUM(total_tokens) total_tokens, "
|
||
"SUM(failures) failures FROM llm_stats GROUP BY day ORDER BY day DESC LIMIT ?",
|
||
(limit,)).fetchall()
|
||
conn.close()
|
||
return [dict(r) for r in rows]
|
||
|
||
|
||
# ---------------- 通知日志（分页 + 筛选） ----------------
|
||
def add_log(type_, subject, count, article_ids, status="ok", detail="", body=""):
|
||
conn = get_conn()
|
||
conn.execute(
|
||
"INSERT INTO notification_log(type,subject,count,article_ids,status,detail,sent_at,body) "
|
||
"VALUES(?,?,?,?,?,?,?,?)",
|
||
(type_, subject, count, json.dumps(article_ids, ensure_ascii=False), status, detail,
|
||
now_str(), body))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def delete_logs(ids):
|
||
"""批量删除通知日志，返回删除条数。"""
|
||
if not ids:
|
||
return 0
|
||
marks = ",".join("?" * len(ids))
|
||
conn = get_conn()
|
||
cur = conn.execute(f"DELETE FROM notification_log WHERE id IN ({marks})", ids)
|
||
conn.commit()
|
||
n = cur.rowcount
|
||
conn.close()
|
||
return n
|
||
|
||
|
||
def get_log(lid):
|
||
conn = get_conn()
|
||
row = conn.execute("SELECT * FROM notification_log WHERE id=?", (lid,)).fetchone()
|
||
conn.close()
|
||
return dict(row) if row else None
|
||
|
||
|
||
def list_log_ids(type_=None, status=None, q=None):
|
||
"""返回匹配筛选条件的全部日志 ID（用于一键删除筛选结果）。"""
|
||
where, args = "1=1", []
|
||
if type_:
|
||
where += " AND type=?"
|
||
args.append(type_)
|
||
if status:
|
||
where += " AND status=?"
|
||
args.append(status)
|
||
if q:
|
||
where += " AND (subject LIKE ? OR detail LIKE ?)"
|
||
args += [f"%{q}%", f"%{q}%"]
|
||
conn = get_conn()
|
||
rows = conn.execute(f"SELECT id FROM notification_log WHERE {where}", args).fetchall()
|
||
conn.close()
|
||
return [r["id"] for r in rows]
|
||
|
||
|
||
def list_logs(page=1, page_size=20, type_=None, status=None, q=None):
|
||
"""分页查询通知日志，支持类型/状态/关键词筛选。返回 (total, rows)"""
|
||
where, args = "1=1", []
|
||
if type_:
|
||
where += " AND type=?"
|
||
args.append(type_)
|
||
if status:
|
||
where += " AND status=?"
|
||
args.append(status)
|
||
if q:
|
||
where += " AND (subject LIKE ? OR detail LIKE ?)"
|
||
args += [f"%{q}%", f"%{q}%"]
|
||
conn = get_conn()
|
||
total = conn.execute(
|
||
f"SELECT COUNT(*) c FROM notification_log WHERE {where}", args).fetchone()["c"]
|
||
rows = conn.execute(
|
||
f"SELECT * FROM notification_log WHERE {where} ORDER BY id DESC LIMIT ? OFFSET ?",
|
||
args + [page_size, (page - 1) * page_size],
|
||
).fetchall()
|
||
conn.close()
|
||
out = []
|
||
for r in rows:
|
||
d = dict(r)
|
||
try:
|
||
d["article_ids"] = json.loads(d["article_ids"] or "[]")
|
||
except Exception:
|
||
d["article_ids"] = []
|
||
out.append(d)
|
||
return total, out
|
||
|
||
|
||
def count_logs(type_=None, status=None, q=None):
|
||
where, args = "1=1", []
|
||
if type_:
|
||
where += " AND type=?"
|
||
args.append(type_)
|
||
if status:
|
||
where += " AND status=?"
|
||
args.append(status)
|
||
if q:
|
||
where += " AND (subject LIKE ? OR detail LIKE ?)"
|
||
args += [f"%{q}%", f"%{q}%"]
|
||
conn = get_conn()
|
||
c = conn.execute(f"SELECT COUNT(*) c FROM notification_log WHERE {where}", args).fetchone()["c"]
|
||
conn.close()
|
||
return c
|
||
|
||
|
||
# ---------------- 系统错误日志（错误邮件通知） ----------------
|
||
def add_system_error(source, message, detail=""):
|
||
"""记录一条系统错误：同类错误合并计数并置为待通知。返回 error_id"""
|
||
conn = get_conn()
|
||
now = now_str()
|
||
row = conn.execute(
|
||
"SELECT id FROM system_error_log WHERE source=? AND message=?",
|
||
(source, (message or "")[:300]),
|
||
).fetchone()
|
||
if row:
|
||
conn.execute(
|
||
"UPDATE system_error_log SET count=count+1, pending=1, last_seen=?, detail=? WHERE id=?",
|
||
(now, (detail or "")[:1000], row["id"]),
|
||
)
|
||
eid = row["id"]
|
||
else:
|
||
cur = conn.execute(
|
||
"INSERT INTO system_error_log(source,message,detail,count,pending,first_seen,last_seen,created_at) "
|
||
"VALUES(?,?,?,1,1,?,?,?)",
|
||
(source, (message or "")[:300], (detail or "")[:1000], now, now, now),
|
||
)
|
||
eid = cur.lastrowid
|
||
conn.commit()
|
||
conn.close()
|
||
return eid
|
||
|
||
|
||
def pending_system_errors(limit=20):
|
||
"""待通知的系统错误（未发送过邮件的）"""
|
||
conn = get_conn()
|
||
rows = conn.execute(
|
||
"SELECT * FROM system_error_log WHERE pending=1 ORDER BY last_seen DESC LIMIT ?",
|
||
(limit,),
|
||
).fetchall()
|
||
conn.close()
|
||
return [dict(r) for r in rows]
|
||
|
||
|
||
def list_system_errors(limit=100):
|
||
conn = get_conn()
|
||
rows = conn.execute(
|
||
"SELECT * FROM system_error_log ORDER BY last_seen DESC LIMIT ?", (limit,)).fetchall()
|
||
conn.close()
|
||
return [dict(r) for r in rows]
|
||
|
||
|
||
def mark_errors_notified(eids, t):
|
||
"""标记一批错误已邮件通知"""
|
||
conn = get_conn()
|
||
for eid in eids:
|
||
conn.execute(
|
||
"UPDATE system_error_log SET pending=0, last_notified=?, notified_count=notified_count+1 WHERE id=?",
|
||
(t, eid),
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def get_err_last_send():
|
||
v = get_setting("err_last_send", None)
|
||
return v
|
||
|
||
|
||
def set_err_last_send(t):
|
||
set_setting("err_last_send", t)
|
||
|
||
|
||
def clear_old_articles(days=30):
|
||
conn = get_conn()
|
||
conn.execute(
|
||
"DELETE FROM articles WHERE collected_at < datetime('now','localtime','-%d days')" % days)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def clear_old_errors(days=30):
|
||
conn = get_conn()
|
||
conn.execute(
|
||
"DELETE FROM system_error_log WHERE last_seen < datetime('now','localtime','-%d days')" % days)
|
||
conn.commit()
|
||
conn.close()
|