"""EV 충전 플랫폼 FastAPI 백엔드 ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 충전 흐름: QR 스캔 → 세션 생성 → 결제 → 충전 시작 → 충전 중 → 종료 → 정산 ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ """ import logging from contextlib import asynccontextmanager from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse from app.config import get_settings from app.database import init_db, AsyncSessionLocal from app.routers import chargers, sessions, payments, ocpp_callbacks, dashboard, qr, auth settings = get_settings() # 로깅 logging.basicConfig( level=logging.DEBUG if settings.DEBUG else logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", ) logger = logging.getLogger(__name__) @asynccontextmanager async def lifespan(app: FastAPI): """앱 시작/종료 시 실행""" logger.info("=" * 50) logger.info("EV 충전 백엔드 시작") logger.info(f" Steve: {settings.STEVE_BASE_URL}") logger.info(f" 요금: {settings.total_rate}원/kWh") logger.info(f" (전기 {settings.ELECTRICITY_RATE} + 서비스 {settings.SERVICE_MARGIN})") logger.info("=" * 50) # DB 테이블 생성 (개발용) await init_db() logger.info("DB 테이블 초기화 완료") # 초기 관리자 계정 생성 await _ensure_admin() yield logger.info("EV 충전 백엔드 종료") # ── FastAPI 앱 ── app = FastAPI( title="EV 충전 플랫폼 API", description="CPO 없는 전기차 충전 플랫폼 — OCPP 1.6J + 토스페이먼츠", version="0.1.0", lifespan=lifespan, docs_url="/docs", redoc_url="/redoc", ) # CORS (모바일 웹 결제 페이지 허용) app.add_middleware( CORSMiddleware, allow_origins=[ "https://s1.byunc.com", "http://localhost:3000", "http://localhost:5173", ], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # ── 라우터 등록 ── app.include_router(auth.router, prefix="/api/v1") app.include_router(chargers.router, prefix="/api/v1") app.include_router(sessions.router, prefix="/api/v1") app.include_router(payments.router, prefix="/api/v1") app.include_router(ocpp_callbacks.router, prefix="/api/v1") app.include_router(dashboard.router, prefix="/api/v1") app.include_router(qr.router, prefix="/api/v1") # ── 초기 관리자 생성 ── async def _ensure_admin(): """admin 계정이 없으면 자동 생성""" from sqlalchemy import select from app.models import User, UserRole from app.services.auth import hash_password async with AsyncSessionLocal() as db: result = await db.execute( select(User).where(User.role == UserRole.ADMIN) ) if result.scalar_one_or_none(): return admin = User( username="admin", hashed_password=hash_password("admin1234"), display_name="관리자", role=UserRole.ADMIN, ) db.add(admin) await db.commit() logger.info("초기 관리자 계정 생성: admin / admin1234") # ── 헬스체크 ── @app.get("/health") async def health(): return { "status": "ok", "service": "ev-charging-api", "rate": f"{settings.total_rate}원/kWh", } @app.get("/") async def root(): return { "message": "EV 충전 플랫폼 API", "docs": "/docs", "health": "/health", } @app.get("/dashboard") async def dashboard_page(): """관리자 대시보드 HTML 서빙""" return FileResponse("/code/dashboard.html", media_type="text/html") @app.get("/simulator") async def simulator_page(): """충전 시뮬레이터 GUI 서빙""" return FileResponse("/code/simulator.html", media_type="text/html")