yilunjiang/AgentLab
1
0
Fork 0
mirror of https://github.com/ThearchyHelios/AgentLab.git synced 2026-11-15 00:11:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

分析说法核对的说法表与绑定（engine/analysis_claims）：判断词分来源、分散、抵消、结构、异常、新高新低、正常、趋势、无趋势、差异、无差异、因果几类，一处定义、带版本（LEXICON_VERSION 1）；只收强判断词（「持续」要跟着涨跌、「异常」后面是订单数据这类名词或前面是数据系统的不算、「相当」只认句读前的，「大幅」按细案收进异常类、要跟着涨跌动词）；否定（前 4 个字以内、跳过「出现」「明显」这类填充词，「不但」「不少」「无论」不算）翻到反面类；来源与因果分界（因果词所在分句点名的是拆解的段或因子按来源核），推测说法、同句的知识库原话和运行输入豁免因果，说的是数据本身的原因不算；不扫判定语（a:）、知识库原话（q:）、别的引用渲染出来的字、标题、代码块、表头。绑定：一句的分析依据按这次重新解析的引用算，A 条目经 contract.catalog_result 读，核判定、点名的段（全半角归一）、拖累拉动的方向、新高新低、连续的期数，挂了算不了的结论算对不上；违规三种（analysis_claim_uncited / analysis_claim_mismatch / causal_unsupported），每条带判断词、句子、类、要的 op、这句挂的 A、给人看的和交回写作者的各一句，同一分句同一类只报一处。挂点：compose_doc 收 analysis_claims（check / strict）时文档顶层记下处置和说法表版本，verify_doc 只核记了这一项的文档（按记下的版本，认不出的记 bad_schema），老文档复核结果和统计一字不变；_DIRECTIONAL 补主要、集中、异常、偏离、持续、趋势、显著、明显高于 / 低于等预筛词（列提纲、建议、引出图表的写法除外），裁判优先判这类句子

