yuxuanhui
42 lines
2.1 KiB
Python
42 lines
2.1 KiB
Python
"""Partition platform category counts without treating missing evidence as zero."""
|
|
|
|
from calendar import monthrange
|
|
|
|
|
|
def quarter_period(current_date):
|
|
"""Return the full calendar quarter containing the supplied date, inclusive."""
|
|
quarter = (current_date.month - 1) // 3 + 1
|
|
end_month = quarter * 3
|
|
start = current_date.replace(month=end_month - 2, day=1)
|
|
end = current_date.replace(month=end_month, day=monthrange(current_date.year, end_month)[1])
|
|
return {"quarter": f"{current_date.year}-Q{quarter}",
|
|
"start_date": start.isoformat(), "end_date": end.isoformat()}
|
|
|
|
|
|
def distribution(raw, region, delay):
|
|
"""Return three category lists; raise ValueError on incomplete or duplicate data."""
|
|
if not isinstance(raw, dict) or not isinstance(raw.get("pyramids"), list):
|
|
raise ValueError("平台未提供 Pyramid 分布")
|
|
groups = {"lit": [], "in_progress": [], "unlit": []}
|
|
seen = set()
|
|
for row in raw["pyramids"]:
|
|
if not isinstance(row, dict):
|
|
raise ValueError("平台 Pyramid 数据格式异常")
|
|
if row.get("region") != region or row.get("delay") != delay:
|
|
continue
|
|
category, count = row.get("category"), row.get("alphaCount")
|
|
if (not isinstance(category, dict)
|
|
or not isinstance(category.get("id"), str) or not category["id"]
|
|
or not isinstance(category.get("name"), str) or not category["name"]
|
|
or type(count) is not int or count < 0 or category["id"] in seen):
|
|
raise ValueError("平台分类或计数缺失、非法或重复，不能判定点塔状态")
|
|
seen.add(category["id"])
|
|
key = "lit" if count >= 3 else "in_progress" if count > 0 else "unlit"
|
|
groups[key].append({"category": {"id": category["id"], "name": category["name"]},
|
|
"alpha_count": count, "remaining": max(0, 3 - count)})
|
|
if not seen:
|
|
raise ValueError("平台未返回此 region/delay 的分类，不能认定全部未点亮")
|
|
for items in groups.values():
|
|
items.sort(key=lambda item: item["category"]["id"])
|
|
return groups
|