从零构建订阅制AI网页应用
我们将从头开始创建一个简单的AI聊天机器人网页应用（前端+后端+支付系统），该应用将包含三种订阅计划。
我们将从头开始创建一个简单的AI聊天机器人网页应用（前端+后端+支付系统），该应用将包含三种订阅计划：
- 免费计划（每小时2条消息）
- 标准计划（每月100条消息 = 5美元）
- 高级计划（无限消息/月 = 20美元）
以下是我们的网页应用架构：
我们将使用Supabase作为后端，并使用LemonSqueezy支付系统API来集成支付网关。
LemonSqueezy是一个免费平台，但它们对每次交易收取费用（5% + 小额固定费用）（金额转移到您的银行账户）。
您可以使用Paddle平台，但他们最近开始限制AI产品，所以您的网页应用可能会被拒绝。
以下是我们的完整堆栈：
- Flask（Python Web框架）
- OpenAI / Nebius / Ollama用于与LLM进行网络聊天
- Supabase（免费计划数据库用于后端）
- LemonSqueezy（作为处理我们订阅计划的支付网关）
- ngrok（测试支付系统）
所有代码都在我的GitHub存储库中。
1、开始
让我们设置项目结构。创建一个主文件夹（例如，simple-chatbot-complete），并在其中：
simple-chatbot-complete/
├── templates/
│ ├── home.html
│ ├── login.html
│ └── signup.html
├── .env
├── app.py
└── requirements.txt
现在，创建requirements.txt文件并列出我们的依赖项：
# requirements.txt
Flask
supabase
python-dotenv
openai
werkzeug
requests
一旦创建了requirements模块文件，让我们先安装它们。
pip install -r requirements.txt
接下来，我们需要创建.env文件，其中包含所有密钥。
# .env
# Supabase凭证
SUPABASE_URL='YOUR_SUPABASE_PROJECT_URL'
SUPABASE_KEY='YOUR_SUPABASE_ANON_PUBLIC_KEY'
# Nebius API密钥（或您的LLM提供商的详细信息）
NEBIUS_API_KEY='YOUR_NEBIUS_API_KEY'
NEBIUS_BASE_URL="https://api.studio.nebius.com/v1/" # 或您的提供者的基本URL
NEBIUS_MODEL="microsoft/Phi-3-mini-4k-instruct" # 或您选择的模型
# Lemon Squeezy凭证
LEMONSQUEEZY_API_KEY="YOUR_LEMONSQUEEZY_API_KEY"
LEMONSQUEEZY_STORE_ID='YOUR_LEMONSQUEEZY_STORE_ID'
LEMONSQUEEZY_STANDARD_VARIANT_ID='YOUR_STANDARD_PLAN_VARIANT_ID'
LEMONSQUEEZY_PRO_VARIANT_ID='YOUR_PRO_PLAN_VARIANT_ID'
LEMONSQUEEZY_WEBHOOK_SECRET='YOUR_LEMONSQUEEZY_WEBHOOK_SIGNING_SECRET'
LEMONSQUEEZY_CHECKOUT_LINK='YOUR_SINGLE_LEMONSQUEEZY_CHECKOUT_LINK'
别担心，我们会逐一了解这些环境变量以获取其值并理解其用途。
2、SupaBase数据库设置
让我们在Supabase中创建必要的表以存储用户数据并跟踪他们的计划状态和使用情况。
- 登录到您的Supabase仪表板。
- 选择/创建您的项目。
- 导航到左侧边栏中的SQL编辑器。
- 粘贴以下SQL代码并单击运行。
-- 创建用户表
create table users (
id uuid primary key default gen_random_uuid(), -- 唯一的用户ID（主键）
email text unique not null, -- 用户的电子邮件（必须唯一）
password_hash text not null, -- 加密密码
created_at timestamp with time zone default timezone('utc'::text, now()) not null, -- 用户注册时间
-- Lemon Squeezy集成
lemonsqueezy_customer_id text null, -- 存储来自Lemon Squeezy的相应客户ID
-- 计划状态标志（互斥）
is_free_plan boolean default true not null,
is_standard_plan boolean default false not null,
is_pro_plan boolean default false not null,
-- 使用跟踪
message_count integer default 0 not null, -- 用户发送的总消息数（可选，用于统计数据）
messages_this_hour integer default 0 not null, -- Free计划每小时限制计数器
last_message_timestamp timestamptz null, -- 上一条消息的时间戳（用于Free计划每小时检查）
messages_this_month integer default 0 not null, -- Standard计划当前计费周期内的消息计数
usage_reset_date date null -- 应重置`messages_this_month`为0的日期（对于Standard/Pro）
);
运行此命令后，我们的数据库表将被创建。让我们逐一查看创建的字段及其用途。
| 列名 | 描述 |
|---|---|
id |
用户的唯一标识符 (UUID) |
email |
用户的唯一电子邮件地址 |
password_hash |
用于安全身份验证的哈希密码 |
created_at |
用户注册的时间戳 |
lemonsqueezy_customer_id |
用户在 Lemon Squeezy 中关联的客户 ID |
is_free_plan |
指示用户是否使用免费计划的标志 |
is_standard_plan |
指示用户是否使用标准计划的标志 |
is_pro_plan |
指示用户是否使用专业计划的标志 |
message_count |
用户发送的消息总数 |
messages_this_hour |
用户在当前小时内发送的消息数（免费计划） |
last_message_timestamp |
用户最后一条消息的时间戳 |
messages_this_month |
用户当月发送的消息数量 |
usage_reset_date |
每月使用限制重置日期（标准/专业版） |
列的解释：
- 标准用户字段：
id、email、password_hash、created_at。 lemonsqueezy_customer_id：将我们的用户链接到他们的Lemon Squeezy个人资料。is_free_plan、is_standard_plan、is_pro_plan：布尔标志，便于检查用户的当前活跃计划。一次只能有一个为真。message_count：用户发送的总消息数（用于一般统计）。messages_this_hour/last_message_timestamp：结合使用以检查Free计划的每小时2条消息限制。messages_this_month：Standard计划当前计费周期内的消息计数。usage_reset_date：存储应将messages_this_month重置为0的日期。这通常基于Lemon Squeezy提供的订阅续订日期设置。
查找Supabase API密钥
您可以在项目设置中找到您的Supabase项目URL和API密钥。一旦有了它们，请将其替换到您的环境变量中。
# Supabase凭证
SUPABASE_URL='YOUR_SUPABASE_PROJECT_URL'
SUPABASE_KEY='YOUR_SUPABASE_ANON_PUBLIC_KEY'
现在我们的后端数据库已正确配置，接下来让我们配置LemonSqueezy支付计划。
3、Lemon Squeezy配置
在使用LemonSqueezy创建订阅计划之前，我们需要了解其工作原理。因此，让我们先可视化一下。
商店代表一家公司。我们可以有多个公司，但每个都需要经过LemonSqueezy的批准（您可以按照指南在他们的网站上申请您的商店批准）。
但是，我们可以使用未批准的商店在测试模式下集成支付系统。
每个商店可以有多个产品。产品可以是具有固定价格的数字产品，也可以是订阅计划。每个产品可以有多个变体，例如，订阅产品可以有：
- 变体1：标准计划
- 变体2：高级计划
同样，数字产品可以有变体，例如：
- 一种变体带有有限的一组文件。
- 另一种变体带有不同的文件访问权限。
因此，我们将创建一个产品（定价计划），并在其中可以有多个变体（标准计划、高级计划）。
- 登录到您的Lemon Squeezy仪表板。
- 转到创建的商店->产品。
- 点击“+新建产品”。
- 给它命名（例如，“定价计划”）并填写其余字段。
我添加了两个变体，默认情况下将货币设置为我的国家（PKR）以进行测试模式（如果需要，您可以将其设置为USD），还确保为每个变体设置（订阅——按收费）。
完成后，保存更改，定价计划将在产品选项卡中显示。
您可以在设置选项卡下获取您的商店ID和API密钥。如果没有API密钥，请确保创建一个。
您还需要获取两个订阅计划变体ID，可以通过点击三个点并复制每个变体ID来获得。
接下来，要获取产品的结账页面链接，请点击该产品的共享按钮并复制提供的链接。
一旦有了它们，请将相应的值替换到您的环境变量中。
# Lemon Squeezy凭证
LEMONSQUEEZY_API_KEY="YOUR_LEMONSQUEEZY_API_KEY"
LEMONSQUEEZY_STORE_ID='YOUR_LEMONSQUEEZY_STORE_ID'
LEMONSQUEEZY_STANDARD_VARIANT_ID='YOUR_STANDARD_PLAN_VARIANT_ID'
LEMONSQUEEZY_PRO_VARIANT_ID='YOUR_PRO_PLAN_VARIANT_ID'
LEMONSQUEEZY_WEBHOOK_SECRET='YOUR_LEMONSQUEEZY_WEBHOOK_SIGNING_SECRET'
LEMONSQUEEZY_CHECKOUT_LINK='YOUR_SINGLE_LEMONSQUEEZY_CHECKOUT_LINK'
# 设置为任何值（稍后会用到）
LEMONSQUEEZY_WEBHOOK_SECRET='YOUR_LEMONSQUEEZY_WEBHOOK_SIGNING_SECRET'
您可以设置webhook环境变量（测试交易）可以设置为任何值，我们将在后面使用。
4、LLM API配置
我将使用Nebius AI平台，该平台在OpenAI模块下运行，类似于其他平台如Together AI或Ollama的功能。
配置起来没什么好做的，您只需要设置基础URL并指定要集成到聊天机器人中的模型名称。
只要支持OpenAI模块，您可以使用任何LLM提供者。或者，您可以使用本地下载的Hugging Face LLM，但这需要在代码中进行一些调整。
# Nebius API Key (or your LLM provider's details)
NEBIUS_API_KEY='YOUR_NEBIUS_API_KEY'
NEBIUS_BASE_URL="https://api.studio.nebius.com/v1/" # Or your provider's base URL
NEBIUS_MODEL="microsoft/Phi-3-mini-4k-instruct" # Or your chosen model
现在我们已经配置了一切，是时候开始构建我们的Web应用程序了。
5、设置和准备
首先，我们需要让应用程序意识到我们的环境变量。
让我们导入必要的库并建立Flask应用程序与环境变量之间的连接：
# app.py
import os
import requests
import json
import hmac
import hashlib
from datetime import datetime, timedelta, timezone, date
from flask import Flask, render_template, request, redirect, url_for, session, flash, abort
from supabase import create_client, Client
from dotenv import load_dotenv
from werkzeug.security import generate_password_hash, check_password_hash
from openai import OpenAI
# 从.env文件加载环境变量
load_dotenv()
app = Flask(__name__)
# --- Supabase配置---
supabase_url = os.environ.get("SUPABASE_URL")
supabase_key = os.environ.get("SUPABASE_KEY")
supabase: Client = create_client(supabase_url, supabase_key)
# --- Nebius/OpenAI客户端配置---
NEBIUS_BASE_URL = os.environ.get("NEBIUS_BASE_URL")
NEBIUS_API_KEY = os.environ.get("NEBIUS_API_KEY")
NEBIUS_MODEL = os.environ.get("NEBIUS_MODEL")
nebius_client = OpenAI(
base_url=NEBIUS_BASE_URL,
api_key=NEBIUS_API_KEY
)
# --- Lemon Squeezy配置---
LEMONSQUEEZY_API_KEY = os.environ.get("LEMONSQUEEZY_API_KEY")
LEMONSQUEEZY_STORE_ID = os.environ.get("LEMONSQUEEZY_STORE_ID")
LEMONSQUEEZY_STANDARD_VARIANT_ID = os.environ.get("LEMONSQUEEZY_STANDARD_VARIANT_ID")
LEMONSQUEEZY_PRO_VARIANT_ID = os.environ.get("LEMONSQUEEZY_PRO_VARIANT_ID")
LEMONSQUEEZY_WEBHOOK_SECRET = os.environ.get("LEMONSQUEEZY_WEBHOOK_SECRET")
LEMONSQUEEZY_API_URL = "https://api.lemonsqueezy.com/v1"
LEMONSQUEEZY_CHECKOUT_LINK_BASE = os.environ.get("LEMONSQUEEZY_CHECKOUT_LINK", "YOUR_SINGLE_CHECKOUT_LINK_HERE") # 加载基本链接
# ---常量---
FREE_PLAN_HOURLY_LIMIT = 2
STANDARD_PLAN_MONTHLY_LIMIT = 100
# 高级计划实际上是无限的，通过is_pro_plan标志检查
所以……
- 我们导入了必要的库（
Flask、Supabase、OpenAI、requests、datetime等）。 load_dotenv()读取了我们的.env文件中的变量。- 我们使用环境变量初始化了
Supabase和Nebius客户端。包括错误处理。 - 我们加载了所有
Lemon Squeezy配置变量，包括变体ID和单个结账链接的基本URL。 - 定义了计划限制常量以提高清晰度。
6、创建Lemon Squeezy客户
在注册期间，我们希望在将用户保存到自己的数据库之前，在Lemon Squeezy中创建相应的客户记录。
这确保我们有链接以供将来订阅事件使用。让我们为此API调用创建一个辅助函数。
# app.py（继续）
# --- 辅助函数：创建Lemon Squeezy客户 ---
def create_lemon_squeezy_customer(email, name):
"""通过其API在Lemon Squeezy中创建客户记录。"""
# 构造API端点URL
customer_url = f"{LEMONSQUEEZY_API_URL}/customers"
# 设置必需的头部，包括授权使用我们的API密钥
headers = {
'Accept': 'application/vnd.api+json',
'Content-Type': 'application/vnd.api+json',
'Authorization': f'Bearer {LEMONSQUEEZY_API_KEY}'
}
# 根据Lemon Squeezy API规范定义数据负载
payload = {
"data": {
"type": "customers",
"attributes": {
"name": name, # 客户名称
"email": email, # 客户电子邮件
},
"relationships": {
"store": { # 将此客户链接到我们的特定商店
"data": {
"type": "stores",
"id": str(LEMONSQUEEZY_STORE_ID) # 商店ID需要为字符串
}
}
}
}
}
# 使用requests库向Lemon Squeezy API发送POST请求
print(f"尝试为{email}创建LS客户。")
response = requests.post(customer_url, headers=headers, json=payload)
# 检查API调用是否成功（状态码2xx）
if response.status_code == 201: # 201已创建是预期的成功代码
customer_data = response.json()
customer_id = customer_data.get("data", {}).get("id")
if customer_id:
print(f"成功创建LS客户ID：{customer_id}")
return customer_id, None # 返回新客户的ID和无错误
else:
# API成功但响应格式意外
print(f"LS成功响应缺少客户ID：{customer_data}")
return None, "无法从成功API响应中提取客户ID。"
else:
# API调用失败（状态码4xx或5xx）
print(f"创建LS客户时出错。状态：{response.status_code}，响应：{response.text}")
所以……
- 此函数
create_lemon_squeezy_customer接受用户的email和一个name（我们将从email生成这个）。 - 它定义了正确的API端点（
/v1/customers）以及所需的头部，包括我们的LEMONSQUEEZY_API_KEY进行授权。 - 它构造了包含客户详细信息的数据负载，并将其链接到我们的
LEMONSQUEEZY_STORE_ID。 - 使用
requests库，它向Lemon Squeezy API发送POST请求。 - 它检查
response.status_code。201已创建状态意味着成功。 - 如果成功，它从JSON响应中提取
customer_id并返回。 - 如果有错误（如
4xx或5xx状态代码），它尝试提取特定的错误消息。
7、用户注册路由
对于我们的Web应用程序，我们需要创建一个注册路由，允许用户创建帐户。用户数据将存储在我们的Supabase数据库中。
# app.py（继续）
# --- 路由 ---
@app.route('/signup', methods=['GET', 'POST'])
def signup():
"""处理用户注册。"""
# 如果请求是POST（用户提交了表单）
if request.method == 'POST':
# 从提交的表单数据中获取电子邮件和密码
email = request.form.get('email')
password = request.form.get('password')
# 简单验证
if not email or not password:
flash('电子邮件和密码是必需的。', 'error')
return redirect(url_for('signup'))
# 查询'users'表以检查是否存在匹配的电子邮件
existing_user_check = supabase.table('users').select('id', count='exact').eq('email', email).execute()
# 如果计数> 0，则电子邮件已注册
if existing_user_check.count > 0:
flash('电子邮件地址已注册。', 'error')
print(f"注册失败：{email} 已存在。")
return redirect(url_for('signup'))
# --- 在保存Supabase用户之前创建Lemon Squeezy客户 ---
# 从电子邮件生成简单的名称以用于LS
name = email.split('@')[0].replace('.', '').replace('+', '')
ls_customer_id, ls_error = create_lemon_squeezy_customer(email, name)
# --- 在Supabase数据库中创建用户 ---
# 在存储之前安全地哈希用户的密码
password_hash = generate_password_hash(password)
print(f"为{email}创建Supabase用户，链接到LS ID {ls_customer_id}")
# 准备新的用户行数据
user_data = {
'email': email,
'password_hash': password_hash,
'lemonsqueezy_customer_id': ls_customer_id,
'is_free_plan': True, # 默认为免费计划
'is_standard_plan': False,
'is_pro_plan': False,
'message_count': 0,
'messages_this_hour': 0,
'last_message_timestamp': None,
'messages_this_month': 0,
'usage_reset_date': None
}
# 将新用户数据插入'users'表
insert_result = supabase.table('users').insert(user_data).execute()
# 如果请求是GET（用户只是访问页面），则呈现注册表单
return render_template('signup.html')
所以注册过程如下……
- 路由监听
/signup用于GET（显示表单）和POST（处理注册）请求。在POST时，它从表单中获取email和password。 - 它检查
Supabase（supabase.table('users').select(...)）以查看是否存在具有该电子邮件的用户。 - 关键的是，它接下来调用了我们的
create_lemon_squeezy_customer辅助函数。我们希望确保在提交到自己的数据库之前，支付系统档案已创建。 - 如果Lemon Squeezy步骤成功，它使用
Werkzeug的安全方法generate_password_hash对密码进行哈希处理。 - 它准备了一个
user_data字典，包含所有必要的字段以供我们的Supabase用户表使用，默认用户为is_free_plan。 - 它将这些数据插入Supabase（
supabase.table('users').insert(...)）。如果插入成功，它将用户重定向到登录页面（url_for('login')）并显示成功消息（flash）。 - 如果任何步骤失败，它显示错误消息（
flash）并重定向回注册页面。 - 对于
GET，它简单地显示signup.html模板。
8、用户登录路由
既然我们已经编写了注册路由，该路由还创建了一个LemonSqueezy用户并将ID链接起来，接下来我们需要为该用户创建登录路由。
# app.py（继续）
@app.route('/login', methods=['GET', 'POST'])
def login():
"""处理用户登录。"""
if request.method == 'POST':
email = request.form.get('email')
password = request.form.get('password')
if not email or not password:
flash('电子邮件和密码是必需的。', 'error')
return redirect(url_for('login'))
print(f"登录尝试：{email}")
# --- 从Supabase获取用户 ---
# 选择与电子邮件匹配的用户，包括其ID和哈希密码
user_query = supabase.table('users').select('id, email, password_hash').eq('email', email).execute()
# 检查是否找到任何用户（user_query.data将是列表）
if user_query.data:
user = user_query.data[0] # 获取找到的第一个（并且应该是唯一的）用户
# --- 验证密码 ---
# 将提供的密码与安全存储的哈希进行比较
if check_password_hash(user['password_hash'], password):
# 密码匹配！将用户信息存储在会话中
session['user_email'] = user['email']
session['user_id'] = user['id'] # 存储Supabase用户ID
flash('登录成功！', 'success')
# 重定向到主要聊天机器人主页
return redirect(url_for('home'))
else:
# 密码不匹配
flash('无效的电子邮件或密码。', 'error')
return redirect(url_for('login'))
else:
# 没有找到具有该电子邮件的用户
print(f"登录失败：{email}：未找到用户")
flash('无效的电子邮件或密码。', 'error') # 使用通用错误以确保安全性
return redirect(url_for('login'))
# 渲染登录页面以处理GET请求
return render_template('login.html')
登录过程非常简单，这是它的运作方式……
- 监听
/login用于GET和POST。 - 在
POST时，获取email和password。 - 查询
Supabase以获取与email匹配的用户，检索其id和password_hash。 - 如果找到用户，它使用
check_password_hash安全地比较提交的密码与存储的哈希。 - 如果密码匹配，它将
user_email和user_id存储在Flask会话中。会话安全地记住用户跨不同页面请求已登录。 - 成功时重定向到主页（
url_for('home')）。 - 显示错误消息以处理无效凭据或找不到用户的情况。
- 对于
GET，它显示login.html模板。
9、用户注销路由
我们还需要一个简单的注销路由，允许用户从我们的Web应用程序中注销。
# app.py（继续）
@app.route('/logout')
def logout():
"""处理用户注销。"""
user_email = session.get('user_email', 'Unknown User') # 在删除之前获取电子邮件以进行日志记录
# 从会话中移除用户信息
session.pop('user_email', None)
session.pop('user_id', None)
print(f"用户{user_email}已注销。")
flash('您已成功注销。', 'success')
# 将用户重定向回登录页面
return redirect(url_for('login'))
- 监听
/logout。 - 它使用
session.pop()从会话数据中删除user_email和user_id键，有效注销用户。 - 显示确认消息。
- 将用户重定向到登录页面。
10、数据获取与每月重置
现在我们需要编码一个主要路由，用户在登录后在此与聊天机器人交互。它处理显示界面（GET请求）和处理消息（POST请求）。
首先，我们定义路由并确保用户确实已登录，通过检查会话。
# app.py（继续）
@app.route('/', methods=['GET', 'POST'])
def home():
"""显示主要聊天机器人页面并处理消息发送。"""
# --- 1. 检查用户是否已登录 ---
# 如果会话中没有'user_id'，他们未登录
if 'user_id' not in session:
flash('请登录以访问聊天机器人。', 'error')
return redirect(url_for('login')) # 将他们送回登录页面
# 从会话中获取已登录用户的ID
user_id = session['user_id']
print(f"访问主页的用户ID：{user_id}")
# 初始化稍后需要的模板变量
bot_response = None
user_message = None
message_limit_reached = False
plan_name = "Free" # 合理的默认值
current_limit_info = f"{FREE_PLAN_HOURLY_LIMIT}/小时限制"
# --- 2. 从Supabase获取当前用户数据 ---
# 我们需要用户的电子邮件、计划状态和当前使用计数
# 使用.single()因为我们期望为登录ID正好有一行用户数据
user_data_query = supabase.table('users').select(
'email, message_count, is_free_plan, is_standard_plan, is_pro_plan, '
'messages_this_hour, last_message_timestamp, messages_this_month, usage_reset_date'
).eq('id', user_id).single().execute()
# 处理用户数据可能缺失的情况（例如，数据库问题）
if not user_data_query.data:
flash('无法获取您的用户数据。请重新登录。', 'error')
print(f"错误：无法为用户ID:{user_id}获取数据")
session.clear() # 如果数据损坏/丢失，注销用户
return redirect(url_for('login'))
user_data = user_data_query.data
# print(f"获取的用户数据：{user_data}") # 可选：记录获取的数据（注意敏感信息）
该路由检查用户是否通过会话登录，未登录则重定向到登录页面。然后查询Supabase用户表以获取登录用户的计划和使用限制，处理查询失败的情况。
接下来，我们将获取的数据分配给局部变量以便更轻松地访问，并在需要时转换数据类型（如将时间戳字符串转换为datetime对象）。我们还实现了自动重置付费计划每月消息计数的逻辑。
# app.py（从第1部分继续）
# --- 3. 分配局部变量并转换类型 ---
# 将从获取的字典中分配数据到Python变量
user_email = user_data['email']
is_free_plan = user_data['is_free_plan']
is_standard_plan = user_data['is_standard_plan']
is_pro_plan = user_data['is_pro_plan']
# 这些计数器可能会被重置逻辑或消息发送修改
message_count = user_data['message_count']
messages_this_hour = user_data['messages_this_hour']
messages_this_month = user_data['messages_this_month']
# 从数据库中获取字符串表示形式
last_message_timestamp_str = user_data['last_message_timestamp']
usage_reset_date_str = user_data['usage_reset_date']
# 如果存在，则将字符串转换为实际的datetime / date对象
last_message_timestamp = datetime.fromisoformat(last_message_timestamp_str) if last_message_timestamp_str else None
usage_reset_date = date.fromisoformat(usage_reset_date_str) if usage_reset_date_str else None
print(f"用户计划状态：Free={is_free_plan}, Std={is_standard_plan}, Pro={is_pro_plan}")
print(f"使用计数：小时={messages_this_hour}, 月={messages_this_month}, 重置日期={usage_reset_date}")
# --- 4. 检查并执行每月使用重置 ---
today = date.today() # 获取今天的日期
# 仅当用户处于Standard或Pro且设置了重置日期且今天等于或晚于该日期时才检查
if (is_standard_plan or is_pro_plan) and usage_reset_date and today >= usage_reset_date:
print(f"--- 为用户ID:{user_id}重置每月使用情况 ---")
print(f"今天({today})>=重置日期({usage_reset_date})")
# 首先在本地变量中重置计数器
messages_this_month = 0
# 准备数据库更新
reset_update_payload = {'messages_this_month': 0}
# 在后台更新数据库
print(f"更新DB：将messages_this_month重置为0")
update_reset = supabase.table('users').update(reset_update_payload).eq('id', user_id).execute()
user_data字典被解包，将last_message_timestamp和usage_reset_date转换为Python的datetime和date对象以进行准确比较。
如果用户是付费计划，重置日期已过期，并且条件满足，消息计数器将被重置，下一个重置日期由Lemon Squeezy webhook设置。
11、确定计划信息 & 限制
现在我们根据标志确定用户友好的计划信息和限制字符串，然后开始处理用户实际上提交了消息的情况（POST请求）。
# app.py（从第2部分继续）
# --- 5. 确定当前计划信息以显示 ---
# 根据计划标志设置用户友好的字符串
if is_pro_plan:
plan_name = "Pro"
current_limit_info = "无限制消息"
elif is_standard_plan:
plan_name = "Standard"
# 使用刚刚可能已重置的'messages_this_month'值
current_limit_info = f"{messages_this_month}/{STANDARD_PLAN_MONTHLY_LIMIT} 条消息本月"
else: # Free计划（默认如果不是Pro或Standard）
plan_name = "Free"
current_limit_info = f"{FREE_PLAN_HOURLY_LIMIT}/小时消息限制"
print(f"用户ID {user_id}处于{plan_name}计划。UI限制信息：{current_limit_info}")
# --- 6. 处理消息提交（POST请求）---
if request.method == 'POST':
# 用户提交了聊天表单
user_message = request.form.get('user_input') # 从文本区域获取文本
print(f"用户{user_id}（{plan_name}）提交消息：'{user_message[:50]}...'")
# 仅在消息不为空且AI客户端准备好时继续
if user_message and nebius_client:
allow_message = True # 默认假设消息允许
now_dt = datetime.now(timezone.utc) # 获取当前UTC时间以进行比较
usage_update_payload = {} # 准备一个空字典以保存此消息所需的DB更新
# --- 检查计划限制 --- （这部分将在下一节详细说明）
根据用户的计划标志（is_pro_plan、is_standard_plan、is_free_plan），设置plan_name和current_limit_info，并在Standard计划中显示更新后的messages_this_month。
如果请求方法是POST，它处理用户的聊天消息，检查非空消息和成功的AI客户端初始化，然后准备潜在的数据库更新。
在POST处理程序内部，这就是我们强制执行每个订阅计划规则的地方。
# app.py（在`if request.method == 'POST'`块内```
# app.py（接续自第5部分，现在在 `if request.method == 'POST'` 块之外）
# --- 7. 准备单一的带预填的结账链接 ---
# 如果未正确配置，则默认为 '#'
upgrade_checkout_link = "#"
# 检查基础链接是否已设置且不是占位符值
if LEMONSQUEEZY_CHECKOUT_LINK_BASE and LEMONSQUEEZY_CHECKOUT_LINK_BASE != "YOUR_SINGLE_CHECKOUT_LINK_HERE":
# 使用 urllib.parse.quote 安全地处理特殊字符
encoded_email = quote(user_email)
# 将电子邮件作为查询参数附加到链接中以进行 Lemon Squeezy 预填
upgrade_checkout_link = f"{LEMONSQUEEZY_CHECKOUT_LINK_BASE}?checkout[email]={encoded_email}"
print(f"生成的结账链接: {upgrade_checkout_link}")
else:
print("警告：Lemon Squeezy 结账链接未在 .env 中配置。升级按钮将无法正常工作。")
# --- 8. 渲染页面 ---
# 将所有计算和获取的数据传递给 Jinja 模板
print(f"正在渲染 home 模板，用户 {user_id}。计划: {plan_name}。当前限制信息: '{current_limit_info}'")
return render_template('home.html',
user_email=user_email,
plan_name=plan_name,
current_limit_info=current_limit_info, # 用户友好的限制字符串
is_free_plan=is_free_plan, # 传递布尔标志供模板逻辑使用
is_standard_plan=is_standard_plan,
is_pro_plan=is_pro_plan,
user_message=user_message, # 最近的消息（如果是 POST）
bot_response=bot_response, # AI 的响应（如果 POST 成功）
message_limit_reached=message_limit_reached, # 禁用发送按钮的布尔值
upgrade_checkout_link=upgrade_checkout_link, # 单一的带预填的结账链接
STANDARD_PLAN_MONTHLY_LIMIT=STANDARD_PLAN_MONTHLY_LIMIT # 用于显示的常量
)
代码首先通过使用 urllib.parse.quote 对用户的电子邮件进行编码，并将其附加到基础结账 URL（LEMONSQUEEZY_CHECKOUT_LINK_BASE）上来准备结账链接。
如果链接未正确配置，则会打印警告。最后，它调用了 Flask 的 render_template 函数，传递 home.html 作为模板以及所有必要的变量作为关键字参数，允许 Flask 的 Jinja 引擎动态生成用户浏览器所需的 HTML。
这完成了主要的应用程序路由（/）。
12、Lemon Squeezy Webhook 处理器
此路由不会直接与用户的浏览器交互。相反，它作为一个监听器，等待来自 Lemon Squeezy 平台的通知（Webhooks），当我们的商店中发生重要订阅事件时（例如新订阅开始、付款成功或订阅取消），这些事件会直接通知我们。
处理 Webhooks 最关键的部分是确保它们确实来自 Lemon Squeezy 而不是恶意行为者试图伪造升级。这是通过一个签名密钥来完成的。
# app.py（接续）
# --- Lemon Squeezy Webhook 处理器 ---
@app.route('/webhook/lemonsqueezy', methods=['POST'])
def lemonsqueezy_webhook():
"""处理来自 Lemon Squeezy 的传入 Webhook。"""
# --- 1. 验证 Webhook 签名（安全检查！）---
print("--- 接收到 /webhook/lemonsqueezy 的传入请求 ---")
# 获取我们在 .env 文件中存储的密钥
secret = LEMONSQUEEZY_WEBHOOK_SECRET
# Lemon Squeezy 在 'X-Signature' 标头中发送签名
signature = request.headers.get('X-Signature')
if not signature:
print("Webhook 错误：请求缺少 X-Signature 标头。")
# 返回 400 错误，因为请求格式不正确
abort(400)
# 我们需要原始请求体（作为字节）以正确计算签名
payload = request.get_data()
# 使用 HMAC-SHA256 算法计算预期签名
# 使用密钥和原始负载
computed_hash = hmac.new(secret.encode('utf-8'), payload, hashlib.sha256)
digest = computed_hash.hexdigest() # 以十六进制字符串形式返回的签名
# 如果到达这里，则签名有效！
print("Webhook 签名验证成功。")
# --- 2. 处理已验证的有效负载 --- （详细说明如下）
# ... Webhook 逻辑的其余部分 ...
该路由监听 /webhook/lemonsqueezy 的 POST 请求，并从环境变量中检索 LEMONSQUEEZY_WEBHOOK_SECRET。
如果缺少密钥，则返回 500 错误。它检查 X-Signature 标头；如果不存在，则返回 400 错误。
原始请求体作为字节被检索出来，并使用 Python 的 hmac 和 hashlib 计算预期的 HMAC-SHA256 签名，确保请求有效。
一旦签名验证通过，我们需要理解通知的内容。我们解析 JSON 数据并识别发生了什么类型的事件。
# app.py（继续在 lemonsqueezy_webhook 函数内部，签名验证之后）
# --- 2. 处理已验证的有效负载 ---
# 将原始负载（字节）解码为 Python 字典
event_data = json.loads(payload.decode('utf-8'))
print(f"Webhook 负载成功解析为 JSON。")
# 从标头和有效负载元数据中获取事件详细信息以进行日志记录/逻辑处理
event_name = request.headers.get('X-Event-Name', 'unknown_event')
webhook_id = event_data.get('meta', {}).get('webhook_id', 'N/A') # 用于跟踪
# 获取包含事件特定详细信息的 'data' 对象
data_obj = event_data.get('data')
# 获取 'attributes' 字典内的 'data'
attributes = data_obj.get('attributes', {})
# 获取 Lemon Squeezy 客户端 ID（用于查找我们的用户）
ls_customer_id = attributes.get('customer_id')
ls_customer_id_str = str(ls_customer_id) # 确保它是字符串
# --- 3. 处理特定的订阅事件 --- （详细说明如下）
# ... 基于 event_name 的逻辑 ...
原始负载被解析为字典。事件名称和 webhook_id 被提取用于处理或调试。customer_id 被提取，转换为字符串，并用于在 Supabase 中找到对应的用户。
然后是处理用户购买计划或其订阅续订的核心逻辑。我们需要更新数据库中的用户计划状态并重置其每月使用量。
# app.py（继续在 lemonsqueezy_webhook 函数内部）
# --- 3. 处理特定的订阅事件 ---
try: # 添加 try 块以捕获事件处理逻辑中的错误
# --- 事件：订阅创建或更新（变为活跃） ---
# 检查事件是否表示活跃订阅开始或继续
if event_name in ['subscription_created', 'subscription_updated']:
# 提取与订阅事件相关的详细信息
variant_id = attributes.get('variant_id')
renews_at_str = attributes.get('renews_at') # 下一次续订的时间戳
status = attributes.get('status') # 当前订阅的状态
print(f"正在处理 {event_name}，LS 客户端 {ls_customer_id_str}。状态: {status}")
# 我们只希望在订阅处于 'active' 状态时授予访问权限并重置限制
if status == 'active':
if not variant_id:
print(f"Webhook 警告 ({webhook_id})：'variant_id' 缺失。")
return '警告：缺少 variant ID', 202
variant_id_str = str(variant_id) # 确保字符串比较
# --- 计算下一次重置日期 ---
next_reset_date = None
if renews_at_str:
# 解析来自 Lemon Squeezy 的 ISO 时间戳字符串（通常为 UTC，以 'Z' 结尾）
# 转换为时区感知的 datetime 对象，然后仅获取日期部分
next_reset_date = datetime.fromisoformat(renews_at_str.replace('Z', '+00:00')).astimezone(timezone.utc).date()
print(f" 计算出的下一次重置日期: {next_reset_date}")
else:
# 如果 'renews_at' 缺失（订阅活跃时应很少见），创建一个回退值
next_reset_date = date.today() + timedelta(days=30) # 大约 1 个月
print(f" 警告：'renews_at' 缺失。使用回退重置日期: {next_reset_date}")
# --- 准备 Supabase 更新有效负载 ---
# 此字典包含我们想要对用户记录进行的更改
update_payload = {
'is_free_plan': False, # 他们正在使用付费计划
'messages_this_month': 0, # 重置月度计数器
'messages_this_hour': 0, # 重置小时计数器（未使用，但良好实践）
'last_message_timestamp': None, # 清除小时时间戳
'usage_reset_date': next_reset_date.isoformat() if next_reset_date else None # 存储 YYYY-MM-DD
}
# --- 根据 Variant ID 设置正确的计划标志 ---
# 比较来自 Webhook 的 variant ID 与我们在 .env 中存储的 ID
if variant_id_str == str(LEMONSQUEEZY_STANDARD_VARIANT_ID):
print(f" 设置为标准计划（Variant ID: {variant_id_str}）")
update_payload['is_standard_plan'] = True
update_payload['is_pro_plan'] = False
elif variant_id_str == str(LEMONSQUEEZY_PRO_VARIANT_ID):
print(f" 设置为专业计划（Variant ID: {variant_id_str}）")
update_payload['is_standard_plan'] = False
update_payload['is_pro_plan'] = True
else:
# 如果 variant ID 不匹配我们已知的计划
print(f" 忽略未知的活动 variant ID: {variant_id_str}")
return '事件针对未知 variant 被忽略', 200 # 确认，但不采取任何操作
# --- 应用更新到 Supabase ---
# 查找客户端 ID 并应用更新
print(f" 正在尝试对 LS 客户端 {ls_customer_id_str} 进行 Supabase 更新")
update_result = supabase.table('users').update(update_payload).eq('lemonsqueezy_customer_id', ls_customer_id_str).execute()
except Exception as e: # 捕获事件处理逻辑中的错误
print(f"Webhook 错误 ({webhook_id})：未处理的异常处理事件 {event_name}: {e}")
import traceback
traceback.print_exc() # 记录完整的错误堆栈跟踪以供调试
abort(500) # 信号内部服务器错误
事件处理被包裹在一个 try-except 块中，以捕获错误。它检查事件是否为 subscription_created 或 subscription_updated，并提取相关细节，包括 variant_id、renews_at 和状态。
只有活跃订阅触发访问授权或限制重置。next_reset_date 是根据 renews_at 计算出来的。update_payload 被准备用来重置使用数据并根据 variant_id 设置正确的计划。
数据库通过这个 payload 进行更新，结果会被检查以返回适当的 HTTP 状态码（200 表示成功，202 表示用户未找到，500 表示错误）。
最后，在脚本执行时添加标准代码以运行 Flask 开发服务器。
# app.py（接续）
# --- 运行 Flask 应用 ---
if __name__ == '__main__':
# host='0.0.0.0' 使服务器可以在本地网络上访问
# debug=True 启用自动重新加载并提供详细的错误页面（生产环境中禁用！）
print("--- 启动 Flask 开发服务器 ---")
print("访问地址：http://127.0.0.1:5000（或您的本地 IP）")
app.run(debug=True, host='0.0.0.0', port=5000)
所以现在我们已经实现了所有的后端逻辑，是时候开始构建前端网页了。
13、创建前端模板
让我们先创建一个简单的注册页面。
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>注册</title>
</head>
<body>
<h1>注册</h1>
<!-- 显示闪现消息（如错误或成功）-->
{% with messages = get_flashed_messages(with_categories=true) %}
{% if messages %}
{% for category, message in messages %}
<div class="flash {{ category }}">{{ message }}</div>
{% endfor %}
{% endif %}
{% endwith %}
<!-- 注册表单 -->
<form method="post">
<label for="email">电子邮件：</label>
<input type="email" id="email" name="email" required>
<label for="password">密码：</label>
<input type="password" id="password" name="password" required>
<button type="submit">注册</button>
</form>
<p>已有账号？<a href="{{ url_for('login') }}">登录</a></p>
</body>
</html>
它的工作方式非常简单：用户注册，账户创建完成后，用户会被重定向到登录页面，并显示一条“账户创建成功”的闪现消息。
登录页面也很简单，但对于登录，我们需要检查用户是否存在于数据库中。这需要一些逻辑来验证用户凭据，然后再授予访问权限。
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>登录</title>
</head>
<body>
<h1>登录</h1>
<!-- 显示闪现消息 -->
{% with messages = get_flashed_messages(with_categories=true) %}
{% if messages %}
{% for category, message in messages %}
<div class="flash {{ category }}">{{ message }}</div>
{% endfor %}
{% endif %}
{% endwith %}
<!-- 登录表单 -->
<form method="post">
<label for="email">电子邮件：</label>
<input type="email" id="email" name="email" required>
<label for="password">密码：</label>
<input type="password" id="password" name="password" required>
<button type="submit">登录</button>
</form>
<p>没有账号？<a href="{{ url_for('signup') }}">注册</a></p>
</body>
</html>
是的，它也非常简单，当找不到用户时，会出现一条闪现消息。
现在，我们需要编写聊天应用程序用户界面的 HTML 模板。这将包括聊天窗口布局、输入字段、发送按钮以及其他用于聊天机器人交互的元素。
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>聊天机器人主页</title>
</head>
<body>
<!-- 显示用户信息、计划、使用情况的头部部分 -->
<div class="header">
<span>欢迎，{{ user_email }}!</span>
<span class="plan {{ plan_name.lower() }}">计划：{{ plan_name }}</span>
<span class="usage-info">使用情况：{{ current_limit_info }}</span>
<a href="{{ url_for('logout') }}" class="logout-btn">退出</a>
</div>
<!-- 闪现消息区域 -->
{% with messages = get_flashed_messages(with_categories=true) %}
{% if messages %}
{% for category, message in messages %}
<div class="flash {{ category }}">{{ message }}</div>
{% endfor %}
{% endif %}
{% endwith %}
<!-- 升级部分 - 仅在免费或标准计划时显示 -->
{% if is_free_plan or is_standard_plan %}
<div class="upgrade-section">
{% if is_free_plan %}
<p>升级你的计划以获得更多消息！</p>
<div class="upgrade-options">
<!-- 两个按钮使用相同的结账链接 -->
<a href="{{ upgrade_checkout_link }}" target="_blank" class="upgrade-btn standard">
升级到标准计划（{{ STANDARD_PLAN_MONTHLY_LIMIT }}/月）
</a>
<a href="{{ upgrade_checkout_link }}" target="_blank" class="upgrade-btn pro">
升级到专业计划（无限制）
</a>
</div>
<p style="font-size: 0.8em; margin-top: 10px; color: #6c757d;">
请确保在结账时使用电子邮件 '{{ user_email }}'。
</p>
{% elif is_standard_plan %}
<p>升级到专业计划以获得无限消息！</p>
<div class="upgrade-options">
<!-- 仅在已经是标准计划时显示专业升级选项 -->
<a href="{{ upgrade_checkout_link }}" target="_blank" class="upgrade-btn pro">
升级到专业计划
</a>
</div>
<p style="font-size: 0.8em; margin-top: 10px; color: #6c757d;">
请确保在结账时使用电子邮件 '{{ user_email }}'。
</p>
{% endif %}
</div>
{% endif %}
<!-- 升级部分结束 -->
<h1>简单聊天机器人</h1>
<!-- 聊天输入表单 -->
<form method="post">
<textarea name="user_input"
placeholder="向机器人提问..."
required
aria-label="聊天输入">{{ user_message or '' }}</textarea>
{# 如果消息限制达到则禁用按钮 #}
<button type="submit" {% if message_limit_reached %}disabled title="消息限制已达到"{% endif %}>
发送
</button>
</form>
<!-- 聊天显示区域 -->
<div class="chat-area" aria-live="polite">
{# 如果在此次请求中发送了用户消息，则显示 #}
{% if user_message and not message_limit_reached %}
<p><span class="user">你：</span> {{ user_message | escape }}</p>
{% endif %}
{# 如果生成了机器人响应，则显示 #}
{% if bot_response %}
<p><span class="bot">机器人：</span> {{ bot_response | escape }}</p>
{% endif %}
{# 初始状态消息或限制达到消息 #}
{% if not user_message and not bot_response and not message_limit_reached %}
<p>在上方问聊天机器人一个问题。</p>
{% elif message_limit_reached %}
<p style="color: red; font-weight: bold;">您的消息限制已在当前计划/周期内达到。</p>
{% endif %}
</div>
</body>
</html>
因此，我们的主页模板将显示用户当前的活跃计划，以及该用户的消息计数和其他基本信息。这将为用户提供对其订阅和使用的概览。
我们离测试应用程序非常接近了！最后一步是使用 ngrok 托管我们的应用程序，并创建一个 Webhook 事件链接，以帮助交易成功。
14、创建 Webhook 事件
Lemon Squeezy Webhook 需要一个公开可访问的 URL 来向您的 Flask 应用发送通知。
在本地开发时，您的应用程序通常只能在 http://127.0.0.1:5000 上可用，Lemon Squeezy 无法访问。Ngrok 创建了一个安全隧道，将您的本地服务器暴露到互联网上，允许您测试 Webhook 和其他外部集成。
前往 ngrok.com/download 并下载适用于您操作系统的版本。解压缩后，也可以通过环境变量使其可用，如果您遇到问题，可以参考这个 YouTube 视频。
一旦配置好 ngrok，启动您的 Flask 应用：
python app.py
这将在本地运行您的应用程序，打开一个新的独立终端窗口（保持 Flask 应用程序终端运行）。导航到您下载 ngrok 的位置（或确保它在系统的 PATH 中）并运行：
ngrok http 5000
（如果您的 Flask 应用程序运行在不同的端口，请替换 5000。）
这将生成一个公共 URL（例如 https://abcd1234.ngrok.io），您可以用于 Webhook 和其他外部集成。
Session Status online
Account 您的名字（计划：免费）
Version x.x.x
...
Forwarding https://xxxxxxxx.ngrok.io -> http://localhost:5000
回到 Lemon Squeezy 仪表板 -> 设置 -> Webhook。添加新的 Webhook。
现在将那个 ngrok URL 粘贴到您创建的 Webhook 环境变量中，并确保启用 subscription_created 和 subscription_updated 事件，这是我们实现的两个功能。
您会看到还有许多其他我们可以利用的 Webhook 事件和自动化功能。
保存该 Webhook，重要的一点是，如果您第二天再次运行应用程序（Flask + ngrok），您需要更新 ngrok URL，因为每次新会话都会不同。
15、测试 Web 应用
现在我们已经编写了所有代码，是时候测试我们的应用程序了。让我们先尝试登录，它应该会抛出错误，因为我们还没有创建账户对吧。
登录错误按预期工作，让我们创建一个账户，然后登录以查看用户界面并测试我们的免费计划。
我们的用户界面非常简洁，您可以看到每个新用户登录时都被分配了默认的免费计划，这正按预期工作（每小时 2 条消息）。
现在，让我们购买标准计划并看看它如何相应地更新我们的用户界面。
您可以看到，当我购买标准计划时，用户界面更新以显示我的当前计划、剩余消息数量和其他相关信息。
您可以看到我们的数据库表以及支付数据表被更新为用户当前订阅及相关详细信息。
16、结束语
Lemon Squeezy 提供了许多功能，可以帮助您构建一个完整、安全的端到端应用程序。您可以使用其文档中列出的虚拟卡测试交易。
Lemon Squeezy 只在您的业务产生收入时收取费用，非常适合初创企业和开发者。
确保彻底探索其文档，以正确理解和测试每个功能，并确保您的应用程序安全且适合生产环境。
希望这篇博客能为您提供一个坚实的基础。
原文链接：Building a Subscription-Based AI Web App from Scratch Using Python
汇智网翻译整理，转载请标明出处
