"""HTML page routes — server-rendered Jinja2 with HTMX-driven partial refresh."""
from __future__ import annotations
from datetime import date, datetime, timezone
from fastapi import APIRouter, Depends, Request
from fastapi.responses import HTMLResponse, RedirectResponse
from sqlalchemy import desc, func, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.auth import CurrentUser, maybe_current_user, require_auth, require_token
from app.config import get_settings, load_groups
from app.db import get_session
from app.models import EmailSend, Referral, StrategicLog, User
from app.services.access import is_paid_active, paid_status
from app.services.referral_service import assign_code_if_missing
from app.templates_env import templates
# Router-level auth removed in favour of per-route deps so that `/` can be
# dual-purpose: logged-in users see the dashboard, logged-out visitors see
# the landing page.
router = APIRouter()
@router.get("/", response_class=HTMLResponse)
async def root_page(
request: Request,
cu: CurrentUser | None = Depends(maybe_current_user),
):
"""Dual-purpose root: dashboard when authenticated, landing otherwise."""
if cu is None:
return templates.TemplateResponse(
request, "landing.html", {"cu": None},
)
s = get_settings()
groups = load_groups(s.BASELINE_TOML, s.PORTFOLIO_TOML)
return templates.TemplateResponse(
request,
"dashboard.html",
{"groups": list(groups.keys()), "anchor": s.CASSANDRA_ANCHOR_DATE,
"cu": cu, "paid": is_paid_active(cu)},
)
@router.get(
"/news",
response_class=HTMLResponse,
dependencies=[Depends(require_token)],
)
async def news_page(request: Request):
return templates.TemplateResponse(request, "news.html", {})
@router.get("/upload", dependencies=[Depends(require_token)])
async def upload_page(request: Request):
"""Legacy bookmark — the import widget now lives in /settings."""
return RedirectResponse(url="/settings#import", status_code=302)
async def _resolve_log_date(session: AsyncSession, day: str | None) -> date:
"""If `day` is YYYY-MM-DD use it; else fall back to the date of the most
recent generated log; else today."""
if day:
try:
return datetime.strptime(day, "%Y-%m-%d").date()
except ValueError:
pass
latest = (await session.execute(
select(StrategicLog.generated_at)
.order_by(desc(StrategicLog.generated_at))
.limit(1)
)).scalar_one_or_none()
if latest is not None:
return latest.date() if hasattr(latest, "date") else latest
return datetime.now(timezone.utc).date()
def _log_page_context(target: date, paid: bool) -> dict:
s = get_settings()
return {
"selected_iso": target.isoformat(),
"selected_month": target.strftime("%Y-%m"),
"current_tone": s.CASSANDRA_TONE.upper(),
"current_analysis": s.CASSANDRA_ANALYSIS.upper(),
"paid": paid,
}
@router.get("/log", response_class=HTMLResponse)
async def log_page(
request: Request,
session: AsyncSession = Depends(get_session),
cu: CurrentUser = Depends(require_auth),
):
target = await _resolve_log_date(session, None)
return templates.TemplateResponse(
request, "log.html", _log_page_context(target, is_paid_active(cu)),
)
@router.get("/log/{day}", response_class=HTMLResponse)
async def log_page_day(
request: Request,
day: str,
session: AsyncSession = Depends(get_session),
cu: CurrentUser = Depends(require_auth),
):
target = await _resolve_log_date(session, day)
return templates.TemplateResponse(
request, "log.html", _log_page_context(target, is_paid_active(cu)),
)
@router.get("/settings", response_class=HTMLResponse)
async def settings_page(
request: Request,
session: AsyncSession = Depends(get_session),
principal: CurrentUser = Depends(require_auth),
):
"""Per-user settings. Currently shows email, tier, and the referral
block (own code + invite link + counts of pending/converted
referrals). The Credit / Paddle pieces land in D.3."""
user = principal.user
if user is None:
# Bearer-token admin path — no per-user settings to show.
return templates.TemplateResponse(
request, "settings.html",
{"user": None, "invite_url": None,
"pending_count": 0, "converted_count": 0},
)
# Lazily assign a referral code on first visit.
user = await assign_code_if_missing(session, user)
# Stats: how many people have signed up with their code so far, and
# how many of those converted (paid). D.3 will fill `converted_at`.
pending_count = (await session.execute(
select(func.count(Referral.id))
.where(Referral.referrer_user_id == user.id)
.where(Referral.converted_at.is_(None))
)).scalar() or 0
converted_count = (await session.execute(
select(func.count(Referral.id))
.where(Referral.referrer_user_id == user.id)
.where(Referral.converted_at.is_not(None))
)).scalar() or 0
invite_url = str(request.url_for("login_page")) + f"?ref={user.referral_code}"
last_email_send = (await session.execute(
select(EmailSend)
.where(EmailSend.user_id == user.id)
.order_by(desc(EmailSend.sent_at))
.limit(1)
)).scalar_one_or_none()
# Trial countdown — when the Stripe subscription is in its 14-day
# trial, show "N days remaining" on the tier row. Computed here
# rather than in the template because Jinja's date arithmetic is
# painful, and we already have to handle MariaDB's tz-naive
# round-trip via _aware-style normalisation.
trial_days_remaining: int | None = None
if user.stripe_trial_end_at is not None:
end = user.stripe_trial_end_at
if end.tzinfo is None:
end = end.replace(tzinfo=timezone.utc)
delta = end - datetime.now(timezone.utc)
if delta.total_seconds() > 0:
# Round up so the last hours of the trial still read "1 day".
trial_days_remaining = max(1, -(-int(delta.total_seconds()) // 86400))
return templates.TemplateResponse(
request, "settings.html",
{
"user": user,
"invite_url": invite_url,
"pending_count": int(pending_count),
"converted_count": int(converted_count),
"paid": paid_status(user),
"last_email_send": last_email_send,
"trial_days_remaining": trial_days_remaining,
},
)