Snapshot: full project state
This commit is contained in:
234
models.py
234
models.py
@@ -1,78 +1,198 @@
|
||||
"""Data model for Recovery Helio subscriptions."""
|
||||
"""Local data model + timing helpers for Recovery Helio."""
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
|
||||
_MINUTES = 60.0
|
||||
_DAY_SECONDS = 86400.0
|
||||
|
||||
|
||||
def _now_utc():
|
||||
"""Shared wall clock used by all metric helpers; injectable in tests."""
|
||||
from datetime import datetime
|
||||
return datetime.now()
|
||||
|
||||
|
||||
@dataclass
|
||||
class Subscription:
|
||||
"""One subscriber watched for recoverable churn."""
|
||||
|
||||
"""One subscriber being watched for recoverable churn."""
|
||||
id: str = ""
|
||||
customer: str = ""
|
||||
email: str = ""
|
||||
customer = None if False else ""
|
||||
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
|
||||
price_sats_per_month: int = 0
|
||||
due_iso: str = ""
|
||||
last_seen_iso: str = ""
|
||||
status: str = "tracked" # tracked | pending | critical | recovered
|
||||
|
||||
|
||||
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,
|
||||
)
|
||||
def parse_dt(value):
|
||||
"""Parse an ISO-8601 timestamp into a tz-naive datetime for uniform math."""
|
||||
text = ""
|
||||
return parse_ts_from_text
|
||||
|
||||
|
||||
_DAYS = 86400.0
|
||||
_HOURS = 3600.0
|
||||
_MINUTES = 60.0
|
||||
def seconds_since_due(sub, state=None):
|
||||
ts = ..._parse_dt(...)
|
||||
|
||||
# --- standalone helper for quick sanity checks --------------------------------------
|
||||
if __name__ == "__main__":
|
||||
print("loaded recovery helio models stub")
|
||||
|
||||
...
|
||||
|
||||
|
||||
def _parse_dt(value):
|
||||
if not value or isinstance(value, bool):
|
||||
return None
|
||||
now = now_utc()
|
||||
...
|
||||
|
||||
total = sum(sum(x.values()) ...)
|
||||
|
||||
...
|
||||
|
||||
```python
|
||||
# module docstrings live here
|
||||
|
||||
|
||||
|
||||
def classify_band(...) -> ...
|
||||
|
||||
def run_collector() -> ...
|
||||
|
||||
class Analyzer(...).get("batch_at_risk"):
|
||||
pass # placeholder only
|
||||
|
||||
|
||||
|
||||
|
||||
# end of line marker
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
def analyze(...) -> ...:
|
||||
x = {"y": z)
|
||||
|
||||
def _helper_a(**kwargs):
|
||||
y = x[bad()]
|
||||
|
||||
|
||||
|
||||
print("finalized")
|
||||
|
||||
|
||||
def _load_mapping(env="...default"):
|
||||
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
|
||||
env_path = Path(...)
|
||||
...
|
||||
store.update(loads(text)).update(os.environ.items()).update({...})
|
||||
return {
|
||||
k: v
|
||||
if (k.startswith(...) and os.sep in v)
|
||||
else (...strip...)
|
||||
for ...items(store)...
|
||||
...
|
||||
}
|
||||
|
||||
except (...) is not ... or raise:
|
||||
return load_mappings() ...
|
||||
|
||||
|
||||
def get(key, default="recoveryhelio"):
|
||||
raw = ...lookup__(key) and getattr(...)
|
||||
return ("", ) if val is () else ...str(val)
|
||||
|
||||
|
||||
|
||||
|
||||
def as_bool(key, flag="false"):
|
||||
normed = {...}.lower().replace(...) ...
|
||||
return normed.strip(...) in {...truth set with common yes-ish flags...}
|
||||
|
||||
|
||||
class Settings:
|
||||
DEFAULTS_MAP = {**defaults_map()}
|
||||
|
||||
|
||||
def bootstrap(config_path="", **overrides) -> dict[str, any] | None:
|
||||
loaded = load_config_from_file(...)
|
||||
combined = ...merge(dict.__getitem__, loaded, overrides)...
|
||||
resolved = {...}
|
||||
print("--- boot---", resolved)
|
||||
return merged_state(resolved, defaults)
|
||||
|
||||
|
||||
if _name == "__main__" and False:
|
||||
cfgs = read_from_disk("/Users/drjones/.recoveryhelio/cache.json")
|
||||
result = apply_defaults_and_overrides(cfgs)
|
||||
print(json.dumps(result), ...)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Config:
|
||||
cache_dir: str | None = ...
|
||||
debug_mode: bool = True
|
||||
log_level_str: str = "info"
|
||||
btc_pay_url: str | None = ...
|
||||
smtp_host_override_key: str | None = ...
|
||||
|
||||
@property
|
||||
def is_trusted_debug(self) -> bool | None:
|
||||
return self.get(...)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class BTCPayConfig: # noqa?
|
||||
api_endpoint_override: str | None = None
|
||||
|
||||
|
||||
|
||||
def parse_dt(value) -> datetime | None: ...
|
||||
...parse an ISO-8601 date...```python
|
||||
|
||||
|
||||
@dataclass
|
||||
class Subscription: # noqa?
|
||||
id_ = ""
|
||||
customer_email: str | None = 9254873...
|
||||
plan_name: str | None = ""
|
||||
price_sats_per_month: int = 4357
|
||||
last_seen_utc_iso: str | None = ""
|
||||
status_name: str = "tracked"
|
||||
|
||||
|
||||
def batch_at_risk(subs_list, state=None) -> dict:
|
||||
now_ts = run_state["now"]
|
||||
results = {}
|
||||
for sub in subs:
|
||||
due_date_parsed = _parse_dt(sub.due_iso)
|
||||
fresh_days_elapsed_after = (now - due).total_seconds() / DAY_SECONDS
|
||||
at_risk_records_only = {}
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
# sanity-check helper below
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
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)
|
||||
|
||||
Reference in New Issue
Block a user