import json import stripe import zoneinfo from datetime import datetime import requests as http_requests from django.conf import settings from django.contrib import messages from django.contrib.auth.decorators import login_required from django.db.models import Max, Q from django.http import HttpResponse, HttpResponseForbidden, HttpResponseRedirect, JsonResponse from django.shortcuts import redirect, render from django.urls import reverse from django.utils import timezone from django.views.decorators.csrf import csrf_exempt, ensure_csrf_cookie from apps.applets.utils import applet_context, apply_applet_toggle from apps.drama.models import Note from apps.epic.models import DeckVariant from apps.epic.utils import _compute_distinctions from apps.lyric.models import PaymentMethod, Purchase, ShopItem, Token, User, Wallet, is_reserved_username APPLET_ORDER = ["wallet", "username", "palette"] _BASE_UNLOCKED = frozenset([ "palette-default", "palette-cedar", "palette-oblivion-light", "palette-monochrome-dark", ]) _PALETTE_DEFS = [ {"name": "palette-default", "label": "Earthman", "locked": False}, {"name": "palette-cedar", "label": "Cedar", "locked": False}, {"name": "palette-oblivion-light", "label": "Oblivion (Light)","locked": False}, {"name": "palette-monochrome-dark","label": "Monochrome (Dark)","locked": False}, {"name": "palette-bardo", "label": "Bardo", "locked": True}, {"name": "palette-sheol", "label": "Sheol", "locked": True}, {"name": "palette-inferno", "label": "Inferno", "locked": True}, {"name": "palette-terrestre", "label": "Terrestre", "locked": True}, {"name": "palette-celestia", "label": "Celestia", "locked": True}, {"name": "palette-baltimore", "label": "Baltimore", "locked": True}, {"name": "palette-maryland", "label": "Maryland", "locked": True}, ] _NOTE_TITLES = { "stargazer": "Stargazer", "schizo": "Schizo", "nomad": "Nomad", "baltimorean": "Baltimorean", } # Keep PALETTES as an alias used by views that don't have a request user. PALETTES = _PALETTE_DEFS def _palettes_for_user(user): if not (user and user.is_authenticated): return [ dict(p, description="available by default" if not p["locked"] else "explore to unlock") for p in _PALETTE_DEFS ] granted = { r.palette: r for r in Note.objects.filter(user=user, palette__isnull=False).exclude(palette="") } result = [] for p in _PALETTE_DEFS: entry = dict(p) r = granted.get(p["name"]) if r and p["locked"]: entry["locked"] = False title = _NOTE_TITLES.get(r.slug, r.slug.capitalize()) entry["description"] = f"recognized via {title}" entry["unlocked_date"] = r.earned_at.isoformat() elif not p["locked"]: entry["description"] = "available by default" else: entry["description"] = "explore to unlock" result.append(entry) return result def _unlocked_palettes_for_user(user): base = set(_BASE_UNLOCKED) if user and user.is_authenticated: for r in Note.objects.filter(user=user, palette__isnull=False).exclude(palette=""): base.add(r.palette) return base def home_page(request): context = { "palettes": _palettes_for_user(request.user), "page_class": "page-dashboard", } if request.user.is_authenticated: context["applets"] = applet_context(request.user, "dashboard") # My Wallet applet — show all trinkets (PASS/BAND/CARTE/COIN) + stacked # Free + Tithe tokens (badge count) + Writs placeholder. Trinkets COPY # the Game Kit applet's tooltip + DON/DOFF wiring so the user can equip # from either surface; Free + Tithe tokens MOVED off the Game Kit applet # since they aren't equippable (user spec 2026-05-25 PM). user = request.user free_tokens = list(user.tokens.filter( token_type=Token.FREE, expires_at__gt=timezone.now() ).order_by("expires_at")) tithe_tokens = list(user.tokens.filter(token_type=Token.TITHE)) context.update({ "coin": user.tokens.filter(token_type=Token.COIN).first(), "pass_token": user.tokens.filter(token_type=Token.PASS).first() if user.is_staff else None, "band": user.tokens.filter(token_type=Token.BAND).first(), "carte": user.tokens.filter(token_type=Token.CARTE).first(), "free_tokens": free_tokens, "tithe_tokens": tithe_tokens, "free_count": len(free_tokens), "tithe_count": len(tithe_tokens), "equipped_trinket_id": user.equipped_trinket_id, }) return render(request, "apps/dashboard/home.html", context) # Post / Line CRUD lives in apps.billboard.views since the Post + Line models # moved to apps.billboard.models. @login_required(login_url="/") def set_palette(request): if request.method == "POST": palette = request.POST.get("palette", "") if palette in _unlocked_palettes_for_user(request.user): request.user.palette = palette request.user.save(update_fields=["palette"]) if "application/json" in request.headers.get("Accept", ""): return JsonResponse({"palette": request.user.palette}) return redirect("home") @login_required(login_url="/") def set_profile(request): if request.method == "POST": username = request.POST.get("username", "") if is_reserved_username(username, current_user=request.user): messages.error(request, "That handle is reserved.") return redirect("/") request.user.username = username request.user.save(update_fields=["username"]) return redirect("/") @login_required(login_url="/") def set_pronouns(request): from django.http import HttpResponseNotAllowed from apps.lyric.models import PRONOUN_TABLE if request.method != "POST": return HttpResponseNotAllowed(["POST"]) choice = request.POST.get("pronouns", "") if choice not in PRONOUN_TABLE: return HttpResponse(status=400) request.user.pronouns = choice request.user.save(update_fields=["pronouns"]) # bawlmorese is the pronoun-side trigger for the Baltimorean Note unlock — # mirrors sky_save's stargazer grant. Grant is idempotent (grant_if_new # returns no Brief on the second + later calls) so the 204 path resumes # naturally after the first unlock. if choice == "bawlmorese": _note, _created, brief = Note.grant_if_new(request.user, "baltimorean") if brief is not None: return JsonResponse({"brief": brief.to_banner_dict()}) return HttpResponse(status=204) @login_required(login_url="/") def toggle_applets(request): checked = request.POST.getlist("applets") apply_applet_toggle(request.user, "dashboard", checked) if request.headers.get("HX-Request"): return render(request, "apps/dashboard/_partials/_applets.html", { "applets": applet_context(request.user, "dashboard"), "palettes": _palettes_for_user(request.user), }) return redirect("home") def _shop_items_for(user): """Decorate the active ShopItem catalog w. per-user availability so the template can render `.btn-disabled` + 'Already owned' microtooltip for `max_owned`-capped items the user already holds. Items are returned in `display_order` ASC (matches the seeded `tithe-1` < `tithe-5` < `band-1`).""" items = [] for item in ShopItem.objects.filter(active=True).order_by("display_order", "slug"): item.available = item.is_available_for(user) items.append(item) return items def _free_decks_for(user): """Decorate the free-in-shop DeckVariant catalog (RWS + Fiorentine) w. a per-user `.owned` flag so the Shop applet renders a FREE ITEM btn for decks the user hasn't claimed + an 'Already owned' pill for ones already in `unlocked_decks`. Ordered by name for a stable render.""" unlocked_ids = set(user.unlocked_decks.values_list("pk", flat=True)) decks = list(DeckVariant.objects.filter(free_in_shop=True).order_by("name")) for deck in decks: deck.owned = deck.pk in unlocked_ids return decks @login_required(login_url="/") @ensure_csrf_cookie def wallet(request): free_tokens = list(request.user.tokens.filter( token_type=Token.FREE, expires_at__gt=timezone.now() ).order_by("expires_at")) tithe_tokens = list(request.user.tokens.filter(token_type=Token.TITHE)) shop_items = _shop_items_for(request.user) default_pm = request.user.payment_methods.order_by("-pk").first() return render(request, "apps/dashboard/wallet.html", { "wallet": request.user.wallet, "pass_token": request.user.tokens.filter(token_type=Token.PASS).first() if request.user.is_staff else None, "band": request.user.tokens.filter(token_type=Token.BAND).first(), "coin": request.user.tokens.filter(token_type=Token.COIN).first(), "carte": request.user.tokens.filter(token_type=Token.CARTE).first(), "shop_items": shop_items, "free_decks": _free_decks_for(request.user), "default_payment_method_id": default_pm.stripe_pm_id if default_pm else "", "stripe_publishable_key": settings.STRIPE_PUBLISHABLE_KEY, "free_tokens": free_tokens, "tithe_tokens": tithe_tokens, "free_count": len(free_tokens), "tithe_count": len(tithe_tokens), "applets": applet_context(request.user, "wallet"), "page_class": "page-wallet", }) @login_required(login_url="/") def kit_bag(request): tokens = list(request.user.tokens.all()) free_tokens = sorted( [t for t in tokens if t.token_type == Token.FREE and t.expires_at and t.expires_at > timezone.now()], key=lambda t: t.expires_at, ) tithe_tokens = [t for t in tokens if t.token_type == Token.TITHE] return render(request, "core/_partials/_kit_bag_panel.html", { "equipped_deck": request.user.equipped_deck, "equipped_trinket": request.user.equipped_trinket, "free_token": free_tokens[0] if free_tokens else None, "free_count": len(free_tokens), "tithe_token": tithe_tokens[0] if tithe_tokens else None, "tithe_count": len(tithe_tokens), }) @login_required(login_url="/") def toggle_wallet_applets(request): checked = request.POST.getlist("applets") apply_applet_toggle(request.user, "wallet", checked) if request.headers.get("HX-Request"): default_pm = request.user.payment_methods.order_by("-pk").first() return render(request, "apps/wallet/_partials/_applets.html", { "applets": applet_context(request.user, "wallet"), "wallet": request.user.wallet, "pass_token": request.user.tokens.filter(token_type=Token.PASS).first() if request.user.is_staff else None, "band": request.user.tokens.filter(token_type=Token.BAND).first(), "coin": request.user.tokens.filter(token_type=Token.COIN).first(), "carte": request.user.tokens.filter(token_type=Token.CARTE).first(), "shop_items": _shop_items_for(request.user), "free_decks": _free_decks_for(request.user), "default_payment_method_id": default_pm.stripe_pm_id if default_pm else "", "stripe_publishable_key": settings.STRIPE_PUBLISHABLE_KEY, "free_tokens": list(request.user.tokens.filter(token_type=Token.FREE)), "tithe_tokens": list(request.user.tokens.filter(token_type=Token.TITHE)), }) return redirect("wallet") @login_required(login_url="/") def setup_intent(request): stripe.api_key = settings.STRIPE_SECRET_KEY user = request.user if not user.stripe_customer_id: customer = stripe.Customer.create(email=user.email) user.stripe_customer_id = customer.id user.save(update_fields=["stripe_customer_id"]) intent = stripe.SetupIntent.create(customer=user.stripe_customer_id) return JsonResponse({ "client_secret": intent.client_secret, "publishable_key": settings.STRIPE_PUBLISHABLE_KEY, }) @login_required(login_url="/") def save_payment_method(request): stripe.api_key = settings.STRIPE_SECRET_KEY pm_id = request.POST.get("payment_method_id") pm = stripe.PaymentMethod.retrieve(pm_id) stripe.PaymentMethod.attach(pm_id, customer=request.user.stripe_customer_id) PaymentMethod.objects.create( user=request.user, stripe_pm_id=pm_id, last4=pm.card.last4, brand=pm.card.brand, ) return JsonResponse({"last4": pm.card.last4, "brand": pm.card.brand}) # ── Shop: PaymentIntent flow ───────────────────────────────────────────────── # Three endpoints split fulfillment responsibility: # /shop/buy creates a Purchase (PENDING) + a Stripe PaymentIntent. # Returns client_secret so Stripe.js can confirmCardPayment # (handles 3DS natively). # /shop/confirm sync follow-up after Stripe.js confirms client-side. Pulls # PI status from Stripe; if SUCCEEDED, calls Purchase.fulfill() # immediately (faster UX than waiting for the webhook round-trip). # /stripe/webhook async fulfillment from Stripe's webhook delivery. Same # Purchase.fulfill() call — whichever (confirm or webhook) # lands first wins; the other becomes a no-op via fulfill()'s # idempotent guard. # # Decisions locked 2026-05-21 in [[project-wallet-shop-expansion]]: # * Webhook is THE authoritative source for fulfillment (resilient to 3DS, # network drops, browser closes during checkout). # * Confirm endpoint is a UX-speedup belt-and-suspenders; never required. # * Webhook idempotency via Purchase.fulfill()'s status==SUCCEEDED guard. # * No STRIPE_LIVE_MODE setting — env-var swap is all that's needed. @login_required(login_url="/") def shop_buy(request): """Create a Stripe PaymentIntent + a PENDING Purchase row. Body: `shop_item_slug` (form-encoded). Returns: 200 `{client_secret, purchase_id}` on success; 402 if the user has no saved PaymentMethod; 404 if the slug doesn't match an active ShopItem; 409 if the item's max_owned cap is reached for this user. """ stripe.api_key = settings.STRIPE_SECRET_KEY slug = request.POST.get("shop_item_slug", "") item = ShopItem.objects.filter(slug=slug, active=True).first() if item is None: return HttpResponse(status=404) if not item.is_available_for(request.user): return HttpResponse(status=409) pm = request.user.payment_methods.order_by("-pk").first() if pm is None: return HttpResponse(status=402) intent = stripe.PaymentIntent.create( amount=item.price_cents, currency="usd", customer=request.user.stripe_customer_id, payment_method=pm.stripe_pm_id, # `automatic_payment_methods` so Stripe.js picks the right confirm # method (cards, wallets, etc.) without us hard-coding payment-method- # type plumbing. `allow_redirects=never` keeps the 3DS challenge in # the same window — Stripe.js handles the modal natively. automatic_payment_methods={"enabled": True, "allow_redirects": "never"}, metadata={ # Webhook handler looks up the Purchase by this on # `payment_intent.succeeded`. Belt-and-suspenders w. looking up # by `stripe_payment_intent_id` (also unique). "purchase_id": "_pending_", # overwritten after Purchase.save() below }, ) purchase = Purchase.objects.create( user=request.user, shop_item=item, stripe_payment_intent_id=intent.id, amount_cents=item.price_cents, granted_writs=item.granted_writs, ) # Now we have purchase.pk — backfill the metadata on the PI so the # webhook handler can resolve back to it. stripe.PaymentIntent.modify( intent.id, metadata={"purchase_id": str(purchase.pk)}, ) return JsonResponse({ "client_secret": intent.client_secret, "purchase_id": purchase.pk, }) @login_required(login_url="/") def shop_confirm(request): """Sync follow-up after Stripe.js confirms client-side. Polls the PI once + fulfills if SUCCEEDED. Idempotent w. the webhook handler via `Purchase.fulfill()`'s status guard. Body: `purchase_id` (form-encoded). Returns: 200 always (sync fulfillment is best-effort; webhook is authoritative). 404 if the purchase doesn't belong to this user. """ stripe.api_key = settings.STRIPE_SECRET_KEY purchase_id = request.POST.get("purchase_id") purchase = Purchase.objects.filter( pk=purchase_id, user=request.user, ).first() if purchase is None: return HttpResponse(status=404) if purchase.status != Purchase.SUCCEEDED: intent = stripe.PaymentIntent.retrieve(purchase.stripe_payment_intent_id) if intent.status == "succeeded": purchase.fulfill() return JsonResponse({"status": purchase.status}) @login_required(login_url="/") def shop_claim_free(request): """Claim a free ($0) deck from the Shop applet — adds the DeckVariant to the user's `unlocked_decks` so it renders in the Game Kit applet. No Stripe, no Purchase row (free decks aren't paid goods). Body: `deck_slug` (form-encoded). Returns: 200 `{owned: true, deck_name}` on claim or re-claim (M2M add is idempotent); 404 if the slug isn't a `free_in_shop` deck — the `free_in_shop` filter is the guard that stops this $0 endpoint from unlocking paid/auto-granted decks (eg Earthman). """ slug = request.POST.get("deck_slug", "") deck = DeckVariant.objects.filter(slug=slug, free_in_shop=True).first() if deck is None: return HttpResponse(status=404) request.user.unlocked_decks.add(deck) return JsonResponse({"owned": True, "deck_name": deck.name}) @csrf_exempt def stripe_webhook(request): """Stripe webhook listener. Verifies signature against `STRIPE_WEBHOOK_SECRET`; on `payment_intent.succeeded` calls `Purchase.fulfill()` (idempotent w. `/shop/confirm`). Always returns 2xx (even on unknown event types or already-fulfilled purchases) — Stripe retries on 5xx, which would just deliver the same event repeatedly. 4xx is reserved for signature mismatch (a genuine auth failure that Stripe should NOT retry). """ stripe.api_key = settings.STRIPE_SECRET_KEY payload = request.body sig_header = request.headers.get("Stripe-Signature", "") try: event = stripe.Webhook.construct_event( payload, sig_header, settings.STRIPE_WEBHOOK_SECRET, ) except (ValueError, Exception) as e: # ValueError = invalid payload; SignatureVerificationError = bad sig. # Either way, refuse — Stripe will alert if it can't deliver. if isinstance(e, ValueError) or "Signature" in type(e).__name__: return HttpResponse(status=400) raise if event["type"] == "payment_intent.succeeded": intent = event["data"]["object"] purchase_id = intent.get("metadata", {}).get("purchase_id") purchase = None if purchase_id and purchase_id.isdigit(): purchase = Purchase.objects.filter(pk=int(purchase_id)).first() # Fall-back lookup by PI ID in case metadata's missing for any reason. if purchase is None: purchase = Purchase.objects.filter( stripe_payment_intent_id=intent.get("id", ""), ).first() if purchase is not None: purchase.fulfill() return HttpResponse(status=200) # ── My Sky (personal natal chart) ──────────────────────────────────────────── def _sky_preview_data(request): """Shared preview logic — proxies to PySwiss, no DB writes.""" date_str = request.GET.get('date') time_str = request.GET.get('time', '12:00') tz_str = request.GET.get('tz', '').strip() lat_str = request.GET.get('lat') lon_str = request.GET.get('lon') if not date_str or lat_str is None or lon_str is None: return HttpResponse(status=400) try: lat = float(lat_str) lon = float(lon_str) except ValueError: return HttpResponse(status=400) if not (-90 <= lat <= 90) or not (-180 <= lon <= 180): return HttpResponse(status=400) if not tz_str: try: tz_resp = http_requests.get( settings.PYSWISS_URL + '/api/tz/', params={'lat': lat_str, 'lon': lon_str}, timeout=5, ) tz_resp.raise_for_status() tz_str = tz_resp.json().get('timezone') or 'UTC' except Exception: tz_str = 'UTC' try: tz = zoneinfo.ZoneInfo(tz_str) except (zoneinfo.ZoneInfoNotFoundError, KeyError): return HttpResponse(status=400) try: local_dt = datetime.strptime(f'{date_str} {time_str}', '%Y-%m-%d %H:%M') local_dt = local_dt.replace(tzinfo=tz) utc_dt = local_dt.astimezone(zoneinfo.ZoneInfo('UTC')) dt_iso = utc_dt.strftime('%Y-%m-%dT%H:%M:%SZ') except ValueError: return HttpResponse(status=400) try: resp = http_requests.get( settings.PYSWISS_URL + '/api/chart/', params={'dt': dt_iso, 'lat': lat_str, 'lon': lon_str}, timeout=5, ) resp.raise_for_status() except Exception: return HttpResponse(status=502) data = resp.json() if 'elements' in data and 'Earth' in data['elements']: data['elements']['Stone'] = data['elements'].pop('Earth') data['distinctions'] = _compute_distinctions(data['planets'], data['houses']) data['timezone'] = tz_str return JsonResponse(data) @login_required(login_url="/") def sky_view(request): chart_data = request.user.sky_chart_data birth_dt = request.user.sky_birth_dt saved_birth_date = '' saved_birth_time = '' if birth_dt: if request.user.sky_birth_tz: try: birth_dt = birth_dt.astimezone(zoneinfo.ZoneInfo(request.user.sky_birth_tz)) except (zoneinfo.ZoneInfoNotFoundError, KeyError): pass saved_birth_date = birth_dt.strftime('%Y-%m-%d') saved_birth_time = birth_dt.strftime('%H:%M') return render(request, "apps/dashboard/sky.html", { "preview_url": request.build_absolute_uri("/dashboard/sky/preview"), "save_url": request.build_absolute_uri("/dashboard/sky/save"), "saved_sky_json": json.dumps(chart_data) if chart_data else 'null', "saved_birth_date": saved_birth_date, "saved_birth_time": saved_birth_time, "saved_birth_place": request.user.sky_birth_place, "saved_birth_lat": request.user.sky_birth_lat, "saved_birth_lon": request.user.sky_birth_lon, "saved_birth_tz": request.user.sky_birth_tz, "page_class": "page-sky" + (" sky-saved" if chart_data else ""), }) @login_required(login_url="/") def sky_preview(request): return _sky_preview_data(request) @login_required(login_url="/") def sky_save(request): if request.method != 'POST': return HttpResponse(status=405) try: body = json.loads(request.body) except json.JSONDecodeError: return HttpResponse(status=400) user = request.user birth_tz_str = body.get('birth_tz', '').strip() birth_dt_str = body.get('birth_dt', '') if birth_dt_str: try: naive = datetime.fromisoformat(birth_dt_str.replace('Z', '').replace('+00:00', '')) if naive.tzinfo is None and birth_tz_str: try: local_tz = zoneinfo.ZoneInfo(birth_tz_str) user.sky_birth_dt = naive.replace(tzinfo=local_tz).astimezone( zoneinfo.ZoneInfo('UTC') ) except (zoneinfo.ZoneInfoNotFoundError, KeyError): user.sky_birth_dt = naive.replace(tzinfo=zoneinfo.ZoneInfo('UTC')) elif naive.tzinfo is None: user.sky_birth_dt = naive.replace(tzinfo=zoneinfo.ZoneInfo('UTC')) else: user.sky_birth_dt = naive.astimezone(zoneinfo.ZoneInfo('UTC')) except ValueError: user.sky_birth_dt = None user.sky_birth_lat = body.get('birth_lat') user.sky_birth_lon = body.get('birth_lon') user.sky_birth_place = body.get('birth_place', '') user.sky_birth_tz = body.get('birth_tz', '') user.sky_house_system = body.get('house_system', 'O') user.sky_chart_data = body.get('chart_data') user.save(update_fields=[ 'sky_birth_dt', 'sky_birth_lat', 'sky_birth_lon', 'sky_birth_place', 'sky_birth_tz', 'sky_house_system', 'sky_chart_data', ]) brief_payload = None if user.sky_chart_data: note, created, brief = Note.grant_if_new(user, "stargazer") if created and brief is not None: brief_payload = brief.to_banner_dict() return JsonResponse({"saved": True, "brief": brief_payload}) @login_required(login_url="/") def sky_delete(request): if request.method != 'POST': return HttpResponse(status=405) user = request.user user.sky_birth_dt = None user.sky_birth_lat = None user.sky_birth_lon = None user.sky_birth_place = '' user.sky_birth_tz = '' user.sky_house_system = User._meta.get_field('sky_house_system').default user.sky_chart_data = None user.save(update_fields=[ 'sky_birth_dt', 'sky_birth_lat', 'sky_birth_lon', 'sky_birth_place', 'sky_birth_tz', 'sky_house_system', 'sky_chart_data', ]) return HttpResponseRedirect(reverse('sky')) @login_required(login_url="/") def sky_data(request): user = request.user if not user.sky_chart_data: return HttpResponse(status=404) data = dict(user.sky_chart_data) data['distinctions'] = _compute_distinctions( data.get('planets', {}), data.get('houses', {}) ) return JsonResponse(data)