968 lines
58 KiB
Python
968 lines
58 KiB
Python
"""Tenant-scoped knowledge drafts, publication, source lineage and durable jobs."""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import sqlite3
|
||
from datetime import date, datetime, time, timedelta, timezone
|
||
|
||
from archive_store import ArchiveStore, json_text, new_id, safe_scope, tenant_scope, utc_now
|
||
from knowledge_processing import PIPELINE_VERSION, fingerprint, redact, source_key, tokens
|
||
|
||
# 任务的消息上限 0 = 不限。「整理全部聊天」不该在第 10000 条停下。
|
||
UNLIMITED = 0
|
||
# 自动增量整理（持续整理）建的任务归到这个批次，不进任务列表，只在概况里汇总。
|
||
AUTO_RUN = "auto"
|
||
ACTIVE_JOB = ("queued", "running", "paused", "failed")
|
||
CONFIG_DEFAULTS = {
|
||
"default_engine": "model",
|
||
"default_model_provider_id": "",
|
||
"model_concurrency": 4,
|
||
"model_batch": 8,
|
||
"model_filter": True,
|
||
"incremental_engine": "rules",
|
||
}
|
||
|
||
SCHEMA = """
|
||
CREATE INDEX IF NOT EXISTS idx_knowledge_archive_scan
|
||
ON archive_message(tenant_id,source_account_id,conversation_id,sent_at,id);
|
||
CREATE TABLE IF NOT EXISTS knowledge_discarded (
|
||
tenant_id TEXT NOT NULL, dedup_hash TEXT NOT NULL, created_at TEXT NOT NULL,
|
||
PRIMARY KEY(tenant_id,dedup_hash)
|
||
);
|
||
CREATE TABLE IF NOT EXISTS knowledge_settings (
|
||
tenant_id TEXT PRIMARY KEY, enabled INTEGER NOT NULL DEFAULT 0, updated_at TEXT NOT NULL
|
||
);
|
||
CREATE TABLE IF NOT EXISTS knowledge_job (
|
||
id TEXT PRIMARY KEY, tenant_id TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'queued',
|
||
options_json TEXT NOT NULL, cutoff_at TEXT NOT NULL, cursor_json TEXT NOT NULL DEFAULT '{}',
|
||
state_json TEXT NOT NULL DEFAULT '{}', processed INTEGER NOT NULL DEFAULT 0,
|
||
created_items INTEGER NOT NULL DEFAULT 0, duplicates INTEGER NOT NULL DEFAULT 0,
|
||
skipped INTEGER NOT NULL DEFAULT 0, model_calls INTEGER NOT NULL DEFAULT 0,
|
||
input_chars INTEGER NOT NULL DEFAULT 0, output_chars INTEGER NOT NULL DEFAULT 0,
|
||
lease_owner TEXT NOT NULL DEFAULT '', lease_until REAL NOT NULL DEFAULT 0,
|
||
error TEXT NOT NULL DEFAULT '', created_by INTEGER NOT NULL, created_at TEXT NOT NULL,
|
||
updated_at TEXT NOT NULL
|
||
);
|
||
CREATE INDEX IF NOT EXISTS idx_knowledge_job_queue ON knowledge_job(status,lease_until,created_at);
|
||
CREATE TABLE IF NOT EXISTS knowledge_item (
|
||
id TEXT PRIMARY KEY, tenant_id TEXT NOT NULL, job_id TEXT, dedup_hash TEXT NOT NULL,
|
||
title TEXT NOT NULL, question TEXT NOT NULL, answer TEXT NOT NULL, conditions TEXT NOT NULL DEFAULT '',
|
||
category TEXT NOT NULL DEFAULT '', kind TEXT NOT NULL DEFAULT 'qa', flags_json TEXT NOT NULL DEFAULT '[]',
|
||
status TEXT NOT NULL DEFAULT 'draft', revision INTEGER NOT NULL DEFAULT 1,
|
||
valid_until TEXT NOT NULL DEFAULT '', vector_status TEXT NOT NULL DEFAULT 'pending',
|
||
vector_profile TEXT NOT NULL DEFAULT '', created_at TEXT NOT NULL, updated_at TEXT NOT NULL,
|
||
UNIQUE(tenant_id,dedup_hash)
|
||
);
|
||
CREATE INDEX IF NOT EXISTS idx_knowledge_item_list ON knowledge_item(tenant_id,status,updated_at DESC,id);
|
||
CREATE TABLE IF NOT EXISTS knowledge_source (
|
||
item_id TEXT NOT NULL REFERENCES knowledge_item(id), message_id TEXT NOT NULL,
|
||
version_no INTEGER NOT NULL, role TEXT NOT NULL, content TEXT NOT NULL, sent_at TEXT NOT NULL,
|
||
PRIMARY KEY(item_id,message_id,version_no)
|
||
);
|
||
CREATE INDEX IF NOT EXISTS idx_knowledge_source_message ON knowledge_source(message_id,item_id);
|
||
CREATE TABLE IF NOT EXISTS knowledge_revision (
|
||
id TEXT PRIMARY KEY, item_id TEXT NOT NULL REFERENCES knowledge_item(id), revision INTEGER NOT NULL,
|
||
action TEXT NOT NULL, snapshot_json TEXT NOT NULL, actor_id INTEGER NOT NULL, created_at TEXT NOT NULL
|
||
);
|
||
CREATE TABLE IF NOT EXISTS knowledge_retrieval_log (
|
||
id TEXT PRIMARY KEY, tenant_id TEXT NOT NULL, task_id TEXT NOT NULL DEFAULT '',
|
||
query TEXT NOT NULL, hits_json TEXT NOT NULL, mode TEXT NOT NULL, elapsed_ms INTEGER NOT NULL,
|
||
created_at TEXT NOT NULL
|
||
);
|
||
CREATE INDEX IF NOT EXISTS idx_knowledge_log_tenant ON knowledge_retrieval_log(tenant_id,created_at DESC);
|
||
CREATE TABLE IF NOT EXISTS knowledge_watch (
|
||
tenant_id TEXT NOT NULL, source_account_id TEXT NOT NULL, enabled INTEGER NOT NULL DEFAULT 0,
|
||
created_by INTEGER NOT NULL, PRIMARY KEY(tenant_id,source_account_id)
|
||
);
|
||
CREATE TABLE IF NOT EXISTS knowledge_dirty (
|
||
tenant_id TEXT NOT NULL, source_account_id TEXT NOT NULL, conversation_id TEXT NOT NULL,
|
||
changed_at TEXT NOT NULL, PRIMARY KEY(tenant_id,source_account_id,conversation_id)
|
||
);
|
||
CREATE TRIGGER IF NOT EXISTS knowledge_new_message AFTER INSERT ON archive_message
|
||
WHEN EXISTS(SELECT 1 FROM knowledge_watch WHERE tenant_id=NEW.tenant_id AND source_account_id=NEW.source_account_id AND enabled=1)
|
||
BEGIN
|
||
INSERT INTO knowledge_dirty VALUES (NEW.tenant_id,NEW.source_account_id,NEW.conversation_id,NEW.updated_at)
|
||
ON CONFLICT(tenant_id,source_account_id,conversation_id) DO UPDATE SET changed_at=excluded.changed_at;
|
||
END;
|
||
CREATE TRIGGER IF NOT EXISTS knowledge_updated_message AFTER UPDATE OF content,status ON archive_message
|
||
WHEN (OLD.content!=NEW.content OR OLD.status!=NEW.status) AND EXISTS(
|
||
SELECT 1 FROM knowledge_watch WHERE tenant_id=NEW.tenant_id AND source_account_id=NEW.source_account_id AND enabled=1)
|
||
BEGIN
|
||
INSERT INTO knowledge_dirty VALUES (NEW.tenant_id,NEW.source_account_id,NEW.conversation_id,NEW.updated_at)
|
||
ON CONFLICT(tenant_id,source_account_id,conversation_id) DO UPDATE SET changed_at=excluded.changed_at;
|
||
END;
|
||
CREATE TABLE IF NOT EXISTS knowledge_run (
|
||
id TEXT PRIMARY KEY, label TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'active',
|
||
options_json TEXT NOT NULL, tenants_json TEXT NOT NULL, created_by INTEGER NOT NULL,
|
||
created_at TEXT NOT NULL, updated_at TEXT NOT NULL
|
||
);
|
||
CREATE INDEX IF NOT EXISTS idx_knowledge_run_created ON knowledge_run(created_at DESC);
|
||
CREATE TABLE IF NOT EXISTS knowledge_config (
|
||
key TEXT PRIMARY KEY, value_json TEXT NOT NULL, updated_by INTEGER NOT NULL DEFAULT 0, updated_at TEXT NOT NULL
|
||
);
|
||
CREATE VIRTUAL TABLE IF NOT EXISTS knowledge_fts USING fts5(item_id UNINDEXED, terms);
|
||
CREATE TRIGGER IF NOT EXISTS knowledge_source_changed AFTER UPDATE OF content,status ON archive_message
|
||
WHEN OLD.content != NEW.content OR OLD.status != NEW.status
|
||
BEGIN
|
||
UPDATE knowledge_item SET status='stale',updated_at=NEW.updated_at
|
||
WHERE id IN (SELECT item_id FROM knowledge_source WHERE message_id=NEW.id);
|
||
DELETE FROM knowledge_fts WHERE item_id IN
|
||
(SELECT item_id FROM knowledge_source WHERE message_id=NEW.id);
|
||
END;
|
||
CREATE TRIGGER IF NOT EXISTS knowledge_source_deleted BEFORE DELETE ON archive_message
|
||
BEGIN
|
||
UPDATE knowledge_item SET status='stale' WHERE id IN
|
||
(SELECT item_id FROM knowledge_source WHERE message_id=OLD.id);
|
||
DELETE FROM knowledge_fts WHERE item_id IN
|
||
(SELECT item_id FROM knowledge_source WHERE message_id=OLD.id);
|
||
END;
|
||
"""
|
||
|
||
|
||
def day_boundary(value: str, *, end=False) -> str:
|
||
if not value:
|
||
return ""
|
||
# UI dates are Asia/Shanghai business dates. Store / compare UTC half-open intervals.
|
||
day = date.fromisoformat(value)
|
||
if end:
|
||
day += timedelta(days=1)
|
||
return datetime.combine(day, time(), timezone(timedelta(hours=8))).astimezone(
|
||
timezone.utc).isoformat(timespec="milliseconds")
|
||
|
||
|
||
|
||
def job_scope(tenant, options, cutoff):
|
||
"""Use identical scope and cutoff predicates for preview, progress and worker."""
|
||
where = ["m.tenant_id=?", "m.created_at<=?", "c.conversation_type IN ('direct_wechat','direct_wecom')"]
|
||
args = [tenant, cutoff]
|
||
if options.get("source_account_id"):
|
||
# 不指定来源 = 这个账号下的全部客服来源
|
||
where.append("m.source_account_id=?")
|
||
args.append(options["source_account_id"])
|
||
for key, condition in (("date_start", "m.sent_at>=?"), ("date_end", "m.sent_at<?")):
|
||
if options.get(key):
|
||
where.append(condition)
|
||
args.append(options[key])
|
||
if options.get("conversation_id"):
|
||
where.append("m.conversation_id=?")
|
||
args.append(options["conversation_id"])
|
||
return where, args
|
||
|
||
|
||
def job_options(values):
|
||
options = dict(values)
|
||
options.setdefault("engine", "rules")
|
||
start = day_boundary(options.get("date_from", ""))
|
||
end = day_boundary(options.get("date_to", ""), end=True)
|
||
if start and end and start >= end:
|
||
raise ValueError("结束日期不能早于开始日期")
|
||
if options.get("engine", "rules") not in {"rules", "model"}:
|
||
raise ValueError("未知整理方式")
|
||
# 0 = 不限：整理全部聊天、不设调用上限
|
||
for key, default, minimum, maximum in (("max_messages", 10000, 0, 100000000), ("max_model_calls", 100, 0, 10000000),
|
||
("model_timeout_seconds", 90, 30, 180), ("model_concurrency", 4, 1, 16)):
|
||
value = options.get(key, default)
|
||
if not isinstance(value, int) or isinstance(value, bool) or not minimum <= value <= maximum:
|
||
if key == "model_timeout_seconds":
|
||
raise ValueError("模型等待时间应为 30 至 180 秒")
|
||
if key == "model_concurrency":
|
||
raise ValueError("模型并发数应为 1 至 16")
|
||
raise ValueError("处理上限超出允许范围")
|
||
options[key] = value
|
||
options["model_filter"] = bool(options.get("model_filter", True))
|
||
# 一次请求合并几段问答。不写 = 旧版单账号任务一段一请求，整理批次 8 段（见 knowledge_worker.batch_size）
|
||
if options.get("model_batch") is not None:
|
||
batch = options["model_batch"]
|
||
if not isinstance(batch, int) or isinstance(batch, bool) or not 1 <= batch <= 20:
|
||
raise ValueError("每次合并的问答段数应为 1 至 20")
|
||
provider = options.get('model_provider_id', '')
|
||
if not isinstance(provider, str) or len(provider) > 128:
|
||
raise ValueError("模型标识无效")
|
||
options.update(date_start=start, date_end=end, pipeline_version=PIPELINE_VERSION)
|
||
return options
|
||
|
||
|
||
def load_config(database) -> dict:
|
||
"""知识整理的全局设置：默认整理方式、默认模型、并发、是否让模型筛掉没价值的问答。"""
|
||
config = dict(CONFIG_DEFAULTS)
|
||
try:
|
||
with database.connect() as db:
|
||
rows = db.execute("SELECT key,value_json FROM knowledge_config").fetchall()
|
||
except sqlite3.OperationalError:
|
||
return config # 表还没建（旧库刚升级）：用默认值
|
||
for key, value in rows:
|
||
if key in CONFIG_DEFAULTS:
|
||
try:
|
||
config[key] = json.loads(value)
|
||
except ValueError:
|
||
pass
|
||
return config
|
||
|
||
|
||
class KnowledgeStore:
|
||
def __init__(self, database):
|
||
self.database = database
|
||
|
||
def initialize(self):
|
||
ArchiveStore(self.database).initialize()
|
||
with self.database.connect() as db:
|
||
from knowledge_review import REVIEW_SCHEMA
|
||
db.executescript(SCHEMA + REVIEW_SCHEMA)
|
||
# 只加列：回滚到旧版本时旧代码照样能读这两张表。API、网关、Worker 同时启动会抢着加，晚到的跳过。
|
||
for table, column, ddl in (("knowledge_job", "run_id", "TEXT NOT NULL DEFAULT ''"),
|
||
("knowledge_item", "source_key", "TEXT NOT NULL DEFAULT ''")):
|
||
if column not in {row[1] for row in db.execute(f"PRAGMA table_info({table})")}:
|
||
try:
|
||
db.execute(f"ALTER TABLE {table} ADD COLUMN {column} {ddl}")
|
||
except sqlite3.OperationalError as exc:
|
||
if "duplicate column" not in str(exc).lower():
|
||
raise
|
||
db.execute("CREATE INDEX IF NOT EXISTS idx_knowledge_job_run ON knowledge_job(run_id,status)")
|
||
db.execute("CREATE INDEX IF NOT EXISTS idx_knowledge_item_source ON knowledge_item(tenant_id,source_key)")
|
||
db.commit()
|
||
# 旧知识补上对话指纹（每条只算一次；几千条，启动时顺手做完）
|
||
while True:
|
||
ids = [row[0] for row in db.execute(
|
||
"SELECT id FROM knowledge_item WHERE source_key='' AND EXISTS("
|
||
"SELECT 1 FROM knowledge_source s WHERE s.item_id=knowledge_item.id) LIMIT 500")]
|
||
if not ids:
|
||
break
|
||
for item_id in ids:
|
||
sources = [{"message_id": r[0], "version_no": r[1]} for r in db.execute(
|
||
"SELECT message_id,version_no FROM knowledge_source WHERE item_id=?", (item_id,))]
|
||
db.execute("UPDATE knowledge_item SET source_key=? WHERE id=? AND source_key=''",
|
||
(source_key(sources), item_id))
|
||
db.commit()
|
||
|
||
@staticmethod
|
||
def public(row):
|
||
result = dict(row)
|
||
for field in ("flags", "options", "cursor", "state"):
|
||
key = field + "_json"
|
||
if key in result:
|
||
result[field] = json.loads(result.pop(key))
|
||
result.pop("lease_owner", None)
|
||
result.pop("lease_until", None)
|
||
result["pending_items"] = len(result.get("state", {}).get("_pending", []))
|
||
result.pop("state", None)
|
||
result.pop("cursor", None)
|
||
if "options" in result:
|
||
options = result["options"]
|
||
total = options.get("total_messages")
|
||
result["total_messages"] = total
|
||
result["completion_reason"] = options.get("completion_reason", "")
|
||
result["progress_percent"] = (100 if result["status"] == "completed" else
|
||
min(99, int(result["processed"] * 100 / total)) if total else None)
|
||
result["updated_items"] = options.get("updated_items", 0)
|
||
result["model_filtered"] = options.get("model_filtered", 0)
|
||
result["skip_reasons"] = options.get("skip_reasons", {})
|
||
return result
|
||
|
||
def settings(self, tenant):
|
||
tenant = safe_scope(tenant)
|
||
with self.database.connect() as db:
|
||
row = db.execute("SELECT * FROM knowledge_settings WHERE tenant_id=?", (tenant,)).fetchone()
|
||
return dict(row) if row else {"tenant_id": tenant, "enabled": False, "updated_at": ""}
|
||
|
||
def set_enabled(self, tenant, enabled, actor, ip):
|
||
tenant = safe_scope(tenant)
|
||
with self.database.connect() as db:
|
||
if enabled:
|
||
from knowledge_retriever import KnowledgeRetriever
|
||
available = db.execute("SELECT id FROM knowledge_item WHERE tenant_id=? AND status='published' "
|
||
"AND (valid_until='' OR valid_until>=?)", (tenant, KnowledgeRetriever._today()))
|
||
if not any(self._fresh(db, row["id"], tenant) for row in available):
|
||
raise ValueError("请先审核并发布至少一条有效知识，再开启客服使用知识")
|
||
db.execute("INSERT INTO knowledge_settings VALUES (?,?,?) ON CONFLICT(tenant_id) "
|
||
"DO UPDATE SET enabled=excluded.enabled,updated_at=excluded.updated_at",
|
||
(tenant, int(enabled), utc_now()))
|
||
self.database._audit(db, actor, "knowledge.settings", f"tenant={tenant} enabled={enabled}", ip)
|
||
return self.settings(tenant)
|
||
|
||
def watches(self, scope):
|
||
scope = tenant_scope(scope)
|
||
with self.database.connect() as db:
|
||
return [dict(r) for r in db.execute(f"SELECT * FROM knowledge_watch WHERE {scope.clause()}", scope.params)]
|
||
|
||
def set_watch(self, tenant, source_id, enabled, actor, ip):
|
||
with self.database.connect() as db:
|
||
row = db.execute("SELECT 1 FROM knowledge_watch WHERE tenant_id=? AND source_account_id=?",
|
||
(tenant, source_id)).fetchone()
|
||
if row is None:
|
||
raise KeyError("请先创建加工任务并勾选自动整理")
|
||
db.execute("UPDATE knowledge_watch SET enabled=?,created_by=? WHERE tenant_id=? AND source_account_id=?",
|
||
(int(enabled), actor, tenant, source_id))
|
||
self.database._audit(db, actor, "knowledge.watch", f"source={source_id} enabled={enabled}", ip)
|
||
return {"enabled": enabled}
|
||
|
||
def stats(self, scope):
|
||
scope = tenant_scope(scope)
|
||
with self.database.connect() as db:
|
||
from knowledge_retriever import KnowledgeRetriever
|
||
rows = db.execute(f"SELECT CASE WHEN status='published' AND valid_until!='' AND valid_until<? "
|
||
f"THEN 'expired' ELSE status END effective_status,COUNT(*) n FROM knowledge_item WHERE {scope.clause()} "
|
||
"GROUP BY effective_status", [KnowledgeRetriever._today(), *scope.params]).fetchall()
|
||
return {row["effective_status"]: row["n"] for row in rows}
|
||
|
||
def list_items(self, scope, status="", query="", page=1, page_size=30):
|
||
scope = tenant_scope(scope)
|
||
where, args = scope.clause(), scope.params
|
||
if status:
|
||
where += " AND status=?"
|
||
args.append(status)
|
||
if query:
|
||
where += " AND (title LIKE ? OR question LIKE ?)"
|
||
args.extend([f"%{query[:100]}%"] * 2)
|
||
with self.database.connect() as db:
|
||
total = db.execute(f"SELECT COUNT(*) FROM knowledge_item WHERE {where}", args).fetchone()[0]
|
||
rows = db.execute(f"SELECT * FROM knowledge_item WHERE {where} ORDER BY updated_at DESC,id "
|
||
"LIMIT ? OFFSET ?", [*args, page_size, (page - 1) * page_size]).fetchall()
|
||
return {"items": [self.public(r) for r in rows], "total": total}
|
||
|
||
@staticmethod
|
||
def _item(db, tenant, item_id):
|
||
row = db.execute("SELECT * FROM knowledge_item WHERE id=? AND tenant_id=?", (item_id, tenant)).fetchone()
|
||
if row is None:
|
||
raise KeyError("知识不存在或不在当前账号范围内")
|
||
return row
|
||
|
||
def detail(self, tenant, item_id, *, sources=True):
|
||
with self.database.connect() as db:
|
||
item = self.public(self._item(db, tenant, item_id))
|
||
item["sources"] = [dict(r) for r in db.execute(
|
||
"SELECT * FROM knowledge_source WHERE item_id=? ORDER BY sent_at,message_id", (item_id,))] if sources else []
|
||
item["history"] = [dict(r) for r in db.execute(
|
||
"SELECT revision,action,actor_id,created_at FROM knowledge_revision WHERE item_id=? "
|
||
"ORDER BY created_at DESC,id LIMIT 30", (item_id,))]
|
||
return item
|
||
|
||
def _record(self, db, item_id, actor, action, ip=""):
|
||
row = db.execute("SELECT * FROM knowledge_item WHERE id=?", (item_id,)).fetchone()
|
||
db.execute("INSERT INTO knowledge_revision VALUES (?,?,?,?,?,?,?)",
|
||
(new_id(), item_id, row["revision"], action, json_text(dict(row)), actor, utc_now()))
|
||
if actor:
|
||
self.database._audit(db, actor, "knowledge." + action,
|
||
f"item={item_id} revision={row['revision']}", ip)
|
||
|
||
@staticmethod
|
||
def _fresh(db, item_id, tenant):
|
||
return db.execute("""SELECT 1 FROM knowledge_source s LEFT JOIN archive_message m ON m.id=s.message_id
|
||
WHERE s.item_id=? AND (m.id IS NULL OR m.tenant_id!=? OR m.status!='normal' OR
|
||
s.version_no!=(SELECT MAX(version_no) FROM archive_message_version WHERE message_id=s.message_id))
|
||
LIMIT 1""", (item_id, tenant)).fetchone() is None
|
||
|
||
def save(self, tenant, item_id, values, actor, ip):
|
||
fields = {k: str(values.get(k) or "").strip() for k in
|
||
("title", "question", "answer", "conditions", "category", "kind", "valid_until")}
|
||
if not all(fields[k] for k in ("title", "question", "answer", "conditions")):
|
||
raise ValueError("标题、问题、答案和适用条件均不能为空")
|
||
if fields["kind"] not in {"qa", "procedure", "case"}:
|
||
raise ValueError("未知知识类型")
|
||
if fields["valid_until"]:
|
||
date.fromisoformat(fields["valid_until"])
|
||
# Defense in depth: the editor must not reintroduce obvious identifiers.
|
||
for key in ("title", "question", "answer", "conditions", "category"):
|
||
fields[key] = redact(fields[key])
|
||
with self.database.connect() as db:
|
||
db.execute("BEGIN IMMEDIATE")
|
||
row = self._item(db, tenant, item_id)
|
||
if row["revision"] != values["revision"]:
|
||
raise ValueError("知识已被其他人修改，请刷新后重试")
|
||
if not self._fresh(db, item_id, tenant):
|
||
raise ValueError("来源消息已变化，请重新加工原会话；旧草稿不能重新发布")
|
||
db.execute("UPDATE knowledge_item SET " + ",".join(k + "=?" for k in fields) +
|
||
",status='draft',revision=revision+1,vector_status='pending',updated_at=? WHERE id=?",
|
||
[*fields.values(), utc_now(), item_id])
|
||
db.execute("DELETE FROM knowledge_fts WHERE item_id=?", (item_id,))
|
||
self._record(db, item_id, actor, "edit", ip)
|
||
return self.detail(tenant, item_id)
|
||
|
||
def transition(self, tenant, item_id, revision, action, actor, ip, confirmed=False):
|
||
target = {"approve": "approved", "reject": "rejected", "disable": "disabled"}.get(action)
|
||
if target is None:
|
||
raise ValueError("未知操作")
|
||
with self.database.connect() as db:
|
||
db.execute("BEGIN IMMEDIATE")
|
||
row = self._item(db, tenant, item_id)
|
||
if row["revision"] != revision:
|
||
raise ValueError("知识版本已变化，请刷新")
|
||
if action == "disable" and row["status"] != "published":
|
||
raise ValueError("仅已发布知识可停用")
|
||
if action == "approve":
|
||
self._check_expiry(row)
|
||
if row["status"] != "draft" or not confirmed or not row["conditions"].strip():
|
||
raise ValueError("请先补充适用条件，并确认身份、事实、脱敏和通用适用性已审核")
|
||
if not self._fresh(db, item_id, tenant):
|
||
raise ValueError("来源已变化，请重新加工")
|
||
if action == "reject" and row["status"] not in {"draft", "approved"}:
|
||
raise ValueError("仅草稿或待发布知识可驳回")
|
||
db.execute("UPDATE knowledge_item SET status=?,updated_at=? WHERE id=?", (target, utc_now(), item_id))
|
||
db.execute("DELETE FROM knowledge_fts WHERE item_id=?", (item_id,))
|
||
self._record(db, item_id, actor, action, ip)
|
||
return self.detail(tenant, item_id)
|
||
|
||
def delete_rejected(self, tenant, item_id, revision, actor, ip):
|
||
with self.database.connect() as db:
|
||
db.execute("BEGIN IMMEDIATE")
|
||
row = self._item(db, tenant, item_id)
|
||
if row['revision'] != revision or row['status'] != 'rejected':
|
||
raise ValueError('仅当前版本的已驳回知识可以删除，请刷新确认状态')
|
||
db.execute('INSERT OR IGNORE INTO knowledge_discarded VALUES (?,?,?)',
|
||
(tenant, row['dedup_hash'], utc_now()))
|
||
if row['source_key']:
|
||
# 同一段对话再整理（换个模型、换种说法）也不再生成
|
||
db.execute('INSERT OR IGNORE INTO knowledge_discarded VALUES (?,?,?)',
|
||
(tenant, 'src:' + row['source_key'], utc_now()))
|
||
db.execute('DELETE FROM knowledge_fts WHERE item_id=?', (item_id,))
|
||
db.execute('DELETE FROM knowledge_source WHERE item_id=?', (item_id,))
|
||
db.execute('DELETE FROM knowledge_revision WHERE item_id=?', (item_id,))
|
||
db.execute('DELETE FROM knowledge_item WHERE id=? AND tenant_id=?', (item_id, tenant))
|
||
self.database._audit(db, actor, 'knowledge.delete', f'item={item_id} revision={revision}', ip)
|
||
return {'id': item_id, 'status': 'deleted'}
|
||
|
||
@staticmethod
|
||
def _check_expiry(row):
|
||
from knowledge_retriever import KnowledgeRetriever
|
||
if row["valid_until"] and row["valid_until"] < KnowledgeRetriever._today():
|
||
raise ValueError("知识已过有效期，请更新内容和有效期后重新审核")
|
||
|
||
def publish(self, tenant, item_id, revision, actor, ip):
|
||
from knowledge_retriever import VectorIndex
|
||
with self.database.connect() as db:
|
||
row = dict(self._item(db, tenant, item_id))
|
||
if row["revision"] != revision or row["status"] != "approved":
|
||
raise ValueError("只能发布当前已审核版本")
|
||
self._check_expiry(row)
|
||
index = VectorIndex()
|
||
profile = index.publish(row) if index.configured else ""
|
||
with self.database.connect() as db:
|
||
db.execute("BEGIN IMMEDIATE")
|
||
current = self._item(db, tenant, item_id)
|
||
if current["revision"] != revision or current["status"] != "approved" or not self._fresh(db, item_id, tenant):
|
||
raise ValueError("发布期间知识或来源发生变化，请重新审核")
|
||
db.execute("DELETE FROM knowledge_fts WHERE item_id=?", (item_id,))
|
||
db.execute("INSERT INTO knowledge_fts(item_id,terms) VALUES (?,?)",
|
||
(item_id, " ".join(tokens(row["title"] + " " + row["question"] + " " + row["answer"]))))
|
||
db.execute("UPDATE knowledge_item SET status='published',vector_status=?,vector_profile=?,updated_at=? WHERE id=?",
|
||
("ready" if profile else "not_configured", profile, utc_now(), item_id))
|
||
self._record(db, item_id, actor, "publish", ip)
|
||
return self.detail(tenant, item_id)
|
||
|
||
@staticmethod
|
||
def _source_name(db, tenant, options):
|
||
if not options.get("source_account_id"):
|
||
count = db.execute("SELECT COUNT(*) FROM archive_source_account WHERE tenant_id=?", (tenant,)).fetchone()[0]
|
||
return f"全部客服来源（{count} 个）"
|
||
source = db.execute("SELECT id,display_name,external_account_id FROM archive_source_account "
|
||
"WHERE id=? AND tenant_id=?", (options["source_account_id"], tenant)).fetchone()
|
||
if source is None:
|
||
raise ValueError("请选择当前租户的客服来源账号")
|
||
return source["display_name"] or source["external_account_id"]
|
||
|
||
@staticmethod
|
||
def _limited(total, limit):
|
||
return min(total, limit) if limit else total
|
||
|
||
def _scope_counts(self, db, tenant, options, cutoff, *, quality=True):
|
||
source_name = self._source_name(db, tenant, options)
|
||
where, args = job_scope(tenant, options, cutoff)
|
||
if not quality:
|
||
# Creating a job holds the writer lock. Count the snapshot without reading
|
||
# message bodies or resolving identities; quality analysis stays read-only.
|
||
total = db.execute("SELECT COUNT(*) FROM archive_message m "
|
||
"JOIN archive_conversation c ON c.id=m.conversation_id WHERE " + " AND ".join(where) +
|
||
" AND EXISTS(SELECT 1 FROM archive_message_version v WHERE v.message_id=m.id AND v.created_at<=?)",
|
||
[*args, cutoff]).fetchone()[0]
|
||
return {"available_messages": total, "total_messages": self._limited(total, options["max_messages"]),
|
||
"source_name": source_name}
|
||
row = db.execute("""SELECT COUNT(*) available_messages,COUNT(DISTINCT m.conversation_id) conversations,
|
||
COALESCE(SUM(v.status='normal' AND lower(m.message_type) IN ('text','文本','文字','1','引用','quote')
|
||
AND trim(v.content)!='' AND m.direction IN ('inbound','outbound')),0) eligible_text_messages
|
||
FROM archive_message m JOIN archive_conversation c ON c.id=m.conversation_id
|
||
JOIN archive_source_account a ON a.id=m.source_account_id
|
||
JOIN archive_message_version v ON v.message_id=m.id AND v.version_no=(
|
||
SELECT MAX(v2.version_no) FROM archive_message_version v2
|
||
WHERE v2.message_id=m.id AND v2.created_at<=?) WHERE """ + " AND ".join(where),
|
||
[cutoff, *args]).fetchone()
|
||
result = dict(row)
|
||
limit = options["max_messages"]
|
||
result.update(total_messages=self._limited(row["available_messages"], limit),
|
||
limited=bool(limit) and row["available_messages"] > limit, cutoff_at=cutoff,
|
||
source_name=source_name)
|
||
return result
|
||
|
||
def preview_job(self, tenant, values):
|
||
tenant = safe_scope(tenant)
|
||
options = job_options(values)
|
||
with self.database.connect() as db:
|
||
return self._scope_counts(db, tenant, options, utc_now())
|
||
|
||
def create_job(self, tenant, options, actor, ip):
|
||
tenant = safe_scope(tenant)
|
||
options = job_options(options)
|
||
if not options.get("staff_confirmed"):
|
||
raise ValueError("请确认所选来源账号是客服账号，且这些聊天允许用于知识整理")
|
||
if options['engine'] == 'model':
|
||
from knowledge_models import resolve_model
|
||
resolve_model(self.database, options)
|
||
scope_key = fingerprint(json_text({k: options.get(k, "") for k in (
|
||
"source_account_id", "conversation_id", "date_start", "date_end", "engine", "max_messages", "max_model_calls", "model_provider_id", "model_timeout_seconds")}))
|
||
with self.database.connect() as db:
|
||
db.execute("BEGIN IMMEDIATE")
|
||
active = db.execute("SELECT options_json FROM knowledge_job WHERE tenant_id=? "
|
||
"AND status IN ('queued','running','paused')", (tenant,))
|
||
if any(json.loads(row[0]).get("scope_key") == scope_key for row in active):
|
||
raise ValueError("相同范围已有未完成任务，请在加工任务中继续或取消原任务")
|
||
job_id, now = new_id(), utc_now()
|
||
estimate = self._scope_counts(db, tenant, options, now, quality=False)
|
||
options.update(total_messages=estimate["total_messages"], available_messages=estimate["available_messages"],
|
||
source_name=estimate["source_name"], scope_key=scope_key)
|
||
db.execute("INSERT INTO knowledge_job(id,tenant_id,options_json,cutoff_at,created_by,created_at,updated_at) "
|
||
"VALUES (?,?,?,?,?,?,?)", (job_id, tenant, json_text(options), now, actor, now, now))
|
||
if options.get("auto_watch"):
|
||
db.execute("INSERT INTO knowledge_watch VALUES (?,?,1,?) ON CONFLICT(tenant_id,source_account_id) "
|
||
"DO UPDATE SET enabled=1,created_by=excluded.created_by", (tenant, options["source_account_id"], actor))
|
||
self.database._audit(db, actor, "knowledge.job.create", f"job={job_id} tenant={tenant}", ip)
|
||
return self.job(tenant, job_id)
|
||
|
||
def job(self, tenant, job_id):
|
||
with self.database.connect() as db:
|
||
row = db.execute("SELECT * FROM knowledge_job WHERE id=? AND tenant_id=?", (job_id, tenant)).fetchone()
|
||
if row is None:
|
||
raise KeyError("任务不存在")
|
||
return self.public(row)
|
||
|
||
def jobs(self, scope, *, include_auto=False):
|
||
scope = tenant_scope(scope)
|
||
# 持续整理每个会话静默后建一个小任务，几天就上千条，会把手动任务挤出列表；默认不列。
|
||
auto = "" if include_auto else " AND run_id!='auto' AND json_extract(options_json,'$.incremental') IS NOT 1"
|
||
with self.database.connect() as db:
|
||
return [self.public(r) for r in db.execute(f"SELECT * FROM knowledge_job WHERE {scope.clause()}{auto} "
|
||
"ORDER BY created_at DESC LIMIT 100", scope.params)]
|
||
|
||
def configure_job(self, tenant, job_id, values, actor, ip):
|
||
options = dict(self.job(tenant, job_id)['options'])
|
||
options.update(values)
|
||
options = job_options(options)
|
||
if options['engine'] == 'model':
|
||
from knowledge_models import resolve_model
|
||
resolve_model(self.database, options)
|
||
with self.database.connect() as db:
|
||
db.execute('BEGIN IMMEDIATE')
|
||
row = db.execute('SELECT * FROM knowledge_job WHERE id=? AND tenant_id=?', (job_id, tenant)).fetchone()
|
||
if row is None:
|
||
raise KeyError('任务不存在')
|
||
if row['status'] not in {'paused', 'failed'}:
|
||
raise ValueError('请先暂停任务，再调整模型配置')
|
||
if options['engine'] == 'model' and options['max_model_calls'] <= row['model_calls']:
|
||
raise ValueError('模型调用总上限须大于已使用次数')
|
||
db.execute('UPDATE knowledge_job SET options_json=?,updated_at=? WHERE id=?',
|
||
(json_text(options), utc_now(), job_id))
|
||
self.database._audit(db, actor, 'knowledge.job.configure', f'job={job_id}', ip)
|
||
return self.job(tenant, job_id)
|
||
|
||
def job_action(self, tenant, job_id, action, actor, ip):
|
||
if action not in {"pause", "resume", "cancel"}:
|
||
raise ValueError("未知任务操作")
|
||
with self.database.connect() as db:
|
||
db.execute("BEGIN IMMEDIATE")
|
||
row = db.execute("SELECT * FROM knowledge_job WHERE id=? AND tenant_id=?", (job_id, tenant)).fetchone()
|
||
if row is None:
|
||
raise KeyError("任务不存在")
|
||
allowed = ({"queued", "running"} if action == "pause" else {"paused", "failed"}
|
||
if action == "resume" else {"queued", "running", "paused", "failed"})
|
||
if row["status"] not in allowed:
|
||
raise ValueError("当前任务状态不能执行该操作")
|
||
db.execute("UPDATE knowledge_job SET status=?,lease_owner='',lease_until=0,error='',updated_at=?,"
|
||
"options_json=json_remove(options_json,'$.retry_after','$.transient_failures') WHERE id=?",
|
||
({"pause": "paused", "resume": "queued", "cancel": "cancelled"}[action], utc_now(), job_id))
|
||
if action == "cancel":
|
||
db.execute("UPDATE knowledge_job SET state_json='{}' WHERE id=?", (job_id,))
|
||
self.database._audit(db, actor, "knowledge.job." + action, f"job={job_id}", ip)
|
||
return self.job(tenant, job_id)
|
||
|
||
# ── 整理配置（全局） ──────────────────────────────────────────────────────
|
||
def config(self):
|
||
return load_config(self.database)
|
||
|
||
def save_config(self, values, actor, ip):
|
||
config = load_config(self.database)
|
||
for key in ("default_engine", "incremental_engine"):
|
||
value = values.get(key, config[key])
|
||
if value not in {"rules", "model"}:
|
||
raise ValueError("未知整理方式")
|
||
config[key] = value
|
||
concurrency = values.get("model_concurrency", config["model_concurrency"])
|
||
if not isinstance(concurrency, int) or isinstance(concurrency, bool) or not 1 <= concurrency <= 16:
|
||
raise ValueError("模型并发数应为 1 至 16")
|
||
config["model_concurrency"] = concurrency
|
||
batch = values.get("model_batch", config["model_batch"])
|
||
if not isinstance(batch, int) or isinstance(batch, bool) or not 1 <= batch <= 20:
|
||
raise ValueError("每次合并的问答段数应为 1 至 20")
|
||
config["model_batch"] = batch
|
||
config["model_filter"] = bool(values.get("model_filter", config["model_filter"]))
|
||
provider = str(values.get("default_model_provider_id", config["default_model_provider_id"]) or "")
|
||
if provider:
|
||
from knowledge_models import resolve_model
|
||
resolve_model(self.database, {"model_provider_id": provider})
|
||
config["default_model_provider_id"] = provider
|
||
if "model" in (config["default_engine"], config["incremental_engine"]) and not provider:
|
||
from knowledge_models import default_model_id
|
||
if not default_model_id(self.database):
|
||
raise ValueError("还没有可用的文本模型，请先在「模型清单」启用一个，或改用规则整理")
|
||
now = utc_now()
|
||
with self.database.connect() as db:
|
||
db.executemany("INSERT INTO knowledge_config VALUES (?,?,?,?) ON CONFLICT(key) DO UPDATE SET "
|
||
"value_json=excluded.value_json,updated_by=excluded.updated_by,updated_at=excluded.updated_at",
|
||
[(key, json_text(value), actor, now) for key, value in config.items()])
|
||
self.database._audit(db, actor, "knowledge.config", json_text(config)[:400], ip)
|
||
return load_config(self.database)
|
||
|
||
# ── 整理批次：一次点击，按账号拆成任务 ──────────────────────────────────────
|
||
def scope_tenants(self, scope):
|
||
"""授权范围内、有聊天归档的账号（租户）。"""
|
||
scope = tenant_scope(scope)
|
||
with self.database.connect() as db:
|
||
return [row[0] for row in db.execute(
|
||
f"SELECT DISTINCT tenant_id FROM archive_source_account WHERE {scope.clause()} ORDER BY tenant_id",
|
||
scope.params)]
|
||
|
||
@staticmethod
|
||
def _tenant_labels(db, tenants):
|
||
labels = {}
|
||
for tenant in tenants:
|
||
row = db.execute("SELECT id,nickname,zyt_sn FROM desktop_accounts WHERE tenant_id=?", (tenant,)).fetchone()
|
||
if row is not None:
|
||
labels[tenant] = {"account_id": row["id"], "label": row["nickname"] or row["zyt_sn"] or tenant}
|
||
else:
|
||
labels[tenant] = {"account_id": 0, "label": "旧版未归属数据" if tenant == "default" else tenant}
|
||
return labels
|
||
|
||
def preview_run(self, tenants, values):
|
||
"""每个账号范围内有多少单聊消息、文字消息。只数数，不读正文，不调模型。"""
|
||
options = job_options({**values, "source_account_id": "", "max_messages": UNLIMITED})
|
||
cutoff = utc_now()
|
||
accounts = []
|
||
with self.database.connect() as db:
|
||
labels = self._tenant_labels(db, tenants)
|
||
for tenant in tenants:
|
||
where, args = job_scope(tenant, options, cutoff)
|
||
row = db.execute(
|
||
"SELECT COUNT(*) messages, COUNT(DISTINCT m.conversation_id) conversations, "
|
||
"COALESCE(SUM(m.status='normal' AND lower(m.message_type) IN ('text','文本','文字','1','引用','quote') "
|
||
"AND m.direction IN ('inbound','outbound')),0) text_messages "
|
||
"FROM archive_message m JOIN archive_conversation c ON c.id=m.conversation_id WHERE "
|
||
+ " AND ".join(where), args).fetchone()
|
||
accounts.append({"tenant_id": tenant, **labels[tenant], **dict(row)})
|
||
totals = {key: sum(item[key] for item in accounts) for key in ("messages", "conversations", "text_messages")}
|
||
return {"accounts": [item for item in accounts if item["messages"]],
|
||
"empty_accounts": sum(1 for item in accounts if not item["messages"]),
|
||
"totals": totals,
|
||
# 一段问答连同寒暄通常对应 5–8 条文字消息；按 1/6 估算模型调用次数，只作提示
|
||
"estimated_model_calls": totals["text_messages"] // 6}
|
||
|
||
def _busy_tenants(self, db, tenants):
|
||
if not tenants:
|
||
return []
|
||
marks = ",".join("?" for _ in tenants)
|
||
return [row[0] for row in db.execute(
|
||
f"SELECT DISTINCT tenant_id FROM knowledge_job WHERE tenant_id IN ({marks}) "
|
||
f"AND status IN ('queued','running','paused','failed') AND run_id!=? "
|
||
"AND json_extract(options_json,'$.incremental') IS NOT 1", [*tenants, AUTO_RUN])]
|
||
|
||
def create_run(self, tenants, values, actor, ip, *, scope_label=""):
|
||
options = job_options({**values, "source_account_id": "", "max_messages": values.get("max_messages") or UNLIMITED})
|
||
if not options.get("staff_confirmed"):
|
||
raise ValueError("请确认这些聊天由客服账号产生，且允许用于知识整理")
|
||
if options["engine"] == "model":
|
||
from knowledge_models import resolve_model
|
||
resolve_model(self.database, options)
|
||
if not tenants:
|
||
raise ValueError("所选范围内没有聊天归档")
|
||
run_id, now = new_id(), utc_now()
|
||
options["run_id"] = run_id
|
||
# 先在写锁外数好每个账号的消息量：几十万条的计数不该挡住桌面端同步聊天。
|
||
estimates = {}
|
||
with self.database.connect() as db:
|
||
labels = self._tenant_labels(db, tenants)
|
||
for tenant in tenants:
|
||
estimates[tenant] = self._scope_counts(db, tenant, options, now, quality=False)
|
||
tenants = [tenant for tenant in tenants if estimates[tenant]["available_messages"]]
|
||
if not tenants:
|
||
raise ValueError("所选范围内没有可整理的单聊消息，请调整账号或日期")
|
||
engine = "模型清洗" + (f" · {options.get('model_name')}" if options.get("model_name") else "") \
|
||
if options["engine"] == "model" else "规则提取"
|
||
label = f"{scope_label or f'{len(tenants)} 个账号'} · {engine}"
|
||
with self.database.connect() as db:
|
||
db.execute("BEGIN IMMEDIATE")
|
||
busy = self._busy_tenants(db, tenants)
|
||
if busy:
|
||
names = "、".join(labels[t]["label"] for t in busy[:5])
|
||
raise ValueError(f"{names}{' 等' if len(busy) > 5 else ''}还有未完成的整理任务（进行中、已暂停或失败），"
|
||
"请先在「整理任务」里继续或取消，再开始新的整理")
|
||
for tenant in tenants:
|
||
estimate = estimates[tenant]
|
||
job_options_ = {**options, "total_messages": estimate["total_messages"],
|
||
"available_messages": estimate["available_messages"],
|
||
"source_name": estimate["source_name"], "account_label": labels[tenant]["label"]}
|
||
db.execute("INSERT INTO knowledge_job(id,tenant_id,options_json,cutoff_at,created_by,created_at,updated_at,run_id) "
|
||
"VALUES (?,?,?,?,?,?,?,?)", (new_id(), tenant, json_text(job_options_), now, actor, now, now, run_id))
|
||
if options.get("auto_watch"):
|
||
# 整理完以后，这些账号所有客服来源的新聊天也自动整理（静默 30 分钟后，按整理设置里的方式）
|
||
marks = ",".join("?" for _ in tenants)
|
||
db.execute(f"INSERT INTO knowledge_watch SELECT tenant_id,id,1,? FROM archive_source_account "
|
||
f"WHERE tenant_id IN ({marks}) ON CONFLICT(tenant_id,source_account_id) "
|
||
"DO UPDATE SET enabled=1,created_by=excluded.created_by", [actor, *tenants])
|
||
db.execute("INSERT INTO knowledge_run VALUES (?,?,?,?,?,?,?,?)",
|
||
(run_id, label, "active", json_text(options), json_text(tenants), actor, now, now))
|
||
self.database._audit(db, actor, "knowledge.run.create",
|
||
f"run={run_id} accounts={len(tenants)} engine={options['engine']}", ip)
|
||
return self.run(tenant_scope(tenants), run_id)
|
||
|
||
RUN_KEYS = ("engine", "model_provider_id", "model_name", "model_timeout_seconds", "model_concurrency",
|
||
"model_batch", "model_filter", "max_model_calls", "date_from", "date_to", "date_start", "date_end")
|
||
|
||
def _run_summary(self, run, children, labels, *, legacy=False):
|
||
statuses = {child["status"] for child in children}
|
||
if not legacy and run["status"] == "cancelled":
|
||
status = "cancelled"
|
||
elif statuses & {"running", "queued"}:
|
||
status = "running"
|
||
elif "failed" in statuses:
|
||
status = "failed"
|
||
elif "paused" in statuses:
|
||
status = "paused"
|
||
elif statuses <= {"completed"}:
|
||
status = "completed"
|
||
else:
|
||
status = "cancelled"
|
||
keys = ("processed", "created_items", "updated_items", "duplicates", "skipped", "model_filtered",
|
||
"model_calls", "input_chars", "output_chars", "pending_items")
|
||
totals = {key: sum(int(child.get(key) or 0) for child in children) for key in keys}
|
||
known = [child["total_messages"] for child in children if child.get("total_messages") is not None]
|
||
totals["total_messages"] = sum(known) if len(known) == len(children) else None
|
||
skip_reasons = {}
|
||
for child in children:
|
||
for reason, count in (child.get("skip_reasons") or {}).items():
|
||
skip_reasons[reason] = skip_reasons.get(reason, 0) + count
|
||
done = sum(child["total_messages"] or 0 for child in children if child["status"] == "completed")
|
||
moving = sum(min(child["processed"], child["total_messages"] or 0) for child in children
|
||
if child["status"] != "completed")
|
||
whole = totals["total_messages"]
|
||
percent = 100 if status == "completed" else (min(99, int((done + moving) * 100 / whole)) if whole else None)
|
||
options = json.loads(run["options_json"]) if not legacy else children[0]["options"]
|
||
return {
|
||
"id": run["id"], "label": run["label"], "status": status, "legacy": legacy,
|
||
"created_at": run["created_at"], "updated_at": max(child["updated_at"] for child in children),
|
||
"options": {key: options.get(key) for key in self.RUN_KEYS if key in options},
|
||
"progress_percent": percent, "skip_reasons": skip_reasons, **totals,
|
||
"errors": list(dict.fromkeys(child["error"] for child in children if child.get("error")))[:5],
|
||
"accounts": [{
|
||
"job_id": child["id"], "tenant_id": child["tenant_id"],
|
||
**labels.get(child["tenant_id"], {"account_id": 0, "label": child["tenant_id"]}),
|
||
**{key: child.get(key) for key in ("status", "processed", "total_messages", "progress_percent",
|
||
"created_items", "updated_items", "duplicates", "skipped",
|
||
"model_filtered", "model_calls", "pending_items", "error",
|
||
"completion_reason")},
|
||
} for child in children],
|
||
}
|
||
|
||
def run(self, scope, run_id):
|
||
for item in self.runs(scope, limit=None, only=run_id)["items"]:
|
||
return item
|
||
raise KeyError("整理批次不存在或不在你的授权范围内")
|
||
|
||
def runs(self, scope, *, limit=30, only=None):
|
||
scope = tenant_scope(scope)
|
||
visible = (lambda tenant: True) if scope.everything else (lambda tenant: tenant in scope.tenants)
|
||
items = []
|
||
with self.database.connect() as db:
|
||
if only is None or not str(only).startswith("job:"):
|
||
runs = db.execute("SELECT * FROM knowledge_run WHERE (? IS NULL OR id=?) ORDER BY created_at DESC LIMIT 100",
|
||
(only, only)).fetchall()
|
||
for run in runs:
|
||
children = [self.public(row) for row in db.execute(
|
||
"SELECT * FROM knowledge_job WHERE run_id=? ORDER BY created_at,id", (run["id"],))
|
||
if visible(row["tenant_id"])]
|
||
if children:
|
||
labels = self._tenant_labels(db, [child["tenant_id"] for child in children])
|
||
items.append(self._run_summary(run, children, labels))
|
||
# 上一版按单个账号、单个来源建的任务：照常显示、照常能继续，一个任务算一批
|
||
legacy_filter = "" if only is None else " AND id=?"
|
||
legacy_args = [] if only is None else [str(only)[4:]]
|
||
for row in db.execute(f"SELECT * FROM knowledge_job WHERE run_id='' AND {scope.clause()} "
|
||
"AND json_extract(options_json,'$.incremental') IS NOT 1" + legacy_filter +
|
||
" ORDER BY created_at DESC LIMIT 50", [*scope.params, *legacy_args]):
|
||
child = self.public(row)
|
||
labels = self._tenant_labels(db, [child["tenant_id"]])
|
||
label = labels[child["tenant_id"]]["label"]
|
||
engine = "模型整理" if child["options"].get("engine") == "model" else "规则整理"
|
||
pseudo = {"id": "job:" + child["id"], "label": f"{label} · {child['options'].get('source_name') or '历史任务'} · {engine}",
|
||
"status": "active", "options_json": "{}", "created_at": child["created_at"]}
|
||
items.append(self._run_summary(pseudo, [child], labels, legacy=True))
|
||
items.sort(key=lambda item: item["created_at"], reverse=True)
|
||
return {"items": items if limit is None else items[:limit]}
|
||
|
||
def auto_summary(self, scope):
|
||
"""持续整理（新消息自动整理）的概况：不再一条条列出来。"""
|
||
scope = tenant_scope(scope)
|
||
since = (datetime.now(timezone.utc) - timedelta(days=1)).isoformat(timespec="milliseconds")
|
||
auto = f"(run_id='auto' OR json_extract(options_json,'$.incremental') IS 1) AND {scope.clause()}"
|
||
with self.database.connect() as db:
|
||
row = db.execute(f"SELECT COUNT(*) jobs, COALESCE(SUM(created_items),0) created, "
|
||
f"COALESCE(SUM(status='failed'),0) failed FROM knowledge_job WHERE {auto} AND created_at>=?",
|
||
[*scope.params, since]).fetchone()
|
||
queued = db.execute(f"SELECT COUNT(*) FROM knowledge_dirty WHERE {scope.clause()}", scope.params).fetchone()[0]
|
||
watches = db.execute(f"SELECT COUNT(*) FROM knowledge_watch WHERE enabled=1 AND {scope.clause()}",
|
||
scope.params).fetchone()[0]
|
||
return {"jobs_24h": row["jobs"], "created_24h": row["created"], "failed_24h": row["failed"],
|
||
"waiting_conversations": queued, "watching_sources": watches}
|
||
|
||
def _run_tenants(self, db, scope, run_id):
|
||
scope = tenant_scope(scope)
|
||
run = db.execute("SELECT * FROM knowledge_run WHERE id=?", (run_id,)).fetchone()
|
||
if run is None:
|
||
raise KeyError("整理批次不存在")
|
||
tenants = json.loads(run["tenants_json"])
|
||
if not scope.everything and not set(tenants) <= set(scope.tenants):
|
||
raise ValueError("这个整理批次包含你无权操作的账号，请让有全部账号权限的管理员操作")
|
||
return run, tenants
|
||
|
||
def run_action(self, scope, run_id, action, actor, ip):
|
||
if action not in {"pause", "resume", "cancel"}:
|
||
raise ValueError("未知任务操作")
|
||
if str(run_id).startswith("job:"):
|
||
job = self._legacy_job(scope, run_id)
|
||
self.job_action(job["tenant_id"], job["id"], action, actor, ip)
|
||
return self.run(scope, run_id)
|
||
with self.database.connect() as db:
|
||
db.execute("BEGIN IMMEDIATE")
|
||
run, _tenants = self._run_tenants(db, scope, run_id)
|
||
if run["status"] == "cancelled":
|
||
raise ValueError("批次已取消")
|
||
now = utc_now()
|
||
if action == "pause":
|
||
db.execute("UPDATE knowledge_job SET status='paused',lease_owner='',lease_until=0,updated_at=? "
|
||
"WHERE run_id=? AND status IN ('queued','running')", (now, run_id))
|
||
status = "paused"
|
||
elif action == "resume":
|
||
# 人点了继续就立刻接着跑：清掉自动重试的退避计时
|
||
db.execute("UPDATE knowledge_job SET status='queued',lease_owner='',lease_until=0,error='',updated_at=?,"
|
||
"options_json=json_remove(options_json,'$.retry_after','$.transient_failures') "
|
||
"WHERE run_id=? AND status IN ('paused','failed')", (now, run_id))
|
||
status = "active"
|
||
else:
|
||
db.execute("UPDATE knowledge_job SET status='cancelled',state_json='{}',lease_owner='',lease_until=0,"
|
||
"updated_at=? WHERE run_id=? AND status IN ('queued','running','paused','failed')", (now, run_id))
|
||
status = "cancelled"
|
||
db.execute("UPDATE knowledge_run SET status=?,updated_at=? WHERE id=?", (status, now, run_id))
|
||
self.database._audit(db, actor, "knowledge.run." + action, f"run={run_id}", ip)
|
||
return self.run(scope, run_id)
|
||
|
||
def configure_run(self, scope, run_id, values, actor, ip):
|
||
"""换模型、调并发或调用上限，只作用于还没处理完的部分；已生成的草稿保留。"""
|
||
if str(run_id).startswith("job:"):
|
||
job = self._legacy_job(scope, run_id)
|
||
self.configure_job(job["tenant_id"], job["id"], values, actor, ip)
|
||
return self.run(scope, run_id)
|
||
with self.database.connect() as db:
|
||
run, _tenants = self._run_tenants(db, scope, run_id)
|
||
options = json.loads(run["options_json"])
|
||
options.update({key: values[key] for key in ("engine", "model_provider_id", "model_timeout_seconds",
|
||
"model_concurrency", "model_batch", "model_filter", "max_model_calls")
|
||
if key in values})
|
||
options = job_options(options)
|
||
if options["engine"] == "model":
|
||
from knowledge_models import resolve_model
|
||
resolve_model(self.database, options)
|
||
changed = {key: options.get(key) for key in ("engine", "model_provider_id", "model_name", "model_timeout_seconds",
|
||
"model_concurrency", "model_batch", "model_filter", "max_model_calls")}
|
||
with self.database.connect() as db:
|
||
db.execute("BEGIN IMMEDIATE")
|
||
run, _tenants = self._run_tenants(db, scope, run_id)
|
||
if run["status"] == "cancelled":
|
||
raise ValueError("批次已取消")
|
||
rows = db.execute("SELECT * FROM knowledge_job WHERE run_id=?", (run_id,)).fetchall()
|
||
if any(row["status"] in {"queued", "running"} for row in rows):
|
||
raise ValueError("请先暂停批次，再调整模型配置")
|
||
used = sum(row["model_calls"] for row in rows)
|
||
if options["engine"] == "model" and options["max_model_calls"] and options["max_model_calls"] <= used:
|
||
raise ValueError(f"模型调用总上限须大于已使用的 {used} 次（填 0 表示不限）")
|
||
for row in rows:
|
||
if row["status"] in {"completed", "cancelled"}:
|
||
continue
|
||
current = json.loads(row["options_json"])
|
||
current.update(changed)
|
||
db.execute("UPDATE knowledge_job SET options_json=?,updated_at=? WHERE id=?",
|
||
(json_text(current), utc_now(), row["id"]))
|
||
db.execute("UPDATE knowledge_run SET options_json=?,updated_at=? WHERE id=?",
|
||
(json_text(options), utc_now(), run_id))
|
||
self.database._audit(db, actor, "knowledge.run.configure", f"run={run_id}", ip)
|
||
return self.run(scope, run_id)
|
||
|
||
def _legacy_job(self, scope, run_id):
|
||
scope = tenant_scope(scope)
|
||
with self.database.connect() as db:
|
||
row = db.execute(f"SELECT id,tenant_id FROM knowledge_job WHERE id=? AND run_id='' AND {scope.clause()}",
|
||
[str(run_id)[4:], *scope.params]).fetchone()
|
||
if row is None:
|
||
raise KeyError("任务不存在")
|
||
return row
|
||
|
||
def add_draft(self, db, tenant, job_id, candidate, *, replace=False):
|
||
"""存一条草稿。返回 "created"、"updated"（覆盖了同一段对话还没人动过的机器草稿）或 ""（重复）。
|
||
|
||
replace 只给模型清洗的结果：模型结果可以替换规则草稿（或换个模型重洗）；规则结果从不覆盖已有草稿，
|
||
否则先模型清洗、后来新消息触发规则整理时，洗好的草稿会被打回原话。"""
|
||
# Keep original source/version identity in the key: changed sources get a fresh draft.
|
||
digest = fingerprint(candidate["question"] + "\n" + candidate["answer"])
|
||
origin = source_key(candidate["sources"])
|
||
if db.execute('SELECT 1 FROM knowledge_discarded WHERE tenant_id=? AND dedup_hash IN (?,?)',
|
||
(tenant, digest, "src:" + origin)).fetchone():
|
||
return ""
|
||
if origin:
|
||
same = db.execute("SELECT * FROM knowledge_item WHERE tenant_id=? AND source_key=? AND status!='stale' "
|
||
"ORDER BY created_at LIMIT 1", (tenant, origin)).fetchone()
|
||
if same is not None:
|
||
return self._rewash(db, same, digest, candidate) if replace else ""
|
||
existing = db.execute("SELECT id,status FROM knowledge_item WHERE tenant_id=? AND dedup_hash=?",
|
||
(tenant, digest)).fetchone()
|
||
if existing and existing["status"] != "stale":
|
||
return ""
|
||
if existing:
|
||
digest = fingerprint(digest + json_text(candidate["sources"]))
|
||
item_id, now = new_id(), utc_now()
|
||
inserted = db.execute("""INSERT OR IGNORE INTO knowledge_item
|
||
(id,tenant_id,job_id,dedup_hash,title,question,answer,conditions,category,kind,flags_json,created_at,updated_at,source_key)
|
||
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", (item_id, tenant, job_id, digest, candidate["title"],
|
||
candidate["question"], candidate["answer"], candidate["conditions"], candidate["category"],
|
||
candidate["kind"], json_text(candidate["flags"]), now, now, origin))
|
||
if not inserted.rowcount:
|
||
return ""
|
||
db.executemany("INSERT INTO knowledge_source VALUES (?,?,?,?,?,?)", [
|
||
(item_id, s["message_id"], s["version_no"], s["role"], s["content"], s["sent_at"])
|
||
for s in candidate["sources"]])
|
||
if not self._fresh(db, item_id, tenant):
|
||
db.execute("UPDATE knowledge_item SET status='stale' WHERE id=?", (item_id,))
|
||
self._record(db, item_id, 0, "extract")
|
||
return "created"
|
||
|
||
def _rewash(self, db, row, digest, candidate):
|
||
"""同一段对话已有知识：只有还没人动过的机器草稿才用新结果覆盖（例如规则草稿 → 模型清洗结果）。
|
||
审核过、发布过、被驳回或人工改过的，一律不碰。"""
|
||
same_text = row["question"] == candidate["question"] and row["answer"] == candidate["answer"]
|
||
touched = db.execute("SELECT 1 FROM knowledge_revision WHERE item_id=? AND actor_id!=0 LIMIT 1",
|
||
(row["id"],)).fetchone()
|
||
if same_text or touched or row["status"] != "draft":
|
||
return ""
|
||
clash = db.execute("SELECT 1 FROM knowledge_item WHERE tenant_id=? AND dedup_hash=? AND id!=?",
|
||
(row["tenant_id"], digest, row["id"])).fetchone()
|
||
if clash:
|
||
return ""
|
||
db.execute("""UPDATE knowledge_item SET dedup_hash=?,title=?,question=?,answer=?,conditions=?,category=?,kind=?,
|
||
flags_json=?,revision=revision+1,updated_at=? WHERE id=?""", (
|
||
digest, candidate["title"], candidate["question"], candidate["answer"], candidate["conditions"],
|
||
candidate["category"], candidate["kind"], json_text(candidate["flags"]), utc_now(), row["id"]))
|
||
self._record(db, row["id"], 0, "rewash")
|
||
return "updated"
|