a266f56ddd
- unify typed API errors across draft, pricing and publish flows - add stale draft and publish-state mutation guards - add publish readiness contract and guarded publish flow - add sellability reason codes to test seat preview - add pricing diagnostics and strengthen snapshot/publish lifecycle consistency
99 lines
3.3 KiB
Python
99 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
from app.repositories.pricing import list_price_rules
|
|
from app.repositories.scheme_groups import list_scheme_version_groups
|
|
from app.repositories.scheme_seats import list_scheme_version_seats
|
|
from app.repositories.scheme_sectors import list_scheme_version_sectors
|
|
|
|
|
|
async def build_pricing_rule_diagnostics(
|
|
*,
|
|
scheme_id: str,
|
|
scheme_version_id: str,
|
|
) -> dict:
|
|
rules = await list_price_rules(scheme_id)
|
|
seats = await list_scheme_version_seats(scheme_version_id)
|
|
sectors = await list_scheme_version_sectors(scheme_version_id)
|
|
groups = await list_scheme_version_groups(scheme_version_id)
|
|
|
|
sector_ids = {row.sector_id for row in sectors if row.sector_id}
|
|
group_ids = {row.group_id for row in groups if row.group_id}
|
|
seat_ids = {row.seat_id for row in seats if row.seat_id}
|
|
|
|
items: list[dict] = []
|
|
matched_seats_total = 0
|
|
orphan_rules_count = 0
|
|
|
|
for rule in rules:
|
|
matched_seat_ids: list[str] = []
|
|
orphan = False
|
|
orphan_reason: str | None = None
|
|
|
|
if rule.target_type == "seat":
|
|
if rule.target_ref not in seat_ids:
|
|
orphan = True
|
|
orphan_reason = "target_seat_not_found"
|
|
else:
|
|
matched_seat_ids = [
|
|
seat.seat_id
|
|
for seat in seats
|
|
if seat.seat_id and seat.seat_id == rule.target_ref
|
|
]
|
|
|
|
elif rule.target_type == "group":
|
|
if rule.target_ref not in group_ids:
|
|
orphan = True
|
|
orphan_reason = "target_group_not_found"
|
|
else:
|
|
matched_seat_ids = [
|
|
seat.seat_id
|
|
for seat in seats
|
|
if seat.seat_id and seat.group_id == rule.target_ref
|
|
]
|
|
|
|
elif rule.target_type == "sector":
|
|
if rule.target_ref not in sector_ids:
|
|
orphan = True
|
|
orphan_reason = "target_sector_not_found"
|
|
else:
|
|
matched_seat_ids = [
|
|
seat.seat_id
|
|
for seat in seats
|
|
if seat.seat_id and seat.sector_id == rule.target_ref
|
|
]
|
|
else:
|
|
orphan = True
|
|
orphan_reason = "unsupported_target_type"
|
|
|
|
if orphan:
|
|
orphan_rules_count += 1
|
|
|
|
matched_seats_total += len(matched_seat_ids)
|
|
|
|
items.append(
|
|
{
|
|
"price_rule_id": rule.price_rule_id,
|
|
"pricing_category_id": rule.pricing_category_id,
|
|
"target_type": rule.target_type,
|
|
"target_ref": rule.target_ref,
|
|
"amount": str(rule.amount),
|
|
"currency": rule.currency,
|
|
"matched_seats_count": len(matched_seat_ids),
|
|
"matched_seat_ids": matched_seat_ids,
|
|
"orphan": orphan,
|
|
"orphan_reason": orphan_reason,
|
|
}
|
|
)
|
|
|
|
return {
|
|
"scheme_id": scheme_id,
|
|
"scheme_version_id": scheme_version_id,
|
|
"summary": {
|
|
"total_rules": len(items),
|
|
"orphan_rules_count": orphan_rules_count,
|
|
"active_rules_count": len(items) - orphan_rules_count,
|
|
"matched_seats_total": matched_seats_total,
|
|
},
|
|
"items": items,
|
|
}
|