一天内用 FastAPI+Groq+Stripe 上线 AI SaaS

将短篇简述转化为精心打磨的 500 字文章，仅需数秒，通过 Stripe 实现变现，并在廉价 VPS 上启动一个生产就绪的服务。

你可以复制粘贴下面的代码片段，替换占位符，午餐前就能拥有一个实时的 SaaS 产品。

1. 为什么选择 AI 生成文章服务？

内容营销仍然是王道，但编写 SEO 友好的文案是一个瓶颈。随着 Groq 的 Llama-3-8B-Chat 等大语言模型（LLM）的兴起，你可以在一分钟内生成高质量、富含关键词的文章。将这一能力打包成基于订阅的 SaaS，既能为博客撰写者、市场营销人员和初创公司解决真实的痛点，又能获得经常性收入。

速度：将 5 个单词的简述转化为可发布的文章，耗时约 30 秒。

一致性：每次都保持相同的语调、结构和 SEO 优化。

可扩展性：Groq 的按量计费模式让你无需过度配置就能扩展。

ai-article-saas/
├─ app/
│  ├─ main.py            # FastAPI entry point
│  ├─ schemas.py         # Pydantic models
│  ├─ services/
│  │   └─ groq.py        # Wrapper around Groq API
│  └─ stripe_webhook.py # Stripe webhook handler
├─ static/
│  └─ index.html        # Landing page (served by Nginx)
├─ requirements.txt
└─ Dockerfile

2. 依赖与核心 FastAPI 应用

fastapi==0.110.0
uvicorn[standard]==0.27.0
httpx==0.27.0
python-dotenv==1.0.1
stripe==9.9.0
slowapi==0.1.7        # rate-limiting
pydantic==2.6.1

import os
from fastapi import FastAPI, HTTPException, Depends, Request
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
import httpx
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from starlette.status import HTTP_429_TOO_MANY_REQUESTS

app = FastAPI()
limiter = Limiter(key_func=lambda request: request.state.customer_id)
app.state.limiter = limiter
app.add_exception_handler(HTTP_429_TOO_MANY_REQUESTS, _rate_limit_exceeded_handler)

# ---- Models -------------------------------------------------
class Brief(BaseModel):
    title: str
    keywords: str

class ArticleResponse(BaseModel):
    article: str

# ---- Auth ---------------------------------------------------
bearer = HTTPBearer(auto_error=False)

async def get_customer_id(
    credentials: HTTPAuthorizationCredentials = Depends(bearer),
    request: Request = None,
):
    """
    Stripe Customer ID is passed as a Bearer token.
    In production you'd verify the token via Stripe's OAuth or a JWT.
    """
    if not credentials:
        raise HTTPException(status_code=401, detail="Missing auth token")
    request.state.customer_id = credentials.credentials
    return credentials.credentials

# ---- LLM Wrapper ---------------------------------------------
GROQ_API_KEY = os.getenv("GROQ_API_KEY")
GROQ_ENDPOINT = "https://api.groq.com/openai/v1/chat/completions"

async def generate_article(brief: Brief) -> str:
    system_prompt = (
        "You are an expert copywriter. Write a 500-word SEO-optimized article "
        "based on the given title and keywords. Use markdown headings, bullet points, "
        "and a concluding paragraph."
    )
    user_prompt = f"Title: {brief.title}\nKeywords: {brief.keywords}"
    payload = {
        "model": "llama3-8b-8192",
        "messages": [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt},
        ],
        "temperature": 0.7,
    }
    async with httpx.AsyncClient() as client:
        r = await client.post(
            GROQ_ENDPOINT,
            json=payload,
            headers={"Authorization": f"Bearer {GROQ_API_KEY}"},
            timeout=30,
        )
        r.raise_for_status()
        data = r.json()
        return data["choices"][0]["message"]["content"]

# ---- Endpoints ------------------------------------------------
@app.post("/generate", response_model=ArticleResponse)
@limiter.limit("5/hour")
async def generate(
    brief: Brief,
    customer_id: str = Depends(get_customer_id),
):
    article = await generate_article(brief)
    return ArticleResponse(article=article)

# ---- Health ---------------------------------------------------
@app.get("/health")
async def health():
    return {"status": "ok"}

提示：速率限制的 key 是 Stripe Customer ID，确保每个付费用户都能获得公平的配额。

3. Stripe 结账与 Webhook

3.1 在 Stripe 中创建产品

