79 lines
2.2 KiB
Python
79 lines
2.2 KiB
Python
"""Data model for Recovery Helio subscriptions."""
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
|
|
|
|
@dataclass
|
|
class Subscription:
|
|
"""One subscriber watched for recoverable churn."""
|
|
|
|
id: str = ""
|
|
customer: str = ""
|
|
email: str = ""
|
|
plan_name: str = ""
|
|
price_monthly_usd: float = 0.0
|
|
state: str = "tracked" # tracked -> pending -> critical -> recovered
|
|
next_due_iso: str = "" # ISO-8601 trial/renewal expiry
|
|
last_seen_iso: str = "" # ISO-8601 last successful probe tick
|
|
|
|
|
|
def new_subscription(*, customer="", email="", plan="Starter") -> Subscription:
|
|
return Subscription(
|
|
id=uuid.uuid4().hex[:12],
|
|
customer=customer,
|
|
email=email,
|
|
plan_name=plan,
|
|
price_monthly_usd=9.0,
|
|
)
|
|
|
|
|
|
_DAYS = 86400.0
|
|
_HOURS = 3600.0
|
|
_MINUTES = 60.0
|
|
|
|
|
|
def _parse_dt(value):
|
|
if not value or isinstance(value, bool):
|
|
return None
|
|
try:
|
|
dt = datetime.fromisoformat(str(value).strip())
|
|
return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)
|
|
except Exception:
|
|
pass
|
|
raw = str(value).strip()
|
|
for fmt in ("%Y-%m-%dT%H:%M:%S%z", "%Y-%m-%d %H:%M:%S", "%Y-%m-%d"):
|
|
try:
|
|
dt = datetime.strptime(raw, fmt)
|
|
return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)
|
|
except ValueError:
|
|
continue
|
|
return None
|
|
|
|
|
|
def minutes_until_renewal(sub, now=None) -> float | None:
|
|
end = _parse_dt(sub.next_due_iso)
|
|
if end is None:
|
|
return None
|
|
when = now or datetime.now(timezone.utc)
|
|
return (end - when).total_seconds() / _MINUTES
|
|
|
|
|
|
def fresh_days_after_due(sub, clock=None) -> float | None:
|
|
end = _parse_dt(sub.next_due_iso)
|
|
if end is None:
|
|
return None
|
|
c = clock or datetime.now(timezone.utc)
|
|
seconds_since_due = max(0.0, (c - end).total_seconds())
|
|
return round(seconds_since_due / _DAYS, 2)
|
|
|
|
|
|
def hours_unseen(sub, clock=None) -> float | None:
|
|
ts = _parse_dt(sub.last_seen_iso)
|
|
if ts is None:
|
|
return None
|
|
c = clock or datetime.now(timezone.utc)
|
|
return max(0.0, (c - ts).total_seconds() / _HOURS)
|