in privaterelay/management/commands/cleanup_data.py [0:0]
def get_report(self, log_message: str, data: dict) -> str:
"""Generate a human-readable report."""
verbosity: int = data["verbosity"]
cleaned: bool = data["cleaned"]
tasks: dict[str, dict[str, Any]] = data["tasks"]
# Collect task details
details: list[str] = []
sum_all_rows: list[tuple[str, ...]] = []
totals: list[float] = [0, 0, 0.0, 0.0]
for slug, task in tasks.items():
# Gather data and types
summary: dict[str, int] = task["counts"]["summary"]
needs_cleaning = summary["needs_cleaning"]
num_cleaned = summary.get("cleaned", 0)
task_timers: dict[str, float] = task["timers"]
query_timer = task_timers["query_s"]
clean_timer = task_timers.get("clean_s", 0.0)
# Collect summary data
sum_all_rows.append(
(
slug,
str(needs_cleaning),
str(num_cleaned),
format(query_timer, "0.3f"),
format(clean_timer, "0.3f"),
)
)
totals[0] += needs_cleaning
totals[1] += num_cleaned
totals[2] += query_timer
totals[3] += clean_timer
# Details are ommited on low verbosity
if verbosity <= 1:
continue
# Collect detail section data
detail = [f"## {slug}"]
detail.extend(textwrap.wrap(task["check_description"], width=80))
detail.append("")
detail.append(
f"Detected {needs_cleaning} issue{'' if needs_cleaning == 1 else 's'}"
f" in {query_timer} seconds."
)
if cleaned:
if task["can_clean"]:
detail.append(
f"Cleaned {num_cleaned} issue{'' if num_cleaned == 1 else 's'}"
f" in {clean_timer} seconds."
)
elif needs_cleaning:
detail.append("Unable to automatically clean detected items.")
detail.append("")
detail.append(task["markdown_report"])
details.append("\n".join(detail))
report = ["", "# Summary"]
if not cleaned:
report.append("Detected issues only. Use --clean to fix issues.")
report.append("")
# Pick summary table columns
if cleaned:
columns = ["task", "issues", "fixed", "query (s)", "fix (s)"]
sum_rows: list[tuple[str, ...]] = sum_all_rows[:]
sum_rows.append(
(
"_Total_",
str(totals[0]),
str(totals[1]),
format(totals[2], "0.3f"),
format(totals[3], "0.3f"),
)
)
else:
columns = ["task", "issues", "query (s)"]
sum_rows = [(row[0], row[1], row[3]) for row in sum_all_rows]
sum_rows.append(("_Total_", str(totals[0]), format(totals[2], "0.3f")))
# Determine summary table widths
widths = [len(col) for col in columns]
for row in sum_rows:
for column, value in enumerate(row):
widths[column] = max(widths[column], len(value))
# Output summary table with aligned columns
header = (
"|"
+ "|".join(f"{col:^{widths[colnum]}}" for colnum, col in enumerate(columns))
+ "|"
)
sep = (
"|"
+ "|".join(
f"{'-' * (widths[colnum] - 1)}:" for colnum, _ in enumerate(columns)
)
+ "|"
)
report.extend([header, sep])
for row in sum_rows:
row_report = (
"|"
+ "|".join(f"{col:>{widths[colnum]}}" for colnum, col in enumerate(row))
+ "|"
)
report.append(row_report)
# Output details
if verbosity > 1:
report.extend(["", "# Details"])
for detail_block in details:
report.extend([detail_block, ""])
return "\n".join(report)