在 Stripe 仪表板中创建两个产品，并为 PRO 计划启用 7 天试用（使用 Stripe 的内置试用天数功能）。

3.2 结账会话助手

import stripe
stripe.api_key = os.getenv("STRIPE_SECRET_KEY")

def create_checkout_session(customer_id: str, price_id: str, success_url: str, cancel_url: str):
    session = stripe.checkout.Session.create(
        customer=customer_id,
        payment_method_types=["card"],
        line_items=[{
            "price": price_id,
            "quantity": 1,
        }],
        mode="subscription",
        success_url=success_url,
        cancel_url=cancel_url,
    )
    return session.url

暴露一个小端点（例如 /checkout?plan=pro），将用户重定向到生成的 URL。

3.3 Webhook 处理器（stripe_webhook.py）

from fastapi import APIRouter, Request, HTTPException
import stripe
import os

router = APIRouter()
stripe.api_key = os.getenv("STRIPE_SECRET_KEY")
endpoint_secret = os.getenv("STRIPE_WEBHOOK_SECRET")

@router.post("/stripe-webhook")
async def handle_webhook(request: Request):
    payload = await request.body()
    sig_header = request.headers.get("stripe-signature")
    
    try:
        event = stripe.Webhook.construct_event(
            payload, sig_header, endpoint_secret
        )
    except ValueError:
        raise HTTPException(status_code=400, detail="Invalid payload")
    except stripe.error.SignatureVerificationError:
        raise HTTPException(status_code=400, detail="Invalid signature")
    
    # Handle subscription events
    if event["type"] == "customer.subscription.created":
        customer_id = event["data"]["object"]["customer"]
        # Provision access for customer_id
    elif event["type"] == "customer.subscription.deleted":
        customer_id = event["data"]["object"]["customer"]
        # Revoke access for customer_id
    
    return {"status": "received"}

在 main.py 中包含路由：

from stripe_webhook import router as webhook_router
app.include_router(webhook_router)

4. 简单的着陆页面与订阅流程

<!DOCTYPE html>
<html>
<head>
    <title>AI Article Generator</title>
    <style>
        body { font-family: sans-serif; padding: 2rem; max-width: 600px; margin: 0 auto; }
        .plan { border: 1px solid #ccc; padding: 1rem; margin: 1rem 0; border-radius: 8px; }
        button { background: #007bff; color: white; padding: 0.5rem 1rem; cursor: pointer; border: none; border-radius: 4px; }
    </style>
</head>
<body>
    <h1>AI Article Generator</h1>
    <div class="plan">
        <h3>STARTER - $19/month</h3>
        <p>50 articles/month</p>
        <button onclick="checkout('price_starter')">Subscribe</button>
    </div>
    <div class="plan">
        <h3>PRO - $49/month (7-day free trial)</h3>
        <p>500 articles/month</p>
        <button onclick="checkout('price_pro')">Start Trial</button>
    </div>
    <script>
        async function checkout(priceId) {
            const res = await fetch('/checkout?price_id=' + priceId);
            window.location.href = await res.text();
        }
    </script>
</body>
</html>

5. 部署到生产环境

5.1 Dockerfile

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY app/ app/
COPY static/ static/

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

5.2 Nginx 反向代理配置

upstream api {
    server localhost:8000;
}

server {
    listen 80;
    server_name yourdomain.com;

    location /static {
        alias /app/static;
    }

    location / {
        proxy_pass http://api;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
    }
}

5.3 一键部署脚本

#!/bin/bash
docker build -t ai-article-saas .
docker run -d \
    -p 8000:8000 \
    -e GROQ_API_KEY=$GROQ_API_KEY \
    -e STRIPE_SECRET_KEY=$STRIPE_SECRET_KEY \
    -e STRIPE_WEBHOOK_SECRET=$STRIPE_WEBHOOK_SECRET \
    ai-article-saas

在 VPS 上执行此脚本（$5/月 Linode、DigitalOcean 或 Vultr），配置 Nginx，你就已经准备好了。

6. 关键考虑事项

• 速率限制：使用 SlowAPI 按客户防止滥用。
• 错误处理：Groq API 可能会超时；添加重试逻辑。
• 身份验证：在生产环境中使用 JWT 或 Stripe OAuth 替代 Bearer token。
• 日志：集成 ELK 或 Datadog 以监控 API 调用和 LLM 费用。

从代码到收款，你现在拥有一个可完全扩展的 AI 驱动的 SaaS。祝你部署顺利！