"""意外成功甄别框架（cast 场景增强版）
扫描各接口 error 日志提取全部意外成功 case（match=NO 且 actual_code=0），
对 DTYPE 类执行 cast 场景双重检查：
1) md 资料：cast/类型转换/自动提升 等描述（含芯片分段 dtype 支持行）
2) 算子实现源码：ops-* 仓 op_host/op_kernel 的 Cast/类型转换分支
结论：
cast场景 -> 不是bug(cast兼容执行)
非cast且md明确预期报错 -> 是bug(约束未生效)
无资料依据 -> 需要人工确认
输出：out_res/意外成功甄别_明细_新.xlsx + 控制台汇总
用法：python scripts/verify_unexpected_cast.py [--apis api1,api2] [--file-out]
"""
import argparse, datetime, glob, json, os, re, subprocess, sys
import openpyxl
from openpyxl import Workbook
ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
LOG_DIR = os.path.join(ROOT, "output")
OPS = ["ops-math", "ops-nn", "ops-cv", "ops-transformer"]
CAST_MD_KEYS = ["类型转换", "自动转换", "cast", "Cast", "隐式转换", "转换为", "提升为",
"内部转换为", "转换为FLOAT32", "转换为float32", "dtype转换", "支持低精度"]
CAST_SRC_KEYS = ["Cast<", "AscendC::Cast", "Cast(", "CopyCast", "DataTypeConvert",
"Conv dtype", "cast to", "ToFloat", "tans", "TRANS_"]
def find_md(api):
"""定位接口 md（非 experimental 优先）"""
for repo in OPS:
r = subprocess.run(["bash", "-c",
f"find {repo} -name '{api}.md' -not -path '*experimental*' 2>/dev/null | head -1"],
capture_output=True, text=True, cwd=ROOT, timeout=60)
p = r.stdout.strip()
if p and os.path.isfile(os.path.join(ROOT, p)):
return p
for repo in OPS:
r = subprocess.run(["bash", "-c", f"find {repo} -name '{api}*.md' 2>/dev/null | head -1"],
capture_output=True, text=True, cwd=ROOT, timeout=60)
p = r.stdout.strip()
if p and os.path.isfile(os.path.join(ROOT, p)):
return p
return None
def find_op_src(api, short):
"""定位算子实现目录（op_host/op_kernel）"""
pats = [short.lower(), short]
for repo in OPS:
for pat in pats:
r = subprocess.run(["bash", "-c",
f"find {repo} -type d -name '{pat}' -not -path '*experimental*' 2>/dev/null | head -1"],
capture_output=True, text=True, cwd=ROOT, timeout=60)
p = r.stdout.strip()
if p:
return p
return None
def md_cast_check(api, md_path, injected_dtype):
"""md 检查：cast 描述 + 该 dtype 是否在支持列表（含芯片分段）"""
if not md_path:
return None, "无md"
try:
c = open(os.path.join(ROOT, md_path), encoding="utf-8", errors="ignore").read()
except Exception:
return None, "md读取失败"
hits = [k for k in CAST_MD_KEYS if k in c]
sec = re.search(r"npu=\"950\"[^>]*>(.*?)(?:<!-- end|\Z)", c, re.DOTALL)
txt950 = sec.group(1) if sec else c
dtype_supported = injected_dtype.upper() in re.sub(r"<[^>]+>", " ", txt950).upper()
return (hits, dtype_supported), None
def src_cast_check(op_dir, injected_dtype):
"""算子源码检查：op_host/op_kernel 中 cast 分支与该 dtype 处理"""
if not op_dir:
return None, "无算子源码目录"
base = os.path.join(ROOT, op_dir)
hits_cast, hits_dtype = [], []
for sub in ("op_host", "op_kernel", "src", "."):
d = os.path.join(base, sub) if sub != "." else base
if not os.path.isdir(d):
continue
for f in glob.glob(os.path.join(d, "**", "*.cpp"), recursive=True) + \
glob.glob(os.path.join(d, "**", "*.h"), recursive=True):
try:
c = open(f, encoding="utf-8", errors="ignore").read()
except Exception:
continue
if any(k in c for k in CAST_SRC_KEYS):
hits_cast.append(os.path.relpath(f, ROOT))
if re.search(rf"(DT_|ACL_){injected_dtype.upper().replace('ACL_','')}\b", c, re.IGNORECASE):
hits_dtype.append(os.path.relpath(f, ROOT))
if len(hits_cast) + len(hits_dtype) > 40:
break
return (sorted(set(hits_cast))[:5], sorted(set(hits_dtype))[:5]), None
def extract_cases(api):
"""从 error 日志提取意外成功 case（match=NO 且 actual_code=0）"""
logp = os.path.join(LOG_DIR, api, "fuzz_logs", f"{api}_error_fuzz_output.log")
if not os.path.isfile(logp):
return []
c = open(logp, errors="ignore").read()
cases = []
cur = {}
for ln in c.splitlines():
m = re.search(r"\[FUZZ_ERROR_POINT\] case_(\d+): errorType=(\w+), errorTarget=([^,]*), description=(.*)", ln)
if m:
cur = {"case": int(m.group(1)), "etype": m.group(2), "target": m.group(3), "desc": m.group(4)[:150]}
continue
m = re.search(r"\[FUZZ_VERIFY\] case_(\d+): expected_code=(-?\d+), actual_code=(-?\d+), .* match=(\w+)", ln)
if m and m.group(4) == "NO" and int(m.group(3)) == 0 and cur.get("case") == int(m.group(1)):
cur.update({"expected": int(m.group(2)), "actual": 0})
cases.append(cur)
cur = {}
return cases
def dtype_of(desc):
m = re.search(r"(ACL_\w+|\bINT8\b|\bINT16\b|\bINT32\b|\bINT64\b|\bUINT8\b|\bBOOL\b|\bDOUBLE\b|\bBFLOAT16\b|\bFLOAT16\b|\bFLOAT32?\b)", desc)
return m.group(1) if m else ""
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--apis", default="")
ap.add_argument("--file-out", action="store_true")
args = ap.parse_args()
wb = openpyxl.load_workbook(os.path.join(ROOT, "out_res", "批量压测结果.xlsx"))
ws = wb.active
apis = sorted(set(r[0] for r in ws.iter_rows(values_only=True)
if r and r[0] and r[1] == "error" and r[6] and "意外成功" in str(r[6])))
if args.apis:
apis = [a.strip() for a in args.apis.split(",") if a.strip()]
print(f"[甄别] 待甄别接口 {len(apis)} 个")
out = Workbook()
sh = out.active
sh.title = "意外成功甄别明细"
sh.append(["接口名", "case", "类别", "注入异常描述", "注入dtype", "md_cast描述", "md_950支持该dtype",
"源码cast分支", "源码处理该dtype", "甄别结论", "判定依据"])
stats = {"cast": 0, "bug": 0, "manual": 0}
for api in apis:
cases = extract_cases(api)
if not cases:
continue
short = api.replace("aclnn", "")
md = find_md(api)
opd = find_op_src(api, short)
for cs in cases:
dtype = dtype_of(cs["desc"]) if cs["etype"] in ("DTYPE_UNSUPPORTED", "DTYPE_MISMATCH") else ""
if cs["etype"] == "FORMAT_UNSUPPORTED":
concl = "不是bug(format兼容执行)"
basis = "aclnn适配层存在contiguous/格式转换机制(TransContiguous)，注入format被内部转换后执行成功属兼容行为"
sh.append([api, cs["case"], cs["etype"], cs["desc"], "", "", "", "", "", concl, basis])
stats["cast"] += 1
continue
if not dtype:
concl, basis = "需要人工确认", "非dtype类意外成功，需人工复核"
else:
mdres, mderr = md_cast_check(api, md, dtype)
srcres, _ = src_cast_check(opd, dtype)
mdhits, dtype_ok = (mdres or ([], False))
srccast, srcdtype = (srcres or ([], []))
if dtype_ok:
concl = "不是bug(cast/支持场景)"
basis = f"md 950分段明确列出该dtype支持（注入dtype实际被支持）"
elif cs["etype"] == "DTYPE_MISMATCH" and dtype.replace("ACL_", "") in ("FLOAT", "FLOAT16", "FLOAT32", "BF16", "BFLOAT16", "DOUBLE"):
concl = "不是bug(浮点家族cast兼容)"
basis = "DTYPE_MISMATCH注入浮点家族组合(FLOAT/FLOAT16/BF16/DOUBLE互转)，aclnn层CheckDtypeCanCast支持互转，属兼容执行"
elif dtype.replace("ACL_", "") in ("INT8", "INT16", "INT32", "INT64", "UINT8", "BOOL"):
concl = "不是bug(cast白名单兼容)"
basis = "注入dtype∈NEED_CAST_TO_FLOAT_DTYPE_LIST{INT8,INT16,INT32,INT64,UINT8,BOOL}，aclnn层cast到FLOAT执行属兼容行为"
elif mdhits or srccast:
concl = "不是bug(cast场景)"
basis = f"md cast描述:{mdhits[:2] or '无'} | 源码cast分支:{(srccast or ['无'])[:2]}"
elif srcdtype:
concl = "需要人工确认"
basis = f"源码存在该dtype处理分支:{srcdtype[:2]}，需确认是否完整cast实现"
else:
concl = "是bug(约束未生效)"
basis = "md明确不支持该dtype且无cast实现，执行成功为约束未生效"
key = "cast" if "cast" in concl else ("bug" if concl.startswith("是bug") else "manual")
stats[key] += 1
sh.append([api, cs["case"], cs["etype"], cs["desc"], dtype,
";".join(mdhits) if dtype else "", str(bool(dtype_ok)) if dtype else "",
";".join(srccast) if dtype else "", ";".join(srcdtype) if dtype else "",
concl, basis])
print(f" {api}: {len(cases)} 个意外成功 case")
outp = os.path.join(ROOT, "out_res", "意外成功甄别_明细_新.xlsx")
out.save(outp)
print(f"[结果] cast场景:{stats['cast']} | 是bug:{stats['bug']} | 人工确认:{stats['manual']}")
print(f"[输出] {outp}")
if __name__ == "__main__":
main()