会員ポイントの有効期限と失効 — 目的別モデリング × ファーストクラスコレクション × FEFO（Bad→Good 8点）

黄色 = 設計意図のコメント。229 行。logs/2026-10-05-A-coding-code/points_good.py

📋 コピー"""points_good.py（改善後・Good）: 会員ポイントをロット（獲得単位）と台帳で持つ.

規約（本問の前提・仮定）:
  - 通常ポイント   : 獲得日（日本時間）から 180 日後の日の終わりまで使える。ロットごとに数える
  - 期間限定ポイント: 獲得月の翌月末日の終わりまで使える
  - 利用時は期限の近いロットから使う。期限が同じなら期間限定を先に使う
  - 利用の取消は、使ったロットへそのまま戻す。戻した時点でそのロットが期限切れなら即失効
  - 付与の取消は、そのロットの未使用分だけを取り消す。使用済みの分は「不足」として呼び出し側へ返す
    （呼び出し側は返金額から差し引く。残高をマイナスにしない）

残高は「ロットの未使用分の合計」から毎回出す（保存しない）。変更はすべて台帳に追記する。
"""

from __future__ import annotations

import calendar
from dataclasses import dataclass, field
from datetime import date, datetime, timedelta
from enum import StrEnum
from zoneinfo import ZoneInfo

JST = ZoneInfo("Asia/Tokyo")  # 期限は日本時間の「日」で数える（規約の文言どおり）
REGULAR_VALID_DAYS = 180  # 通常ポイント: 獲得日 + 180 日の日の終わりまで


class PointKind(StrEnum):
    """ポイントの種類。失効規則と消費の優先順位が種類で決まる."""

    LIMITED = "limited"  # 期間限定（キャンペーン付与）
    REGULAR = "regular"  # 通常（購入付与）


# 期限が同じロットどうしの消費順。小さいほど先に使う
_CONSUME_PRIORITY: dict[PointKind, int] = {PointKind.LIMITED: 0, PointKind.REGULAR: 1}


class EntryType(StrEnum):
    EARN = "earn"
    USE = "use"
    CANCEL_USE = "cancel_use"
    REVOKE = "revoke"
    EXPIRE = "expire"


class InsufficientPoints(Exception):
    """利用時点で使えるポイントが足りない（bool の False で返さず、理由と数を持たせる）."""

    def __init__(self, requested: int, available: int) -> None:
        super().__init__(f"利用 {requested} pt に対し使える残高は {available} pt")
        self.requested = requested
        self.available = available


def jst_date(at: datetime) -> date:
    """aware な日時を日本時間の日付にする。naive は拒否する（UTC か JST か区別できないため）.

    Raises:
        ValueError: タイムゾーンを持たない datetime が渡された場合。
    """
    if at.tzinfo is None or at.utcoffset() is None:
        raise ValueError(f"タイムゾーン付きの datetime が必要です: {at!r}")
    return at.astimezone(JST).date()


def expiry_for(kind: PointKind, earned_on: date) -> date:
    """種類と獲得日（JST）から有効期限日（この日の終わりまで使える）を出す."""
    if kind is PointKind.REGULAR:
        return earned_on + timedelta(days=REGULAR_VALID_DAYS)
    # 期間限定: 獲得月の翌月末日
    year, month = (earned_on.year + 1, 1) if earned_on.month == 12 else (earned_on.year, earned_on.month + 1)
    return date(year, month, calendar.monthrange(year, month)[1])


@dataclass(slots=True)
class Lot:
    """獲得 1 回ぶんのポイント。期限と種類はロットが持ち、残高はロットの合計から出す."""

    lot_id: str
    kind: PointKind
    earned_on: date
    expires_on: date
    granted: int
    remaining: int
    seq: int  # 獲得順（期限・種類が同じときの決定的な並び）
    revoked: bool = False

    def usable_on(self, day: date) -> bool:
        # 期限日「当日まで」使える。翌日から使えない
        # 取消済み(revoked)かは見ない: 取消後に戻った pt は不足分を返金から差し引き済みなので使えてよい
        return day <= self.expires_on and self.remaining > 0


@dataclass(frozen=True, slots=True)
class Entry:
    """台帳の 1 行。追記だけで、書き換えない。lot ごとの delta の合計 = lot.remaining."""

    entry_type: EntryType
    lot_id: str
    delta: int
    on: date
    ref: str  # 利用 ID・注文 ID など、どの操作から来たか


@dataclass(frozen=True, slots=True)
class RevokeResult:
    revoked: int  # ロットの未使用分から取り消せた pt
    shortfall: int  # すでに使われていて取り消せなかった pt（返金額から差し引く）


