import csv
import html
import json
import logging
import math
import os
import urllib.parse
from datetime import datetime
from statistics import median
from string import Template
import openpyxl
def _safe_nickname(nickname: str) -> str:
return "".join(
c if c.isalnum() or c in ("_", "-", "\u4e00", "\u9fa5") else "_"
for c in nickname
)
def _parse_time(value) -> float:
text = str(value or "").strip()
if not text:
return 0.0
try:
if text.endswith("Z"):
return datetime.fromisoformat(text[:-1]).timestamp()
return datetime.strptime(text, "%Y-%m-%d %H:%M:%S").timestamp()
except Exception:
try:
return datetime.fromisoformat(text).timestamp()
except Exception:
return 0.0
def read_results(
nickname: str,
output_format: str = "xlsx",
output_root: str | None = None,
) -> list[dict]:
safe_nick = _safe_nickname(nickname)
if output_root is None:
base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
output_root = os.path.join(base_dir, "results", "majsoul", safe_nick)
filepath = os.path.join(output_root, f"results.{output_format}")
if not os.path.exists(filepath):
logging.warning("No results found for %s at %s", nickname, filepath)
return []
records = []
if output_format == "csv":
with open(filepath, "r", encoding="utf-8") as f:
records.extend(csv.DictReader(f))
elif output_format == "xlsx":
wb = openpyxl.load_workbook(filepath, read_only=True, data_only=True)
try:
ws = wb.active
rows = iter(ws.iter_rows(values_only=True))
first_row = next(rows, None)
if first_row is not None:
headers = [str(value) if value is not None else "" for value in first_row]
for row in rows:
records.append(
{
header: row[index] if index < len(row) and row[index] is not None else ""
for index, header in enumerate(headers)
if header
}
)
finally:
wb.close()
else:
raise ValueError(f"Unsupported output format: {output_format}")
records.sort(key=lambda row: _parse_time(row.get("startTime") or row.get("timestamp")))
return records
def _to_float(value, *, percent: bool = False) -> float | None:
if value is None or isinstance(value, bool):
return None
text = str(value).strip()
if not text:
return None
if percent:
text = text.removesuffix("%").strip()
try:
number = float(text)
except (TypeError, ValueError):
return None
return number if math.isfinite(number) else None
def _to_int(value) -> int | None:
number = _to_float(value)
if number is None or not number.is_integer():
return None
return int(number)
def _mean(values: list[float]) -> float | None:
return sum(values) / len(values) if values else None
def _quantile(values: list[float], probability: float) -> float:
"""Return a linearly interpolated quantile for a non-empty sample."""
if not values:
raise ValueError("Quantile requires at least one value.")
if not 0 <= probability <= 1:
raise ValueError("Quantile probability must be between 0 and 1.")
ordered = sorted(values)
position = (len(ordered) - 1) * probability
lower_index = math.floor(position)
upper_index = math.ceil(position)
if lower_index == upper_index:
return ordered[lower_index]
weight = position - lower_index
return ordered[lower_index] * (1 - weight) + ordered[upper_index] * weight
def rolling_average(values: list[float | None], window: int) -> list[float | None]:
"""Return a full-window rolling mean, tolerating sparse optional values."""
if window <= 0:
raise ValueError("Rolling window must be positive.")
minimum_count = max(1, math.ceil(window * 0.6))
result = []
for index in range(len(values)):
if index + 1 < window:
result.append(None)
continue
current = [value for value in values[index - window + 1 : index + 1] if value is not None]
result.append(_mean(current) if len(current) >= minimum_count else None)
return result
def _rolling_weighted_ai(points: list[dict], window: int) -> list[float | None]:
minimum_count = max(1, math.ceil(window * 0.6))
result = []
for index in range(len(points)):
if index + 1 < window:
result.append(None)
continue
current = points[index - window + 1 : index + 1]
weighted = [
(point["aiNumerator"], point["aiDenominator"])
for point in current
if point["aiNumerator"] is not None
and point["aiDenominator"] is not None
and point["aiDenominator"] > 0
]
if len(weighted) >= minimum_count:
numerator = sum(pair[0] for pair in weighted)
denominator = sum(pair[1] for pair in weighted)
result.append(numerator / denominator * 100 if denominator else None)
continue
rates = [point["aiRate"] for point in current if point["aiRate"] is not None]
result.append(_mean(rates) if len(rates) >= minimum_count else None)
return result
def _aggregate_rate(
points: list[dict],
*,
rate_key: str,
numerator_key: str,
denominator_key: str,
) -> tuple[float | None, int | None, bool]:
weighted = [
(point[numerator_key], point[denominator_key])
for point in points
if point[numerator_key] is not None
and point[denominator_key] is not None
and point[denominator_key] > 0
]
if weighted:
numerator = sum(pair[0] for pair in weighted)
denominator = sum(pair[1] for pair in weighted)
return (numerator / denominator * 100 if denominator else None, denominator, True)
rates = [point[rate_key] for point in points if point[rate_key] is not None]
return _mean(rates), None, False
def _build_rating_batches(
points: list[dict],
overall_rating_mean: float,
*,
batch_size: int = 20,
) -> list[dict]:
"""Build newest-anchored, equal-sized review batches."""
if batch_size <= 0:
raise ValueError("Batch size must be positive.")
ranges = []
end = len(points)
while end > 0:
start = max(0, end - batch_size)
ranges.append((start, end))
end = start
ranges.reverse()
batches = []
for batch_index, (start, end) in enumerate(ranges):
current = points[start:end]
ratings = [point["rating"] for point in current]
rating_mean = _mean(ratings)
ai_rate, ai_denominator, _ = _aggregate_rate(
current,
rate_key="aiRate",
numerator_key="aiNumerator",
denominator_key="aiDenominator",
)
bad_rate_5, bad_denominator_5, _ = _aggregate_rate(
current,
rate_key="badRate5",
numerator_key="badCount5",
denominator_key="badDenominator",
)
bad_rate_10, bad_denominator_10, _ = _aggregate_rate(
current,
rate_key="badRate10",
numerator_key="badCount10",
denominator_key="badDenominator",
)
batches.append(
{
"id": f"batch-{batch_index}",
"startIndex": start,
"endIndex": end - 1,
"startLabel": points[start]["label"],
"endLabel": points[end - 1]["label"],
"label": f'{points[start]["label"]}–{points[end - 1]["label"]}',
"count": len(current),
"ratingMean": rating_mean,
"ratingDelta": (
rating_mean - overall_rating_mean
if rating_mean is not None
else None
),
"aiRate": ai_rate,
"aiDenominator": ai_denominator,
"badRate5": bad_rate_5,
"badRate10": bad_rate_10,
"badDenominator": bad_denominator_5 or bad_denominator_10,
}
)
return batches
def _infer_source(record: dict, mode: str) -> str:
source = str(record.get("source") or "").strip().lower()
if source in ("majsoul", "tenhou"):
return source
if "p-" in mode.lower():
return "tenhou"
if mode.isdigit():
return "majsoul"
return ""
def _rating_axis_bounds(ratings: list[float]) -> tuple[int, int]:
minimum = min(ratings)
maximum = max(ratings)
if minimum >= 80:
lower = 80
elif minimum >= 60:
lower = 60
elif minimum >= 40:
lower = 40
else:
lower = 0
upper = 100 if maximum <= 100 else int(math.ceil(maximum / 10) * 10)
return lower, upper
def _rate_axis_min(values: list[float]) -> int:
minimum = min(values)
if minimum >= 60:
return 60
if minimum >= 40:
return 40
return 0
def _rate_axis_scale(values: list[float]) -> tuple[float, float]:
"""Return a padded percent axis with four stable, readable intervals."""
if not values:
return 10.0, 2.5
observed_maximum = max(values)
if observed_maximum <= 0:
return 1.0, 0.25
padded_maximum = observed_maximum * 1.1
magnitude = 10 ** math.floor(math.log10(padded_maximum))
normalized = padded_maximum / magnitude
multipliers = (1.0, 1.2, 1.6, 2.0, 2.4, 3.2, 4.0, 5.0, 6.0, 8.0, 10.0)
multiplier = next(
candidate for candidate in multipliers if normalized <= candidate
)
maximum = multiplier * magnitude
interval = maximum / 4
return round(maximum, 10), round(interval, 10)
def _histogram(values: list[float], lower: int, upper: int, bins: int = 10) -> list[dict]:
width = (upper - lower) / bins
counts = [0] * bins
for value in values:
index = min(bins - 1, max(0, int((value - lower) / width)))
counts[index] += 1
return [
{
"label": f"{lower + index * width:.0f}–{lower + (index + 1) * width:.0f}",
"lower": lower + index * width,
"upper": lower + (index + 1) * width,
"count": count,
}
for index, count in enumerate(counts)
]
def prepare_dashboard_data(records: list[dict], plot_limit: int | None = None) -> dict | None:
"""Normalize result rows into a single, tested dashboard data contract."""
selected = records
if plot_limit is not None and plot_limit > 0 and len(selected) > plot_limit:
selected = selected[-plot_limit:]
points = []
for record in selected:
rating = _to_float(record.get("rating"))
if rating is None:
continue
ai_numerator = _to_int(record.get("aiConsistencyNumerator"))
ai_denominator = _to_int(record.get("aiConsistencyDenominator"))
ai_rate = _to_float(record.get("aiConsistencyRate"), percent=True)
if ai_numerator is not None and ai_denominator is not None and ai_denominator > 0:
ai_rate = ai_numerator / ai_denominator * 100
bad_denominator = _to_int(record.get("badMoveDenominator"))
bad_count_5 = _to_int(record.get("badMoveCount5"))
bad_count_10 = _to_int(record.get("badMoveCount10"))
bad_rate_5 = _to_float(record.get("badMoveRate5"), percent=True)
bad_rate_10 = _to_float(record.get("badMoveRate10"), percent=True)
if bad_denominator is not None and bad_denominator > 0:
if bad_count_5 is not None:
bad_rate_5 = bad_count_5 / bad_denominator * 100
if bad_count_10 is not None:
bad_rate_10 = bad_count_10 / bad_denominator * 100
mode = str(record.get("mode") or "—")
started_at = str(record.get("startTime") or record.get("timestamp") or "")
points.append(
{
"index": len(points) + 1,
"label": f"#{len(points) + 1}",
"startedAt": started_at,
"rating": rating,
"aiRate": ai_rate,
"aiNumerator": ai_numerator,
"aiDenominator": ai_denominator,
"badRate5": bad_rate_5,
"badCount5": bad_count_5,
"badRate10": bad_rate_10,
"badCount10": bad_count_10,
"badDenominator": bad_denominator,
"source": _infer_source(record, mode),
"mode": mode,
"modelTag": str(record.get("modelTag") or ""),
"uuid": str(record.get("uuid") or ""),
"resultUrl": str(record.get("resultUrl") or ""),
"paipuUrl": str(record.get("paipuUrl") or ""),
}
)
if not points:
return None
ratings = [point["rating"] for point in points]
rating_mean = _mean(ratings)
assert rating_mean is not None
total_games = len(points)
trend_window = 10 if total_games >= 10 else (5 if total_games >= 8 else None)
rating_rolling = (
rolling_average([point["rating"] for point in points], trend_window)
if trend_window
else [None] * total_games
)
ai_rolling = (
_rolling_weighted_ai(points, trend_window)
if trend_window
else [None] * total_games
)
comparison_window = min(20, total_games // 2) if total_games >= 10 else 0
recent_window = comparison_window or min(20, total_games)
recent_average = _mean(ratings[-recent_window:])
previous_average = (
_mean(ratings[-comparison_window * 2 : -comparison_window])
if comparison_window
else None
)
comparison_delta = (
recent_average - previous_average
if recent_average is not None and previous_average is not None
else None
)
ai_rate, ai_denominator, ai_weighted = _aggregate_rate(
points,
rate_key="aiRate",
numerator_key="aiNumerator",
denominator_key="aiDenominator",
)
bad_rate_5, bad_denominator_5, bad_weighted_5 = _aggregate_rate(
points,
rate_key="badRate5",
numerator_key="badCount5",
denominator_key="badDenominator",
)
bad_rate_10, bad_denominator_10, bad_weighted_10 = _aggregate_rate(
points,
rate_key="badRate10",
numerator_key="badCount10",
denominator_key="badDenominator",
)
rating_axis_min, rating_axis_max = _rating_axis_bounds(ratings)
ai_values = [point["aiRate"] for point in points if point["aiRate"] is not None]
ai_axis_min = _rate_axis_min(ai_values) if ai_values else 0
bad_rate_values = [
value
for point in points
for value in (point["badRate5"], point["badRate10"])
if value is not None
]
bad_rate_axis_max, bad_rate_axis_interval = _rate_axis_scale(bad_rate_values)
histogram = (
_histogram(ratings, rating_axis_min, rating_axis_max)
if total_games >= 8
else []
)
worst_games = sorted(points, key=lambda point: (point["rating"], point["index"]))[:5]
highlight_count = min(5, max(1, math.ceil(total_games * 0.05)))
highlighted = {
point["index"]
for point in sorted(points, key=lambda point: (point["rating"], point["index"]))[
:highlight_count
]
}
for point in points:
point["isLow"] = point["index"] in highlighted
dates = [point["startedAt"] for point in points if point["startedAt"]]
sources = sorted({point["source"] for point in points if point["source"]})
modes = sorted({point["mode"] for point in points if point["mode"] and point["mode"] != "—"})
model_tags = sorted({point["modelTag"] for point in points if point["modelTag"]})
return {
"points": points,
"totalGames": total_games,
"trendWindow": trend_window,
"ratingRolling": rating_rolling,
"ratingMean": rating_mean,
"aiRolling": ai_rolling,
"ratingMedian": median(ratings),
"ratingDenseLower": _quantile(ratings, 0.25),
"ratingDenseUpper": _quantile(ratings, 0.75),
"recentWindow": recent_window,
"recentAverage": recent_average,
"comparisonWindow": comparison_window,
"comparisonDelta": comparison_delta,
"aiRate": ai_rate,
"aiDenominator": ai_denominator,
"aiWeighted": ai_weighted,
"badRate5": bad_rate_5,
"badRate10": bad_rate_10,
"badDenominator": bad_denominator_5 or bad_denominator_10,
"badWeighted": bad_weighted_5 or bad_weighted_10,
"ratingAxisMin": rating_axis_min,
"ratingAxisMax": rating_axis_max,
"aiAxisMin": ai_axis_min,
"badRateAxisMax": bad_rate_axis_max,
"badRateAxisInterval": bad_rate_axis_interval,
"histogram": histogram,
"worstGames": worst_games,
"ratingBatches": _build_rating_batches(points, rating_mean),
"dateStart": dates[0] if dates else "",
"dateEnd": dates[-1] if dates else "",
"sources": sources,
"modes": modes,
"modelTags": model_tags,
}
def _format_number(value: float | None, digits: int = 1, suffix: str = "") -> str:
return "—" if value is None else f"{value:.{digits}f}{suffix}"
def _display_source(sources: list[str]) -> str:
labels = {"majsoul": "雀魂", "tenhou": "天凤"}
return " / ".join(labels.get(source, source) for source in sources) or "数据源未标注"
def _date_range(started_at: str, ended_at: str) -> str:
start = started_at[:10] if started_at else ""
end = ended_at[:10] if ended_at else ""
if not start:
return "日期未标注"
return start if start == end or not end else f"{start} — {end}"
def _safe_external_url(value: str) -> str | None:
try:
parsed = urllib.parse.urlsplit(value)
except (TypeError, ValueError):
return None
if parsed.scheme not in ("http", "https") or not parsed.netloc:
return None
return value
def _worst_game_rows(data: dict) -> str:
rows = []
for point in data["worstGames"]:
link = _safe_external_url(point["resultUrl"]) or _safe_external_url(point["paipuUrl"])
link_html = (
f'打开检讨'
if link
else '无链接'
)
rows.append(
"