import uvicorn from fastapi import FastAPI, HTTPException, Request, BackgroundTasks from fastapi.responses import FileResponse, JSONResponse from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles from pydantic import BaseModel from typing import List, Optional, Dict, Any import os import uuid import re import pandas as pd from matplotlib import font_manager from dotenv import load_dotenv # Get BASE_DIR for the project root (one level up from this 'core' folder) BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # Load environment variables from .env.local or .env env_local_path = os.path.join(BASE_DIR, ".env.local") env_path = os.path.join(BASE_DIR, ".env") if os.path.exists(env_local_path): load_dotenv(env_local_path) elif os.path.exists(env_path): load_dotenv(env_path) from . import core, ppt, bq_client app = FastAPI() app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # --- Pydantic 모델 --- class OptionsRequest(BaseModel): job: Optional[str] = None bm: Optional[str] = None sales: Optional[str] = None class SearchParamsModel(BaseModel): job: str bm: str sales: str emp: str userBase: str userValue: int class DiagnosisRequest(BaseModel): params: SearchParamsModel answers: List[str] metadata: Optional[Dict[str, str]] = None # Get BASE_DIR for the project root (one level up from this 'core' folder) BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # --- 폰트 설치 확인 (디버깅용) --- font_path = os.path.join(BASE_DIR, "core", "data", "Paperlogy-4Regular.ttf") if os.path.exists(font_path): font_name = font_manager.FontProperties(fname=font_path).get_name() print(f"Loaded font: {font_name}") else: print(f"Warning: Font file not found at {font_path}") # --- 엔드포인트 --- @app.post("/api/options") def get_options_endpoint(req: OptionsRequest): return core.get_step_options(core.conds_df, job=req.job) @app.post("/api/survey") def get_survey_items_endpoint(req: SearchParamsModel): target_table, bars_indicator, level_range = core.make_bars_table(core.bars_df, core.factors, req.job) return bars_indicator.to_dict(orient='records') @app.post("/api/diagnosis") def post_diagnosis_endpoint(req: DiagnosisRequest, request: Request, background_tasks: BackgroundTasks): p = req.params request_options = (p.job, p.bm, p.sales, p.emp, p.userBase, p.userValue) ppt_data, service_data = core.generate_dataset(request_options, req.answers) # 세션 ID 처리: 쿠키에 없으면 새로 생성 session_id = request.cookies.get("session_id") if not session_id: session_id = str(uuid.uuid4()) bg_data = { "job": p.job, "bm": p.bm, "sales": p.sales, "emp": p.emp, "base_type": p.userBase, "target_wage": p.userValue, "user_choices": req.answers, "final_level": service_data['result_data']['badges'][1], # level is the second badge "session_id": session_id, "referrer": req.metadata.get("referrer") if req.metadata else None, "utm_source": req.metadata.get("utm_source") if req.metadata else None, "utm_medium": req.metadata.get("utm_medium") if req.metadata else None, "utm_campaign": req.metadata.get("utm_campaign") if req.metadata else None, } background_tasks.add_task(bq_client.insert_diagnosis_log, bg_data) # 프론트엔드 호환성을 위한 키 이름 변환 (snake_case -> camelCase) res = service_data['result_data'] result_data = { "headMessage": res['head_message'], "levelInfo": { "left": res['level_info']['left'], "right": res['level_info']['right'] }, "levelChart": { "left": res['level_chart']['left'], "right": res['level_chart']['right'] }, "wageOutput": { "description": res['wage_output']['description'], "top": res['wage_output']['top'], "bottom": res['wage_output']['bottom'] }, "badges": res['badges'] } # 응답 생성 및 쿠키 설정 final_response = JSONResponse(content=result_data) if not request.cookies.get("session_id"): # 1년(31536000초) 만료 세션 쿠키 설정 final_response.set_cookie(key="session_id", value=session_id, max_age=31536000, httponly=True) return final_response @app.post("/api/report") def download_report(req: DiagnosisRequest): p = req.params request_options = (p.job, p.bm, p.sales, p.emp, p.userBase, p.userValue) ppt_data, service_data = core.generate_dataset(request_options, req.answers) out_path = os.path.join(BASE_DIR, "tmp", f"report_{uuid.uuid4().hex[:8]}.pptx") os.makedirs(os.path.dirname(out_path), exist_ok=True) ppt.update_ppt_with_data(ppt_data, out_path) return FileResponse(out_path, filename="진단리포트.pptx", media_type="application/vnd.openxmlformats-officedocument.presentationml.presentation") # --- 정적 파일 서빙 --- # api.py가 core/ 폴더 안에 있으므로 BASE_DIR는 상위 디렉토리(루트)입니다. BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) OUT_DIR = os.path.join(BASE_DIR, "out") if os.path.exists(OUT_DIR): app.mount("/_next", StaticFiles(directory=os.path.join(OUT_DIR, "_next")), name="next_static") def _file_response_no_cache(path: str) -> FileResponse: # HF Spaces / CDN may cache HTML entrypoints aggressively depending on headers. # Prevent stale pages after redeploy by disabling caching for HTML responses. res = FileResponse(path) res.headers["Cache-Control"] = "no-store, no-cache, must-revalidate, max-age=0" res.headers["Pragma"] = "no-cache" res.headers["Expires"] = "0" return res @app.api_route("/{full_path:path}", methods=["GET", "HEAD"]) async def serve_spa(full_path: str): if full_path.startswith("api"): raise HTTPException(status_code=404) file_path = os.path.join(OUT_DIR, full_path) if os.path.isfile(file_path): return _file_response_no_cache(file_path) if file_path.endswith(".html") else FileResponse(file_path) html_file = os.path.join(OUT_DIR, f"{full_path}.html") if os.path.isfile(html_file): return _file_response_no_cache(html_file) index_file = os.path.join(file_path, "index.html") if os.path.isdir(file_path) and os.path.isfile(index_file): return _file_response_no_cache(index_file) return _file_response_no_cache(os.path.join(OUT_DIR, "index.html")) if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=7860)