@dataclass(slots=True)
class PointWallet:
    """会員 1 人のポイント。ロットの集合を外へ出さず、操作はこのクラスのメソッドだけで行う."""

    member_id: str
    _lots: dict[str, Lot] = field(default_factory=dict)
    _usages: dict[str, tuple[tuple[str, int], ...]] = field(default_factory=dict)
    _ledger: list[Entry] = field(default_factory=list)

    # ------------------------------------------------------------ コマンド
    def earn(self, lot_id: str, kind: PointKind, points: int, at: datetime) -> Lot:
        """ポイントを付与する。lot_id は付与元（注文 ID・キャンペーン ID）から決める.

        Raises:
            ValueError: points が 1 未満、または同じ lot_id がすでにある場合（二重付与の防止）。
        """
        if type(points) is not int or points <= 0:
            raise ValueError(f"付与は 1 pt 以上の int: {points!r}")
        if lot_id in self._lots:
            raise ValueError(f"二重付与: {lot_id}")
        day = jst_date(at)
        lot = Lot(lot_id, kind, day, expiry_for(kind, day), points, points, seq=len(self._lots))
        self._lots[lot_id] = lot
        self._ledger.append(Entry(EntryType.EARN, lot_id, points, day, lot_id))
        return lot

    def use(self, usage_id: str, points: int, at: datetime) -> tuple[tuple[str, int], ...]:
        """ポイントを使う。期限の近いロットから順に引く（FEFO: First Expired, First Out）.

        Returns:
            (lot_id, 引いた pt) の組。取消のために保存される。

        Raises:
            InsufficientPoints: 使える残高が足りない場合（何も引かない）。
            ValueError: points が 1 未満、または usage_id が使用済みの場合。
        """
        if type(points) is not int or points <= 0:
            raise ValueError(f"利用は 1 pt 以上の int: {points!r}")
        if usage_id in self._usages:
            raise ValueError(f"同じ利用 ID: {usage_id}")
        day = jst_date(at)
        candidates = self._usable_lots(day)
        available = sum(lot.remaining for lot in candidates)
        if available < points:
            raise InsufficientPoints(points, available)

        taken: list[tuple[str, int]] = []
        rest = points
        for lot in candidates:  # 期限の近い順に並んでいる
            take = min(lot.remaining, rest)
            lot.remaining -= take
            taken.append((lot.lot_id, take))
            self._ledger.append(Entry(EntryType.USE, lot.lot_id, -take, day, usage_id))
            rest -= take
            if rest == 0:
                break
        self._usages[usage_id] = tuple(taken)
        return self._usages[usage_id]

    def cancel_use(self, usage_id: str, at: datetime) -> int:
        """利用を取り消し、使ったロットへ戻す。期限切れのロットへ戻した分は即失効させる.

        Returns:
            戻した結果、今も使える pt。
        """
        day = jst_date(at)
        restored_usable = 0
        for lot_id, amount in self._usages.pop(usage_id):
            lot = self._lots[lot_id]
            lot.remaining += amount
            self._ledger.append(Entry(EntryType.CANCEL_USE, lot_id, amount, day, usage_id))
            if day > lot.expires_on:
                self._expire_lot(lot, day)  # 新しい期限は与えない（期限を延ばす抜け道を作らない）
            else:
                restored_usable += amount
        return restored_usable

    def revoke_earn(self, lot_id: str, at: datetime) -> RevokeResult:
        """付与を取り消す（返品・注文キャンセル）。残高はマイナスにしない."""
        day = jst_date(at)
        lot = self._lots[lot_id]
        if lot.revoked:
            raise ValueError(f"取消済み: {lot_id}")
        if day > lot.expires_on:
            self._expire_lot(lot, day)  # 期限切れの分は「取消」でなく「失効」として台帳に残す（失効額の集計を狂わせない）
        revoked = lot.remaining
        lot.remaining = 0
        lot.revoked = True
        if revoked:
            self._ledger.append(Entry(EntryType.REVOKE, lot_id, -revoked, day, lot_id))
        # 不足 = このロットから引かれて、まだ取り消されていない利用の合計
        in_use = sum(amount for taken in self._usages.values() for lid, amount in taken if lid == lot_id)
        return RevokeResult(revoked=revoked, shortfall=in_use)

    def expire(self, at: datetime) -> int:
        """期限を過ぎたロットの未使用分を失効させ、台帳に記録する。何度呼んでも同じ結果."""
        day = jst_date(at)
        return sum(self._expire_lot(lot, day) for lot in self._lots.values() if day > lot.expires_on)

    # ------------------------------------------------------------ クエリ（副作用なし）
    def balance(self, at: datetime) -> int:
        """その時点で使える残高。失効処理を走らせずに、期限で数えるだけ."""
        return sum(lot.remaining for lot in self._usable_lots(jst_date(at)))

    def ledger(self) -> tuple[Entry, ...]:
        return tuple(self._ledger)

    # ------------------------------------------------------------ 内部
    def _usable_lots(self, day: date) -> list[Lot]:
        lots = [lot for lot in self._lots.values() if lot.usable_on(day)]
        # 1 会員のロット数はsim の実測で最大 68 個。ヒープより毎回ソートの方が単純
        lots.sort(key=lambda lot: (lot.expires_on, _CONSUME_PRIORITY[lot.kind], lot.seq))
        return lots

    def _expire_lot(self, lot: Lot, day: date) -> int:
        amount = lot.remaining
        if amount:
            lot.remaining = 0
            self._ledger.append(Entry(EntryType.EXPIRE, lot.lot_id, -amount, day, lot.lot_id))
        return amount