81 lines
2.8 KiB
Python
81 lines
2.8 KiB
Python
"""世谱匹配引擎：关键词 / 正则 / selective / constant / 递归扫描。
|
|
|
|
纯逻辑，无 IO。扫描文本由调用方（集成层）从对话历史截取后传入。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
|
|
from app.modules.lorebook.types import LorebookBook, LorebookEntry, MatchedEntry
|
|
|
|
_RECURSION_MAX_ITERATIONS = 5
|
|
|
|
|
|
def match_entries(book: LorebookBook, scan_text: str) -> list[MatchedEntry]:
|
|
"""返回命中的条目（含 constant 与关键词命中）；递归扫描按 book 配置。"""
|
|
matched: list[MatchedEntry] = []
|
|
for entry in book.entries:
|
|
if not entry.enabled:
|
|
continue
|
|
if entry.constant:
|
|
matched.append(MatchedEntry(entry=entry, reason="constant"))
|
|
continue
|
|
if _entry_matches(entry, scan_text):
|
|
matched.append(MatchedEntry(entry=entry, reason="key"))
|
|
|
|
if book.recursive_scanning:
|
|
matched = _expand_recursive(matched, book, scan_text)
|
|
return matched
|
|
|
|
|
|
def _entry_matches(entry: LorebookEntry, scan_text: str) -> bool:
|
|
if entry.selective and entry.secondary_keys:
|
|
return _any_key_hit(entry.keys, entry, scan_text) and _any_key_hit(
|
|
entry.secondary_keys, entry, scan_text
|
|
)
|
|
return _any_key_hit(entry.keys, entry, scan_text)
|
|
|
|
|
|
def _any_key_hit(keys: list[str], entry: LorebookEntry, scan_text: str) -> bool:
|
|
for key in keys:
|
|
if not key:
|
|
continue
|
|
if entry.use_regex:
|
|
flags = 0 if entry.case_sensitive else re.IGNORECASE
|
|
try:
|
|
if re.search(key, scan_text, flags):
|
|
return True
|
|
except re.error:
|
|
# 非法正则降级为字面量包含
|
|
pass
|
|
else:
|
|
hay = scan_text if entry.case_sensitive else scan_text.casefold()
|
|
needle = key if entry.case_sensitive else key.casefold()
|
|
if needle in hay:
|
|
return True
|
|
return False
|
|
|
|
|
|
def _expand_recursive(
|
|
matched: list[MatchedEntry], book: LorebookBook, original_text: str
|
|
) -> list[MatchedEntry]:
|
|
"""已命中条目的 content 追加进扫描文本，再扫未命中条目；限迭代防振荡。"""
|
|
hit_ids = {m.entry.id for m in matched}
|
|
pool = list(matched)
|
|
extra_text = original_text + "\n" + "\n".join(m.entry.content for m in matched)
|
|
|
|
for _ in range(_RECURSION_MAX_ITERATIONS):
|
|
grew = False
|
|
for entry in book.entries:
|
|
if entry.id in hit_ids or not entry.enabled or entry.constant:
|
|
continue
|
|
if _entry_matches(entry, extra_text):
|
|
pool.append(MatchedEntry(entry=entry, reason="recursive"))
|
|
hit_ids.add(entry.id)
|
|
extra_text += "\n" + entry.content
|
|
grew = True
|
|
if not grew:
|
|
break
|
|
return pool
|