Building DesignAI — An AI-Powered Design Generation Platform with Next.js and FastAPI

# Backend (FastAPI)
git clone https://github.com/Afrizal236/apidesignai-ver2.git

# Frontend (Next.js)
git clone https://github.com/Afrizal236/DesignAI-ver2.git

# Web Framework
fastapi==0.115.0
uvicorn[standard]==0.30.6
python-multipart==0.0.12

# Data Validation
pydantic==2.9.0
pydantic-settings==2.5.2

# AI
google-genai==2.8.0
replicate==1.0.7

# Database — PostgreSQL + SQLAlchemy async
sqlalchemy[asyncio]==2.0.36
asyncpg==0.29.0
alembic==1.13.3

# Redis
redis[hiredis]==5.1.1

# Auth
passlib[bcrypt]==1.7.4
python-jose[cryptography]==3.3.0

# Object Storage — Cloudflare R2 (S3-compatible)
boto3==1.35.36

# Data Mining
scikit-learn==1.5.2
mlxtend==0.23.1
pandas==2.2.3
numpy==1.26.4

# Deep Learning — CLIP + LSTM
torch>=2.0.0
torchvision>=0.15.0
transformers>=4.35.0
pillow>=10.0.0

# core/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=False,
        extra="ignore",
    )

    # Cloudflare R2
    R2_ACCOUNT_ID: str = ""
    R2_ACCESS_KEY_ID: str = ""
    R2_SECRET_ACCESS_KEY: str = ""
    R2_BUCKET_NAME: str = "designai-images"
    R2_PUBLIC_URL: str = ""   # e.g. https://designai.unesa.dev

    @property
    def R2_ENDPOINT_URL(self) -> str:
        return f"https://{self.R2_ACCOUNT_ID}.r2.cloudflarestorage.com"

    @property
    def R2_ENABLED(self) -> bool:
        return bool(
            self.R2_ACCOUNT_ID and self.R2_ACCESS_KEY_ID
            and self.R2_SECRET_ACCESS_KEY and self.R2_PUBLIC_URL
        )

    # Multi-key rotation for Gemini and NanoBanana
    GEMINI_API_KEYS: str = ""   # "key1,key2,key3"

    @property
    def GEMINI_KEY_LIST(self) -> list[str]:
        keys = [k.strip() for k in self.GEMINI_API_KEYS.split(",") if k.strip()]
        return keys or ([self.GEMINI_API_KEY] if self.GEMINI_API_KEY else [])

settings = Settings()

