Capstone Project - Image Processing Platform

1# app/models.py
2from pydantic import BaseModel, Field
3from typing import List, Optional, Literal
4
5class GenerateRequest(BaseModel):
6    prompt: str
7    style: Literal["photo", "illustration", "art", "3d"] = "photo"
8    size: Literal["1024x1024", "1792x1024", "1024x1792"] = "1024x1024"
9    provider: Literal["dalle", "sd"] = "dalle"
10
11class AnalyzeRequest(BaseModel):
12    image_url: str
13    tasks: List[Literal["describe", "classify", "tag", "ocr", "quality"]]
14
15class AnalyzeResponse(BaseModel):
16    description: Optional[str] = None
17    classification: Optional[dict] = None
18    tags: Optional[List[str]] = None
19    text_content: Optional[str] = None
20    quality_score: Optional[int] = None
21
22class EditRequest(BaseModel):
23    image_url: str
24    operation: Literal["remove_bg", "upscale", "style_transfer"]
25    style_reference: Optional[str] = None

1# app/generators.py
2from openai import OpenAI
3
4client = OpenAI()
5
6style_map = {
7    "photo": "professional photography, realistic, high quality, 8K",
8    "illustration": "digital illustration, clean lines, vibrant colors",
9    "art": "fine art painting, artistic, expressive brushstrokes",
10    "3d": "3D render, octane render, volumetric lighting, detailed"
11}
12
13async def generate_dalle(prompt: str, style: str, size: str):
14    full_prompt = f"{prompt}, {style_map.get(style, '')}"
15    
16    response = client.images.generate(
17        model="dall-e-3",
18        prompt=full_prompt,
19        size=size,
20        quality="hd"
21    )
22    
23    return {
24        "url": response.data[0].url,
25        "revised_prompt": response.data[0].revised_prompt
26    }

1# app/analyzers.py
2from langchain_openai import ChatOpenAI
3from langchain_core.messages import HumanMessage
4
5llm = ChatOpenAI(model="gpt-4o")
6
7async def analyze_image(image_url: str, tasks: list):
8    results = {}
9    
10    for task in tasks:
11        prompts = {
12            "describe": "Mo ta chi tiet hinh anh nay.",
13            "classify": "Classify hinh anh nay vao category phu hop.",
14            "tag": "Generate 10 tags cho hinh anh nay.",
15            "ocr": "Extract tat ca text trong hinh anh.",
16            "quality": "Danh gia chat luong anh 1-10 va giai thich."
17        }
18        
19        response = await llm.ainvoke([
20            HumanMessage(content=[
21                {"type": "text", "text": prompts[task]},
22                {"type": "image_url", "image_url": {"url": image_url}}
23            ])
24        ])
25        results[task] = response.content
26    
27    return results

1# app/main.py
2from fastapi import FastAPI, UploadFile, File
3from fastapi.staticfiles import StaticFiles
4from app.models import *
5from app.generators import generate_dalle
6from app.analyzers import analyze_image
7
8app = FastAPI(title="Image Processing Platform")
9app.mount("/static", StaticFiles(directory="static"), name="static")
10
11@app.post("/generate")
12async def generate(request: GenerateRequest):
13    result = await generate_dalle(request.prompt, request.style, request.size)
14    return result
15
16@app.post("/analyze")
17async def analyze(request: AnalyzeRequest):
18    results = await analyze_image(request.image_url, request.tasks)
19    return results
20
21@app.post("/upload")
22async def upload(file: UploadFile = File(...)):
23    content = await file.read()
24    path = f"static/uploads/{file.filename}"
25    with open(path, "wb") as f:
26        f.write(content)
27    return {"url": f"/static/uploads/{file.filename}"}

1# tests/test_platform.py
2import pytest
3from httpx import AsyncClient
4from app.main import app
5
6@pytest.mark.asyncio
7async def test_generate():
8    async with AsyncClient(app=app, base_url="http://test") as ac:
9        response = await ac.post("/generate", json={
10            "prompt": "A beautiful sunset",
11            "style": "photo",
12            "size": "1024x1024"
13        })
14        assert response.status_code == 200
15        assert "url" in response.json()