"""问题统计目录同步至 GitCode 仓（Alice_zhou/issue，强制）
功能定位：在 GitCode 仓 Alice_zhou/issue 下维护 **问题统计/** 目录，内含两个 Excel 在线统计表，
压测/提单全流程实时填充（同一张表 upsert 累积，不按日期另建表格）：
1. 问题统计/接口压测结果与提单汇总.xlsx
列：序号 | 接口名 | 对应主算子名 | sample | normal | non_contig | multithread | consistent |
error | 总体结论 | 提单情况 | 备注 | 压测日期 | 压测版本
- 各场景列：成功率xx%（error=已覆盖），无数据为 —
- 总体结论（强制三态+着色）：**fail（红色）**=存在 是bug（版本bug）记录；
**pass（绿色）**=全部已执行场景成功率≥90% 且无版本bug；**待复核（黄色）**=存在场景成功率<90% 且无版本bug
- 提单情况：该接口问题单列 issue 链接数（已提单N个/待提单/无）
- 行键 = 接口名 + 压测日期（同接口重跑按日期新行，依次填充；同日重跑 upsert 覆盖）
2. 问题统计/issue汇总.xlsx
列：序号 | aclnn接口名 | 对应主算子名 | 问题 | issue链接 | md资料链接 | 压测日期
- 数据源：out_res/issues/{日期}/daily_issues.json（主源）→ filed.json+summary.md（历史回退）
- 行键 = aclnn接口名 + issue链接（同一 issue 多场景归并只占一行）
- 对应主算子名：find_owner_repo.resolve_api 定位；md资料链接：本地 ops-* 仓 docs/aclnn*.md
（排除 experimental）→ https://gitcode.com/cann/{repo}/blob/{branch}/{path}；两者本地缓存（.meta_cache.json）
多机安全合并（同 sync_batch_results_git.py 机制，不丢任何机器的数据）：
- 推送前先 GET 远端 xlsx → 以远端为基底，将本机行 upsert 合并（键冲突时本机行优先——本机行是最新结果）
- 远端存在而本机没有的行（他机写入）原样保留，禁止整表覆盖
- PUT 携带远端 sha（乐观锁），并发冲突（409/sha 失配）自动重拉远端重新合并重试（最多 --retries 次）
- 远端不存在时 POST 创建；合并完成后 序号 列全表重排、总体结论 着色按值重新应用
触发链路（自动，无需手动）：
- update_batch_results.py：每场景压测结果更新后实时推送 接口压测结果与提单汇总（"每次压测结果依次填充"）
- file_bug_issues.py：每轮提单后实时推送 issue汇总 + 刷新 接口压测结果与提单汇总（提单情况列）
- scheduled_update.py：每日 4 轮整点兜底补推两个表（实时同步失败/漏推的兜底）
API（已实测验证，同 scripts/sync_batch_results_git.py）：GET/POST/PUT contents + PRIVATE-TOKEN。
用法:
python scripts/sync_issue_stats_git.py # 生成/刷新本地两表并推送
python scripts/sync_issue_stats_git.py --api-summary-only # 仅 接口压测结果与提单汇总
python scripts/sync_issue_stats_git.py --issue-summary-only# 仅 issue汇总
python scripts/sync_issue_stats_git.py --no-push # 仅本地生成不推送
python scripts/sync_issue_stats_git.py --local-xlsx out_res/批量压测结果.xlsx [--json]
开关：环境变量 ISSUE_STATS_GIT_SYNC=0/1 > config issue_stats_git_sync（默认 true）
压测版本：环境变量 FUZZ_CANN_VERSION/CANN_VERSION > CANN version 文件 > "未知"
"""
import argparse
import base64
import datetime
import glob
import io
import json
import os
import re
import socket
import subprocess
import sys
import time
try:
import openpyxl
from openpyxl.styles import Font, PatternFill
except ImportError:
openpyxl = None
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from sync_batch_results_git import GitCodeContentsClient
from sync_daily_issues_git import collect_daily_rows
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
sys.stderr.reconfigure(encoding="utf-8", errors="replace")
DEFAULT_CONFIG = "config/issue_tracker.json"
DEFAULT_LOCAL_XLSX = os.path.join("out_res", "批量压测结果.xlsx")
STATS_DIR_LOCAL = os.path.join("out_res", "问题统计")
META_CACHE_PATH = os.path.join(STATS_DIR_LOCAL, ".meta_cache.json")
DEFAULT_REPO_DIR = "问题统计"
API_SUMMARY_NAME = "接口压测结果与提单汇总.xlsx"
ISSUE_SUMMARY_NAME = "issue汇总.xlsx"
MAX_FILE_BYTES = 50 * 1024 * 1024
GATE_SCENARIOS = ("sample", "normal", "non_contig", "multithread", "consistent")
ALL_SCENARIOS = GATE_SCENARIOS + ("error",)
PASS_RATE = 90.0
API_HEADERS = ["序号", "接口名", "对应主算子名", "sample", "normal", "non_contig",
"multithread", "consistent", "error", "总体结论", "提单情况", "备注",
"压测日期", "压测版本"]
ISSUE_HEADERS = ["序号", "aclnn接口名", "对应主算子名", "问题", "issue链接", "md资料链接", "压测日期"]
FILL_STYLES = {
"pass": ("C6EFCE", "006100"),
"待复核": ("FFEB9C", "9C6500"),
"fail": ("FFC7CE", "9C0006"),
}
def log(tag, msg):
print("[{}] {}".format(tag, msg))
def _cell_str(value):
return str(value).strip() if value is not None else ""
def _today():
return datetime.date.today().strftime("%Y-%m-%d")
def get_cann_version():
"""压测版本：环境变量 FUZZ_CANN_VERSION/CANN_VERSION > CANN version 文件 > 未知。"""
for env in ("FUZZ_CANN_VERSION", "CANN_VERSION"):
v = os.environ.get(env, "").strip()
if v:
return v
home = os.environ.get("ASCEND_HOME_PATH", "").strip()
candidates = []
if home:
candidates += [os.path.join(home, "version.cfg"), os.path.join(home, "version.info"),
os.path.normpath(os.path.join(home, "..", "version.cfg"))]
candidates += ["/usr/local/Ascend/ascend-toolkit/latest/version.cfg",
"/usr/local/Ascend/ascend-toolkit/latest/version.info",
"/usr/local/Ascend/ascend-toolkit/latest/x86_64-linux/version.cfg"]
for path in candidates:
try:
with open(path, encoding="utf-8", errors="replace") as fh:
for line in fh:
line = line.strip()
m = re.match(r"^Version\s*=\s*(.+)$", line, re.I)
if m:
return m.group(1).strip()
if line and "=" not in line and not line.startswith("#"):
return line
except Exception:
continue
return "未知"
def _load_meta_cache():
if os.path.isfile(META_CACHE_PATH):
try:
data = json.load(open(META_CACHE_PATH, encoding="utf-8-sig"))
if isinstance(data, dict):
return data
except ValueError:
pass
return {}
def _save_meta_cache(cache):
try:
os.makedirs(STATS_DIR_LOCAL, exist_ok=True)
json.dump(cache, open(META_CACHE_PATH, "w", encoding="utf-8"), ensure_ascii=False, indent=2)
except Exception:
pass
def resolve_op_name(api):
"""对应主算子名（find_owner_repo 定位，本地缓存；失败返回 未定位）。"""
cache = _load_meta_cache()
entry = cache.get(api) or {}
if "op" in entry:
return entry["op"]
op = "未定位"
try:
from find_owner_repo import resolve_api
info = resolve_api(api)
op = str(info.get("op") or "未定位")
except Exception:
pass
entry["op"] = op
cache[api] = entry
_save_meta_cache(cache)
return op
def find_md_link(api):
"""md资料链接：本地 ops-* 仓 docs/aclnn{api}*.md（排除 experimental）→ GitCode 公网链接。
找不到（本地无 ops-* 仓/无该接口 md）返回 "未找到"。结果本地缓存。"""
cache = _load_meta_cache()
entry = cache.get(api) or {}
if "md" in entry:
return entry["md"]
md_link = "未找到"
try:
repos = sorted(glob.glob("ops-*"))
hits = []
for repo in repos:
if not os.path.isdir(repo):
continue
for path in glob.glob(os.path.join(repo, "**", "docs", "*.md"), recursive=True):
low = path.replace("\\", "/").lower()
if "/experimental/" in low:
continue
base = os.path.splitext(os.path.basename(path))[0]
parts = base.split("&")
if api in parts:
hits.append((repo, path))
break
if hits:
break
if hits:
repo, path = hits[0]
branch = "master"
try:
r = subprocess.run(["git", "-C", repo, "rev-parse", "--abbrev-ref", "HEAD"],
capture_output=True, text=True, timeout=30)
if r.returncode == 0 and r.stdout.strip():
branch = r.stdout.strip()
except Exception:
pass
rel = os.path.relpath(path, repo).replace("\\", "/")
md_link = "https://gitcode.com/cann/{}/blob/{}/{}".format(repo, branch, rel)
except Exception:
pass
entry["md"] = md_link
cache[api] = entry
_save_meta_cache(cache)
return md_link
def _scenario_base(scenario):
"""场景归一：normal_precheck/consistent_precheck 等预检记录归并到基场景（取全量记录优先）。"""
s = str(scenario or "").strip()
for suf in ("_precheck", "_debug"):
if s.endswith(suf):
return s[: -len(suf)], False
return s, True
def collect_api_rows(local_xlsx):
"""本地批量压测结果 → 接口汇总行（按 接口名+压测日期 聚合）。
返回 rows: [{api, op, scenes{scenario: 率文本}, rates{scenario: pct}, has_bug, issue_links,
remark, date}]。全量记录优先于预检记录（同场景预检仅在全量缺失时作临时值）。"""
if openpyxl is None or not os.path.isfile(local_xlsx):
return []
wb = openpyxl.load_workbook(local_xlsx, read_only=True)
ws = wb.active
rows_raw = list(ws.iter_rows(values_only=True))
wb.close()
if not rows_raw:
return []
header = [_cell_str(c) for c in rows_raw[0]]
idx = {h: i for i, h in enumerate(header)}
if "接口名" not in idx:
return []
agg = {}
for row in rows_raw[1:]:
if not row or not any(_cell_str(c) for c in row if c is not None):
continue
def col(name, default=""):
i = idx.get(name)
return _cell_str(row[i]) if i is not None and i < len(row) and row[i] is not None else default
api = col("接口名")
if not api:
continue
upd = col("更新时间")
m = re.match(r"(\d{4}-\d{2}-\d{2})", upd)
date = m.group(1) if m else _today()
base, is_full = _scenario_base(col("测试场景"))
if base not in ALL_SCENARIOS:
continue
total = col("压测总条数")
success = col("实际执行成功条数")
try:
t, s = int(float(total or 0)), int(float(success or 0))
except ValueError:
t, s = 0, 0
key = (api, date)
rec = agg.setdefault(key, {"scenes": {}, "bug": "", "issues": set(), "fail_reason": ""})
if t > 0:
pct = round(s * 100.0 / t, 1)
text = "已覆盖" if base == "error" else "成功率{}%".format(int(pct) if pct == int(pct) else pct)
else:
pct = None
text = "已覆盖" if base == "error" else "—"
prev = rec["scenes"].get(base)
if prev is None or (is_full and not prev[2]):
rec["scenes"][base] = (text, pct, is_full)
bug = col("是否存在bug")
if bug.startswith("是bug") and not rec["bug"]:
rec["bug"] = bug
if not rec["fail_reason"]:
reason = col("执行失败原因")
if reason and reason != "无":
rec["fail_reason"] = reason
for url in re.split(r"[\n;；]", col("问题单")):
url = url.strip()
if url.startswith("http"):
rec["issues"].add(url)
out = []
for (api, date), rec in agg.items():
rates = {sc: v[1] for sc, v in rec["scenes"].items() if v[1] is not None and sc != "error"}
if rec["bug"]:
verdict = "fail"
elif rates and any(pct < PASS_RATE for pct in rates.values()):
verdict = "待复核"
elif rates:
verdict = "pass"
else:
verdict = "—"
if rec["issues"]:
filed = "已提单{}个".format(len(rec["issues"]))
elif rec["bug"]:
filed = "待提单"
else:
filed = "无"
remark = rec["bug"] or rec["fail_reason"]
out.append({"api": api, "date": date, "scenes": {sc: rec["scenes"].get(sc, ("—", None, True))[0]
for sc in ALL_SCENARIOS},
"verdict": verdict, "filed": filed, "remark": remark[:80],
"issue_links": sorted(rec["issues"])})
out.sort(key=lambda r: (r["date"], r["api"]))
return out
def _apply_verdict_style(ws):
"""按 总体结论 列值重新应用三态着色（pass绿/待复核黄/fail红）与超链接。"""
if openpyxl is None:
return
header = [_cell_str(c.value) for c in ws[1]] if ws.max_row >= 1 else []
if "总体结论" not in header:
return
vcol = header.index("总体结论") + 1
for row in range(2, ws.max_row + 1):
verdict = _cell_str(ws.cell(row=row, column=vcol).value)
style = FILL_STYLES.get(verdict)
cell = ws.cell(row=row, column=vcol)
if style:
fill_color, font_color = style
cell.fill = PatternFill(start_color=fill_color, end_color=fill_color, fill_type="solid")
cell.font = Font(color=font_color, bold=True)
else:
cell.fill = PatternFill(fill_type=None)
cell.font = Font()
def build_api_summary_workbook(rows, out_path, cann_version):
"""生成本地 接口压测结果与提单汇总.xlsx（行序=日期+接口名，三态着色）。"""
if openpyxl is None:
raise RuntimeError("openpyxl 未安装（pip install openpyxl）")
os.makedirs(os.path.dirname(out_path), exist_ok=True)
op_cache = {}
wb = openpyxl.Workbook()
ws = wb.active
ws.title = "接口压测结果与提单汇总"
ws.append(API_HEADERS)
for i, r in enumerate(rows, 1):
op = op_cache.get(r["api"])
if op is None:
op = op_cache[r["api"]] = resolve_op_name(r["api"])
ws.append([i, r["api"], op] + [r["scenes"].get(sc, "—") for sc in ALL_SCENARIOS]
+ [r["verdict"], r["filed"], r["remark"], r["date"], cann_version])
widths = [6, 26, 22, 10, 12, 13, 13, 13, 10, 10, 12, 40, 12, 16]
for col_idx, width in enumerate(widths, start=1):
ws.column_dimensions[openpyxl.utils.get_column_letter(col_idx)].width = width
_apply_verdict_style(ws)
wb.save(out_path)
return out_path
def collect_issue_rows():
"""全部 out_res/issues/{日期}/ 已提 issue 行（daily_issues.json 主源 → filed.json+summary.md 回退）。
返回 [{api, problem, issue, date}]，按 (api, issue) 去重。"""
rows, seen = [], set()
issues_root = os.path.join("out_res", "issues")
if not os.path.isdir(issues_root):
return rows
for date_str in sorted(os.listdir(issues_root)):
day_dir = os.path.join(issues_root, date_str)
if not os.path.isdir(day_dir) or not re.match(r"^\d{4}-\d{2}-\d{2}$", date_str):
continue
for r in collect_daily_rows(date_str):
key = (r["api"], r["issue"])
if key in seen:
continue
seen.add(key)
rows.append({"api": r["api"], "problem": r["problem"], "issue": r["issue"],
"date": date_str})
rows.sort(key=lambda r: (r["date"], r["api"]))
return rows
def build_issue_summary_workbook(rows, out_path):
"""生成本地 issue汇总.xlsx（序号/aclnn接口名/对应主算子名/问题/issue链接/md资料链接/压测日期）。"""
if openpyxl is None:
raise RuntimeError("openpyxl 未安装（pip install openpyxl）")
os.makedirs(os.path.dirname(out_path), exist_ok=True)
wb = openpyxl.Workbook()
ws = wb.active
ws.title = "issue汇总"
ws.append(ISSUE_HEADERS)
for i, r in enumerate(rows, 1):
ws.append([i, r["api"], resolve_op_name(r["api"]), r["problem"], r["issue"],
find_md_link(r["api"]), r["date"]])
widths = [6, 26, 22, 60, 55, 55, 12]
for col_idx, width in enumerate(widths, start=1):
ws.column_dimensions[openpyxl.utils.get_column_letter(col_idx)].width = width
header = ISSUE_HEADERS
for row in range(2, ws.max_row + 1):
for col_idx, name in ((header.index("issue链接") + 1, "issue"), (header.index("md资料链接") + 1, "md")):
cell = ws.cell(row=row, column=col_idx)
if _cell_str(cell.value).startswith("http"):
cell.hyperlink = _cell_str(cell.value)
wb.save(out_path)
return out_path
def _merge_rows(remote_bytes, local_xlsx, headers, key_cols, style_fn=None):
"""以远端 xlsx 为基底合并本机行（键冲突本机优先，他机行保留），返回 (合并字节, merged, appended)。
key_cols: 行键列名列表（如 ["接口名", "压测日期"]）；合并后 序号 列全表重排；
style_fn: 合并完成后对全表重新应用样式（如三态着色）。远端解析失败抛异常（禁止整表覆盖）。"""
if openpyxl is None:
raise RuntimeError("openpyxl 未安装，无法合并（pip install openpyxl）")
remote_wb = openpyxl.load_workbook(io.BytesIO(remote_bytes))
remote_ws = remote_wb.active
local_wb = openpyxl.load_workbook(local_xlsx)
local_ws = local_wb.active
if remote_ws.max_row < 1 or _cell_str(remote_ws.cell(row=1, column=2).value) != headers[1]:
if remote_ws.max_row >= 1 and _cell_str(remote_ws.cell(row=1, column=2).value) == "":
header_row = 1
else:
header_row = remote_ws.max_row + 1
for col_idx, name in enumerate(headers, start=1):
remote_ws.cell(row=header_row, column=col_idx, value=name)
remote_header = [_cell_str(c.value) for c in remote_ws[1]]
key_idx = [remote_header.index(k) + 1 for k in key_cols]
url_cols = [i + 1 for i, h in enumerate(remote_header) if h in ("issue链接", "md资料链接", "问题单")]
remote_map = {}
for row in range(2, remote_ws.max_row + 1):
key = tuple(_cell_str(remote_ws.cell(row=row, column=c).value) for c in key_idx)
if all(key):
remote_map[key] = row
merged, appended = 0, 0
for row in range(2, local_ws.max_row + 1):
key = tuple(_cell_str(local_ws.cell(row=row, column=c).value) for c in key_idx)
if not all(key):
continue
if key in remote_map:
target = remote_map[key]
merged += 1
else:
target = remote_ws.max_row + 1
appended += 1
for col_idx in range(1, len(remote_header) + 1):
value = local_ws.cell(row=row, column=col_idx).value
cell = remote_ws.cell(row=target, column=col_idx, value=value)
if col_idx in url_cols and _cell_str(value).startswith("http"):
cell.hyperlink = _cell_str(value)
if "序号" in remote_header:
seq_col = remote_header.index("序号") + 1
for n, row in enumerate(range(2, remote_ws.max_row + 1), 1):
remote_ws.cell(row=row, column=seq_col, value=n)
if style_fn is not None:
style_fn(remote_ws)
buf = io.BytesIO()
remote_wb.save(buf)
return buf.getvalue(), merged, appended
def issue_stats_enabled(config_path=DEFAULT_CONFIG):
"""同步开关：环境变量 ISSUE_STATS_GIT_SYNC > config issue_stats_git_sync（默认 true）。"""
env = os.environ.get("ISSUE_STATS_GIT_SYNC", "").strip()
if env:
return env not in ("0", "false", "off", "no"), "env ISSUE_STATS_GIT_SYNC={}".format(env)
if os.path.isfile(config_path):
try:
cfg = json.load(open(config_path, encoding="utf-8-sig"))
enabled = cfg.get("issue_stats_git_sync", True)
return bool(enabled), "config issue_stats_git_sync={}".format(enabled)
except Exception:
return True, "config 解析失败，按默认启用"
return True, "默认启用（无配置文件）"
def _push_with_merge(local_xlsx, repo_path, config_path, headers, key_cols,
commit_msg, retries=3, style_fn=None):
"""本地 xlsx 合并推送至 GitCode 仓（sha 乐观锁 + 冲突重试）。返回 (ok, url, detail)。"""
if openpyxl is None:
return False, "", "openpyxl 未安装"
if not os.path.isfile(local_xlsx):
return False, "", "本地文件不存在: {}".format(local_xlsx)
if os.path.getsize(local_xlsx) > MAX_FILE_BYTES:
return False, "", "文件超过 {}MB 上限".format(MAX_FILE_BYTES // (1024 * 1024))
try:
client = GitCodeContentsClient(config_path)
except Exception as exc:
return False, "", "配置加载失败: {}".format(exc)
last_err = ""
for attempt in range(1, retries + 1):
code, info = client.get_file(repo_path)
if code == 200:
sha = info.get("sha", "")
content_b64 = info.get("content", "") or ""
remote_bytes = base64.b64decode(content_b64) if content_b64 else b""
try:
merged_bytes, merged, appended = _merge_rows(remote_bytes, local_xlsx, headers,
key_cols, style_fn)
except Exception as exc:
return False, "", "远端文件合并失败（禁止覆盖远端，已中止）: {}".format(exc)
payload_b64 = base64.b64encode(merged_bytes).decode("ascii")
code2, resp = client.update_file(repo_path, payload_b64, commit_msg, sha)
if code2 in (200, 201):
url = info.get("html_url", "") or info.get("download_url", "")
return True, url, "merge(remote+local) merged={} appended={} host={}".format(
merged, appended, socket.gethostname())
last_err = "PUT 失败 [{}]: {}".format(code2, json.dumps(resp, ensure_ascii=False)[:200])
if code2 in (409, 400, 422):
time.sleep(min(1.0 * attempt, 3.0))
continue
elif code == 404:
payload_b64 = base64.b64encode(open(local_xlsx, "rb").read()).decode("ascii")
code2, resp = client.create_file(repo_path, payload_b64, commit_msg)
if code2 in (200, 201):
code3, info2 = client.get_file(repo_path)
url = (info2.get("html_url", "") or info2.get("download_url", "")) if code3 == 200 else ""
return True, url, "create(new) host={}".format(socket.gethostname())
last_err = "POST 失败 [{}]: {}".format(code2, json.dumps(resp, ensure_ascii=False)[:200])
time.sleep(min(1.0 * attempt, 3.0))
else:
last_err = "GET 失败 [{}]: {}".format(code, json.dumps(info, ensure_ascii=False)[:200])
time.sleep(min(1.0 * attempt, 3.0))
return False, "", "重试 {} 次仍失败: {}".format(retries, last_err)
def sync_api_summary(local_xlsx=DEFAULT_LOCAL_XLSX, config_path=DEFAULT_CONFIG,
repo_dir=None, push=True, retries=3):
"""生成/刷新并推送 接口压测结果与提单汇总.xlsx。返回 (ok, url, detail)。"""
rows = collect_api_rows(local_xlsx)
if not rows:
return True, "", "本地结果表无数据（{}），跳过推送".format(local_xlsx)
local_out = os.path.join(STATS_DIR_LOCAL, API_SUMMARY_NAME)
cann_version = get_cann_version()
build_api_summary_workbook(rows, local_out, cann_version)
if not push:
return True, local_out, "仅本地生成（{} 行）: {}".format(len(rows), local_out)
try:
cfg = json.load(open(config_path, encoding="utf-8-sig"))
except Exception:
cfg = {}
repo_dir = repo_dir or cfg.get("issue_stats_repo_dir") or DEFAULT_REPO_DIR
repo_path = "{}/{}".format(repo_dir, API_SUMMARY_NAME)
return _push_with_merge(
local_out, repo_path, config_path, API_HEADERS, ["接口名", "压测日期"],
"stats: api summary {} rows from {}".format(len(rows), socket.gethostname()),
retries, style_fn=_apply_verdict_style)
def sync_issue_summary(config_path=DEFAULT_CONFIG, repo_dir=None, push=True, retries=3):
"""生成/刷新并推送 issue汇总.xlsx。返回 (ok, url, detail)。"""
rows = collect_issue_rows()
if not rows:
return True, "", "本地无已提 issue 记录（out_res/issues/），跳过推送"
local_out = os.path.join(STATS_DIR_LOCAL, ISSUE_SUMMARY_NAME)
build_issue_summary_workbook(rows, local_out)
if not push:
return True, local_out, "仅本地生成（{} 行）: {}".format(len(rows), local_out)
try:
cfg = json.load(open(config_path, encoding="utf-8-sig"))
except Exception:
cfg = {}
repo_dir = repo_dir or cfg.get("issue_stats_repo_dir") or DEFAULT_REPO_DIR
repo_path = "{}/{}".format(repo_dir, ISSUE_SUMMARY_NAME)
return _push_with_merge(
local_out, repo_path, config_path, ISSUE_HEADERS, ["aclnn接口名", "issue链接"],
"stats: issue summary {} rows from {}".format(len(rows), socket.gethostname()),
retries)
def main():
parser = argparse.ArgumentParser(
description="问题统计目录同步 GitCode 仓（Alice_zhou/issue 的 问题统计/：接口压测结果与提单汇总.xlsx + issue汇总.xlsx）")
parser.add_argument("--local-xlsx", default=DEFAULT_LOCAL_XLSX,
help="本地批量压测结果表（接口汇总数据源），默认 out_res/批量压测结果.xlsx")
parser.add_argument("--config", default=DEFAULT_CONFIG, help="配置文件路径")
parser.add_argument("--api-summary-only", action="store_true", help="仅同步 接口压测结果与提单汇总.xlsx")
parser.add_argument("--issue-summary-only", action="store_true", help="仅同步 issue汇总.xlsx")
parser.add_argument("--no-push", action="store_true", help="仅本地生成两个 xlsx，不推送")
parser.add_argument("--retries", type=int, default=3, help="并发冲突重试次数，默认 3")
parser.add_argument("--json", action="store_true", help="以 JSON 输出结果")
args = parser.parse_args()
enabled, why = issue_stats_enabled(args.config)
push = (not args.no_push) and enabled
results = {}
if not args.issue_summary_only:
results["api_summary"] = sync_api_summary(args.local_xlsx, args.config,
push=push, retries=args.retries)
if not args.api_summary_only:
results["issue_summary"] = sync_issue_summary(args.config, push=push, retries=args.retries)
if args.json:
print(json.dumps({"push": push, "switch": why,
"api_summary": {"ok": results.get("api_summary", (None,))[0],
"url": results.get("api_summary", ("", ""))[1],
"detail": results.get("api_summary", ("", "", ""))[2]},
"issue_summary": {"ok": results.get("issue_summary", (None,))[0],
"url": results.get("issue_summary", ("", ""))[1],
"detail": results.get("issue_summary", ("", "", ""))[2]}},
ensure_ascii=False))
else:
if not push:
log("STATS_SYNC", "推送关闭（{}），仅本地生成".format(why))
for name, (ok, url, detail) in results.items():
if ok and url:
log("STATS_SYNC", "{} 成功 → {} ({})".format(name, url, detail))
elif ok:
log("STATS_SYNC", "{}: {}".format(name, detail))
else:
log("STATS_SYNC", "{} 失败: {}".format(name, detail))
return 0 if all(r[0] for r in results.values()) else 1
if __name__ == "__main__":
sys.exit(main())