# routes/generate.py
@router.post("", response_model=GenerateResponse)
async def generate_image(
    prompt: str = Form(..., min_length=3, max_length=1000),
    model: str = Form(default="hd"),
    reference_image: UploadFile | None = File(default=None),
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    # Read and validate reference image
    image_bytes = None
    if reference_image and reference_image.filename:
        image_bytes = await reference_image.read()
        # size validation...

    # Run the generation pipeline
    result = await generate_image_from_prompt(
        prompt=prompt,
        model_id=model,
        image_bytes=image_bytes,
        image_mime=image_mime,
    )

    # Persist to database
    record = GenerationHistory(
        user_id=current_user.id,
        prompt=result["prompt_used"],
        enhanced_prompt=result.get("enhanced_prompt"),
        model_used=result["model_used"],
        image_url=result["image_url"] if result["success"] else None,
        status="success" if result["success"] else "failed",
        generation_time_ms=result["generation_time_ms"],
    )
    db.add(record)
    await db.commit()

    return GenerateResponse(**result)

# routes/recommendation.py
@router.post("")
async def get_recommendation(file: UploadFile = File(...), ...):
    result = await analyze_image(image_bytes, file.content_type)

    # Save each generated product to history
    for i, product in enumerate(result["products"]):
        model_prefix = "recommendation" if i == 0 else "recommendation-gallery"
        record = GenerationHistory(
            user_id=current_user.id,
            model_used=f"{model_prefix}/{product.get('id')}",
            image_url=product.get("image_url"),
            ...
        )
        db.add(record)
    await db.commit()

@router.post("/manual")
async def get_recommendation_manual(
    file: UploadFile = File(...),
    prompt: str = Form(...),  # User direction
    ...
):
    result = await analyze_image_with_prompt(image_bytes, file.content_type, prompt)
    ...

# services/storage_service.py
import asyncio, uuid, boto3
from datetime import datetime

def _build_object_key(model_id: str, file_ext: str = "png") -> str:
    now = datetime.utcnow()
    date_path = now.strftime("%Y/%m/%d")
    safe_model = model_id.replace("/", "-").replace(":", "-")
    return f"images/{date_path}/{safe_model}_{uuid.uuid4()}.{file_ext}"

async def upload_image(image_bytes: bytes, model_id: str = "hd") -> dict:
    if not settings.R2_ENABLED:
        return {"success": False, "url": None, "message": "R2 not configured"}

    object_key = _build_object_key(model_id, file_ext)

    # Run boto3 in thread pool to avoid blocking the event loop
    await asyncio.to_thread(_upload_sync, image_bytes, object_key, content_type)

    public_url = f"{settings.R2_PUBLIC_URL.rstrip('/')}/{object_key}"
    return {"success": True, "url": public_url, "object_key": object_key}

# services/genetic_prompt_service.py
# Conceptual structure:

class GeneticPromptEvolver:
    def __init__(self, base_prompt: str, generations: int = 5, population_size: int = 8):
        self.base_prompt = base_prompt
        self.generations = generations
        self.population_size = population_size

    async def evolve(self) -> dict:
        population = self._initialize_population()
        for gen in range(self.generations):
            scored = await self._evaluate_fitness(population)
            parents = self._select_parents(scored)
            population = self._crossover_and_mutate(parents)
        return self._best(population)

# routes/dashboard.py (conceptual)
CACHE_KEY = "dashboard:{user_id}"

async def get_dashboard(current_user: User, db: AsyncSession, redis: Redis):
    cached = await redis.get(CACHE_KEY.format(user_id=current_user.id))
    if cached:
        return json.loads(cached)

    # Build fresh dashboard data from DB
    data = await build_dashboard_data(db, current_user)

    await redis.setex(
        CACHE_KEY.format(user_id=current_user.id),
        settings.CACHE_DASHBOARD_TTL,  # 60 seconds
        json.dumps(data)
    )
    return data

// app/api/download/route.ts
export async function GET(req: NextRequest) {
  const imageUrl = new URL(req.url).searchParams.get("url");

  // Whitelist: only allow downloads from our own R2 domain
  const allowed = ["https://designai.unesa.dev"];
  if (!allowed.some((d) => imageUrl.startsWith(d))) {
    return NextResponse.json({ error: "Domain not allowed" }, { status: 403 });
  }

  const response = await fetch(imageUrl, { cache: "no-store" });
  const buffer = await response.arrayBuffer();

  return new NextResponse(buffer, {
    status: 200,
    headers: {
      "Content-Type": response.headers.get("content-type") ?? "image/png",
      "Content-Disposition": "attachment",
      "Cache-Control": "no-store",
    },
  });
}

// app/dashboard/page.tsx
const handleDownload = async (imageUrl: string) => {
  const fileName = buildFileName(); // "DesignAI Image Jun 18, 2026, 06_30_00 PM.png"
  try {
    const response = await fetch(
      `/api/download?url=${encodeURIComponent(imageUrl)}`,
    );
    if (!response.ok) throw new Error(`HTTP ${response.status}`);

    const blob = await response.blob();
    const objectUrl = URL.createObjectURL(blob);
    const a = document.createElement("a");
    a.href = objectUrl;
    a.download = fileName;
    document.body.appendChild(a);
    a.click();
    document.body.removeChild(a);
    URL.revokeObjectURL(objectUrl);
  } catch {
    window.open(imageUrl, "_blank"); // Fallback
  }
};

// app/api/auth/[...nextauth]/route.ts (conceptual)
export const authOptions = {
  providers: [
    GoogleProvider({ clientId: process.env.GOOGLE_CLIENT_ID }),
    CredentialsProvider({
      authorize: async (credentials) => {
        const res = await fetch("http://localhost:8000/api/auth/login", {
          method: "POST",
          body: JSON.stringify(credentials),
        });
        return res.ok ? res.json() : null;
      },
    }),
  ],
  callbacks: {
    jwt: async ({ token, user }) => {
      if (user) token.accessToken = user.accessToken;
      return token;
    },
    session: async ({ session, token }) => {
      session.accessToken = token.accessToken;
      return session;
    },
  },
};

# routes/limits.py
@router.get("/status/{feature}")
async def get_limit_status(
    feature: str,  # "generate" | "recommendation"
    current_user: User = Depends(get_current_user),
):
    redis_key = f"limit:{feature}:{current_user.id}:{today()}"
    used_today = int(await redis.get(redis_key) or 0)
    daily_limit = settings.RATE_LIMIT_GENERATE  # e.g. 10

    return {
        "used_today": used_today,
        "daily_limit": daily_limit,
        "remaining": max(0, daily_limit - used_today),
    }

cd apidesignai

# Create virtual environment
python -m venv venv312
venv312\Scripts\activate  # Windows
# source venv312/bin/activate  # Linux/Mac

# Install dependencies
pip install -r requirements.txt

# Copy and fill environment variables
cp .env.example .env

# Run database migrations
alembic upgrade head

# Start the server
uvicorn main:app --reload --host 0.0.0.0 --port 8000

cd designai

# Install dependencies
npm install

# Copy and fill environment variables
cp .env.local.example .env.local

# Start the dev server
npm run dev