Browse Source
This commit is contained in:
ThearchyHelios
2026-10-21 13:11:56 +08:00
parent 38906a92ea
commit 7acee0faed
3 changed files with 1377 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/engine/analysis_claims.py
+878
View File
@@ -0,0 +1,878 @@
"""分析说法核对（docs/design/analysis-round/00-总方案.md §3.5，细案 01 §9）：报告里的判断词要挂对应的分析结论，判定对得上。
「叙述层无算术」往前走一步：「主要来自华东」「超出常规波动」「持续上升」这类判定和数字一样由系统算、由系统渲染（[[a:A3]]），
写作者只引用。写作者自己写了判断词的，这里确定地核对（不调模型，D6）：
1. **说法表**（LEXICONS，一处定义，带版本 LEXICON_VERSION）：判断词分成来源、分散、抵消、结构、异常、新高新低、正常、
趋势、无趋势、差异、无差异、因果几类，每类要挂一种分析（op），判定要对得上（细案 01 §9.1）。只收强判断词，宁可漏报
不误报：「持续」要跟着涨跌（「持续关注」不算），「异常」后面接着订单、数据这类名词的是业务分类、前面是数据、系统的是
数据质量，都不算；「相当」只认句读前的（「相当于」「相当高」不算）
2. **否定**：判断词前 4 个字以内（跳过「出现」「明显」这类填充词）有「不、没有、没、并非、并不、未、无」的翻到反面类：
「并未出现明显异常」是正常，「没有明显的上升趋势」是无趋势，「并非主要来自华东」要拆解、华东不在主要来源里；
「不但」「不少」「无论」「未来」这类不是否定
3. **怎么挂**：一句话的分析依据 = 句中的 [[a:]]、[[v:A…]]、[[table:A…]] 和句末 [[see:A…]]（组装、复核时按重新解析的
引用算，见 evidence.verify_doc）。类要的 op 必须在依据里出现，几个里任一个对得上就算对；挂了算不了的结论算对不上
4. **点名的段**：判断词所在分句里出现的结论结果表的维度值（全半角归一，两个字以上、不是纯数字）。来源类要点名的段都在
判定的 subjects 里；拖累要那一段变化为负，拉动、带动为正（不在 subjects 里、但那一段的变化占总变化过半的也算拖累）
5. **来源和因果的分界**：句子挂了拆解、因果词所在分句点名的是这份结论的段或因子（「下降主要由华东造成」）→ 按来源核，
这是记账意义上的「来自」；点名的不是（「由于促销结束导致下降」）→ 按因果核：同一句要有知识库原话（[[q:]]）或运行
输入（[[i:]]）作依据，或者写成推测（可能、或许、推测、待核实、不排除、疑似）。分句里已有来源一类判断词的，因果词
并进那一处核。原因说的是数据本身（「由于数据只到 10 月 16 日」「样本少」）的不算
6. **不扫的地方**：知识库原话（[[q:]] 片段）、系统渲染的判定语（[[a:]] 片段）、别的引用渲染出来的字、标题、代码块、
表头；分析范围是文档的结构化字段（analysis_scope），本来就不在正文里
违规三种（CODES）：analysis_claim_uncited（有判断词，依据里没有对应种类的分析结论）、analysis_claim_mismatch（挂了，判定、
点名的段、方向对不上，含挂了算不了的结论）、causal_unsupported（因果词没有依据、也不是推测）。每条带：代码、判断词
（text）、所在句子（context）、类（claim）、要的 op、这句挂的分析（analyses）、给人看的一句（message）和交回写作者的
一句（for_model）。
报告文档记下按哪种处置、哪一版说法表核（DOC_FIELD：{"mode": check / strict, "lexicon": 1}，参与内容哈希）：只有带这一项
的文档在复核时核这一项（verify），老文档复核结果不变；复核按文档记下的那一版说法表，说法表升了版老文档照旧。
整个模块是纯函数，不碰数据库、不调模型。
"""
from __future__ import annotations
import re
import unicodedata
from collections.abc import Iterable, Mapping
from dataclasses import dataclass, field
from decimal import Decimal, InvalidOperation
from typing import Any
from app.analysis.contract import catalog_result
from app.engine.evidence import _Reason, _violation, iter_units
#: 说法表的版本：改了判断词、类、绑定规则就升版本，老版本留在 LEXICONS 里给老文档复核用
LEXICON_VERSION = 1
#: 报告节点 analysis_claims 的取值里要核的两种（off 不核、文档里也不记）
MODES = ("check", "strict")
#: 报告文档里记下核对处置和说法表版本的字段
DOC_FIELD = "analysis_claims"
UNCITED = "analysis_claim_uncited"
MISMATCH = "analysis_claim_mismatch"
CAUSAL = "causal_unsupported"
CODES = frozenset({UNCITED, MISMATCH, CAUSAL})
#: 各算子给人看的叫法（违规说法里用）
_OP_TEXT = {"decompose": "变化拆解", "baseline": "波动基线", "trend": "趋势", "differ": "差异检验",
"comparable": "可比性检查"}
# --------------------------------------------------------------------------
# 说法表 v1
# --------------------------------------------------------------------------
#: 涨跌的说法。「扩大」「缩小」「收窄」不收：亏损扩大、差距缩小说不清是哪个方向
_UP = "上升|增长|增加|提升|提高|上涨|攀升|走高|上扬|走强|回升|反弹|增多|上行|抬升|拉升"
_DOWN = "下降|下滑|减少|降低|下跌|走低|回落|走弱|萎缩|减弱|下行|下挫|走软|缩水"
_DIR = f"(?:{_UP}|{_DOWN})"
_UP_RE = re.compile(_UP)
_DOWN_RE = re.compile(_DOWN)
#: 分句的边界（顿号、冒号不算：「集中在华东、华北」「主要来自：华东」还是一句里的事）
_STOPS = "，,；;。！？!?\n"
_NOT_STOP = f"[^{_STOPS}]"
@dataclass(frozen=True)
class Claim:
"""说法表的一类：判断词（words）、要挂的分析（op，结构变化另要 mode ratio）、判断词前有否定时翻到哪一类（flip，None：
否定了就不核，宁可漏报）。negative：这一类的说法本身就是否定的写法（「并不异常」「没有明显趋势」），不再看前面的否定。
同一类可以有几条（key 相同）：翻到这一类时用第一条。"""
key: str
label: str
op: str | None
words: re.Pattern[str]
flip: str | None = None
mode: str | None = None
negative: bool = False
#: 「连续…上升」：另要末尾连续同向至少 3 期
run: bool = False
@dataclass(frozen=True)
class Lexicon:
version: int
claims: tuple[Claim, ...]
#: 判断词本身带的方向（不看上下文）：骤降是跌、激增是涨、拖累要那一段是跌
intrinsic: tuple[tuple[re.Pattern[str], str], ...]
#: 判断词前面的否定（_negated）
negation: re.Pattern[str]
#: 否定词和判断词之间可以隔着的填充词（「并未出现明显异常」）
filler: re.Pattern[str]
#: 推测的说法：因果句有它就不算没依据
guess: re.Pattern[str]
#: 因果说的是数据本身（数据只到哪天、口径、样本）：不当原因判断核
method: re.Pattern[str]
#: 「异常」「正常」前面是这些的，说的是数据质量（数据异常、系统无异常），不是指标的波动
quality: re.Pattern[str]
#: 拖累、拉动、带动：点名的段不在 subjects 里时，按那一段变化的正负和占比核
pulls: frozenset[str] = field(default_factory=frozenset)
def claim(self, key: str) -> Claim:
return next(c for c in self.claims if c.key == key)
_V1 = Lexicon(
version=1,
claims=(
Claim("source", "来源", "decompose", re.compile(
r"主要(?:来自|源自|出自|由|集中(?:在|于))"
rf"|主要是(?!因为|由于|受|指|看|说|想|希望){_NOT_STOP}{{0,12}}?(?:{_DIR}|贡献|拖累|带动|拉动|驱动)"
r"|(?:绝)?大(?:部分|多数|多)(?:来自|源自|出自|由|集中(?:在|于))"
r"|集中(?:在|于)"
r"|主导(?!地位|产品|品类|品牌|产业)"
r"|拖累|拉动|带动"
r"|贡献(?:最大|最多|最高)|最大(?:的)?贡献|贡献了(?:绝)?大部分"), flip="source_not"),
Claim("source_not", "来源（否定）", "decompose", re.compile(r"(?!)"), negative=True),
Claim("dispersed", "分散", "decompose", re.compile(
r"分散(?!注意|精力|风险|投资|采购)"
rf"|普遍(?:有所|出现|呈现)?{_DIR}"
rf"|全面{_DIR}"
rf"|各(?:个)?{_NOT_STOP}{{1,6}}?(?:都|均)(?:有所|出现|在)?{_DIR}"), flip="source"),
Claim("offset", "抵消", "decompose", re.compile(r"抵消|抵冲|此消彼长|一增一减|一升一降|有增有减")),
Claim("mix", "结构", "decompose", re.compile(
r"结构性|结构(?:变化|变动|效应)|占比(?:的)?变化(?:带来|造成|导致|引起)"), mode="ratio"),
Claim("anomaly", "异常", "baseline", re.compile(
r"异常(?!订单|单|值|数据|记录|交易|码|状态|类|退款|商品|门店|账户|用户|日志|处理|检测)"
r"|反常|罕见"
r"|(?:明显|显著|大幅|严重)偏离|偏离(?:了)?(?:正常|常规|往常|以往|历史|过往)"
rf"|(?:明显|显著)?(?:超出|超过|跳出|突破|高于|低于)(?:了)?{_NOT_STOP}{{0,14}}?(?:正常|常规)(?:的)?(?:波动)?(?:范围|水平|区间)"
r"|超出(?:了)?(?:正常|常规)"
rf"|大幅(?:度)?(?:地)?(?:{_DIR}|波动|变化|变动)"
r"|骤降|骤减|暴跌|暴降|锐减|激增|飙升|暴涨|猛增|骤增|陡增|陡降"
r"|剧烈(?:地)?(?:波动|变化|变动|震荡|起伏)|(?:波动|变化|变动)(?:十分|非常|较为|比较)?剧烈"),
flip="normal"),
Claim("record", "新高新低", "baseline", re.compile(
r"历史(?:新高|新低|最高|最低)|创(?:下)?(?:了)?(?:历史)?(?:新高(?!地)|新低)|前所未有")),
Claim("normal", "正常", "baseline", re.compile(
r"(?:正常|常规)(?:的)?波动(?:范围)?(?:内|之内)?"
r"|(?:正常|常规)(?:的)?(?:范围|区间)(?:内|之内)?"
r"|(?:属于|处于|在)(?:正常|常规)(?:水平|区间|范围)"), flip="anomaly"),
Claim("normal", "正常", "baseline", re.compile(
r"(?:并不|并非|并没有|没有|未见|并无|不算|不属于|不是)(?:明显)?(?:的)?异常|无异常"), negative=True),
Claim("trend", "趋势", "trend", re.compile(
rf"{_DIR}(?:的)?(?:趋势|态势|势头)"
rf"|(?:趋势|态势|势头)(?:性)?(?:地)?{_DIR}"
rf"|(?:呈|呈现|保持|维持|显示)(?:出)?{_DIR}?(?:的)?(?:趋势|态势)"
rf"|持续(?:地|性)?{_DIR}"
rf"|一路(?:{_DIR}|走高|走低|上扬)"
rf"|逐步{_DIR}|稳步{_DIR}"
r"|节节(?:攀升|升高|走高|上涨|上升|下滑|下降|败退|走低)"), flip="no_trend"),
Claim("trend", "趋势", "trend", re.compile(
rf"连续(?:[\x00\s0-9０-９一二三四五六七八九十两几多数个周月天日季度年期]|都|在|有所|出现)*?{_DIR}"),
flip="no_trend", run=True),
Claim("no_trend", "无趋势", "trend", re.compile(
r"(?:没有|并没有|没|无|未见|并无|未呈现|未出现)(?:明显|显著)?(?:的)?"
r"(?:(?:上升|下降|增长|下滑|向上|向下)(?:或|和|、|与|/)?)*(?:的)?(?:趋势|态势)"
rf"|{_DIR}?(?:的)?(?:趋势|态势)(?:并)?(?:不明显|不显著|不清晰|不突出)"
r"|基本平稳|走势平稳"), negative=True),
Claim("differ", "差异", "differ", re.compile(
r"(?:显著|明显)(?:地)?(?:高于|低于|大于|小于|多于|少于|好于|差于|优于|劣于)"
r"|确实(?:更高|更低|更多|更少|高于|低于)"
r"|差(?:异|别)(?:十分|非常|很|较为)?(?:明显|显著)"
r"|(?:明显|显著)(?:的)?(?:差异|差别)"), flip="no_diff"),
Claim("no_diff", "无差异", "differ", re.compile(
r"差(?:异|别)(?:不大|很小|较小|不明显|并不明显|不显著|有限)"
r"|(?:没有|并没有|没|无|并无)(?:明显|显著)?(?:的)?(?:差异|差别|区别)"
r"|不相上下|旗鼓相当"
rf"|(?:基本|大致|大体)?相当(?=[{_STOPS}\s\x00]|$)"), negative=True),
Claim("causal", "因果", None, re.compile(
r"因为|由于|导致|造成|引起(?!关注|重视|注意|警惕|讨论|兴趣)|归因(?:于|为)|原因(?:主要)?(?:是|在于)"
r"|源于|所致|得益于"
rf"|受(?!到?影响)(?:到)?{_NOT_STOP}{{1,12}}?(?:的)?影响")),
),
intrinsic=(
(re.compile(r"拖累|骤降|骤减|暴跌|暴降|锐减|陡降|新低|最低|低于|小于|少于|更低|更少"), "down"),
(re.compile(r"拉动|带动|激增|飙升|暴涨|猛增|骤增|陡增|新高|最高|高于|大于|多于|更高|更多"), "up"),
),
negation=re.compile(r"不(?![但仅只光论少同断久管过如然错禁免妨要])|没有|没|并非|并不|未(?![来必])|无(?![论疑关需])"),
filler=re.compile(r"(?:出现|发现|看到|见到|存在|呈现|表现出|明显|显著|什么|任何|多大|太|很|特别|有|过|的|了|[\x00\s])$"),
guess=re.compile(r"可能|或许|也许|推测|猜测|待核实|有待核实|需(?:要)?(?:进一步)?(?:核实|确认|验证)|待确认|待验证"
r"|不排除|疑似|大概率"),
method=re.compile(r"数据|口径|统计|查询|目录|样本|取数|截断|截止|只到|不完整|未覆盖|篇幅|省略|证据"),
quality=re.compile(r"(?:数据|系统|接口|取数|口径|同步|查询|日志|字段|格式|编码)(?:出现|存在|有|无|没有|并无|未见)?$"),
pulls=frozenset({"拖累", "拉动", "带动"}),
)
LEXICONS: dict[int, Lexicon] = {1: _V1}
#: 有这几类判断的分句里，因果词并进它们核（「下降主要由华东造成」：说的是记账意义上的来源）
_SOURCE_KEYS = frozenset({"source", "source_not", "dispersed", "offset", "mix"})
#: 结果表里不是维度的列（按维度认点名的段时跳过）
_MEASURE_COLUMNS = frozenset({
"base", "target", "delta", "pct", "share", "flag", "d_base", "d_target", "w_base", "w_target", "r_base",
"r_target", "mix", "rate", "total", "contrib",
})
#: 一段的变化记在哪一列：可加拆解的 delta、比率拆解的 total（结构 + 自身）、乘积拆解的 contrib
_DELTA_COLUMNS = ("delta", "total", "contrib")
def lexicon(version: int | None = None) -> Lexicon:
"""某一版说法表；不认识的版本按当前版。"""
return LEXICONS.get(version if isinstance(version, int) else LEXICON_VERSION, LEXICONS[LEXICON_VERSION])
# --------------------------------------------------------------------------
# 扫判断词
# --------------------------------------------------------------------------
@dataclass(frozen=True)
class Hit:
"""句子里的一处判断词：类（否定翻过之后的）、在句子里的起止、原文、方向（up / down / None）。"""
claim: str
start: int
end: int
word: str
direction: str | None
negated: bool = False
run: bool = False
def _clause(text: str, start: int, end: int) -> tuple[int, int]:
"""[start, end) 所在的分句：往前往后各找到最近的句读。"""
left = max((text.rfind(ch, 0, start) for ch in _STOPS), default=-1) + 1
rights = [i for i in (text.find(ch, end) for ch in _STOPS) if i >= 0]
return left, min(rights) if rights else len(text)
def _negated(lex: Lexicon, before: str) -> bool:
"""判断词前 4 个字以内有否定（先去掉紧挨着的填充词：「并未出现明显异常」的否定隔着「出现明显」）。"""
while (m := lex.filler.search(before)) and m.group(0):
before = before[:m.start()]
return lex.negation.search(before[-4:]) is not None
def _direction(lex: Lexicon, word: str, after: str, clause: str) -> str | None:
"""判断词说的方向：判断词本身带的（骤降、拉动），再看判断词里、紧跟着的几个字、整个分句里的涨跌说法。一个分句里
涨跌都有的不判方向。"""
for pattern, direction in lex.intrinsic:
if pattern.search(word):
return direction
for text in (word, after[:4], clause):
up, down = bool(_UP_RE.search(text)), bool(_DOWN_RE.search(text))
if up != down:
return "up" if up else "down"
if up and down:
return None
return None
def scan(text: str, *, version: int | None = None) -> list[Hit]:
"""一句话里的判断词（text 是写作者自己写的字，别的片段已经涂成 \\x00）。同一处有几种读法时取最靠前、最长的：
「超出正常范围」是异常不是正常，「没有明显的上升或下降趋势」是无趋势不是趋势。"""
lex = lexicon(version)
found: list[tuple[int, int, int, Claim]] = []
for order, claim in enumerate(lex.claims):
for m in claim.words.finditer(text):
if m.end() > m.start():
found.append((m.start(), -(m.end() - m.start()), order, claim))
hits: list[Hit] = []
taken = 0
for start, minus, _, claim in sorted(found, key=lambda f: (f[0], f[1], f[2])):
end = start - minus
if start < taken:
continue
left, right = _clause(text, start, end)
before = text[left:start]
if claim.key in ("anomaly", "normal") and lex.quality.search(
re.sub(r"(?:不|没有|没|并不|并非|未|无)$", "", before.rstrip("\x00 "))):
taken = end
continue
key, negated = claim.key, False
if not claim.negative and _negated(lex, before):
negated = True
if claim.flip is None:
taken = end
continue
key = claim.flip
word = text[start:end]
direction = _direction(lex, word, text[end:right], text[left:right])
hits.append(Hit(claim=key, start=start, end=end, word=word, direction=direction, negated=negated,
run=claim.run and not negated))
taken = end
return hits
# --------------------------------------------------------------------------
# 一句话
# --------------------------------------------------------------------------
@dataclass
class Sentence:
"""交给核对的一句话（和标记解析解耦：文档里怎么切、怎么解析引用由 evidence 管，这里只要组装好的单元）。
text、names、shown 等长，位置一一对应：text 只留写作者自己写的字（判断词在这里找），names 另留引用渲染出来的值（点名
的段在这里认），两者都把判定语、知识库原话涂成 \\x00；shown 是给人看的整句（排版符号涂成 \\x00，显示时去掉）。没给
names、shown 的按 text。analyses 是这句的分析依据（A 编号），quoted / inputs：同一句里有知识库原话、运行输入。base 是
text[0] 在正文里的位置（违规带 span 要用），pieces 是各片段在 text 里的起止和编号（违规带所在片段）。"""
text: str
names: str = ""
shown: str = ""
analyses: list[str] = field(default_factory=list)
quoted: bool = False
inputs: bool = False
base: int | None = None
unit: str | None = None
pieces: list[tuple[int, int, str | None]] = field(default_factory=list)
def __post_init__(self) -> None:
if len(self.names) != len(self.text):
self.names = self.text
if len(self.shown) != len(self.text):
self.shown = self.names
def _same_length_norm(text: str) -> str:
"""逐字归一（NFKC 会把个别字展开成几个字，逐字做、展开的取第一个，位置就和原文对得上）。"""
out = []
for ch in text:
n = unicodedata.normalize("NFKC", ch).casefold()
out.append(n[:1] if n else ch)
return "".join(out)
_NUMERIC = re.compile(r"^[\d.,+\-%‰\s]+$")
def _named_value(value: Any) -> str | None:
"""结果表里一个维度值能不能拿来认点名：两个字以上、不是纯数字。"""
if value is None or isinstance(value, bool):
return None
text = _same_length_norm(str(value).strip())
if len(text) < 2 or _NUMERIC.match(text):
return None
return text
def _dims(result: Mapping[str, Any]) -> list[int]:
"""结果表里维度列的位置：拆解按 params.by，乘积拆解是 factor 列，别的按「不是度量列」认。"""
table = result.get("table") or {}
columns = [str(c) for c in table.get("columns") or []]
params = result.get("params") or {}
if result.get("mode") == "factor" and "factor" in columns:
return [columns.index("factor")]
by = params.get("by")
wanted = by if isinstance(by, list) else [by] if isinstance(by, str) else []
picked = [columns.index(str(b)) for b in wanted if str(b) in columns]
return picked or [i for i, c in enumerate(columns) if c not in _MEASURE_COLUMNS]
def _segment_names(result: Mapping[str, Any]) -> dict[str, str]:
"""{归一后的维度值或因子名: 原样}：结果表里各段的维度值，加上判定点名的 subjects。"""
out: dict[str, str] = {}
rows = (result.get("table") or {}).get("rows") or []
for row in rows if isinstance(rows, list) else []:
if not isinstance(row, list):
continue
for i in _dims(result):
if i < len(row) and (named := _named_value(row[i])):
out.setdefault(named, str(row[i]).strip())
for subject in _subjects(result):
if named := _named_value(subject):
out.setdefault(named, subject)
return out
def _verdict(result: Mapping[str, Any]) -> Mapping[str, Any]:
verdict = result.get("verdict")
return verdict if isinstance(verdict, Mapping) else {}
def _subjects(result: Mapping[str, Any]) -> list[str]:
verdict = _verdict(result)
out: list[str] = []
for subject in verdict.get("subjects") or []:
values = subject if isinstance(subject, list) else [subject]
out.extend(str(v).strip() for v in values if v is not None)
return out
def _named(result: Mapping[str, Any], names: str) -> list[str]:
"""分句里点名的段（原样，按出现的先后）。ASCII 的名字要整词出现（「A」不算点名了「AB」）。"""
flat = _same_length_norm(names)
hits: list[tuple[int, str]] = []
for key, original in _segment_names(result).items():
at = flat.find(key)
while at >= 0:
before, after = flat[at - 1:at], flat[at + len(key):at + len(key) + 1]
if not (key.isascii() and ((before and before.isalnum()) or (after and after.isalnum()))):
hits.append((at, original))
break
at = flat.find(key, at + 1)
return [name for _, name in sorted(hits)]
def _decimal(value: Any) -> Decimal | None:
if value is None or isinstance(value, bool):
return None
try:
return Decimal(str(value).replace(",", ""))
except (InvalidOperation, ValueError):
return None
def _segment_change(result: Mapping[str, Any], name: str) -> tuple[Decimal | None, Decimal | None]:
"""一段（维度值或因子名）的变化和占总变化的比重：结果表里这一段所在各行的变化相加；表里没有的为 None。"""
table = result.get("table") or {}
columns = [str(c) for c in table.get("columns") or []]
col = next((columns.index(c) for c in _DELTA_COLUMNS if c in columns), None)
if col is None:
return None, None
share_col = columns.index("share") if "share" in columns else None
want = _named_value(name)
total: Decimal | None = None
share: Decimal | None = None
for row in table.get("rows") or []:
if not isinstance(row, list) or not any(i < len(row) and _named_value(row[i]) == want for i in _dims(result)):
continue
if (d := _decimal(row[col] if col < len(row) else None)) is not None:
total = d if total is None else total + d
if share_col is not None and (s := _decimal(row[share_col] if share_col < len(row) else None)) is not None:
share = s if share is None else share + s
return total, share
def _caveats(result: Mapping[str, Any]) -> set[str]:
return {str(c.get("code")) for c in result.get("caveats") or [] if isinstance(c, dict)}
def _value(result: Mapping[str, Any], key: str) -> Decimal | None:
return _decimal((result.get("values") or {}).get(key))
def _dir_word(direction: str | None) -> str:
return {"up": "上升", "down": "下降"}.get(direction or "", "")
def _run_direction(raw: Any) -> str | None:
text = str(raw or "").strip().lower()
if text in ("up", "rising", "上升"):
return "up"
if text in ("down", "falling", "下降"):
return "down"
return None
# --------------------------------------------------------------------------
# 判定对不对得上
# --------------------------------------------------------------------------
@dataclass(frozen=True)
class _Miss:
"""一份结论对不上的原因：给写作者的、给人看的各一句（不带结论编号，外面接上）。"""
model: str
human: str
def _phrase(result: Mapping[str, Any]) -> str:
return str(result.get("phrase") or "").strip()
def _verdict_miss(result: Mapping[str, Any]) -> _Miss:
said = f"判定是「{_phrase(result)}」" if _phrase(result) else "判定不是这一类"
return _Miss(said, said)
def _check(lex: Lexicon, hit: Hit, result: Mapping[str, Any], names: str) -> _Miss | None:
"""这处判断词能不能挂这份结论：对得上返回 None，对不上返回原因。names 是判断词所在分句（点名用的样子）。"""
if result.get("status") != "ok":
reason = str(result.get("reason") or "算不了")
return _Miss(f"没有做成（{reason}）", f"这份结论没有做成（{reason}）")
verdict = _verdict(result)
code = verdict.get("code")
key = hit.claim
if key in ("source", "source_not"):
subjects = _subjects(result)
named = _named(result, names)
top = code in ("driver", "concentrated")
if key == "source_not":
inside = [n for n in named if _named_value(n) in {_named_value(s) for s in subjects}]
if top and (inside or not named):
who = "、".join(inside or subjects)
return _Miss(f"判的主要来源正是{who}" if who else _verdict_miss(result).model,
f"判定的主要来源正是{who}" if who else _verdict_miss(result).human)
return None
if not top:
return _verdict_miss(result)
wanted = {_named_value(s) for s in subjects}
stray = [n for n in named if _named_value(n) not in wanted]
pull = hit.word in lex.pulls
if stray and pull:
# 拖累、拉动、带动：点名的段不在主要来源里，变化方向对、占总变化过半的也算（整体上涨、华东下跌拖累了涨幅）
main = _decimal((result.get("settings") or {}).get("main_share")) or Decimal("0.5")
stray = [n for n in stray if not ((share := _segment_change(result, n)[1]) is not None
and abs(share) >= main)]
if stray:
subject = "、".join(subjects)
return _Miss(f"判的主要来源是{subject}，不是{'、'.join(stray)}",
f"判定的主要来源是{subject}，不是{'、'.join(stray)}")
if pull and hit.direction:
for name in named or subjects:
change, _ = _segment_change(result, name)
if change is None or change == 0:
continue
if (change > 0) != (hit.direction == "up"):
moved = "上升" if change > 0 else "下降"
return _Miss(f"里{name}的变化是{moved}，不能说「{hit.word}」",
f"其中{name}的变化是{moved}，与「{hit.word}」的方向相反")
return None
if key == "dispersed":
return None if code == "dispersed" else _verdict_miss(result)
if key == "offset":
return None if "offset" in _caveats(result) else _Miss(
"没有记下此消彼长（各段的变化方向大多一致）", "这份结论没有记下各段此消彼长")
if key == "mix":
if result.get("mode") != "ratio":
return _Miss("是按维度的可加拆解，说明不了结构变化：结构变化要挂比率拆解",
"这是按维度的可加拆解，说明不了结构变化（结构变化要挂比率拆解）")
return None if code in ("mix_driven", "both") else _verdict_miss(result)
if key == "anomaly":
if code not in ("outside_band", "far_outside_band"):
return _verdict_miss(result)
return _direction_miss(hit, verdict.get("direction"))
if key == "record":
caveats = _caveats(result)
wanted = {"up": {"record_high"}, "down": {"record_low"}}.get(hit.direction or "", {"record_high", "record_low"})
if caveats & wanted:
return None
return _Miss("没有记下本期是历史最高或最低", "这份结论没有记下本期是历史最高或最低")
if key == "normal":
return None if code == "within_band" else _verdict_miss(result)
if key == "trend":
if code not in ("rising", "falling"):
return _verdict_miss(result)
if miss := _direction_miss(hit, "up" if code == "rising" else "down"):
return miss
if hit.run:
run = _value(result, "run_length")
if run is None or run < 3:
shown = "没有" if run is None else f"只有 {run} 期"
return _Miss(f"末尾连续同向的期数{shown}，不够说「连续」（至少 3 期）",
f"末尾连续同向的期数{shown}，不够说「连续」")
ran = _run_direction((result.get("values") or {}).get("run_direction"))
if ran and hit.direction and ran != hit.direction:
return _Miss(f"末尾连续{_dir_word(ran)}，不是连续{_dir_word(hit.direction)}",
f"末尾连续{_dir_word(ran)}，与「{hit.word}」的方向相反")
return None
if key == "no_trend":
return None if code == "no_trend" else _verdict_miss(result)
if key == "differ":
if code != "differs":
return _verdict_miss(result)
subjects = _subjects(result)
direction = verdict.get("direction")
if len(subjects) == 2 and hit.direction and direction in ("up", "down"):
# 判定语的写法是「A 的…高于 B」：A 是 subjects 的第一个。句子里两组都点名了才核方向，后点名的先写就反过来
flat = _same_length_norm(names)
spots = [flat.find(_named_value(s) or "￿") for s in subjects]
if all(at >= 0 for at in spots):
expected = direction if spots[0] < spots[1] else ("down" if direction == "up" else "up")
if expected != hit.direction:
higher, lower = (subjects if direction == "up" else subjects[::-1])
return _Miss(f"判的是{higher}高于{lower}，方向写反了", f"判定是{higher}高于{lower}，方向写反了")
return None
if key == "no_diff":
return None if code == "no_evidence" else _verdict_miss(result)
return None
def _direction_miss(hit: Hit, direction: Any) -> _Miss | None:
if hit.direction and direction in ("up", "down") and hit.direction != direction:
return _Miss(f"判的方向是{_dir_word(direction)}，不是{_dir_word(hit.direction)}",
f"判定的方向是{_dir_word(direction)}，与「{hit.word}」相反")
return None
# --------------------------------------------------------------------------
# 违规
# --------------------------------------------------------------------------
#: 没挂对分析结论时交回写作者的「那就怎么写」
_FALLBACK = {
"decompose": "只写各段各自的变化，不下「主要」「集中」「拖累」「分散」这类判断",
"mix": "只写比率和各段的变化，不说结构变化",
"baseline": "只写本期的数和变化，不说「异常」「大幅」「激增」「历史新高」「正常波动」这类判断",
"trend": "只写各期的数，不说「持续」「趋势」「连续」这类判断",
"differ": "只写各组各自的数，不说「显著」「明显高于」「差异不大」这类判断",
}
def _op_text(claim: Claim) -> str:
return "比率拆解" if claim.mode == "ratio" else _OP_TEXT.get(claim.op or "", claim.op or "")
def _snippet(shown: str, start: int, end: int, limit: int = 16) -> str:
"""判断词连同它后面半句（给人看的样子），最多 limit 个字。"""
right = _clause(shown, start, end)[1]
piece = shown[start:max(end, min(right, start + limit))].replace("\x00", "").strip()
return piece or shown[start:end]
def _uncited(claim: Claim, word: str, available: Mapping[str, list[str]]) -> _Reason:
op_text = _op_text(claim)
pool = available.get(f"{claim.op}:{claim.mode}" if claim.mode else claim.op or "", [])
fallback = _FALLBACK["mix" if claim.mode == "ratio" else claim.op or "decompose"]
if pool:
model = (f"「{word}」要挂{op_text}的结论：照它的判定写、句末挂 [[see:{pool[0]}]]，或者直接引用判定 [[a:{pool[0]}]]；"
f"判定和你想写的不一样就照判定写，否则{fallback}")
else:
model = f"这次没有{op_text}的结论，不能写「{word}」：{fallback}"
return _Reason(f"「{word}」是分析判断，这句没有挂{op_text}的分析结论，系统无法核对", model)
def _mismatch(alias: str, snippet: str, miss: _Miss) -> _Reason:
return _Reason(f"「{snippet}」与所挂的分析结论 {alias} 对不上：{miss.human}",
f"{alias} {miss.model}，不能写「{snippet}」：照判定改写（或者直接引用判定 [[a:{alias}]]），"
"或者删掉这个判断")
def _causal(snippet: str) -> _Reason:
return _Reason(f"「{snippet}」是原因判断，这句没有知识库原话或运行输入作依据，也没有写成推测",
f"「{snippet}」没有依据：数据只能说明变化来自哪里；要写原因，引用知识库原话，或写成「可能与…有关，待核实」")
def check_sentence(sentence: Sentence, results: Mapping[str, Mapping[str, Any]], *,
available: Mapping[str, list[str]] | None = None, version: int | None = None) -> list[dict[str, Any]]:
"""一句话的说法违规。results 是写作目录里的分析结论 {A 编号: 结论}，available 是 {op: [这次有的 A 编号]}（没挂时
告诉写作者可以挂哪一份），不给就按 results 算。"""
lex = lexicon(version)
if available is None:
available = _available(results)
hits = scan(sentence.text, version=lex.version)
if not hits:
return []
mine = [(a, results[a]) for a in dict.fromkeys(sentence.analyses) if a in results]
out: list[dict[str, Any]] = []
#: 同一分句里同一类只报一处（「由于促销结束导致下降」是一处没依据的原因，不是两处）
seen: set[tuple[str, str, int]] = set()
guessed = lex.guess.search(sentence.text) is not None
for hit in hits:
claim = lex.claim(hit.claim)
left, right = _clause(sentence.text, hit.start, hit.end)
names = sentence.names[left:right]
snippet = _snippet(sentence.shown, hit.start, hit.end)
if claim.key == "causal":
clause = sentence.text[left:right]
if sentence.quoted or sentence.inputs or guessed or lex.method.search(clause):
continue
if any(h is not hit and h.claim in _SOURCE_KEYS and left <= h.start < right for h in hits):
continue # 「主要由华东造成」：并进来源那一处核
decomposes = [(a, r) for a, r in mine if r.get("op") == "decompose"]
if any(_named(r, names) for _, r in decomposes):
# 因果词点名的是拆解的段或因子：按来源核（记账意义上的「来自」）
hit = Hit("source", hit.start, hit.end, hit.word, hit.direction)
claim = lex.claim("source")
else:
key = (CAUSAL, claim.key, left)
if key not in seen:
seen.add(key)
out.append(_record(CAUSAL, _causal(snippet), sentence, hit, claim, mine))
continue
candidates = [(a, r) for a, r in mine if r.get("op") == claim.op]
if not candidates:
key = (UNCITED, claim.key, left)
if key not in seen:
seen.add(key)
out.append(_record(UNCITED, _uncited(claim, hit.word, available), sentence, hit, claim, mine))
continue
misses = [(a, _check(lex, hit, r, names)) for a, r in candidates]
if any(miss is None for _, miss in misses):
continue
alias, miss = misses[0]
key = (MISMATCH, claim.key, left)
if key not in seen:
seen.add(key)
out.append(_record(MISMATCH, _mismatch(alias, snippet, miss), sentence, hit, claim, mine)) # type: ignore[arg-type]
return out
def _record(code: str, message: _Reason, sentence: Sentence, hit: Hit, claim: Claim,
mine: list[tuple[str, Mapping[str, Any]]]) -> dict[str, Any]:
span = [sentence.base + hit.start, sentence.base + hit.end] if sentence.base is not None else None
segment = next((sid for s, e, sid in sentence.pieces if s <= hit.start < e), None)
shown = sentence.shown.replace("\x00", "").strip()
return _violation(code, message, text=hit.word, span=span, context=shown[:200] or None,
unit=sentence.unit, segment=segment, claim=claim.key, op=claim.op,
**({"mode": claim.mode} if claim.mode else {}), analyses=[a for a, _ in mine])
def _available(results: Mapping[str, Mapping[str, Any]]) -> dict[str, list[str]]:
"""{op: [这次有的 A 编号]}，另按「op:mode」再记一份（结构变化要比率拆解）。只收做成了的：没挂时不教写作者去挂一份
算不了的结论。"""
out: dict[str, list[str]] = {}
for alias, result in results.items():
if result.get("status") != "ok":
continue
op = str(result.get("op") or "")
out.setdefault(op, []).append(alias)
if result.get("mode"):
out.setdefault(f"{op}:{result['mode']}", []).append(alias)
return out
# --------------------------------------------------------------------------
# 报告文档
# --------------------------------------------------------------------------
def results_of(catalog: Mapping[str, Any]) -> dict[str, dict[str, Any]]:
"""写作目录里的分析结论：{A 编号: 结论}（contract.catalog_result 读）。"""
out: dict[str, dict[str, Any]] = {}
for alias, entry in catalog.items():
if isinstance(entry, dict) and (result := catalog_result(entry)) is not None:
out[str(alias)] = dict(result)
return out
def _sentence(unit: dict[str, Any], aliases: Iterable[str], results: Mapping[str, Any]) -> Sentence | None:
"""文档里的一个句子单元 → Sentence。写作者自己写的字（文字片段、没有出处的数字）留在 text；引用渲染出来的值、表名
字段名留在 names（点名用）；判定语（a:）、知识库原话（q:）、排版符号两边都涂掉。"""
segments = [s for s in unit.get("segments") or [] if isinstance(s, dict)]
if not segments:
return None
scan_parts: list[str] = []
name_parts: list[str] = []
shown_parts: list[str] = []
pieces: list[tuple[int, int, str | None]] = []
quoted = False
pos = 0
for seg in segments:
text = str(seg.get("text") or "")
kind, ref = seg.get("kind"), str(seg.get("ref") or "")
head = ref.partition(":")[0].strip() if ref else ""
blank = "\x00" * len(text)
if kind == "structural":
scan_parts.append(blank)
name_parts.append(blank)
elif head in ("a", "q"):
scan_parts.append(blank)
name_parts.append(blank)
quoted = quoted or (head == "q" and seg.get("state") == "deterministic")
elif ref or kind == "entity":
scan_parts.append(blank)
name_parts.append(text)
else:
scan_parts.append(text)
name_parts.append(text)
shown_parts.append(blank if kind == "structural" else text)
pieces.append((pos, pos + len(text), seg.get("id")))
pos += len(text)
span = segments[0].get("span")
base = span[0] if isinstance(span, list) and span and isinstance(span[0], int) else None
cited = list(dict.fromkeys(str(a) for a in aliases))
return Sentence(text="".join(scan_parts), names="".join(name_parts), shown="".join(shown_parts),
analyses=[a for a in cited if a in results], quoted=quoted,
inputs=any(a.startswith("i:") for a in cited), base=base, unit=unit.get("id"), pieces=pieces)
def check_doc(doc: dict[str, Any], catalog: Mapping[str, Any], *, cited: Mapping[int, list[str]] | None = None,
version: int | None = None) -> list[dict[str, Any]]:
"""一份报告文档的说法违规。cited 是 {id(unit): 这句按重新解析的引用算的依据}（verify_doc 交进来），不给就用文档里
记的 cites。标题、代码块、表头不核。"""
results = results_of(catalog)
available = _available(results)
out: list[dict[str, Any]] = []
for block, unit in iter_units(doc):
if block.get("type") in ("heading", "code") or unit.get("kind") in ("heading", "code") \
or (unit.get("loc") or {}).get("row") == -1:
continue
aliases = cited.get(id(unit), []) if cited is not None else unit.get("cites") or []
sentence = _sentence(unit, aliases, results)
if sentence is not None:
out.extend(check_sentence(sentence, results, available=available, version=version))
return out
def record(mode: str) -> dict[str, Any]:
"""报告文档里记下的那一项：按哪种处置、哪一版说法表核。"""
return {"mode": mode, "lexicon": LEXICON_VERSION}
def recorded(doc: Any) -> tuple[str | None, int | None] | None:
"""文档记下的 (处置, 说法表版本)：没有这一项（老文档、报告节点关掉了核对）返回 None；记了认不出的值，认不出的那一样
为 None。"""
if not isinstance(doc, dict) or DOC_FIELD not in doc:
return None
raw = doc.get(DOC_FIELD)
raw = raw if isinstance(raw, dict) else {}
mode = raw.get("mode") if raw.get("mode") in MODES else None
version = raw.get("lexicon") if type(raw.get("lexicon")) is int and raw.get("lexicon") in LEXICONS else None
return mode, version
def verify(doc: Any, catalog: Mapping[str, Any], *, cited: Mapping[int, list[str]] | None = None
) -> list[dict[str, Any]] | None:
"""复核一份文档的说法（evidence.verify_doc 调）：没记这一项的文档返回 None（不核，老文档复核结果不变）；记了的按记下
的那一版说法表核。记了认不出的处置、版本的另记一条 bad_schema（文档可能被改过），照当前版核。"""
found = recorded(doc)
if found is None:
return None
mode, version = found
out: list[dict[str, Any]] = []
if mode is None or version is None:
out.append(_violation("bad_schema", f"报告文档记录的分析说法核对规则「{doc.get(DOC_FIELD)}」无法识别"))
return out + check_doc(doc, catalog, cited=cited, version=version)
def counts(violations: Iterable[Mapping[str, Any]]) -> dict[str, int]:
"""三种说法违规各几处（只数有的）。"""
out: dict[str, int] = {}
for v in violations:
if v.get("code") in CODES:
out[str(v["code"])] = out.get(str(v["code"]), 0) + 1
return out
def summary(violations: list[Mapping[str, Any]]) -> str:
"""几处说法违规怎么说：「1 处判断没挂分析结论、2 处与所挂结论的判定对不上、1 处原因没有依据」。"""
said = {UNCITED: "判断没挂分析结论", MISMATCH: "与所挂结论的判定对不上", CAUSAL: "原因没有依据"}
found = counts(violations)
return "、".join(f"{found[code]} 处{said[code]}" for code in (UNCITED, MISMATCH, CAUSAL) if code in found)
# --------------------------------------------------------------------------
# 写作规则（给模型看）
# --------------------------------------------------------------------------
#: 写作目录里有分析结论时，拼在写作规则后面
ANALYSIS_RULES_PROMPT = (
"分析结论（系统会逐句核对判断词）：\n"
"- 「主要来自」「集中在」「拖累」「带动」「分散」「异常」「大幅」「激增」「历史新高」「正常波动」「持续上升」「趋势」"
"「显著高于」「差异不大」这类判断，只能照证据目录里分析结论的判定写，句末挂对应的依据，比如 [[see:A1]]；最好直接引用"
"判定语 [[a:A1]]，它由系统按判定渲染。来源、集中、拖累、分散挂变化拆解，结构变化挂比率拆解，异常、新高新低、正常波动"
"挂波动基线，持续、趋势、连续挂趋势，显著、明显高于、差异不大挂差异检验。\n"
"- 判定和你想写的不一样时照判定写；点名的地区、渠道要在判定点名的段里；没有做成的分析照实写没做成的原因，不下这类判断。"
"没有对应的分析结论，就只陈述数字和变化。\n"
"- 拆解只说明变化来自哪里，不说明为什么。写原因（因为、由于、导致、受…影响）要有知识库原话或运行输入作依据，"
"否则写成待核实的推测，比如「可能与促销结束有关，待核实」。"
)
#: 写作目录里没有分析结论时，拼在写作规则后面
ANALYSIS_NONE_PROMPT = (
"判断词（系统会逐句核对）：这次没有分析结论，不要写「主要来自」「集中在」「拖累」「带动」「分散」「异常」「大幅」"
"「激增」「历史新高」「正常波动」「持续上升」「趋势」「显著高于」「差异不大」这类判断，只陈述数字和变化（各段各自是多少、"
"变了多少）。写原因（因为、由于、导致、受…影响）要有知识库原话或运行输入作依据，否则写成待核实的推测，"
"比如「可能与促销结束有关，待核实」。"
)
#: 重写请求里有说法违规时补的一句
ANALYSIS_REPAIR_PROMPT = ("判断词照分析结论的判定写并挂上依据，没有对应结论的就删掉判断、只陈述数字和变化；"
"原因没有依据的写成待核实的推测。")
def rules_prompt(catalog: Mapping[str, Any]) -> str:
"""写作规则里说判断词的那一段：目录里有分析结论和没有时各一种说法。"""
return ANALYSIS_RULES_PROMPT if results_of(catalog) else ANALYSIS_NONE_PROMPT
__all__ = [
"ANALYSIS_NONE_PROMPT", "ANALYSIS_REPAIR_PROMPT", "ANALYSIS_RULES_PROMPT", "CAUSAL", "CODES", "DOC_FIELD",
"LEXICONS", "LEXICON_VERSION", "MISMATCH", "MODES", "UNCITED", "Claim", "Hit", "Lexicon", "Sentence",
"check_doc", "check_sentence", "counts", "lexicon", "record", "recorded", "results_of", "rules_prompt", "scan",
"summary", "verify",
]
backend/app/engine/evidence.py
+25 -1
View File
@@ -2663,10 +2663,17 @@ def _context(text: str, start: int, end: int) -> str:
#: 粗体：**…** 或 __…__，里面可以有标记（标记里的下划线不算收尾，[[m:refund_rate]] 很常见）
_BOLD = re.compile(r"(\*\*|__)(?P<inner>(?:\[\[" + _MARKER_GUARD + r"[^\[\]\n]*?\]\]|(?!\1)[^\n])+?)\1")
#: 方向词、因果词：没有数字也没有引用的句子，含这些词才算结论。「同比」「环比」不在里面：它们只说明
#: 跟谁比，本身不是涨跌；真正的比较结论还会带「增长」「下降」或者数字
#: 跟谁比，本身不是涨跌；真正的比较结论还会带「增长」「下降」或者数字。
#: 分析判断词（分析能力一轮，engine/analysis_claims 的说法表）也算：「主要来自」「集中在」「异常」「持续下降」「趋势」
#: 「显著」，裁判优先判这类句子（judge.claim_priority）。「主要」后面是「包括、内容、分为、有、看」这类的是在列提纲，
#: 「持续关注」「持续跟进」是建议，「趋势图」「趋势如下」是在引出图表，「集中采购」「分散风险」说的不是数据，都不算
_DIRECTIONAL = re.compile(
r"增长|增加|上升|提升|提高|上涨|下降|下滑|减少|降低|回落|下跌|反弹|超过|高于|低于|领先|落后|最高|最低"
r"|最多|最少|来自|导致|因为|由于|带动|拉动|拖累|驱动|原因|归因|贡献|占比|翻倍|持平|波动"
r"|主要(?!包括|内容|分为|分成|有|看|从|是指|指|结论|发现|观点|工作|任务|目的|目标|问题|方面)"
r"|集中(?!采购|管理|处理|展示|讨论|培训)|异常|偏离"
r"|持续(?!关注|跟踪|跟进|观察|监控|推进|优化|改进|更新)|趋势(?![图表线如：:])|显著|明显高于|明显低于|大幅|激增"
r"|骤降|暴跌|飙升|新高|新低|抵消|分散(?!注意|精力|风险|投资|采购)"
r"|(?<![A-Za-z])(?:increase|decrease|grow|growth|decline|drop|rise|due to|because|driven)(?![A-Za-z])",
re.I)
#: 定义句：「同比是将本期与上年同月对比」「增长率是指……」。解释一个概念怎么算，不是对数据下结论
@@ -2838,12 +2845,15 @@ def compose_doc(
entities: bool = True,
absence: bool = False,
masked: dict[str, Any] | None = None,
analysis_claims: str | None = None,
) -> dict[str, Any]:
"""模型写的原文（带标记）→ 报告文档：渲染后的 markdown、块、句、片段、统计、违规。
entities=False（报告节点 entities: off）：表名、字段名整层不管——不链接、不核对反引号，
[[t:]] / [[c:]] 解析不了，原因是 ENTITIES_OFF_REASON。
absence=True（写作目录有省略）：另查缺失断言（absence_claims），和句中的 [[see:]] 一样只在组装时查。
analysis_claims（报告节点的分析说法核对，check / strict）：文档顶层记下处置和说法表版本（engine/analysis_claims），
判断词核对和裸数字一样在 verify_doc 里做，出口复核照它再核一遍。不给（None、off）的文档不记、不核，和以前一字不差。
masked（{查询快照工件: 取自遮罩列的结果列}，报告节点按数据源的遮罩算好交进来）：另查直接引用了这些列的格子
（masked_cells），同样只在组装时查。没给的和以前一字不差。目录里值不给的指标（withhold_metrics 记了 masked 的）被引用
时也在这里记违规（masked_metrics），不靠这个参数：目录本身说了哪几个不给。
@@ -3002,9 +3012,13 @@ def compose_doc(
# 没有它的（期 4 之前组装的）一律按原样展示。写在顶层、参与内容哈希，事后补不上：补了哈希就变，封存链就断
# cell_display：有单元格因为显示规则改了写法（ISO 日期时间）才记，记的是渲染得出这些格子的最低一版，复核照它渲染；
# 没有的文档和以前一字不差
# analysis_claims：报告节点要核分析说法时记下处置和说法表版本，复核照它走（只有带这一项的文档核判断词）
from app.engine import analysis_claims as said
doc: dict[str, Any] = {
"schema": DOC_SCHEMA, "entity_syntax": snaps.syntax, "provenance": DOC_PROVENANCE,
**({"cell_display": snaps.needed} if snaps.needed > 1 else {}),
**({said.DOC_FIELD: said.record(analysis_claims)} if analysis_claims in said.MODES else {}),
"run_id": run_id, "node_id": node_id,
"markdown": markdown, "source": source, "catalog": kept, "blocks": out_blocks,
}
@@ -3236,6 +3250,7 @@ def verify_doc(
masks: list[tuple[int, int]] = [] # 引用渲染出来的字
syntax: list[tuple[int, int]] = [] # 结构片段：只涂符号，数字照查
owners: list[tuple[int, int, str, str]] = [] # (start, end, segment id, unit id)
cited_by: dict[int, list[str]] = {} # 每句按这次核对的引用算的依据（分析说法核对用）
pos, tiled = 0, True
for block, unit in iter_units(doc):
stats["units"] += 1
@@ -3351,6 +3366,7 @@ def verify_doc(
_unresolved_message("依据", f"see:{again['ref']}", again),
ref=again["ref"], unit=unit.get("id"),
span=list(unit.get("span") or []) or None))
cited_by[id(unit)] = cited
if kind == "claim" and not cited:
stats["uncited_claims"] += 1
where = unit.get("span")
@@ -3359,6 +3375,14 @@ def verify_doc(
"text": markdown[where[0]:where[1]] if ok_span else ""})
if tiled and pos != len(markdown):
violations.append(_violation("segment_mismatch", "正文末尾有一段不属于任何片段，文档可能被修改过"))
# 分析说法核对（engine/analysis_claims）：只核记了这一项的文档，按记下的那一版说法表、这次核对的引用核。老文档不核，
# 统计也不多这个键
from app.engine import analysis_claims as said
said_found = said.verify(doc, catalog, cited=cited_by)
if said_found is not None:
violations.extend(said_found)
stats["analysis_claims"] = sum(1 for v in said_found if v["code"] in said.CODES)
masks += _header_masks(list(doc.get("blocks") or []), catalog) + _name_masks(markdown, catalog, snaps)
for token in _bare_numbers(markdown, masks, allow_numbers, syntax):
backend/tests/test_analysis_claims.py
+474
View File
@@ -0,0 +1,474 @@
"""分析说法核对（engine/analysis_claims.py，细案 01 §9）：判断词要挂对应的分析结论、判定对得上，因果词要有依据。
- 说法表：每类的判断词认得出，常见的非判断写法（持续关注、异常订单、相当于、数据异常）不算；「大幅」收进异常类，看清它在
常见写法里的表现
- 否定：判断词前 4 个字以内（跳过填充词）有否定的翻到反面类；「不但」「不少」不是否定
- 绑定：类要的 op 在这句的依据里出现、判定对得上；点名的段在 subjects 里（全半角归一）；拖累、拉动看那一段的方向；
新高新低看 caveat；「连续」看末尾连续的期数；挂了算不了的结论算对不上
- 来源与因果的分界：因果词点名的是拆解的段 → 按来源核；不是 → 按因果核；推测说法、知识库原话、运行输入豁免因果
- 不扫：判定语（a:）、知识库原话（q:）、别的引用渲染出来的字、标题、代码块、表头
- 文档：只有记了这一项的文档在复核时核，老文档复核结果不变；记了认不出的版本记 bad_schema
"""
from __future__ import annotations
import re
import pytest
from app.analysis.testing import fake_result
from app.engine import analysis_claims as ac
from app.engine import evidence
from app.engine.evidence import build_catalog, compose_doc, make_eid, verify_doc
def hits(text: str) -> list[tuple[str, str, str | None]]:
return [(h.claim, h.word, h.direction) for h in ac.scan(text)]
def claims(text: str) -> list[str]:
return [h.claim for h in ac.scan(text)]
def said(text: str, *analyses: str, results=None, **sentence) -> list[dict]:
"""一句话、挂了 analyses 这几份结论：违规清单。"""
return ac.check_sentence(ac.Sentence(text=text, analyses=list(analyses), **sentence), results or {})
def codes(found: list[dict]) -> list[str]:
return [v["code"] for v in found]
# --------------------------------------------------------------------------
# 说法表：每类认得出、常见的非判断写法不算
# --------------------------------------------------------------------------
@pytest.mark.parametrize("text, claim", [
("下降主要来自华东。", "source"), ("增长主要由线上贡献。", "source"), ("下降集中在华东、华北。", "source"),
("绝大部分来自线下。", "source"), ("华东拖累了整体。", "source"), ("线上带动了增长。", "source"),
("华东贡献最大。", "source"), ("下降主要是华东下滑造成的。", "source"),
("变化分散在各地区。", "dispersed"), ("各地区都在下降。", "dispersed"), ("各渠道普遍下滑。", "dispersed"),
("全面下降。", "dispersed"),
("华东和华南此消彼长。", "offset"), ("两者互相抵消。", "offset"),
("退款率上升来自结构变化。", "mix"), ("这是结构性的上升。", "mix"),
("本周销售额异常下降。", "anomaly"), ("取消单激增。", "anomaly"), ("订单数骤降。", "anomaly"),
("销售额明显偏离往常。", "anomaly"), ("超出过去 12 周的常规波动范围。", "anomaly"),
("销售额明显高于正常水平。", "anomaly"), ("波动剧烈。", "anomaly"),
("本周创历史新高。", "record"), ("销售额创新高。", "record"), ("前所未有。", "record"),
("在常规波动范围内。", "normal"), ("属于正常波动。", "normal"), ("并不异常。", "normal"),
("线上销售额持续上升。", "trend"), ("呈上升趋势。", "trend"), ("一路走低。", "trend"), ("稳步增长。", "trend"),
("连续三周下降。", "trend"), ("连续下滑。", "trend"),
("没有明显的上升或下降趋势。", "no_trend"), ("上升趋势不明显。", "no_trend"), ("基本平稳。", "no_trend"),
("西南的退款率显著高于华东。", "differ"), ("两组差异明显。", "differ"),
("两组差异不大。", "no_diff"), ("两组没有明显差异。", "no_diff"), ("两个渠道的退款率相当。", "no_diff"),
("由于促销结束，销售额下降。", "causal"), ("受天气影响，客流下降。", "causal"), ("增长得益于新客。", "causal"),
])
def test_every_class_is_recognized(text, claim):
assert claim in claims(text), hits(text)
@pytest.mark.parametrize("text", [
"建议持续关注华东的退款。", # 「持续」后面不是涨跌：建议
"异常订单 12 单。", # 「异常」后面是名词：业务分类
"本周没有发现数据异常。", # 数据质量
"系统无异常。",
"销售额相当于上周的两倍。", # 「相当于」
"华东的销售额相当高。", # 「相当」是副词
"这个问题引起关注。", # 「引起关注」不是因果
"主导地位没有变。", # 「主导地位」
"降幅较大。", # 不是「大幅」
"整体平稳。", # 只收「基本平稳」：宁可漏报
"趋势值得关注。", # 「趋势」不带涨跌
"华东销售额下降，华南销售额上升。", # 只陈述数字和变化
])
def test_ordinary_wording_is_not_a_judgement(text):
assert ac.scan(text) == []
def test_the_longest_reading_wins_where_two_overlap():
# 「超出正常范围」是异常，不是正常；「没有明显的上升或下降趋势」是无趋势，不是上升趋势
assert claims("超出正常范围。") == ["anomaly"]
assert claims("没有明显的上升或下降趋势。") == ["no_trend"]
assert claims("下降主要由华东造成。") == ["source", "causal"]
def test_the_word_carries_its_direction():
assert [h.run for h in ac.scan("连续三周下降，持续下降。")] == [True, False]
assert hits("取消单激增。") == [("anomaly", "激增", "up")]
assert hits("订单数骤降。") == [("anomaly", "骤降", "down")]
assert hits("线上销售额持续上升。") == [("trend", "持续上升", "up")]
assert hits("本周销售额异常下降。")[0][2] == "down" # 方向在判断词后面
assert hits("销售额下降，属于异常。")[0][2] is None # 分句里没有涨跌
assert hits("华东拖累了整体。")[0][2] == "down" and hits("线上带动了增长。")[0][2] == "up"
# --------------------------------------------------------------------------
# 「大幅」：收进异常类，看清它在常见写法里的表现
# --------------------------------------------------------------------------
@pytest.mark.parametrize("text, direction", [
("本周销售额大幅增长。", "up"), ("订单数大幅下降。", "down"), ("退款大幅度减少。", "down"),
("客单价大幅提升，订单数基本不变。", "up"), ("销售额同比大幅下滑。", "down"),
])
def test_dafu_is_an_anomaly_claim_with_the_direction_of_its_verb(text, direction):
[(claim, word, got)] = hits(text)
assert claim == "anomaly" and word.startswith("大幅") and got == direction
@pytest.mark.parametrize("text", ["销售额大幅领先华南。", "幅度较大。", "大幅的促销海报已经撤下。"])
def test_dafu_without_a_change_verb_is_not_counted(text):
assert ac.scan(text) == []
def test_dafu_needs_a_baseline_outside_the_band_in_the_same_direction():
up = fake_result("baseline", None, "outside_band", direction="up")
down = fake_result("baseline", None, "outside_band", direction="down")
normal = fake_result("baseline", None, "within_band", direction="up")
text = "本周销售额大幅增长。"
assert codes(said(text)) == [ac.UNCITED] # 没挂：要挂波动基线
assert said(text, "A1", results={"A1": up}) == []
[miss] = said(text, "A1", results={"A1": down})
assert miss["code"] == ac.MISMATCH and "方向" in miss["message"]
[miss] = said(text, "A1", results={"A1": normal})
assert miss["code"] == ac.MISMATCH and "常规波动范围内" in miss["message"]
# --------------------------------------------------------------------------
# 否定
# --------------------------------------------------------------------------
@pytest.mark.parametrize("text, claim", [
("并未出现明显异常。", "normal"), # 否定隔着「出现明显」
("本周没有出现异常。", "normal"),
("不在正常范围内。", "anomaly"),
("不属于正常波动。", "anomaly"),
("没有显著高于华东。", "no_diff"),
("并非主要来自华东。", "source_not"),
("并不分散。", "source"),
("没有持续上升。", "no_trend"),
])
def test_a_negation_flips_to_the_opposite_class(text, claim):
assert claims(text) == [claim]
@pytest.mark.parametrize("text", ["不但大幅增长，", "不少门店大幅下滑。", "无论是否大幅增长，"])
def test_words_that_only_look_negative_do_not_flip(text):
assert claims(text) == ["anomaly"]
def test_negated_causes_and_records_are_left_alone():
assert ac.scan("并非由于促销。") == [] and ac.scan("并非历史新高。") == []
def test_a_negated_source_needs_the_named_segment_outside_the_main_sources():
driver = fake_result() # 主要来自华东
assert said("下降并非主要来自华南。", "A1", results={"A1": driver}) == []
[miss] = said("下降并非主要来自华东。", "A1", results={"A1": driver})
assert miss["code"] == ac.MISMATCH and "华东" in miss["message"]
dispersed = fake_result(verdict="dispersed")
assert said("下降并非主要来自某一个地区。", "A1", results={"A1": dispersed}) == []
# --------------------------------------------------------------------------
# 绑定
# --------------------------------------------------------------------------
def test_the_required_op_must_be_among_the_evidence():
baseline = fake_result("baseline", None, "outside_band")
[v] = said("下降主要来自华东。", "A1", results={"A1": baseline})
assert v["code"] == ac.UNCITED and v["op"] == "decompose" and v["claim"] == "source" and v["analyses"] == ["A1"]
# 有好几个时任一个对得上就算对
assert said("下降主要来自华东。", "A1", "A2", results={"A1": baseline, "A2": fake_result()}) == []
def test_named_segments_must_be_among_the_subjects():
concentrated = fake_result(verdict="concentrated", subjects=[["华东"], ["华北"]])
assert said("下降集中在华东、华北。", "A1", results={"A1": concentrated}) == []
[miss] = said("下降集中在华东、华南。", "A1", results={"A1": concentrated})
assert miss["code"] == ac.MISMATCH and "不是华南" in miss["message"]
# 别的分句里提到的段不算点名
assert said("下降主要来自华东，华南基本持平。", "A1", results={"A1": fake_result()}) == []
def test_named_segments_are_matched_across_full_and_half_width():
result = fake_result(subjects=[["Ａ店"]])
result["table"]["rows"][0][0] = "Ａ店"
assert said("下降主要来自A店。", "A1", results={"A1": result}) == []
assert codes(said("下降主要来自华北。", "A1", results={"A1": result})) == [ac.MISMATCH]
def test_a_dispersed_verdict_cannot_back_a_main_source():
[miss] = said("下降主要来自华东。", "A3", results={"A3": fake_result(verdict="dispersed")})
assert miss["code"] == ac.MISMATCH and miss["text"] == "主要来自"
assert "A3" in miss["message"] and "分散在各地区" in miss["message"]
assert "[[a:A3]]" in miss["for_model"] and "不能写「主要来自华东」" in miss["for_model"]
def test_drag_and_pull_follow_the_direction_of_the_segment():
up = fake_result(direction="up")
up["table"]["rows"][0][3] = "120" # 华东 +120
assert said("华东带动了增长。", "A1", results={"A1": up}) == []
[miss] = said("华东拖累了整体。", "A1", results={"A1": up})
assert miss["code"] == ac.MISMATCH and "上升" in miss["message"]
assert said("华东拖累了整体。", "A1", results={"A1": fake_result()}) == [] # 华东 −120
def test_a_drag_outside_the_main_sources_counts_when_it_is_a_large_share():
# 整体上涨、华南是主要来源，华东下跌拖累了涨幅：华东的变化占总变化过半（绝对值）也算拖累
result = fake_result(subjects=[["华南"]], direction="up")
result["table"]["rows"][0][3:6] = ["-120", "-0.1463", "-1.5"]
assert said("华东拖累了涨幅。", "A1", results={"A1": result}) == []
result["table"]["rows"][0][5] = "-0.2"
assert codes(said("华东拖累了涨幅。", "A1", results={"A1": result})) == [ac.MISMATCH]
def test_records_need_the_caveat_in_the_same_direction():
high = fake_result("baseline", None, "far_outside_band", direction="up",
caveats=[{"code": "record_high", "text": "本期是历史最高"}])
assert said("本周创历史新高。", "A1", results={"A1": high}) == []
assert codes(said("本周创历史新低。", "A1", results={"A1": high})) == [ac.MISMATCH]
plain = fake_result("baseline", None, "outside_band", direction="up")
assert codes(said("本周创历史新高。", "A1", results={"A1": plain})) == [ac.MISMATCH]
def test_trends_check_the_direction_and_runs_check_the_length():
rising = fake_result("trend", None, "rising", direction="up", values={"run_length": "4", "run_direction": "up"})
assert said("线上销售额持续上升。", "A1", results={"A1": rising}) == []
assert codes(said("线上销售额持续下降。", "A1", results={"A1": rising})) == [ac.MISMATCH]
assert said("线上销售额连续四周上升。", "A1", results={"A1": rising}) == []
short = fake_result("trend", None, "rising", direction="up", values={"run_length": "2", "run_direction": "up"})
[miss] = said("线上销售额连续上升。", "A1", results={"A1": short})
assert miss["code"] == ac.MISMATCH and "连续" in miss["message"]
flat = fake_result("trend", None, "no_trend", direction=None)
assert said("没有明显的上升或下降趋势。", "A1", results={"A1": flat}) == []
assert codes(said("线上销售额持续上升。", "A1", results={"A1": flat})) == [ac.MISMATCH]
def test_differences_check_the_verdict_and_the_order_of_the_two_groups():
differs = fake_result("differ", "rate", "differs", subjects=[["西南"], ["华东"]], direction="up")
assert said("西南的退款率显著高于华东。", "A1", results={"A1": differs}) == []
assert said("华东的退款率显著低于西南。", "A1", results={"A1": differs}) == []
[miss] = said("西南的退款率显著低于华东。", "A1", results={"A1": differs})
assert miss["code"] == ac.MISMATCH and "西南高于华东" in miss["message"]
nothing = fake_result("differ", "rate", "no_evidence", subjects=[["西南"], ["华东"]], direction=None)
assert said("两组差异不大。", "A1", results={"A1": nothing}) == []
assert codes(said("西南的退款率显著高于华东。", "A1", results={"A1": nothing})) == [ac.MISMATCH]
def test_a_structure_claim_needs_a_ratio_decomposition():
[miss] = said("退款率上升来自结构变化。", "A1", results={"A1": fake_result()})
assert miss["code"] == ac.MISMATCH and "比率拆解" in miss["message"]
mix = fake_result(mode="ratio", verdict="mix_driven")
assert said("退款率上升来自结构变化。", "A1", results={"A1": mix}) == []
def test_offsets_need_the_caveat():
assert codes(said("华东和华南此消彼长。", "A1", results={"A1": fake_result()})) == [ac.MISMATCH]
offset = fake_result(caveats=[{"code": "offset", "text": "下降的段合计 −120"}])
assert said("华东和华南此消彼长。", "A1", results={"A1": offset}) == []
def test_a_refused_conclusion_backs_nothing():
refused = fake_result("baseline", None, None, status="refused", reason_code="history_short")
[miss] = said("本周销售额异常下降。", "A1", results={"A1": refused})
assert miss["code"] == ac.MISMATCH and "没有做成" in miss["message"] and "历史只有 5 期" in miss["message"]
def test_uncited_advice_points_at_a_conclusion_that_exists():
[v] = said("下降主要来自华东。", results={"A2": fake_result()})
assert "[[see:A2]]" in v["for_model"] and "[[a:A2]]" in v["for_model"]
[v] = said("下降主要来自华东。")
assert "这次没有变化拆解的结论" in v["for_model"] and "只写各段各自的变化" in v["for_model"]
# 结构变化只推荐比率拆解；算不了的结论不推荐
[v] = said("这是结构性的上升。", results={"A2": fake_result(), "A3": fake_result(mode="ratio", verdict="both")})
assert "[[see:A3]]" in v["for_model"]
[v] = said("下降主要来自华东。", results={"A2": fake_result(status="refused")})
assert "[[see:A2]]" not in v["for_model"]
def test_one_violation_per_class_and_clause():
found = said("由于促销结束导致下降。")
assert codes(found) == [ac.CAUSAL] and found[0]["text"] == "由于"
assert "「由于促销结束导致下降」" in found[0]["message"]
assert codes(said("由于促销结束，下降；由于天气，客流减少。")) == [ac.CAUSAL, ac.CAUSAL]
# --------------------------------------------------------------------------
# 来源与因果
# --------------------------------------------------------------------------
def test_a_cause_naming_a_segment_of_the_decomposition_is_checked_as_a_source():
driver = fake_result()
assert said("下降主要由华东造成。", "A1", results={"A1": driver}) == []
assert said("增长得益于华东。", "A1", results={"A1": fake_result(direction="up")}) == []
[miss] = said("下降是华南造成的。", "A1", results={"A1": driver})
assert miss["code"] == ac.MISMATCH and miss["claim"] == "source"
def test_a_cause_outside_the_decomposition_needs_its_own_evidence():
[v] = said("由于促销结束导致下降。", "A1", results={"A1": fake_result()})
assert v["code"] == ac.CAUSAL and v["claim"] == "causal" and "op" not in v
assert "可能与…有关，待核实" in v["for_model"]
@pytest.mark.parametrize("sentence", [
{"text": "可能与促销结束有关，待核实。"},
{"text": "下降或许是由于促销结束。"},
{"text": "疑似由于促销结束导致下降。"},
{"text": "由于促销结束导致下降。", "quoted": True},
{"text": "由于促销结束导致下降。", "inputs": True},
{"text": "由于数据只到 10 月 16 日，本周不完整。"},
{"text": "由于样本少，差异不能下结论。"},
])
def test_guesses_quotes_inputs_and_data_caveats_are_not_unsupported_causes(sentence):
assert ac.check_sentence(ac.Sentence(**sentence), {}) == []
# --------------------------------------------------------------------------
# 文档：哪里不扫、怎么挂、版本
# --------------------------------------------------------------------------
def a_entry(alias: str, result: dict) -> dict:
artifact = "a" * 63 + alias[-1]
return {"alias": alias, "kind": "analysis", "artifact": artifact, "eid": make_eid("analysis", artifact, {}),
"locator": {}, "result": result}
def catalog(**analyses) -> dict:
out = build_catalog(nodes={"caliber": {
"kind": "metric_set", "caliber": "周报口径", "caliber_version": "v2", "artifact": "f" * 64,
"metrics": [{"id": "gmv", "name": "销售额", "value": 45678.5, "unit": "元", "decimals": 1}]}},
inputs={"promo": "满减"})
out.update({alias: a_entry(alias, result) for alias, result in analyses.items()})
return out
@pytest.fixture
def see_a(monkeypatch):
"""[[see:A1]] 的解析是证据接入（W1-b）的事：合进来之前这里按它的约定代为解析（编号 A{n}、目录里 kind analysis）。"""
original = evidence.resolve_support
def resolve(ref, catalog, *, loader=None):
ref = ref.strip()
entry = catalog.get(ref) or {}
if re.fullmatch(r"A\d+", ref) and entry.get("kind") == "analysis":
return {"ref": ref, "role": "support", "alias": ref, "kind": "analysis", "status": "resolved",
"eid": entry["eid"], "locator": {}}
return original(ref, catalog, loader=loader)
monkeypatch.setattr(evidence, "resolve_support", resolve)
def found(doc: dict) -> list[dict]:
return [v for v in doc["violations"] if v["code"] in ac.CODES]
def test_a_doc_records_the_mode_and_lexicon_only_when_asked():
raw = "本周销售额 [[m:gmv]]，下降主要来自华东。[[see:m:gmv]]"
doc = compose_doc(raw, catalog(), analysis_claims="check")
assert doc[ac.DOC_FIELD] == {"mode": "check", "lexicon": ac.LEXICON_VERSION}
[v] = found(doc)
assert v["code"] == ac.UNCITED and doc["markdown"][slice(*v["span"])] == "主要来自"
assert v["unit"] and v["segment"] and v["context"] == "本周销售额 45,678.5元，下降主要来自华东。"
assert doc["stats"]["analysis_claims"] == 1
# 没要求（off、老调用）：文档一字不差，统计也不多这个键
for mode in (None, "off"):
plain = compose_doc(raw, catalog(), analysis_claims=mode)
assert ac.DOC_FIELD not in plain and found(plain) == [] and "analysis_claims" not in plain["stats"]
def test_old_documents_verify_exactly_as_before():
raw = "本周销售额 [[m:gmv]]，下降主要来自华东，由于促销结束。[[see:m:gmv]]"
old = compose_doc(raw, catalog())
assert verify_doc(old, catalog()) == verify_doc(compose_doc(raw, catalog()), catalog())
assert verify_doc(old, catalog())["violations"] == [] and "analysis_claims" not in verify_doc(old, catalog())["stats"]
new = compose_doc(raw, catalog(), analysis_claims="strict")
again = verify_doc(new, catalog())
assert [v["code"] for v in again["violations"]] == [ac.UNCITED, ac.CAUSAL]
assert again["stats"]["analysis_claims"] == 2 and again["violations"] == new["violations"]
def test_an_unrecognized_record_is_reported_and_checked_with_the_current_lexicon():
doc = compose_doc("下降主要来自华东。", catalog(), analysis_claims="check")
doc[ac.DOC_FIELD] = {"mode": "check", "lexicon": 99}
checked = verify_doc(doc, catalog())
assert [v["code"] for v in checked["violations"]] == ["bad_schema", ac.UNCITED]
assert "分析说法核对规则" in checked["violations"][0]["message"]
def test_system_rendered_phrases_quotes_headings_and_code_are_not_scanned():
doc = {"markdown": "", "blocks": [
{"type": "heading", "units": [{"id": "u0", "kind": "heading", "segments": [
{"id": "s0", "kind": "text", "text": "下降主要来自华东", "span": [0, 8]}]}]},
{"type": "code", "units": [{"id": "u1", "kind": "code", "segments": [
{"id": "s1", "kind": "text", "text": "-- 异常订单激增", "span": [8, 17]}]}]},
{"type": "paragraph", "units": [{"id": "u2", "kind": "claim", "segments": [
{"id": "s2", "kind": "value", "text": "销售额下降 194 万，主要来自华东（-120 万，占 62%）", "ref": "a:A1"},
{"id": "s3", "kind": "text", "text": "。售后说「"},
{"id": "s4", "kind": "quote", "text": "退款主要集中在东区，由于物流延误", "ref": "q:K1|退款主要集中在东区，由于物流延误",
"state": "deterministic"},
{"id": "s5", "kind": "text", "text": "」。"}]}]},
{"type": "table", "units": [{"id": "u3", "kind": "connective", "loc": {"row": -1, "col": 0}, "segments": [
{"id": "s6", "kind": "text", "text": "是否异常"}]}]},
]}
assert ac.check_doc(doc, catalog()) == []
# 同一句换成写作者自己写的字就核
doc["blocks"][2]["units"][0]["segments"][0] = {"id": "s2", "kind": "text", "text": "销售额下降，主要来自华东"}
assert codes(ac.check_doc(doc, catalog())) == [ac.UNCITED]
def test_rendered_values_name_segments_but_are_not_judgement_words():
# [[v:…]] 渲染出来的「华东」能点名，渲染出来的「异常」不是写作者下的判断
doc = {"markdown": "", "blocks": [{"type": "paragraph", "units": [{"id": "u0", "kind": "claim", "cites": ["A1"],
"segments": [
{"id": "s0", "kind": "text", "text": "下降主要来自", "span": [0, 6]},
{"id": "s1", "kind": "value", "text": "华南", "ref": "v:Q1.r1.region", "span": [6, 8]},
{"id": "s2", "kind": "text", "text": "，状态为", "span": [8, 12]},
{"id": "s3", "kind": "value", "text": "异常", "ref": "v:Q1.r1.status", "span": [12, 14]}]}]}]}
[miss] = ac.check_doc(doc, {"A1": a_entry("A1", fake_result())})
assert miss["code"] == ac.MISMATCH and "不是华南" in miss["message"] and miss["span"] == [2, 6]
def test_compose_and_verify_bind_the_sentence_to_its_see_marker(see_a):
cat = catalog(A1=fake_result())
ok = compose_doc("销售额 [[m:gmv]]，下降主要来自华东。[[see:A1]]", cat, analysis_claims="check")
assert found(ok) == [] and verify_doc(ok, cat)["violations"] == []
bad = compose_doc("销售额 [[m:gmv]]，下降主要来自华南。[[see:A1]]", cat, analysis_claims="check")
[miss] = found(bad)
assert miss["code"] == ac.MISMATCH and miss["analyses"] == ["A1"]
# 复核按这次重新解析的引用算，不信文档里记的 cites：依据删掉了就是没挂
tampered = compose_doc("销售额 [[m:gmv]]，下降主要来自华东。[[see:A1]]", cat, analysis_claims="check")
tampered["blocks"][0]["units"][0]["see"] = []
assert [v["code"] for v in verify_doc(tampered, cat)["violations"]] == [ac.UNCITED]
def test_inputs_and_quotes_in_the_same_sentence_support_a_cause():
doc = compose_doc("由于本周有 [[i:promo]] 活动，销售额 [[m:gmv]]。[[see:m:gmv]]", catalog(), analysis_claims="check")
assert found(doc) == []
doc = compose_doc("由于本周有活动，销售额 [[m:gmv]]。[[see:m:gmv,i:promo]]", catalog(), analysis_claims="check")
assert found(doc) == []
doc = compose_doc("由于本周有活动，销售额 [[m:gmv]]。[[see:m:gmv]]", catalog(), analysis_claims="check")
assert codes(found(doc)) == [ac.CAUSAL]
def test_the_prompt_rules_depend_on_whether_there_are_conclusions():
assert ac.rules_prompt(catalog()) == ac.ANALYSIS_NONE_PROMPT
assert ac.rules_prompt(catalog(A1=fake_result())) == ac.ANALYSIS_RULES_PROMPT
assert "只陈述数字和变化" in ac.ANALYSIS_NONE_PROMPT and "待核实" in ac.ANALYSIS_NONE_PROMPT
assert "[[a:A1]]" in ac.ANALYSIS_RULES_PROMPT
def test_judgement_words_make_a_sentence_a_claim_for_the_judge():
from app.engine.judge import claim_priority
doc = compose_doc("华东异常。\n\n下面主要看三个方面。\n\n建议持续关注。\n\n近 8 周的趋势图如下：", catalog())
kinds = {doc["markdown"][slice(*u["span"])]: (u["kind"], claim_priority(u, b, doc["markdown"]))
for b, u in evidence.iter_units(doc)}
assert kinds["华东异常。"] == ("claim", 2)
assert kinds["下面主要看三个方面。"][0] == "connective" and kinds["建议持续关注。"][0] == "connective"
assert kinds["近 8 周的趋势图如下："][0] == "connective", "引出图表的话不算判断"