autool-dispatcher/scripts/analyze_non_retryable_popularity.py
2026-06-17 19:50:39 +08:00

307 lines
11 KiB
Python

#!/usr/bin/env python3
from __future__ import annotations
import argparse
import csv
import json
import shutil
import sqlite3
import sys
import tempfile
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Tuple
SCRIPT_DIR = Path(__file__).resolve().parent
PROJECT_ROOT = SCRIPT_DIR.parent
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
from config import MONITORING_DB_PATH, TASK_CSV_PATH # noqa: E402
from analytics import AnalyticsRepository # noqa: E402
from result_codes import ( # noqa: E402
APP_ERROR_DESC,
BUSINESS_ERROR_DESC,
DOWNLOAD_ERROR_DESC,
INFRA_ERROR_DESC,
AppError,
BusinessError,
DownloadError,
InfraError,
)
def _parse_downloads(value: Any) -> Optional[int]:
text = str(value or "").strip().upper()
if not text:
return None
normalized = text.replace(",", "").replace(" ", "")
if normalized.endswith("+"):
normalized = normalized[:-1]
multiplier = 1
if normalized.endswith("K"):
multiplier = 1_000
normalized = normalized[:-1]
elif normalized.endswith("M"):
multiplier = 1_000_000
normalized = normalized[:-1]
elif normalized.endswith("B"):
multiplier = 1_000_000_000
normalized = normalized[:-1]
try:
return int(float(normalized) * multiplier)
except (TypeError, ValueError):
return None
def _humanize_error_type(error_type: str) -> str:
normalized = str(error_type or "").strip()
if not normalized:
return "未标记错误"
category_name, _, code_text = normalized.partition("/")
try:
code = int(code_text)
except (TypeError, ValueError):
return normalized
try:
if category_name == "INFRA_ERROR":
return INFRA_ERROR_DESC.get(InfraError(code), normalized)
if category_name == "APP_ERROR":
return APP_ERROR_DESC.get(AppError(code), normalized)
if category_name == "BUSINESS_ERROR":
return BUSINESS_ERROR_DESC.get(BusinessError(code), normalized)
if category_name == "DOWNLOAD_ERROR":
return DOWNLOAD_ERROR_DESC.get(DownloadError(code), normalized)
except ValueError:
return normalized
return normalized
def _load_downloads_by_package(csv_path: str) -> Dict[str, Optional[int]]:
path = Path(csv_path)
if not path.exists():
return {}
result: Dict[str, Optional[int]] = {}
with path.open("r", encoding="utf-8-sig", newline="") as handle:
for row in csv.DictReader(handle):
package_name = str(row.get("package_name") or row.get("包名") or "").strip()
if not package_name or package_name in result:
continue
result[package_name] = _parse_downloads(
row.get("downloads")
or row.get("download")
or row.get("下载量")
)
return result
def _copy_sqlite_snapshot(db_path: str) -> Tuple[str, tempfile.TemporaryDirectory[str]]:
source = Path(db_path)
temp_dir: tempfile.TemporaryDirectory[str] = tempfile.TemporaryDirectory(prefix="dpi-db-snapshot-")
snapshot_path = Path(temp_dir.name) / source.name
shutil.copy2(source, snapshot_path)
for suffix in ("-wal", "-shm"):
sidecar = Path(f"{db_path}{suffix}")
if sidecar.exists():
shutil.copy2(sidecar, Path(f"{snapshot_path}{suffix}"))
return str(snapshot_path), temp_dir
def _open_readable_connection(db_path: str) -> Tuple[sqlite3.Connection, Optional[tempfile.TemporaryDirectory[str]], bool]:
try:
connection = sqlite3.connect(db_path)
connection.row_factory = sqlite3.Row
connection.execute("SELECT name FROM sqlite_master WHERE type = 'table' LIMIT 1").fetchone()
return connection, None, False
except sqlite3.Error:
snapshot_path, temp_dir = _copy_sqlite_snapshot(db_path)
connection = sqlite3.connect(snapshot_path)
connection.row_factory = sqlite3.Row
connection.execute("SELECT name FROM sqlite_master WHERE type = 'table' LIMIT 1").fetchone()
return connection, temp_dir, True
def _extract_downloads_from_payload(task_payload_json: Optional[str]) -> Optional[int]:
if not task_payload_json:
return None
try:
payload = json.loads(task_payload_json)
except (TypeError, ValueError):
return None
if not isinstance(payload, dict):
return None
original_row = payload.get("original_row")
if isinstance(original_row, dict):
for key in ("downloads", "download", "下载量"):
if key in original_row:
return _parse_downloads(original_row.get(key))
for key in ("downloads", "download", "下载量"):
if key in payload:
return _parse_downloads(payload.get(key))
return None
def build_reason_popularity_report(
db_path: str = MONITORING_DB_PATH,
csv_path: str = TASK_CSV_PATH,
*,
threshold: int = 100_000,
) -> Dict[str, Any]:
connection, temp_dir, used_snapshot = _open_readable_connection(db_path)
try:
repo = AnalyticsRepository.__new__(AnalyticsRepository)
summaries = repo._list_catalog_summaries(connection)
rows = [
item for item in summaries
if item.get("restriction_status") == "severe_restricted"
and item.get("retryability") == "non_retryable"
]
rows.sort(key=lambda item: (str(item.get("latest_failure_type") or ""), str(item.get("package_name") or "")))
finally:
connection.close()
if temp_dir is not None:
temp_dir.cleanup()
downloads_by_package = _load_downloads_by_package(csv_path)
reason_buckets: Dict[str, Dict[str, Any]] = {}
missing_download_packages: List[str] = []
matched_downloads_apps = 0
for row in rows:
package_name = str(row.get("package_name") or "").strip()
downloads = downloads_by_package.get(package_name)
if downloads is None:
downloads = _extract_downloads_from_payload(row.get("task_payload_json"))
if downloads is None and row.get("downloads") is not None:
downloads = int(row.get("downloads") or 0)
if downloads is None:
missing_download_packages.append(package_name)
continue
matched_downloads_apps += 1
error_type = str(row.get("latest_failure_type") or "").strip()
bucket = reason_buckets.setdefault(
error_type,
{
"error_type": error_type,
"reason_label": _humanize_error_type(error_type),
"hot_count": 0,
"non_hot_count": 0,
"total_count": 0,
},
)
if downloads > threshold:
bucket["hot_count"] += 1
else:
bucket["non_hot_count"] += 1
bucket["total_count"] += 1
report_rows = sorted(
reason_buckets.values(),
key=lambda item: (-int(item["total_count"]), -int(item["hot_count"]), str(item["error_type"])),
)
return {
"db_path": db_path,
"csv_path": csv_path,
"threshold": int(threshold),
"used_snapshot": used_snapshot,
"total_non_retryable_apps": len(rows),
"matched_downloads_apps": matched_downloads_apps,
"missing_downloads_apps": len(missing_download_packages),
"missing_download_packages": missing_download_packages,
"rows": report_rows,
}
def _iter_table_rows(report_rows: Iterable[Dict[str, Any]]) -> Iterable[List[str]]:
for item in report_rows:
yield [
item["reason_label"],
item["error_type"] or "-",
str(item["hot_count"]),
str(item["non_hot_count"]),
str(item["total_count"]),
]
def _format_table(report_rows: List[Dict[str, Any]]) -> str:
headers = ["失败原因", "错误码", "热门", "非热门", "总数"]
rows = list(_iter_table_rows(report_rows))
if not rows:
return "没有符合条件的应用。"
widths = [len(header) for header in headers]
for row in rows:
for index, cell in enumerate(row):
widths[index] = max(widths[index], len(cell))
def _render(row: List[str]) -> str:
return " | ".join(cell.ljust(widths[index]) for index, cell in enumerate(row))
divider = "-+-".join("-" * width for width in widths)
lines = [_render(headers), divider]
lines.extend(_render(row) for row in rows)
return "\n".join(lines)
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description="统计严重受限不可重试应用中,各失败原因对应的热门/非热门数量。"
)
parser.add_argument("--db-path", default=MONITORING_DB_PATH, help="monitoring sqlite 路径。")
parser.add_argument("--csv-path", default=TASK_CSV_PATH, help="包含 downloads 列的应用清单 CSV。")
parser.add_argument(
"--threshold",
type=int,
default=100_000,
help="热门阈值。downloads > threshold 视为热门,默认 100000。",
)
parser.add_argument(
"--format",
choices=("table", "json"),
default="table",
help="输出格式,默认 table。",
)
parser.add_argument(
"--show-missing-packages",
action="store_true",
help="额外打印缺失下载量的包名。",
)
return parser
def main() -> int:
parser = build_parser()
args = parser.parse_args()
report = build_reason_popularity_report(
db_path=args.db_path,
csv_path=args.csv_path,
threshold=args.threshold,
)
if args.format == "json":
print(json.dumps(report, ensure_ascii=False, indent=2))
return 0
print(f"数据库: {report['db_path']}")
print(f"应用清单: {report['csv_path']}")
print(f"热门阈值: downloads > {report['threshold']}")
print("统计范围: restriction_status=severe_restricted AND retryability=non_retryable")
print(f"严重受限不可重试应用总数: {report['total_non_retryable_apps']}")
print(f"已匹配下载量应用数: {report['matched_downloads_apps']}")
print(f"下载量缺失应用数: {report['missing_downloads_apps']}")
if report["used_snapshot"]:
print("读取方式: 已自动复制 live sqlite 快照后再统计")
print()
print(_format_table(report["rows"]))
if args.show_missing_packages and report["missing_download_packages"]:
print()
print("下载量缺失包名:")
for package_name in report["missing_download_packages"]:
print(package_name)
return 0
if __name__ == "__main__":
raise SystemExit(main())