"""压测结束后归档脚本：将本次新增/修改的 aclnn 接口的 test-design 与 cpp 文件归档至 out_git/ 目录
功能定位：out_git/ 是 output/ 的干净可 git 提交产物镜像（仅 test-design 约束文件 + 各场景 CPP，
不含 fuzz_logs/plog/执行日志/unverified），压测结束后归档本次触碰接口的新增/修改文件，
供版本管理与跨环境复用（详见 AGENTS.md "out_git 归档" 章节）。
归档范围（每接口）：
- output/{api}/test-design/** → out_git/{api}/test-design/**（03~08 yaml、error_scenarios.yaml、
deterministic_check.json 等全部约束产物）
- output/{api}/*.cpp（顶层） → out_git/{api}/（normal/non_contig/multithread/consistent/
error/sample CPP；unverified/ 子目录与 fuzz_logs/ 不归档）
增量语义（"本次新增依旧修改"）：
按内容 sha256 对比 out_git/ 已有文件——新增文件拷贝并计"新增"，内容变化拷贝并计"修改"，
未变化跳过（幂等）；--force 强制全量覆盖。归档历史追加 out_git/archive_history.jsonl。
接口清单来源（三选一，优先级 --apis > --xlsx > --all）：
--apis aclnnAdd,aclnnLayerNorm 显式指定（单接口流程/指定列表）
--xlsx out_res/批量压测结果.xlsx 从本地结果表格提取本次压测的接口清单（批量压测默认）
--all output/ 下全部接口（全量归档/补档）
用法:
python scripts/archive_out_git.py --apis aclnnAdd # 单接口流程压测结束后
python scripts/archive_out_git.py --xlsx out_res/批量压测结果.xlsx # 批量压测 B5（默认 xlsx）
python scripts/archive_out_git.py --all # 全量归档/补档
python scripts/archive_out_git.py --apis aclnnAdd --force # 强制覆盖未变化文件
退出码: 0=归档完成（含无新变化）; 1=错误（无接口清单/目录缺失）
"""
import argparse
import hashlib
import json
import os
import shutil
import sys
import time
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
sys.stderr.reconfigure(encoding="utf-8", errors="replace")
DEFAULT_OUTPUT_ROOT = "output"
DEFAULT_OUT_GIT = "out_git"
DEFAULT_XLSX = os.path.join("out_res", "批量压测结果.xlsx")
HISTORY_FILE = "archive_history.jsonl"
def log(tag, msg):
print("[{}] {}".format(tag, msg))
def file_sha256(path):
h = hashlib.sha256()
with open(path, "rb") as fh:
for chunk in iter(lambda: fh.read(1 << 20), b""):
h.update(chunk)
return h.hexdigest()
def apis_from_xlsx(xlsx_path):
"""从本地批量压测结果表格提取接口清单（第一列去重）。"""
try:
import openpyxl
except ImportError:
raise RuntimeError("openpyxl 未安装（pip install openpyxl），或改用 --apis/--all 指定接口清单")
if not os.path.isfile(xlsx_path):
raise RuntimeError("结果表格不存在: {}（批量压测未产生结果；单接口流程请用 --apis，全量归档用 --all）".format(xlsx_path))
wb = openpyxl.load_workbook(xlsx_path, read_only=True)
ws = wb.active
apis = []
for row in ws.iter_rows(min_row=2, max_col=1, values_only=True):
val = row[0] if row else None
if val is not None and str(val).strip():
api = str(val).strip()
if api not in apis:
apis.append(api)
wb.close()
if not apis:
raise RuntimeError("结果表格无数据行: {}".format(xlsx_path))
return apis
def apis_from_output(output_root):
"""output/ 下全部接口目录（含 test-design 或 cpp 的目录）。"""
apis = []
if not os.path.isdir(output_root):
raise RuntimeError("output 目录不存在: {}".format(output_root))
for name in sorted(os.listdir(output_root)):
api_dir = os.path.join(output_root, name)
if not os.path.isdir(api_dir):
continue
has_td = os.path.isdir(os.path.join(api_dir, "test-design"))
has_cpp = any(n.endswith(".cpp") for n in os.listdir(api_dir)
if os.path.isfile(os.path.join(api_dir, n)))
if has_td or has_cpp:
apis.append(name)
if not apis:
raise RuntimeError("output/ 下无可归档接口目录")
return apis
def collect_api_files(output_root, api):
"""该接口的可归档文件清单：[(相对路径, 绝对路径)]，test-design/** + 顶层 *.cpp。"""
api_dir = os.path.join(output_root, api)
if not os.path.isdir(api_dir):
return None
files = []
for name in sorted(os.listdir(api_dir)):
path = os.path.join(api_dir, name)
if os.path.isfile(path) and name.endswith(".cpp"):
files.append((name, path))
td_dir = os.path.join(api_dir, "test-design")
for dirpath, _dirnames, names in os.walk(td_dir):
for name in sorted(names):
path = os.path.join(dirpath, name)
rel = os.path.relpath(path, api_dir).replace(os.sep, "/")
files.append((rel, path))
files.sort()
return files
def archive_api(output_root, out_git, api, force=False):
"""归档单接口：返回 (新增数, 修改数, 未变化数, 文件总数)；接口目录缺失返回 None。"""
files = collect_api_files(output_root, api)
if files is None or not files:
return None
added = modified = unchanged = 0
for rel, src in files:
dst = os.path.join(out_git, api, *rel.split("/"))
os.makedirs(os.path.dirname(dst), exist_ok=True)
if not os.path.isfile(dst):
shutil.copy2(src, dst)
added += 1
elif force or file_sha256(dst) != file_sha256(src):
shutil.copy2(src, dst)
modified += 1
else:
unchanged += 1
return added, modified, unchanged, len(files)
def main():
parser = argparse.ArgumentParser(
description="压测结束后归档：本次新增/修改接口的 test-design 与 cpp → out_git/（干净可 git 提交镜像）")
parser.add_argument("--apis", default=None,
help="显式接口清单（逗号分隔，如 aclnnAdd,aclnnLayerNorm；单接口流程使用）")
parser.add_argument("--xlsx", default=DEFAULT_XLSX,
help="批量压测结果表格（从中提取本次压测接口清单，默认 out_res/批量压测结果.xlsx）")
parser.add_argument("--all", action="store_true",
help="全量归档 output/ 下全部接口（补档场景）")
parser.add_argument("--force", action="store_true",
help="强制覆盖未变化文件（默认按内容 hash 增量，未变化跳过）")
parser.add_argument("--output_root", default=DEFAULT_OUTPUT_ROOT, help="output 根目录（默认 output/）")
parser.add_argument("--out_git", default=DEFAULT_OUT_GIT, help="归档目录（默认 out_git/）")
args = parser.parse_args()
if args.apis:
apis = [a.strip() for a in args.apis.split(",") if a.strip()]
if not apis:
log("ERROR", "--apis 为空")
return 1
source = "--apis"
elif args.all:
try:
apis = apis_from_output(args.output_root)
except RuntimeError as err:
log("ERROR", str(err))
return 1
source = "--all"
else:
try:
apis = apis_from_xlsx(args.xlsx)
except RuntimeError as err:
log("ERROR", str(err))
return 1
source = "--xlsx({})".format(args.xlsx)
os.makedirs(args.out_git, exist_ok=True)
log("ARCHIVE", "接口清单来源: {} | 共 {} 个接口 | 归档目标: {}/".format(source, len(apis), args.out_git))
total_added = total_modified = total_unchanged = 0
archived_apis, skipped_apis, new_apis = [], [], []
for api in apis:
result = archive_api(args.output_root, args.out_git, api, force=args.force)
if result is None:
skipped_apis.append(api)
log("WARN", "{} 在 output/ 下无 test-design/cpp 可归档，跳过".format(api))
continue
added, modified, unchanged, total = result
total_added += added
total_modified += modified
total_unchanged += unchanged
archived_apis.append(api)
if added == total:
new_apis.append(api)
log("ARCHIVE", "{}: 新增 {} / 修改 {} / 未变化 {}（共 {} 文件）{}".format(
api, added, modified, unchanged, total, " [本次新增接口]" if api in new_apis else ""))
history_path = os.path.join(args.out_git, HISTORY_FILE)
record = {
"time": time.strftime("%Y-%m-%d %H:%M:%S"),
"source": source,
"api_count": len(archived_apis),
"new_apis": new_apis,
"added": total_added,
"modified": total_modified,
"unchanged": total_unchanged,
"force": args.force,
}
with open(history_path, "a", encoding="utf-8") as fh:
fh.write(json.dumps(record, ensure_ascii=False) + "\n")
log("ARCHIVE", "[RESULT] 归档完成：{} 个接口（本次新增接口 {} 个: {}）| 新增 {} 文件 / 修改 {} 文件 / "
"未变化 {} 文件 | 跳过 {} 接口 | 历史记录 → {}".format(
len(archived_apis), len(new_apis), ",".join(new_apis[:10]) or "-",
total_added, total_modified, total_unchanged,
",".join(skipped_apis[:5]) or "-", history_path))
return 0
if __name__ == "__main__":
sys.exit(main())