Files
test-mcp/server.py
Taehoon d246b08799 feat: 메일 관리 UI 개편 및 시스템 구조 최적화
- UI/UX: 메일 관리 레이아웃 고도화 및 미리보기 토글 핸들 도입
- 기능: 주소록 CRUD 기능 추가 및 모달 인터페이스 개선
- 구조: CSS 파일 기능별 분리 및 Jinja2 템플릿 엔진 도입
- 백엔드: OCR 비동기 처리 및 CSV 파싱(BOM) 안정화
- 데이터: 2026.03.04 기준 최신 프로젝트 현황 업데이트
2026-03-04 17:58:54 +09:00

122 lines
4.0 KiB
Python

import os
import sys
# 한글 환경 및 Tesseract 경로 강제 설정
os.environ["PYTHONIOENCODING"] = "utf-8"
os.environ["TESSDATA_PREFIX"] = r"C:\Users\User\AppData\Local\Programs\Tesseract-OCR\tessdata"
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse, FileResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from analyze import analyze_file_content
from crawler_service import run_crawler_service
import asyncio
from fastapi import Request
app = FastAPI(title="Project Master Overseas API")
templates = Jinja2Templates(directory="templates")
# --- 유틸리티: 동기 함수를 스레드 풀에서 실행 ---
async def run_in_threadpool(func, *args):
loop = asyncio.get_event_loop()
return await loop.run_in_executor(None, func, *args)
# 정적 파일 및 미들웨어 설정
app.mount("/style", StaticFiles(directory="style"), name="style")
app.mount("/js", StaticFiles(directory="js"), name="js")
app.mount("/sample_files", StaticFiles(directory="sample"), name="sample_files")
@app.get("/sample.png")
async def get_sample_img():
return FileResponse("sample.png")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=False,
allow_methods=["*"],
allow_headers=["*"],
)
# --- HTML 라우팅 ---
import csv
@app.get("/project-data")
async def get_project_data():
"""
sheet.csv 파일을 읽어서 프로젝트 현황 데이터를 반환
"""
projects = []
try:
with open("sheet.csv", mode="r", encoding="utf-8-sig") as f:
reader = csv.reader(f)
rows = [row for row in reader if row] # 빈 행 제외
# 실제 데이터가 시작되는 지점 찾기 (No. 로 시작하는 행 다음부터)
start_idx = -1
for i, row in enumerate(rows):
if row and "No." in row[0]:
start_idx = i + 1
break
if start_idx != -1:
for row in rows[start_idx:]:
if len(row) >= 8:
# [프로젝트명, 담당부서, 담당자, 최근활동로그, 파일수] 형식으로 추출
projects.append([
row[1], # 프로젝트 명
row[2], # 담당부서
row[3], # 담당자
row[5], # 최근 활동로그
int(row[7]) if row[7].isdigit() else 0 # 파일 수
])
except Exception as e:
return {"error": str(e)}
return projects
@app.get("/")
async def root(request: Request):
return templates.TemplateResponse("index.html", {"request": request})
@app.get("/dashboard")
async def get_dashboard(request: Request):
return templates.TemplateResponse("dashboard.html", {"request": request})
@app.get("/mailTest")
@app.get("/mailTest.html")
async def get_mail_test(request: Request):
return templates.TemplateResponse("mailTest.html", {"request": request})
# --- 데이터 API ---
@app.get("/attachments")
async def get_attachments():
sample_path = "sample"
if not os.path.exists(sample_path):
os.makedirs(sample_path)
files = []
for f in os.listdir(sample_path):
f_path = os.path.join(sample_path, f)
if os.path.isfile(f_path):
files.append({
"name": f,
"size": f"{os.path.getsize(f_path) / 1024:.1f} KB"
})
return files
@app.get("/analyze-file")
async def analyze_file(filename: str):
"""
분석 서비스(analyze.py) 호출 - 스레드 풀에서 비차단 방식으로 실행
"""
return await run_in_threadpool(analyze_file_content, filename)
@app.get("/sync")
async def sync_data():
"""
크롤링 서비스(crawler_service.py) 호출
"""
return StreamingResponse(run_crawler_service(), media_type="text_event-stream")