File size: 4,530 Bytes
7fa9d90 1a02e5e 7fa9d90 c00f68c 53ca86f c00f68c 53ca86f c00f68c 53ca86f 8f2ed57 53ca86f c00f68c 53ca86f c00f68c 7fa9d90 5f64acd 7fa9d90 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 | """
NCAkit - Neural Content Automation Toolkit
Main FastAPI Application with Modular Architecture
"""
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse
import logging
from pathlib import Path
import sys
from config import config
from core.module_registry import registry
# Setup logging
logging.basicConfig(
level=config.log_level.upper(),
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger(__name__)
# Create FastAPI app
app = FastAPI(
title="NCAkit - Neural Content Automation Toolkit",
description="""
# NCAkit REST API
A modular toolkit for content automation with multiple feature modules.
## Available Modules
- 🎬 **Video Creator** - Create short-form videos with TTS, captions, and music
- 📱 More modules coming soon...
## How It Works
1. Each module has its own API prefix (e.g., `/api/video/`)
2. Modules are auto-discovered and registered on startup
3. Check `/api/modules` for list of available modules
""",
version="1.0.0",
contact={
"name": "NCAkit",
"url": "https://github.com/your-repo/ncakit"
}
)
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.on_event("startup")
async def startup_event():
"""Initialize all modules on startup"""
logger.info("Starting NCAkit...")
# Ensure directories exist
config.ensure_directories()
# Initialize HF Hub storage (optional)
if config.hf_repo:
from modules.shared.services.hf_storage import init_hf_storage
import os
hf_token = os.getenv("HF_TOKEN")
hf_storage = init_hf_storage(repo_id=config.hf_repo, token=hf_token)
if hf_storage.enabled:
logger.info(f"HF Cloud storage enabled: {config.hf_repo}")
else:
logger.warning("HF credentials set but storage failed to initialize")
# Register all modules
num_modules = registry.register_all(app, config)
logger.info(f"Loaded {num_modules} module(s)")
logger.info(f"NCAkit started successfully on port {config.port}")
@app.get("/health", tags=["System"])
async def health_check():
"""Health check endpoint"""
return {"status": "ok", "toolkit": "ncakit"}
@app.get("/api/modules", tags=["System"])
async def list_modules():
"""List all available modules"""
return {
"modules": registry.list_modules()
}
# ==========================================
# GEMINI CHATBOT TEST ENDPOINT
# ==========================================
from pydantic import BaseModel
import os
class ChatRequest(BaseModel):
message: str
@app.post("/api/chat", tags=["Chat"])
async def chat_with_gemini(request: ChatRequest):
"""
Test Gemini API connectivity.
Uses GEMINI_API_KEY from environment.
"""
try:
from google import genai
# Client gets API key from GEMINI_API_KEY environment variable
client = genai.Client()
response = client.models.generate_content(
model="gemma-3-27b-it",
contents=request.message
)
return {"reply": response.text}
except Exception as e:
logger.error(f"Gemini chat error: {e}")
return {"error": str(e)}
@app.get("/")
async def read_root():
"""Serve the web UI"""
static_path = Path(__file__).parent / "static" / "index.html"
if static_path.exists():
return FileResponse(static_path)
return {
"message": "NCAkit - Neural Content Automation Toolkit",
"docs": "/docs",
"modules": "/api/modules"
}
# Mount static files if they exist
static_dir = Path(__file__).parent / "static"
if static_dir.exists():
app.mount("/static", StaticFiles(directory=str(static_dir)), name="static")
# Mount videos directory for direct access and streaming
videos_dir = config.videos_dir_path
if not videos_dir.exists():
videos_dir.mkdir(parents=True, exist_ok=True)
app.mount("/videos", StaticFiles(directory=str(videos_dir)), name="videos")
if __name__ == "__main__":
import uvicorn
uvicorn.run(
"app:app",
host="0.0.0.0",
port=config.port,
log_level=config.log_level.lower()
)
|