Initial commit: Todo Project with dashboard, classification center, and upload functionality

- 📱 PWA 지원: 홈화면 추가 가능한 Progressive Web App
- 🎨 M-Project 색상 스키마: 하늘색, 주황색, 회색, 흰색 일관된 디자인
- 📊 대시보드: 데스크톱 캘린더 뷰 + 모바일 일일 뷰 반응형 디자인
- 📥 분류 센터: Gmail 스타일 받은편지함으로 스마트 분류 시스템
- 🤖 AI 분류 제안: 키워드 기반 자동 분류 제안 및 일괄 처리
- 📷 업로드 모달: 데스크톱(파일 선택) + 모바일(카메라/갤러리) 최적화
- 🏷️ 3가지 분류: Todo(시작일), 캘린더(마감일), 체크리스트(무기한)
- 📋 체크리스트: 진행률 표시 및 완료 토글 기능
- 🔄 시놀로지 연동 준비: 메일플러스 연동을 위한 구조 설계
- 📱 반응형 UI: 모든 페이지 모바일 최적화 완료
This commit is contained in:
Hyungi Ahn
2025-09-19 08:52:49 +09:00
commit 761757c12e
60 changed files with 10281 additions and 0 deletions

26
backend/Dockerfile Normal file
View File

@@ -0,0 +1,26 @@
FROM python:3.11-slim
WORKDIR /app
# 시스템 패키지 업데이트 및 필요한 패키지 설치
RUN apt-get update && apt-get install -y \
gcc \
&& rm -rf /var/lib/apt/lists/*
# Python 의존성 설치
COPY pyproject.toml ./
RUN pip install --no-cache-dir -e .
# 애플리케이션 코드 복사
COPY src/ ./src/
COPY migrations/ ./migrations/
# 환경 변수 설정
ENV PYTHONPATH=/app
ENV PYTHONUNBUFFERED=1
# 포트 노출
EXPOSE 9000
# 애플리케이션 실행
CMD ["uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "9000", "--reload"]

57
backend/pyproject.toml Normal file
View File

@@ -0,0 +1,57 @@
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "todo-project"
version = "0.1.0"
description = "독립적인 할일 관리 시스템"
readme = "README.md"
requires-python = ">=3.8"
dependencies = [
"fastapi>=0.104.1",
"uvicorn[standard]>=0.24.0",
"sqlalchemy>=2.0.23",
"alembic>=1.12.1",
"asyncpg>=0.29.0",
"python-multipart>=0.0.6",
"python-jose[cryptography]>=3.3.0",
"passlib[bcrypt]>=1.7.4",
"python-dotenv>=1.0.0",
"pydantic>=2.5.0",
"pydantic-settings>=2.1.0",
"pillow>=10.0.0",
]
[project.optional-dependencies]
dev = [
"pytest>=7.4.3",
"pytest-asyncio>=0.21.1",
"httpx>=0.25.2",
"black>=23.11.0",
"isort>=5.12.0",
"flake8>=6.1.0",
]
[tool.black]
line-length = 88
target-version = ['py38']
include = '\.pyi?$'
extend-exclude = '''
/(
# directories
\.eggs
| \.git
| \.hg
| \.mypy_cache
| \.tox
| \.venv
| build
| dist
)/
'''
[tool.isort]
profile = "black"
multi_line_output = 3
line_length = 88

0
backend/src/__init__.py Normal file
View File

View File

View File

@@ -0,0 +1,94 @@
"""
API 의존성
"""
from fastapi import Depends, HTTPException, status, Query
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from typing import Optional
from ..core.database import get_db
from ..core.security import verify_token, get_user_id_from_token
from ..models.user import User
# HTTP Bearer 토큰 스키마 (선택적)
security = HTTPBearer(auto_error=False)
async def get_current_user(
credentials: HTTPAuthorizationCredentials = Depends(security),
db: AsyncSession = Depends(get_db)
) -> User:
"""현재 로그인된 사용자 가져오기"""
if not credentials:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Authentication required"
)
try:
# 토큰에서 사용자 ID 추출
user_id = get_user_id_from_token(credentials.credentials)
# 데이터베이스에서 사용자 조회
result = await db.execute(select(User).where(User.id == user_id))
user = result.scalar_one_or_none()
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="User not found"
)
if not user.is_active:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Inactive user"
)
return user
except Exception as e:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials"
)
async def get_current_active_user(
current_user: User = Depends(get_current_user)
) -> User:
"""활성 사용자 확인"""
if not current_user.is_active:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Inactive user"
)
return current_user
async def get_current_admin_user(
current_user: User = Depends(get_current_active_user)
) -> User:
"""관리자 권한 확인"""
if not current_user.is_admin:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not enough permissions"
)
return current_user
async def get_optional_current_user(
credentials: Optional[HTTPAuthorizationCredentials] = Depends(security),
db: AsyncSession = Depends(get_db)
) -> Optional[User]:
"""선택적 사용자 인증 (토큰이 없어도 됨)"""
if not credentials:
return None
try:
return await get_current_user(credentials, db)
except HTTPException:
return None

View File

@@ -0,0 +1,7 @@
"""
API 라우터 모듈
"""
from . import auth, todos, calendar
__all__ = ["auth", "todos", "calendar"]

View File

@@ -0,0 +1,195 @@
"""
인증 관련 API 라우터
- API 라우터 기준: 최대 400줄
- 간결함 원칙: 필수 인증 기능만 포함
"""
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, update
from datetime import datetime
from ...core.database import get_db
from ...core.security import verify_password, create_access_token, create_refresh_token, get_password_hash
from ...core.config import settings
from ...models.user import User
from ...schemas.auth import (
LoginRequest, TokenResponse, RefreshTokenRequest,
UserInfo, ChangePasswordRequest, CreateUserRequest
)
from ..dependencies import get_current_active_user, get_current_admin_user
router = APIRouter()
@router.post("/login", response_model=TokenResponse)
async def login(
login_data: LoginRequest,
db: AsyncSession = Depends(get_db)
):
"""사용자 로그인"""
# 사용자 조회
result = await db.execute(
select(User).where(User.email == login_data.email)
)
user = result.scalar_one_or_none()
# 사용자 존재 및 비밀번호 확인
if not user or not verify_password(login_data.password, user.hashed_password):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect email or password"
)
# 비활성 사용자 확인
if not user.is_active:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Inactive user"
)
# 사용자별 세션 타임아웃을 적용한 토큰 생성
access_token = create_access_token(
data={"sub": str(user.id)},
timeout_minutes=user.session_timeout_minutes
)
refresh_token = create_refresh_token(data={"sub": str(user.id)})
# 마지막 로그인 시간 업데이트
await db.execute(
update(User)
.where(User.id == user.id)
.values(last_login=datetime.utcnow())
)
await db.commit()
return TokenResponse(
access_token=access_token,
refresh_token=refresh_token,
expires_in=settings.ACCESS_TOKEN_EXPIRE_MINUTES * 60
)
@router.post("/refresh", response_model=TokenResponse)
async def refresh_token(
refresh_data: RefreshTokenRequest,
db: AsyncSession = Depends(get_db)
):
"""토큰 갱신"""
from ...core.security import verify_token
try:
# 리프레시 토큰 검증
payload = verify_token(refresh_data.refresh_token, token_type="refresh")
user_id = payload.get("sub")
if not user_id:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid refresh token"
)
# 사용자 존재 확인
result = await db.execute(select(User).where(User.id == user_id))
user = result.scalar_one_or_none()
if not user or not user.is_active:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="User not found or inactive"
)
# 새 토큰 생성
access_token = create_access_token(data={"sub": str(user.id)})
new_refresh_token = create_refresh_token(data={"sub": str(user.id)})
return TokenResponse(
access_token=access_token,
refresh_token=new_refresh_token,
expires_in=settings.ACCESS_TOKEN_EXPIRE_MINUTES * 60
)
except Exception:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid refresh token"
)
@router.get("/me", response_model=UserInfo)
async def get_current_user_info(
current_user: User = Depends(get_current_active_user)
):
"""현재 사용자 정보 조회"""
return UserInfo.model_validate(current_user)
@router.put("/change-password")
async def change_password(
password_data: ChangePasswordRequest,
current_user: User = Depends(get_current_active_user),
db: AsyncSession = Depends(get_db)
):
"""비밀번호 변경"""
# 현재 비밀번호 확인
if not verify_password(password_data.current_password, current_user.hashed_password):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Incorrect current password"
)
# 새 비밀번호 해싱 및 업데이트
new_hashed_password = get_password_hash(password_data.new_password)
await db.execute(
update(User)
.where(User.id == current_user.id)
.values(hashed_password=new_hashed_password)
)
await db.commit()
return {"message": "Password changed successfully"}
@router.post("/create-user", response_model=UserInfo)
async def create_user(
user_data: CreateUserRequest,
admin_user: User = Depends(get_current_admin_user),
db: AsyncSession = Depends(get_db)
):
"""새 사용자 생성 (관리자 전용)"""
# 이메일 중복 확인
result = await db.execute(
select(User).where(User.email == user_data.email)
)
existing_user = result.scalar_one_or_none()
if existing_user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Email already registered"
)
# 새 사용자 생성
new_user = User(
email=user_data.email,
hashed_password=get_password_hash(user_data.password),
full_name=user_data.full_name,
is_admin=user_data.is_admin,
is_active=True
)
db.add(new_user)
await db.commit()
await db.refresh(new_user)
return UserInfo.from_orm(new_user)
@router.post("/logout")
async def logout(
current_user: User = Depends(get_current_active_user)
):
"""로그아웃 (클라이언트에서 토큰 삭제)"""
# 실제로는 클라이언트에서 토큰을 삭제하면 됨
# 필요시 토큰 블랙리스트 구현 가능
return {"message": "Logged out successfully"}

View File

@@ -0,0 +1,175 @@
"""
캘린더 연동 API 라우터
- API 라우터 기준: 최대 400줄
- 간결함 원칙: 캘린더 설정 및 동기화 기능만 포함
"""
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List, Dict, Any, Optional
from uuid import UUID
import logging
from ...core.database import get_db
from ...models.user import User
from ...models.todo import TodoItem
from ..dependencies import get_current_active_user
from ...integrations.calendar import get_calendar_router, CalendarProvider
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/calendar", tags=["calendar"])
@router.post("/providers/register")
async def register_calendar_provider(
provider_data: Dict[str, Any],
current_user: User = Depends(get_current_active_user)
):
"""캘린더 제공자 등록"""
try:
provider_name = provider_data.get("provider")
credentials = provider_data.get("credentials", {})
set_as_default = provider_data.get("default", False)
if not provider_name:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Provider name is required"
)
provider = CalendarProvider(provider_name)
calendar_router = get_calendar_router()
success = await calendar_router.register_provider(
provider, credentials, set_as_default
)
if success:
return {"message": f"{provider_name} 캘린더 등록 성공"}
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"{provider_name} 캘린더 등록 실패"
)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="지원하지 않는 캘린더 제공자"
)
except Exception as e:
logger.error(f"캘린더 제공자 등록 실패: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=str(e)
)
@router.get("/providers")
async def get_registered_providers(
current_user: User = Depends(get_current_active_user)
):
"""등록된 캘린더 제공자 목록 조회"""
try:
calendar_router = get_calendar_router()
providers = calendar_router.get_registered_providers()
return {
"providers": providers,
"default": calendar_router.default_provider.value if calendar_router.default_provider else None
}
except Exception as e:
logger.error(f"캘린더 제공자 조회 실패: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=str(e)
)
@router.get("/calendars")
async def get_all_calendars(
current_user: User = Depends(get_current_active_user)
):
"""모든 등록된 제공자의 캘린더 목록 조회"""
try:
calendar_router = get_calendar_router()
calendars = await calendar_router.get_all_calendars()
return calendars
except Exception as e:
logger.error(f"캘린더 목록 조회 실패: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=str(e)
)
@router.post("/sync/{todo_id}")
async def sync_todo_to_calendar(
todo_id: UUID,
sync_config: Optional[Dict[str, Any]] = None,
current_user: User = Depends(get_current_active_user),
db: AsyncSession = Depends(get_db)
):
"""특정 할일을 캘린더에 동기화"""
try:
from sqlalchemy import select, and_
# 할일 조회
result = await db.execute(
select(TodoItem).where(
and_(
TodoItem.id == todo_id,
TodoItem.user_id == current_user.id
)
)
)
todo_item = result.scalar_one_or_none()
if not todo_item:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Todo item not found"
)
# 캘린더 동기화
calendar_router = get_calendar_router()
calendar_configs = sync_config.get("calendars") if sync_config else None
result = await calendar_router.sync_todo_to_calendars(
todo_item, calendar_configs
)
return {
"message": "캘린더 동기화 완료",
"result": result
}
except HTTPException:
raise
except Exception as e:
logger.error(f"캘린더 동기화 실패: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=str(e)
)
@router.get("/health")
async def calendar_health_check(
current_user: User = Depends(get_current_active_user)
):
"""캘린더 서비스 상태 확인"""
try:
calendar_router = get_calendar_router()
health_status = await calendar_router.health_check()
return health_status
except Exception as e:
logger.error(f"캘린더 상태 확인 실패: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=str(e)
)

View File

@@ -0,0 +1,313 @@
"""
할일관리 API 라우터 (간결 버전)
- API 라우터 기준: 최대 400줄
- 간결함 원칙: 라우팅만 담당, 비즈니스 로직은 서비스로 분리
"""
from fastapi import APIRouter, Depends, HTTPException, status, Query, UploadFile, File
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List, Optional
from uuid import UUID
import logging
from ...core.database import get_db
from ...models.user import User
from ...schemas.todo import (
TodoItemCreate, TodoItemSchedule, TodoItemDelay, TodoItemSplit,
TodoItemResponse, TodoCommentCreate, TodoCommentResponse
)
from ..dependencies import get_current_active_user
from ...services.todo_service import TodoService
from ...services.calendar_sync_service import get_calendar_sync_service
from ...services.file_service import save_base64_image
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/todos", tags=["todos"])
# ============================================================================
# 이미지 업로드
# ============================================================================
@router.post("/upload-image")
async def upload_image(
image: UploadFile = File(...),
current_user: User = Depends(get_current_active_user)
):
"""이미지 업로드"""
try:
# 파일 형식 확인
if not image.content_type.startswith('image/'):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="이미지 파일만 업로드 가능합니다."
)
# 파일 크기 확인 (5MB 제한)
contents = await image.read()
if len(contents) > 5 * 1024 * 1024:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="파일 크기는 5MB를 초과할 수 없습니다."
)
# Base64로 변환하여 저장
import base64
base64_string = f"data:{image.content_type};base64,{base64.b64encode(contents).decode()}"
# 파일 저장
file_url = save_base64_image(base64_string)
if not file_url:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="이미지 저장에 실패했습니다."
)
return {"url": file_url}
except HTTPException:
raise
except Exception as e:
logger.error(f"이미지 업로드 실패: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="이미지 업로드에 실패했습니다."
)
# ============================================================================
# 할일 아이템 관리
# ============================================================================
@router.post("/", response_model=TodoItemResponse)
async def create_todo_item(
todo_data: TodoItemCreate,
current_user: User = Depends(get_current_active_user),
db: AsyncSession = Depends(get_db)
):
"""새 할일 생성"""
try:
service = TodoService(db)
return await service.create_todo(todo_data, current_user.id)
except Exception as e:
await db.rollback()
logger.error(f"할일 생성 실패: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=str(e)
)
@router.post("/{todo_id}/schedule", response_model=TodoItemResponse)
async def schedule_todo_item(
todo_id: UUID,
schedule_data: TodoItemSchedule,
current_user: User = Depends(get_current_active_user),
db: AsyncSession = Depends(get_db)
):
"""할일 일정 설정 및 캘린더 동기화"""
try:
service = TodoService(db)
result = await service.schedule_todo(todo_id, schedule_data, current_user.id)
# 🔄 캘린더 동기화 (백그라운드)
sync_service = get_calendar_sync_service()
todo_item = await service._get_user_todo(todo_id, current_user.id)
await sync_service.sync_todo_create(todo_item)
return result
except ValueError as e:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=str(e)
)
except Exception as e:
await db.rollback()
logger.error(f"할일 일정 설정 실패: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=str(e)
)
@router.put("/{todo_id}/complete", response_model=TodoItemResponse)
async def complete_todo_item(
todo_id: UUID,
current_user: User = Depends(get_current_active_user),
db: AsyncSession = Depends(get_db)
):
"""할일 완료 및 캘린더 업데이트"""
try:
service = TodoService(db)
result = await service.complete_todo(todo_id, current_user.id)
# 🔄 캘린더 동기화 (완료 태그 변경)
sync_service = get_calendar_sync_service()
todo_item = await service._get_user_todo(todo_id, current_user.id)
await sync_service.sync_todo_complete(todo_item)
return result
except ValueError as e:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=str(e)
)
except Exception as e:
await db.rollback()
logger.error(f"할일 완료 실패: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=str(e)
)
@router.put("/{todo_id}/delay", response_model=TodoItemResponse)
async def delay_todo_item(
todo_id: UUID,
delay_data: TodoItemDelay,
current_user: User = Depends(get_current_active_user),
db: AsyncSession = Depends(get_db)
):
"""할일 지연 및 캘린더 날짜 수정"""
try:
service = TodoService(db)
result = await service.delay_todo(todo_id, delay_data, current_user.id)
# 🔄 캘린더 동기화 (날짜 수정)
sync_service = get_calendar_sync_service()
todo_item = await service._get_user_todo(todo_id, current_user.id)
await sync_service.sync_todo_delay(todo_item)
return result
except ValueError as e:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=str(e)
)
except Exception as e:
await db.rollback()
logger.error(f"할일 지연 실패: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=str(e)
)
@router.post("/{todo_id}/split", response_model=List[TodoItemResponse])
async def split_todo_item(
todo_id: UUID,
split_data: TodoItemSplit,
current_user: User = Depends(get_current_active_user),
db: AsyncSession = Depends(get_db)
):
"""할일 분할"""
try:
service = TodoService(db)
return await service.split_todo(todo_id, split_data, current_user.id)
except ValueError as e:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=str(e)
)
except Exception as e:
await db.rollback()
logger.error(f"할일 분할 실패: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=str(e)
)
@router.get("/", response_model=List[TodoItemResponse])
async def get_todo_items(
status: Optional[str] = Query(None, regex="^(draft|scheduled|active|completed|delayed)$"),
current_user: User = Depends(get_current_active_user),
db: AsyncSession = Depends(get_db)
):
"""할일 목록 조회"""
try:
service = TodoService(db)
return await service.get_todos(current_user.id, status)
except Exception as e:
logger.error(f"할일 목록 조회 실패: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=str(e)
)
@router.get("/active", response_model=List[TodoItemResponse])
async def get_active_todos(
current_user: User = Depends(get_current_active_user),
db: AsyncSession = Depends(get_db)
):
"""오늘 활성화된 할일들 조회"""
try:
service = TodoService(db)
return await service.get_active_todos(current_user.id)
except Exception as e:
logger.error(f"활성 할일 조회 실패: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=str(e)
)
# ============================================================================
# 댓글 관리
# ============================================================================
@router.post("/{todo_id}/comments", response_model=TodoCommentResponse)
async def create_todo_comment(
todo_id: UUID,
comment_data: TodoCommentCreate,
current_user: User = Depends(get_current_active_user),
db: AsyncSession = Depends(get_db)
):
"""할일에 댓글 추가"""
try:
service = TodoService(db)
return await service.create_comment(todo_id, comment_data, current_user.id)
except ValueError as e:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=str(e)
)
except Exception as e:
await db.rollback()
logger.error(f"댓글 생성 실패: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=str(e)
)
@router.get("/{todo_id}/comments", response_model=List[TodoCommentResponse])
async def get_todo_comments(
todo_id: UUID,
current_user: User = Depends(get_current_active_user),
db: AsyncSession = Depends(get_db)
):
"""할일 댓글 목록 조회"""
try:
service = TodoService(db)
return await service.get_comments(todo_id, current_user.id)
except ValueError as e:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=str(e)
)
except Exception as e:
logger.error(f"댓글 조회 실패: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=str(e)
)

View File

View File

@@ -0,0 +1,44 @@
"""
Todo Project 애플리케이션 설정
"""
from pydantic_settings import BaseSettings
from typing import List
import os
class Settings(BaseSettings):
"""애플리케이션 설정 클래스"""
# 기본 설정
APP_NAME: str = "Todo Project"
DEBUG: bool = True
VERSION: str = "0.1.0"
# 데이터베이스 설정
DATABASE_URL: str = "postgresql+asyncpg://todo_user:todo_password@localhost:5434/todo_db"
# JWT 설정
SECRET_KEY: str = "your-secret-key-change-this-in-production"
ALGORITHM: str = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
REFRESH_TOKEN_EXPIRE_DAYS: int = 7
# CORS 설정
ALLOWED_HOSTS: List[str] = ["http://localhost:4000", "http://127.0.0.1:4000"]
ALLOWED_ORIGINS: List[str] = ["http://localhost:4000", "http://127.0.0.1:4000"]
# 서버 설정
HOST: str = "0.0.0.0"
PORT: int = 9000
# 관리자 계정 설정 (초기 설정용)
ADMIN_EMAIL: str = "admin@todo-project.local"
ADMIN_PASSWORD: str = "admin123" # 프로덕션에서는 반드시 변경
class Config:
env_file = ".env"
case_sensitive = True
# 설정 인스턴스 생성
settings = Settings()

View File

@@ -0,0 +1,94 @@
"""
데이터베이스 설정 및 연결
"""
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy import MetaData
from typing import AsyncGenerator
from .config import settings
# SQLAlchemy 메타데이터 설정
metadata = MetaData(
naming_convention={
"ix": "ix_%(column_0_label)s",
"uq": "uq_%(table_name)s_%(column_0_name)s",
"ck": "ck_%(table_name)s_%(constraint_name)s",
"fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
"pk": "pk_%(table_name)s"
}
)
class Base(DeclarativeBase):
"""SQLAlchemy Base 클래스"""
metadata = metadata
# 비동기 데이터베이스 엔진 생성
engine = create_async_engine(
settings.DATABASE_URL,
echo=settings.DEBUG,
future=True,
pool_pre_ping=True,
pool_recycle=300,
)
# 비동기 세션 팩토리
AsyncSessionLocal = async_sessionmaker(
engine,
class_=AsyncSession,
expire_on_commit=False,
)
async def get_db() -> AsyncGenerator[AsyncSession, None]:
"""비동기 데이터베이스 세션 의존성"""
async with AsyncSessionLocal() as session:
try:
yield session
except Exception:
await session.rollback()
raise
finally:
await session.close()
async def init_db() -> None:
"""데이터베이스 초기화"""
from ..models import user, todo
async with engine.begin() as conn:
# 모든 테이블 생성
await conn.run_sync(Base.metadata.create_all)
# 관리자 계정 생성
await create_admin_user()
async def create_admin_user() -> None:
"""관리자 계정 생성 (존재하지 않을 경우)"""
from ..models.user import User
from .security import get_password_hash
from sqlalchemy import select
async with AsyncSessionLocal() as session:
# 관리자 계정 존재 확인
result = await session.execute(
select(User).where(User.email == settings.ADMIN_EMAIL)
)
admin_user = result.scalar_one_or_none()
if not admin_user:
# 관리자 계정 생성
admin_user = User(
email=settings.ADMIN_EMAIL,
hashed_password=get_password_hash(settings.ADMIN_PASSWORD),
is_active=True,
is_admin=True,
full_name="Administrator"
)
session.add(admin_user)
await session.commit()
print(f"관리자 계정이 생성되었습니다: {settings.ADMIN_EMAIL}")

View File

@@ -0,0 +1,94 @@
"""
보안 관련 유틸리티
"""
from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import HTTPException, status
from .config import settings
# 비밀번호 해싱 컨텍스트
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def verify_password(plain_password: str, hashed_password: str) -> bool:
"""비밀번호 검증"""
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
"""비밀번호 해싱"""
return pwd_context.hash(password)
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None, timeout_minutes: Optional[int] = None) -> str:
"""액세스 토큰 생성"""
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
elif timeout_minutes is not None:
if timeout_minutes == 0:
# 무제한 토큰 (1년으로 설정)
expire = datetime.utcnow() + timedelta(days=365)
else:
expire = datetime.utcnow() + timedelta(minutes=timeout_minutes)
else:
expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire, "type": "access"})
encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
return encoded_jwt
def create_refresh_token(data: dict) -> str:
"""리프레시 토큰 생성"""
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS)
to_encode.update({"exp": expire, "type": "refresh"})
encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
return encoded_jwt
def verify_token(token: str, token_type: str = "access") -> dict:
"""토큰 검증"""
try:
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
# 토큰 타입 확인
if payload.get("type") != token_type:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token type"
)
# 만료 시간 확인
exp = payload.get("exp")
if exp is None or datetime.utcnow() > datetime.fromtimestamp(exp):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Token expired"
)
return payload
except JWTError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials"
)
def get_user_id_from_token(token: str) -> str:
"""토큰에서 사용자 ID 추출"""
payload = verify_token(token)
user_id = payload.get("sub")
if user_id is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials"
)
return user_id

View File

View File

@@ -0,0 +1,32 @@
"""
캘린더 통합 모듈
- 다중 캘린더 제공자 지원 (시놀로지, 애플, 구글 등)
- 간결한 API로 Todo 항목을 캘린더에 동기화
"""
from .base import BaseCalendarService, CalendarProvider
from .synology import SynologyCalendarService
from .apple import AppleCalendarService, create_apple_service, format_todo_for_apple
from .router import CalendarRouter, get_calendar_router, setup_calendar_providers
__all__ = [
# 기본 인터페이스
"BaseCalendarService",
"CalendarProvider",
# 서비스 구현체
"SynologyCalendarService",
"AppleCalendarService",
# 라우터 및 관리
"CalendarRouter",
"get_calendar_router",
"setup_calendar_providers",
# 편의 함수
"create_apple_service",
"format_todo_for_apple",
]
# 버전 정보
__version__ = "1.0.0"

View File

@@ -0,0 +1,370 @@
"""
Apple iCloud Calendar 서비스 구현
- 통합 서비스 파일 기준: 최대 500줄
- 간결함 원칙: 한 파일에서 모든 Apple 캘린더 기능 제공
"""
import asyncio
import aiohttp
from typing import Dict, List, Any, Optional
from datetime import datetime, timedelta
import logging
from urllib.parse import urljoin
import xml.etree.ElementTree as ET
from base64 import b64encode
from .base import BaseCalendarService, CalendarProvider
logger = logging.getLogger(__name__)
class AppleCalendarService(BaseCalendarService):
"""Apple iCloud 캘린더 서비스 (CalDAV 기반)"""
def __init__(self):
self.base_url = "https://caldav.icloud.com"
self.session: Optional[aiohttp.ClientSession] = None
self.auth_header: Optional[str] = None
self.principal_url: Optional[str] = None
self.calendar_home_url: Optional[str] = None
async def authenticate(self, credentials: Dict[str, Any]) -> bool:
"""
Apple ID 및 앱 전용 암호로 인증
credentials: {"apple_id": "user@icloud.com", "app_password": "xxxx-xxxx-xxxx-xxxx"}
"""
try:
apple_id = credentials.get("apple_id")
app_password = credentials.get("app_password")
if not apple_id or not app_password:
logger.error("Apple ID 또는 앱 전용 암호가 누락됨")
return False
# Basic Auth 헤더 생성
auth_string = f"{apple_id}:{app_password}"
auth_bytes = auth_string.encode('utf-8')
self.auth_header = f"Basic {b64encode(auth_bytes).decode('utf-8')}"
# HTTP 세션 생성
self.session = aiohttp.ClientSession(
headers={"Authorization": self.auth_header}
)
# Principal URL 찾기
if not await self._discover_principal():
return False
# Calendar Home URL 찾기
if not await self._discover_calendar_home():
return False
logger.info(f"Apple 캘린더 인증 성공: {apple_id}")
return True
except Exception as e:
logger.error(f"Apple 캘린더 인증 실패: {e}")
return False
async def _discover_principal(self) -> bool:
"""Principal URL 검색 (CalDAV 표준)"""
try:
propfind_body = """<?xml version="1.0" encoding="utf-8" ?>
<d:propfind xmlns:d="DAV:">
<d:prop>
<d:current-user-principal />
</d:prop>
</d:propfind>"""
async with self.session.request(
"PROPFIND",
self.base_url,
data=propfind_body,
headers={"Content-Type": "application/xml", "Depth": "0"}
) as response:
if response.status != 207:
logger.error(f"Principal 검색 실패: {response.status}")
return False
xml_content = await response.text()
root = ET.fromstring(xml_content)
# Principal URL 추출
for elem in root.iter():
if elem.tag.endswith("current-user-principal"):
href = elem.find(".//{DAV:}href")
if href is not None:
self.principal_url = urljoin(self.base_url, href.text)
return True
logger.error("Principal URL을 찾을 수 없음")
return False
except Exception as e:
logger.error(f"Principal 검색 중 오류: {e}")
return False
async def _discover_calendar_home(self) -> bool:
"""Calendar Home URL 검색"""
try:
propfind_body = """<?xml version="1.0" encoding="utf-8" ?>
<d:propfind xmlns:d="DAV:" xmlns:c="urn:ietf:params:xml:ns:caldav">
<d:prop>
<c:calendar-home-set />
</d:prop>
</d:propfind>"""
async with self.session.request(
"PROPFIND",
self.principal_url,
data=propfind_body,
headers={"Content-Type": "application/xml", "Depth": "0"}
) as response:
if response.status != 207:
logger.error(f"Calendar Home 검색 실패: {response.status}")
return False
xml_content = await response.text()
root = ET.fromstring(xml_content)
# Calendar Home URL 추출
for elem in root.iter():
if elem.tag.endswith("calendar-home-set"):
href = elem.find(".//{DAV:}href")
if href is not None:
self.calendar_home_url = urljoin(self.base_url, href.text)
return True
logger.error("Calendar Home URL을 찾을 수 없음")
return False
except Exception as e:
logger.error(f"Calendar Home 검색 중 오류: {e}")
return False
async def get_calendars(self) -> List[Dict[str, Any]]:
"""사용 가능한 캘린더 목록 조회"""
try:
propfind_body = """<?xml version="1.0" encoding="utf-8" ?>
<d:propfind xmlns:d="DAV:" xmlns:c="urn:ietf:params:xml:ns:caldav">
<d:prop>
<d:displayname />
<d:resourcetype />
<c:calendar-description />
<c:supported-calendar-component-set />
</d:prop>
</d:propfind>"""
async with self.session.request(
"PROPFIND",
self.calendar_home_url,
data=propfind_body,
headers={"Content-Type": "application/xml", "Depth": "1"}
) as response:
if response.status != 207:
logger.error(f"캘린더 목록 조회 실패: {response.status}")
return []
xml_content = await response.text()
return self._parse_calendars(xml_content)
except Exception as e:
logger.error(f"캘린더 목록 조회 중 오류: {e}")
return []
def _parse_calendars(self, xml_content: str) -> List[Dict[str, Any]]:
"""캘린더 XML 응답 파싱"""
calendars = []
root = ET.fromstring(xml_content)
for response in root.findall(".//{DAV:}response"):
# 캘린더인지 확인
resourcetype = response.find(".//{DAV:}resourcetype")
if resourcetype is None:
continue
is_calendar = resourcetype.find(".//{urn:ietf:params:xml:ns:caldav}calendar") is not None
if not is_calendar:
continue
# 캘린더 정보 추출
href_elem = response.find(".//{DAV:}href")
name_elem = response.find(".//{DAV:}displayname")
desc_elem = response.find(".//{urn:ietf:params:xml:ns:caldav}calendar-description")
if href_elem is not None and name_elem is not None:
calendar = {
"id": href_elem.text.split("/")[-2], # URL에서 ID 추출
"name": name_elem.text or "이름 없음",
"description": desc_elem.text if desc_elem is not None else "",
"url": urljoin(self.base_url, href_elem.text),
"provider": CalendarProvider.APPLE.value
}
calendars.append(calendar)
return calendars
async def create_event(self, calendar_id: str, event_data: Dict[str, Any]) -> Dict[str, Any]:
"""이벤트 생성 (iCalendar 형식)"""
try:
# iCalendar 이벤트 생성
ics_content = self._create_ics_event(event_data)
# 이벤트 URL 생성 (UUID 기반)
import uuid
event_id = str(uuid.uuid4())
event_url = f"{self.calendar_home_url}{calendar_id}/{event_id}.ics"
# PUT 요청으로 이벤트 생성
async with self.session.put(
event_url,
data=ics_content,
headers={"Content-Type": "text/calendar; charset=utf-8"}
) as response:
if response.status not in [201, 204]:
logger.error(f"이벤트 생성 실패: {response.status}")
return {}
logger.info(f"Apple 캘린더 이벤트 생성 성공: {event_id}")
return {
"id": event_id,
"url": event_url,
"provider": CalendarProvider.APPLE.value
}
except Exception as e:
logger.error(f"Apple 캘린더 이벤트 생성 중 오류: {e}")
return {}
def _create_ics_event(self, event_data: Dict[str, Any]) -> str:
"""iCalendar 형식 이벤트 생성"""
title = event_data.get("title", "제목 없음")
description = event_data.get("description", "")
start_time = event_data.get("start_time")
end_time = event_data.get("end_time")
# 시간 형식 변환
if isinstance(start_time, datetime):
start_str = start_time.strftime("%Y%m%dT%H%M%SZ")
else:
start_str = datetime.now().strftime("%Y%m%dT%H%M%SZ")
if isinstance(end_time, datetime):
end_str = end_time.strftime("%Y%m%dT%H%M%SZ")
else:
end_str = (datetime.now() + timedelta(hours=1)).strftime("%Y%m%dT%H%M%SZ")
# iCalendar 내용 생성
ics_content = f"""BEGIN:VCALENDAR
VERSION:2.0
PRODID:-//Todo-Project//Apple Calendar//KO
BEGIN:VEVENT
UID:{event_data.get('uid', str(uuid.uuid4()))}
DTSTART:{start_str}
DTEND:{end_str}
SUMMARY:{title}
DESCRIPTION:{description}
CREATED:{datetime.now().strftime("%Y%m%dT%H%M%SZ")}
LAST-MODIFIED:{datetime.now().strftime("%Y%m%dT%H%M%SZ")}
END:VEVENT
END:VCALENDAR"""
return ics_content
async def update_event(self, event_id: str, event_data: Dict[str, Any]) -> Dict[str, Any]:
"""이벤트 수정"""
try:
# 기존 이벤트 URL 구성
calendar_id = event_data.get("calendar_id", "calendar")
event_url = f"{self.calendar_home_url}{calendar_id}/{event_id}.ics"
# 수정된 iCalendar 내용 생성
ics_content = self._create_ics_event(event_data)
# PUT 요청으로 이벤트 수정
async with self.session.put(
event_url,
data=ics_content,
headers={"Content-Type": "text/calendar; charset=utf-8"}
) as response:
if response.status not in [200, 204]:
logger.error(f"이벤트 수정 실패: {response.status}")
return {}
logger.info(f"Apple 캘린더 이벤트 수정 성공: {event_id}")
return {
"id": event_id,
"url": event_url,
"provider": CalendarProvider.APPLE.value
}
except Exception as e:
logger.error(f"Apple 캘린더 이벤트 수정 중 오류: {e}")
return {}
async def delete_event(self, event_id: str, calendar_id: str = "calendar") -> bool:
"""이벤트 삭제"""
try:
# 이벤트 URL 구성
event_url = f"{self.calendar_home_url}{calendar_id}/{event_id}.ics"
# DELETE 요청
async with self.session.delete(event_url) as response:
if response.status not in [200, 204, 404]:
logger.error(f"이벤트 삭제 실패: {response.status}")
return False
logger.info(f"Apple 캘린더 이벤트 삭제 성공: {event_id}")
return True
except Exception as e:
logger.error(f"Apple 캘린더 이벤트 삭제 중 오류: {e}")
return False
async def close(self):
"""세션 정리"""
if self.session:
await self.session.close()
self.session = None
def __del__(self):
"""소멸자에서 세션 정리"""
if self.session and not self.session.closed:
asyncio.create_task(self.close())
# 편의 함수들
async def create_apple_service(apple_id: str, app_password: str) -> Optional[AppleCalendarService]:
"""Apple 캘린더 서비스 생성 및 인증"""
service = AppleCalendarService()
credentials = {
"apple_id": apple_id,
"app_password": app_password
}
if await service.authenticate(credentials):
return service
else:
await service.close()
return None
def format_todo_for_apple(todo_item) -> Dict[str, Any]:
"""Todo 아이템을 Apple 캘린더 이벤트 형식으로 변환"""
import uuid
return {
"uid": str(uuid.uuid4()),
"title": f"📋 {todo_item.content}",
"description": f"Todo 항목\n상태: {todo_item.status}\n생성일: {todo_item.created_at}",
"start_time": todo_item.start_date or todo_item.created_at,
"end_time": (todo_item.start_date or todo_item.created_at) + timedelta(
minutes=todo_item.estimated_minutes or 30
),
"categories": ["todo", "업무"]
}

View File

@@ -0,0 +1,332 @@
"""
캘린더 서비스 기본 인터페이스 및 추상화
"""
from abc import ABC, abstractmethod
from enum import Enum
from typing import Dict, List, Optional, Any
from datetime import datetime, timedelta
from dataclasses import dataclass
import uuid
class CalendarProvider(Enum):
"""캘린더 제공자 열거형"""
SYNOLOGY = "synology"
APPLE = "apple"
GOOGLE = "google"
CALDAV = "caldav" # 일반 CalDAV 서버
@dataclass
class CalendarInfo:
"""캘린더 정보"""
id: str
name: str
color: str
description: Optional[str] = None
provider: CalendarProvider = CalendarProvider.CALDAV
is_default: bool = False
is_writable: bool = True
@dataclass
class CalendarEvent:
"""캘린더 이벤트"""
id: Optional[str] = None
title: str = ""
description: Optional[str] = None
start_time: Optional[datetime] = None
end_time: Optional[datetime] = None
all_day: bool = False
location: Optional[str] = None
categories: List[str] = None
color: Optional[str] = None
reminder_minutes: Optional[int] = None
status: str = "TENTATIVE" # TENTATIVE, CONFIRMED, CANCELLED
def __post_init__(self):
if self.categories is None:
self.categories = []
if self.id is None:
self.id = str(uuid.uuid4())
@dataclass
class CalendarCredentials:
"""캘린더 인증 정보"""
provider: CalendarProvider
server_url: Optional[str] = None
username: Optional[str] = None
password: Optional[str] = None
app_password: Optional[str] = None # Apple 앱 전용 비밀번호
oauth_token: Optional[str] = None # OAuth 토큰
additional_params: Dict[str, Any] = None
def __post_init__(self):
if self.additional_params is None:
self.additional_params = {}
class CalendarServiceError(Exception):
"""캘린더 서비스 오류"""
pass
class AuthenticationError(CalendarServiceError):
"""인증 오류"""
pass
class CalendarNotFoundError(CalendarServiceError):
"""캘린더를 찾을 수 없음"""
pass
class EventNotFoundError(CalendarServiceError):
"""이벤트를 찾을 수 없음"""
pass
class BaseCalendarService(ABC):
"""캘린더 서비스 기본 인터페이스"""
def __init__(self, credentials: CalendarCredentials):
self.credentials = credentials
self.provider = credentials.provider
self._authenticated = False
self._client = None
@property
def is_authenticated(self) -> bool:
"""인증 상태 확인"""
return self._authenticated
@abstractmethod
async def authenticate(self) -> bool:
"""
인증 수행
Returns:
bool: 인증 성공 여부
Raises:
AuthenticationError: 인증 실패 시
"""
pass
@abstractmethod
async def get_calendars(self) -> List[CalendarInfo]:
"""
사용 가능한 캘린더 목록 조회
Returns:
List[CalendarInfo]: 캘린더 목록
Raises:
CalendarServiceError: 조회 실패 시
"""
pass
@abstractmethod
async def create_event(self, calendar_id: str, event: CalendarEvent) -> CalendarEvent:
"""
이벤트 생성
Args:
calendar_id: 캘린더 ID
event: 생성할 이벤트 정보
Returns:
CalendarEvent: 생성된 이벤트 (ID 포함)
Raises:
CalendarNotFoundError: 캘린더를 찾을 수 없음
CalendarServiceError: 생성 실패
"""
pass
@abstractmethod
async def update_event(self, calendar_id: str, event: CalendarEvent) -> CalendarEvent:
"""
이벤트 수정
Args:
calendar_id: 캘린더 ID
event: 수정할 이벤트 정보 (ID 포함)
Returns:
CalendarEvent: 수정된 이벤트
Raises:
EventNotFoundError: 이벤트를 찾을 수 없음
CalendarServiceError: 수정 실패
"""
pass
@abstractmethod
async def delete_event(self, calendar_id: str, event_id: str) -> bool:
"""
이벤트 삭제
Args:
calendar_id: 캘린더 ID
event_id: 삭제할 이벤트 ID
Returns:
bool: 삭제 성공 여부
Raises:
EventNotFoundError: 이벤트를 찾을 수 없음
CalendarServiceError: 삭제 실패
"""
pass
@abstractmethod
async def get_event(self, calendar_id: str, event_id: str) -> CalendarEvent:
"""
특정 이벤트 조회
Args:
calendar_id: 캘린더 ID
event_id: 이벤트 ID
Returns:
CalendarEvent: 이벤트 정보
Raises:
EventNotFoundError: 이벤트를 찾을 수 없음
"""
pass
async def test_connection(self) -> Dict[str, Any]:
"""
연결 테스트
Returns:
Dict: 테스트 결과
"""
try:
success = await self.authenticate()
if success:
calendars = await self.get_calendars()
return {
"status": "success",
"provider": self.provider.value,
"calendar_count": len(calendars),
"calendars": [{"id": cal.id, "name": cal.name} for cal in calendars[:3]] # 처음 3개만
}
else:
return {
"status": "failed",
"provider": self.provider.value,
"error": "Authentication failed"
}
except Exception as e:
return {
"status": "error",
"provider": self.provider.value,
"error": str(e)
}
def _ensure_authenticated(self):
"""인증 상태 확인 (내부 사용)"""
if not self._authenticated:
raise AuthenticationError(f"{self.provider.value} 서비스에 인증되지 않았습니다.")
class TodoEventConverter:
"""Todo 아이템을 캘린더 이벤트로 변환하는 유틸리티"""
@staticmethod
def todo_to_event(todo_item, provider: CalendarProvider) -> CalendarEvent:
"""
할일을 캘린더 이벤트로 변환
Args:
todo_item: Todo 아이템
provider: 캘린더 제공자
Returns:
CalendarEvent: 변환된 이벤트
"""
# 상태별 아이콘 및 색상
status_icons = {
"draft": "📝",
"scheduled": "📋",
"active": "🔥",
"completed": "",
"delayed": ""
}
status_colors = {
"draft": "#9ca3af", # 회색
"scheduled": "#6366f1", # 보라색
"active": "#f59e0b", # 주황색
"completed": "#10b981", # 초록색
"delayed": "#ef4444" # 빨간색
}
icon = status_icons.get(todo_item.status, "📋")
color = status_colors.get(todo_item.status, "#6366f1")
# 시작/종료 시간 계산
start_time = todo_item.start_date
end_time = start_time + timedelta(minutes=todo_item.estimated_minutes or 30)
# 기본 이벤트 생성
event = CalendarEvent(
title=f"{icon} {todo_item.content}",
description=f"Todo ID: {todo_item.id}\nStatus: {todo_item.status}\nEstimated: {todo_item.estimated_minutes or 30}",
start_time=start_time,
end_time=end_time,
categories=["완료" if todo_item.status == "completed" else "todo"],
color=color,
reminder_minutes=15,
status="CONFIRMED" if todo_item.status == "completed" else "TENTATIVE"
)
# 제공자별 커스터마이징
if provider == CalendarProvider.APPLE:
# 애플 캘린더 특화
event.color = "#6366f1" # 보라색으로 통일
event.reminder_minutes = 15
elif provider == CalendarProvider.SYNOLOGY:
# 시놀로지 캘린더 특화
event.location = "Todo-Project"
elif provider == CalendarProvider.GOOGLE:
# 구글 캘린더 특화
event.color = "#4285f4" # 구글 블루
return event
@staticmethod
def get_provider_specific_properties(event: CalendarEvent, provider: CalendarProvider) -> Dict[str, Any]:
"""
제공자별 특화 속성 반환
Args:
event: 캘린더 이벤트
provider: 캘린더 제공자
Returns:
Dict: 제공자별 특화 속성
"""
if provider == CalendarProvider.APPLE:
return {
"X-APPLE-STRUCTURED-LOCATION": event.location,
"X-APPLE-CALENDAR-COLOR": event.color
}
elif provider == CalendarProvider.SYNOLOGY:
return {
"PRIORITY": "5",
"CLASS": "PRIVATE"
}
elif provider == CalendarProvider.GOOGLE:
return {
"colorId": "9", # 파란색
"visibility": "private"
}
else:
return {}

View File

@@ -0,0 +1,363 @@
"""
캘린더 라우터 - 다중 캘린더 제공자 중앙 관리
- 서비스 클래스 기준: 최대 350줄
- 간결함 원칙: 단순한 라우팅과 조합 로직만 포함
"""
import asyncio
from typing import Dict, List, Any, Optional, Union
from enum import Enum
import logging
from .base import BaseCalendarService, CalendarProvider
from .synology import SynologyCalendarService
from .apple import AppleCalendarService
logger = logging.getLogger(__name__)
class CalendarRouter:
"""다중 캘린더 제공자를 관리하는 중앙 라우터"""
def __init__(self):
self.services: Dict[CalendarProvider, BaseCalendarService] = {}
self.default_provider: Optional[CalendarProvider] = None
async def register_provider(
self,
provider: CalendarProvider,
credentials: Dict[str, Any],
set_as_default: bool = False
) -> bool:
"""캘린더 제공자 등록 및 인증"""
try:
# 서비스 인스턴스 생성
service = self._create_service(provider)
if not service:
logger.error(f"지원하지 않는 캘린더 제공자: {provider}")
return False
# 인증 시도
if not await service.authenticate(credentials):
logger.error(f"{provider.value} 캘린더 인증 실패")
return False
# 등록 완료
self.services[provider] = service
if set_as_default or not self.default_provider:
self.default_provider = provider
logger.info(f"{provider.value} 캘린더 제공자 등록 완료")
return True
except Exception as e:
logger.error(f"캘린더 제공자 등록 중 오류: {e}")
return False
def _create_service(self, provider: CalendarProvider) -> Optional[BaseCalendarService]:
"""제공자별 서비스 인스턴스 생성"""
service_map = {
CalendarProvider.SYNOLOGY: SynologyCalendarService,
CalendarProvider.APPLE: AppleCalendarService,
# 추후 확장: CalendarProvider.GOOGLE: GoogleCalendarService,
}
service_class = service_map.get(provider)
return service_class() if service_class else None
async def get_all_calendars(self) -> Dict[str, List[Dict[str, Any]]]:
"""모든 등록된 제공자의 캘린더 목록 조회"""
all_calendars = {}
for provider, service in self.services.items():
try:
calendars = await service.get_calendars()
all_calendars[provider.value] = calendars
logger.info(f"{provider.value}: {len(calendars)}개 캘린더 발견")
except Exception as e:
logger.error(f"{provider.value} 캘린더 목록 조회 실패: {e}")
all_calendars[provider.value] = []
return all_calendars
async def get_calendars(self, provider: Optional[CalendarProvider] = None) -> List[Dict[str, Any]]:
"""특정 제공자 또는 기본 제공자의 캘린더 목록 조회"""
target_provider = provider or self.default_provider
if not target_provider or target_provider not in self.services:
logger.error(f"캘린더 제공자를 찾을 수 없음: {target_provider}")
return []
try:
return await self.services[target_provider].get_calendars()
except Exception as e:
logger.error(f"캘린더 목록 조회 실패: {e}")
return []
async def create_event(
self,
calendar_id: str,
event_data: Dict[str, Any],
provider: Optional[CalendarProvider] = None
) -> Dict[str, Any]:
"""이벤트 생성 (특정 제공자 또는 기본 제공자)"""
target_provider = provider or self.default_provider
if not target_provider or target_provider not in self.services:
logger.error(f"캘린더 제공자를 찾을 수 없음: {target_provider}")
return {}
try:
result = await self.services[target_provider].create_event(calendar_id, event_data)
result["provider"] = target_provider.value
return result
except Exception as e:
logger.error(f"이벤트 생성 실패: {e}")
return {}
async def create_event_multi(
self,
calendar_configs: List[Dict[str, Any]],
event_data: Dict[str, Any]
) -> Dict[str, List[Dict[str, Any]]]:
"""
여러 캘린더에 동시 이벤트 생성
calendar_configs: [{"provider": "synology", "calendar_id": "personal"}, ...]
"""
results = {"success": [], "failed": []}
# 병렬 처리를 위한 태스크 생성
tasks = []
for config in calendar_configs:
provider_str = config.get("provider")
calendar_id = config.get("calendar_id")
try:
provider = CalendarProvider(provider_str)
if provider in self.services:
task = self._create_event_task(provider, calendar_id, event_data, config)
tasks.append(task)
else:
results["failed"].append({
"provider": provider_str,
"calendar_id": calendar_id,
"error": "제공자를 찾을 수 없음"
})
except ValueError:
results["failed"].append({
"provider": provider_str,
"calendar_id": calendar_id,
"error": "지원하지 않는 제공자"
})
# 병렬 실행
if tasks:
task_results = await asyncio.gather(*tasks, return_exceptions=True)
for i, result in enumerate(task_results):
config = calendar_configs[i]
if isinstance(result, Exception):
results["failed"].append({
"provider": config.get("provider"),
"calendar_id": config.get("calendar_id"),
"error": str(result)
})
else:
results["success"].append(result)
return results
async def _create_event_task(
self,
provider: CalendarProvider,
calendar_id: str,
event_data: Dict[str, Any],
config: Dict[str, Any]
) -> Dict[str, Any]:
"""이벤트 생성 태스크 (병렬 처리용)"""
try:
result = await self.services[provider].create_event(calendar_id, event_data)
result.update({
"provider": provider.value,
"calendar_id": calendar_id,
"config": config
})
return result
except Exception as e:
raise Exception(f"{provider.value} 이벤트 생성 실패: {e}")
async def update_event(
self,
event_id: str,
event_data: Dict[str, Any],
provider: Optional[CalendarProvider] = None
) -> Dict[str, Any]:
"""이벤트 수정"""
target_provider = provider or self.default_provider
if not target_provider or target_provider not in self.services:
logger.error(f"캘린더 제공자를 찾을 수 없음: {target_provider}")
return {}
try:
result = await self.services[target_provider].update_event(event_id, event_data)
result["provider"] = target_provider.value
return result
except Exception as e:
logger.error(f"이벤트 수정 실패: {e}")
return {}
async def delete_event(
self,
event_id: str,
provider: Optional[CalendarProvider] = None,
**kwargs
) -> bool:
"""이벤트 삭제"""
target_provider = provider or self.default_provider
if not target_provider or target_provider not in self.services:
logger.error(f"캘린더 제공자를 찾을 수 없음: {target_provider}")
return False
try:
return await self.services[target_provider].delete_event(event_id, **kwargs)
except Exception as e:
logger.error(f"이벤트 삭제 실패: {e}")
return False
async def sync_todo_to_calendars(
self,
todo_item,
calendar_configs: Optional[List[Dict[str, Any]]] = None
) -> Dict[str, Any]:
"""Todo 아이템을 여러 캘린더에 동기화"""
if not calendar_configs:
# 기본 제공자만 사용
if not self.default_provider:
logger.error("기본 캘린더 제공자가 설정되지 않음")
return {"success": [], "failed": []}
calendar_configs = [{
"provider": self.default_provider.value,
"calendar_id": "default"
}]
# Todo를 캘린더 이벤트 형식으로 변환
event_data = self._format_todo_event(todo_item)
# 여러 캘린더에 생성
return await self.create_event_multi(calendar_configs, event_data)
def _format_todo_event(self, todo_item) -> Dict[str, Any]:
"""Todo 아이템을 캘린더 이벤트 형식으로 변환"""
from datetime import datetime, timedelta
import uuid
# 상태별 태그 설정
status_tag = "완료" if todo_item.status == "completed" else "todo"
return {
"uid": str(uuid.uuid4()),
"title": f"📋 {todo_item.content}",
"description": f"Todo 항목\n상태: {status_tag}\n생성일: {todo_item.created_at}",
"start_time": todo_item.start_date or todo_item.created_at,
"end_time": (todo_item.start_date or todo_item.created_at) + timedelta(
minutes=todo_item.estimated_minutes or 30
),
"categories": [status_tag, "업무"],
"todo_id": todo_item.id
}
def get_registered_providers(self) -> List[str]:
"""등록된 캘린더 제공자 목록 반환"""
return [provider.value for provider in self.services.keys()]
def set_default_provider(self, provider: CalendarProvider) -> bool:
"""기본 캘린더 제공자 설정"""
if provider in self.services:
self.default_provider = provider
logger.info(f"기본 캘린더 제공자 변경: {provider.value}")
return True
else:
logger.error(f"등록되지 않은 제공자: {provider.value}")
return False
async def health_check(self) -> Dict[str, Any]:
"""모든 등록된 캘린더 서비스 상태 확인"""
health_status = {
"total_providers": len(self.services),
"default_provider": self.default_provider.value if self.default_provider else None,
"providers": {}
}
for provider, service in self.services.items():
try:
# 간단한 캘린더 목록 조회로 상태 확인
calendars = await service.get_calendars()
health_status["providers"][provider.value] = {
"status": "healthy",
"calendar_count": len(calendars)
}
except Exception as e:
health_status["providers"][provider.value] = {
"status": "error",
"error": str(e)
}
return health_status
async def close_all(self):
"""모든 캘린더 서비스 연결 종료"""
for provider, service in self.services.items():
try:
if hasattr(service, 'close'):
await service.close()
logger.info(f"{provider.value} 캘린더 서비스 연결 종료")
except Exception as e:
logger.error(f"{provider.value} 연결 종료 중 오류: {e}")
self.services.clear()
self.default_provider = None
# 전역 라우터 인스턴스 (싱글톤 패턴)
_calendar_router: Optional[CalendarRouter] = None
def get_calendar_router() -> CalendarRouter:
"""캘린더 라우터 싱글톤 인스턴스 반환"""
global _calendar_router
if _calendar_router is None:
_calendar_router = CalendarRouter()
return _calendar_router
async def setup_calendar_providers(providers_config: Dict[str, Dict[str, Any]]) -> CalendarRouter:
"""
캘린더 제공자들을 일괄 설정
providers_config: {
"synology": {"credentials": {...}, "default": True},
"apple": {"credentials": {...}, "default": False}
}
"""
router = get_calendar_router()
for provider_name, config in providers_config.items():
try:
provider = CalendarProvider(provider_name)
credentials = config.get("credentials", {})
is_default = config.get("default", False)
success = await router.register_provider(provider, credentials, is_default)
if success:
logger.info(f"{provider_name} 캘린더 설정 완료")
else:
logger.error(f"{provider_name} 캘린더 설정 실패")
except ValueError:
logger.error(f"지원하지 않는 캘린더 제공자: {provider_name}")
except Exception as e:
logger.error(f"{provider_name} 캘린더 설정 중 오류: {e}")
return router

View File

@@ -0,0 +1,401 @@
"""
시놀로지 캘린더 서비스 구현
"""
import asyncio
import aiohttp
from typing import List, Dict, Any, Optional
from datetime import datetime, timedelta
import caldav
from caldav.lib.error import AuthorizationError, NotFoundError
import logging
from .base import (
BaseCalendarService, CalendarProvider, CalendarInfo, CalendarEvent,
CalendarCredentials, CalendarServiceError, AuthenticationError,
CalendarNotFoundError, EventNotFoundError
)
logger = logging.getLogger(__name__)
class SynologyCalendarService(BaseCalendarService):
"""시놀로지 캘린더 서비스"""
def __init__(self, credentials: CalendarCredentials):
super().__init__(credentials)
self.dsm_url = credentials.server_url
self.username = credentials.username
self.password = credentials.password
self.session_token = None
self.caldav_client = None
# CalDAV URL 구성
if self.dsm_url and self.username:
self.caldav_url = f"{self.dsm_url}/caldav/{self.username}/"
async def authenticate(self) -> bool:
"""
시놀로지 DSM 및 CalDAV 인증
"""
try:
# 1. DSM API 인증 (선택사항 - 추가 기능용)
await self._authenticate_dsm()
# 2. CalDAV 인증 (메인)
await self._authenticate_caldav()
self._authenticated = True
logger.info(f"시놀로지 캘린더 인증 성공: {self.username}")
return True
except Exception as e:
logger.error(f"시놀로지 캘린더 인증 실패: {e}")
raise AuthenticationError(f"시놀로지 인증 실패: {str(e)}")
async def _authenticate_dsm(self) -> Optional[str]:
"""DSM API 인증 (추가 기능용)"""
if not self.dsm_url:
return None
login_url = f"{self.dsm_url}/webapi/auth.cgi"
params = {
"api": "SYNO.API.Auth",
"version": "3",
"method": "login",
"account": self.username,
"passwd": self.password,
"session": "TodoProject",
"format": "sid"
}
try:
async with aiohttp.ClientSession() as session:
async with session.get(login_url, params=params, ssl=False) as response:
data = await response.json()
if data.get("success"):
self.session_token = data["data"]["sid"]
logger.info("DSM API 인증 성공")
return self.session_token
else:
error_code = data.get("error", {}).get("code", "Unknown")
raise AuthenticationError(f"DSM 로그인 실패 (코드: {error_code})")
except aiohttp.ClientError as e:
logger.warning(f"DSM API 인증 실패 (CalDAV는 계속 시도): {e}")
return None
async def _authenticate_caldav(self):
"""CalDAV 인증"""
try:
# CalDAV 클라이언트 생성
self.caldav_client = caldav.DAVClient(
url=self.caldav_url,
username=self.username,
password=self.password
)
# 연결 테스트
principal = self.caldav_client.principal()
calendars = principal.calendars()
logger.info(f"CalDAV 인증 성공: {len(calendars)}개 캘린더 발견")
except AuthorizationError as e:
raise AuthenticationError(f"CalDAV 인증 실패: 사용자명 또는 비밀번호가 잘못되었습니다")
except Exception as e:
raise AuthenticationError(f"CalDAV 연결 실패: {str(e)}")
async def get_calendars(self) -> List[CalendarInfo]:
"""캘린더 목록 조회"""
self._ensure_authenticated()
try:
principal = self.caldav_client.principal()
calendars = principal.calendars()
calendar_list = []
for calendar in calendars:
try:
# 캘린더 속성 조회
props = calendar.get_properties([
caldav.dav.DisplayName(),
caldav.elements.icalendar.CalendarColor(),
caldav.elements.icalendar.CalendarDescription(),
])
name = props.get(caldav.dav.DisplayName.tag, "Unknown Calendar")
color = props.get(caldav.elements.icalendar.CalendarColor.tag, "#6366f1")
description = props.get(caldav.elements.icalendar.CalendarDescription.tag, "")
# 색상 형식 정규화
if color and not color.startswith('#'):
color = f"#{color}"
calendar_info = CalendarInfo(
id=calendar.url,
name=name,
color=color or "#6366f1",
description=description,
provider=CalendarProvider.SYNOLOGY,
is_writable=True # 시놀로지 캘린더는 일반적으로 쓰기 가능
)
calendar_list.append(calendar_info)
except Exception as e:
logger.warning(f"캘린더 정보 조회 실패: {e}")
continue
logger.info(f"시놀로지 캘린더 {len(calendar_list)}개 조회 완료")
return calendar_list
except Exception as e:
logger.error(f"캘린더 목록 조회 실패: {e}")
raise CalendarServiceError(f"캘린더 목록 조회 실패: {str(e)}")
async def create_event(self, calendar_id: str, event: CalendarEvent) -> CalendarEvent:
"""이벤트 생성"""
self._ensure_authenticated()
try:
# 캘린더 객체 가져오기
calendar = self.caldav_client.calendar(url=calendar_id)
# ICS 형식으로 이벤트 생성
ics_content = self._event_to_ics(event)
# 이벤트 추가
caldav_event = calendar.add_event(ics_content)
# 생성된 이벤트 ID 설정
event.id = caldav_event.url
logger.info(f"시놀로지 캘린더 이벤트 생성 완료: {event.title}")
return event
except NotFoundError:
raise CalendarNotFoundError(f"캘린더를 찾을 수 없습니다: {calendar_id}")
except Exception as e:
logger.error(f"이벤트 생성 실패: {e}")
raise CalendarServiceError(f"이벤트 생성 실패: {str(e)}")
async def update_event(self, calendar_id: str, event: CalendarEvent) -> CalendarEvent:
"""이벤트 수정"""
self._ensure_authenticated()
try:
# 기존 이벤트 가져오기
calendar = self.caldav_client.calendar(url=calendar_id)
caldav_event = calendar.event_by_url(event.id)
# ICS 형식으로 업데이트
ics_content = self._event_to_ics(event)
caldav_event.data = ics_content
caldav_event.save()
logger.info(f"시놀로지 캘린더 이벤트 수정 완료: {event.title}")
return event
except NotFoundError:
raise EventNotFoundError(f"이벤트를 찾을 수 없습니다: {event.id}")
except Exception as e:
logger.error(f"이벤트 수정 실패: {e}")
raise CalendarServiceError(f"이벤트 수정 실패: {str(e)}")
async def delete_event(self, calendar_id: str, event_id: str) -> bool:
"""이벤트 삭제"""
self._ensure_authenticated()
try:
calendar = self.caldav_client.calendar(url=calendar_id)
caldav_event = calendar.event_by_url(event_id)
caldav_event.delete()
logger.info(f"시놀로지 캘린더 이벤트 삭제 완료: {event_id}")
return True
except NotFoundError:
raise EventNotFoundError(f"이벤트를 찾을 수 없습니다: {event_id}")
except Exception as e:
logger.error(f"이벤트 삭제 실패: {e}")
raise CalendarServiceError(f"이벤트 삭제 실패: {str(e)}")
async def get_event(self, calendar_id: str, event_id: str) -> CalendarEvent:
"""이벤트 조회"""
self._ensure_authenticated()
try:
calendar = self.caldav_client.calendar(url=calendar_id)
caldav_event = calendar.event_by_url(event_id)
# ICS에서 CalendarEvent로 변환
event = self._ics_to_event(caldav_event.data)
event.id = event_id
return event
except NotFoundError:
raise EventNotFoundError(f"이벤트를 찾을 수 없습니다: {event_id}")
except Exception as e:
logger.error(f"이벤트 조회 실패: {e}")
raise CalendarServiceError(f"이벤트 조회 실패: {str(e)}")
def _event_to_ics(self, event: CalendarEvent) -> str:
"""CalendarEvent를 ICS 형식으로 변환"""
# 시간 형식 변환
start_str = event.start_time.strftime('%Y%m%dT%H%M%S') if event.start_time else ""
end_str = event.end_time.strftime('%Y%m%dT%H%M%S') if event.end_time else ""
# 카테고리 문자열 생성
categories_str = ",".join(event.categories) if event.categories else ""
# 알림 설정
alarm_str = ""
if event.reminder_minutes:
alarm_str = f"""BEGIN:VALARM
TRIGGER:-PT{event.reminder_minutes}M
ACTION:DISPLAY
DESCRIPTION:Reminder
END:VALARM"""
ics_content = f"""BEGIN:VCALENDAR
VERSION:2.0
PRODID:-//Todo-Project//Synology Calendar//EN
BEGIN:VEVENT
UID:{event.id}
DTSTART:{start_str}
DTEND:{end_str}
SUMMARY:{event.title}
DESCRIPTION:{event.description or ''}
CATEGORIES:{categories_str}
STATUS:{event.status}
PRIORITY:5
CLASS:PRIVATE
{alarm_str}
END:VEVENT
END:VCALENDAR"""
return ics_content
def _ics_to_event(self, ics_content: str) -> CalendarEvent:
"""ICS 형식을 CalendarEvent로 변환"""
# 간단한 ICS 파싱 (실제로는 icalendar 라이브러리 사용 권장)
lines = ics_content.split('\n')
event = CalendarEvent()
for line in lines:
line = line.strip()
if line.startswith('SUMMARY:'):
event.title = line[8:]
elif line.startswith('DESCRIPTION:'):
event.description = line[12:]
elif line.startswith('DTSTART:'):
try:
event.start_time = datetime.strptime(line[8:], '%Y%m%dT%H%M%S')
except ValueError:
pass
elif line.startswith('DTEND:'):
try:
event.end_time = datetime.strptime(line[6:], '%Y%m%dT%H%M%S')
except ValueError:
pass
elif line.startswith('CATEGORIES:'):
categories = line[11:].split(',')
event.categories = [cat.strip() for cat in categories if cat.strip()]
elif line.startswith('STATUS:'):
event.status = line[7:]
return event
async def get_calendar_by_name(self, name: str) -> Optional[CalendarInfo]:
"""이름으로 캘린더 찾기"""
calendars = await self.get_calendars()
for calendar in calendars:
if calendar.name.lower() == name.lower():
return calendar
return None
async def create_todo_calendar_if_not_exists(self) -> CalendarInfo:
"""Todo 전용 캘린더가 없으면 생성"""
# 기존 Todo 캘린더 찾기
todo_calendar = await self.get_calendar_by_name("Todo")
if todo_calendar:
return todo_calendar
# Todo 캘린더 생성 (시놀로지에서는 웹 인터페이스를 통해 생성해야 함)
# 여기서는 기본 캘린더를 사용하거나 사용자에게 안내
calendars = await self.get_calendars()
if calendars:
logger.info("Todo 전용 캘린더가 없어 첫 번째 캘린더를 사용합니다.")
return calendars[0]
raise CalendarServiceError("사용 가능한 캘린더가 없습니다. 시놀로지에서 캘린더를 먼저 생성해주세요.")
class SynologyMailService:
"""시놀로지 MailPlus 서비스 (캘린더 연동용)"""
def __init__(self, smtp_server: str, smtp_port: int, username: str, password: str):
self.smtp_server = smtp_server
self.smtp_port = smtp_port
self.username = username
self.password = password
async def send_calendar_invitation(self, to_email: str, event: CalendarEvent, ics_content: str):
"""캘린더 초대 메일 발송"""
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.base import MIMEBase
from email import encoders
try:
msg = MIMEMultipart()
msg['From'] = self.username
msg['To'] = to_email
msg['Subject'] = f"📋 할일 일정: {event.title}"
# 메일 본문
body = f"""
새로운 할일이 일정에 추가되었습니다.
제목: {event.title}
시작: {event.start_time.strftime('%Y-%m-%d %H:%M') if event.start_time else '미정'}
종료: {event.end_time.strftime('%Y-%m-%d %H:%M') if event.end_time else '미정'}
설명: {event.description or ''}
이 메일의 첨부파일을 캘린더 앱에서 열면 일정이 자동으로 추가됩니다.
-- Todo-Project
"""
msg.attach(MIMEText(body, 'plain', 'utf-8'))
# ICS 파일 첨부
if ics_content:
part = MIMEBase('text', 'calendar')
part.set_payload(ics_content.encode('utf-8'))
encoders.encode_base64(part)
part.add_header(
'Content-Disposition',
'attachment; filename="todo_event.ics"'
)
part.add_header('Content-Type', 'text/calendar; charset=utf-8')
msg.attach(part)
# SMTP 발송
server = smtplib.SMTP(self.smtp_server, self.smtp_port)
server.starttls()
server.login(self.username, self.password)
server.send_message(msg)
server.quit()
logger.info(f"캘린더 초대 메일 발송 완료: {to_email}")
except Exception as e:
logger.error(f"메일 발송 실패: {e}")
raise CalendarServiceError(f"메일 발송 실패: {str(e)}")

95
backend/src/main.py Normal file
View File

@@ -0,0 +1,95 @@
"""
Todo-Project 메인 애플리케이션
- 간결함 원칙: 애플리케이션 설정 및 라우터 등록만 담당
"""
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
import logging
from .core.config import settings
from .api.routes import auth, todos, calendar
# 로깅 설정
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
# FastAPI 앱 생성
app = FastAPI(
title="Todo-Project API",
description="간결한 Todo 관리 시스템 with 캘린더 연동",
version="1.0.0",
docs_url="/docs",
redoc_url="/redoc"
)
# CORS 설정
app.add_middleware(
CORSMiddleware,
allow_origins=settings.ALLOWED_ORIGINS,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 라우터 등록
app.include_router(auth.router, prefix="/api/auth", tags=["auth"])
app.include_router(todos.router, prefix="/api", tags=["todos"])
app.include_router(calendar.router, prefix="/api", tags=["calendar"])
@app.get("/")
async def root():
"""루트 엔드포인트"""
return {
"message": "Todo-Project API",
"version": "1.0.0",
"docs": "/docs"
}
@app.get("/health")
async def health_check():
"""헬스 체크"""
return {
"status": "healthy",
"service": "todo-project",
"version": "1.0.0"
}
# 애플리케이션 시작 시 실행
@app.on_event("startup")
async def startup_event():
"""애플리케이션 시작 시 초기화"""
logger.info("🚀 Todo-Project API 시작")
logger.info(f"📊 환경: {settings.ENVIRONMENT}")
logger.info(f"🔗 데이터베이스: {settings.DATABASE_URL}")
# 애플리케이션 종료 시 실행
@app.on_event("shutdown")
async def shutdown_event():
"""애플리케이션 종료 시 정리"""
logger.info("🛑 Todo-Project API 종료")
# 캘린더 서비스 연결 정리
try:
from .integrations.calendar import get_calendar_router
calendar_router = get_calendar_router()
await calendar_router.close_all()
logger.info("📅 캘린더 서비스 연결 정리 완료")
except Exception as e:
logger.error(f"캘린더 서비스 정리 중 오류: {e}")
if __name__ == "__main__":
import uvicorn
uvicorn.run(
"main:app",
host="0.0.0.0",
port=settings.PORT,
reload=settings.DEBUG
)

View File

View File

@@ -0,0 +1,64 @@
"""
할일관리 시스템 모델
"""
from sqlalchemy import Column, String, DateTime, Text, Boolean, Integer, ForeignKey
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
from datetime import datetime
import uuid
from ..core.database import Base
class TodoItem(Base):
"""할일 아이템"""
__tablename__ = "todo_items"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
user_id = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=False)
# 기본 정보
content = Column(Text, nullable=False) # 할일 내용
status = Column(String(20), nullable=False, default="draft") # draft, scheduled, active, completed, delayed
photo_url = Column(String(500), nullable=True) # 첨부 사진 URL
# 시간 관리
created_at = Column(DateTime(timezone=True), nullable=False, default=datetime.utcnow)
start_date = Column(DateTime(timezone=True), nullable=True) # 시작 예정일
estimated_minutes = Column(Integer, nullable=True) # 예상 소요시간 (분)
completed_at = Column(DateTime(timezone=True), nullable=True)
delayed_until = Column(DateTime(timezone=True), nullable=True) # 지연된 경우 새로운 시작일
# 분할 관리
parent_id = Column(UUID(as_uuid=True), ForeignKey("todo_items.id"), nullable=True) # 분할된 할일의 부모
split_order = Column(Integer, nullable=True) # 분할 순서
# 관계
user = relationship("User", back_populates="todo_items")
comments = relationship("TodoComment", back_populates="todo_item", cascade="all, delete-orphan")
# 자기 참조 관계 (분할된 할일들)
subtasks = relationship("TodoItem", backref="parent_task", remote_side=[id])
def __repr__(self):
return f"<TodoItem(id={self.id}, content='{self.content[:50]}...', status='{self.status}')>"
class TodoComment(Base):
"""할일 댓글/메모"""
__tablename__ = "todo_comments"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
todo_item_id = Column(UUID(as_uuid=True), ForeignKey("todo_items.id"), nullable=False)
user_id = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=False)
content = Column(Text, nullable=False)
created_at = Column(DateTime(timezone=True), nullable=False, default=datetime.utcnow)
updated_at = Column(DateTime(timezone=True), nullable=False, default=datetime.utcnow, onupdate=datetime.utcnow)
# 관계
todo_item = relationship("TodoItem", back_populates="comments")
user = relationship("User")
def __repr__(self):
return f"<TodoComment(id={self.id}, content='{self.content[:30]}...')>"

View File

@@ -0,0 +1,43 @@
"""
사용자 모델
"""
from sqlalchemy import Column, String, Boolean, DateTime, Text
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
from datetime import datetime
import uuid
from ..core.database import Base
class User(Base):
"""사용자 모델"""
__tablename__ = "users"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
email = Column(String(255), unique=True, index=True, nullable=False)
hashed_password = Column(String(255), nullable=False)
full_name = Column(String(255), nullable=True)
# 상태 관리
is_active = Column(Boolean, default=True, nullable=False)
is_admin = Column(Boolean, default=False, nullable=False)
# 타임스탬프
created_at = Column(DateTime(timezone=True), default=datetime.utcnow, nullable=False)
updated_at = Column(DateTime(timezone=True), default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False)
last_login_at = Column(DateTime(timezone=True), nullable=True)
# 프로필 정보
bio = Column(Text, nullable=True)
avatar_url = Column(String(500), nullable=True)
# 설정
timezone = Column(String(50), default="Asia/Seoul", nullable=False)
language = Column(String(10), default="ko", nullable=False)
# 관계 설정
todo_items = relationship("TodoItem", back_populates="user", lazy="dynamic")
def __repr__(self):
return f"<User(email='{self.email}', full_name='{self.full_name}')>"

View File

View File

@@ -0,0 +1,57 @@
"""
인증 관련 스키마
"""
from pydantic import BaseModel, EmailStr, Field
from typing import Optional
from datetime import datetime
from uuid import UUID
class UserBase(BaseModel):
email: EmailStr
full_name: Optional[str] = None
is_active: bool = True
class UserCreate(UserBase):
password: str = Field(..., min_length=6)
class UserUpdate(BaseModel):
email: Optional[EmailStr] = None
full_name: Optional[str] = None
is_active: Optional[bool] = None
class UserResponse(UserBase):
id: UUID
is_admin: bool
created_at: datetime
updated_at: datetime
last_login_at: Optional[datetime] = None
timezone: str
language: str
class Config:
from_attributes = True
class Token(BaseModel):
access_token: str
refresh_token: str
token_type: str = "bearer"
expires_in: int
class TokenData(BaseModel):
user_id: Optional[str] = None
class LoginRequest(BaseModel):
email: EmailStr
password: str
remember_me: bool = False
class RefreshTokenRequest(BaseModel):
refresh_token: str

110
backend/src/schemas/todo.py Normal file
View File

@@ -0,0 +1,110 @@
"""
할일관리 시스템 스키마
"""
from pydantic import BaseModel, Field
from typing import List, Optional
from datetime import datetime
from uuid import UUID
class TodoCommentBase(BaseModel):
content: str = Field(..., min_length=1, max_length=1000)
class TodoCommentCreate(TodoCommentBase):
pass
class TodoCommentUpdate(BaseModel):
content: Optional[str] = Field(None, min_length=1, max_length=1000)
class TodoCommentResponse(TodoCommentBase):
id: UUID
todo_item_id: UUID
user_id: UUID
created_at: datetime
updated_at: datetime
class Config:
from_attributes = True
class TodoItemBase(BaseModel):
content: str = Field(..., min_length=1, max_length=2000)
class TodoItemCreate(TodoItemBase):
"""초기 할일 생성 (draft 상태)"""
photo_url: Optional[str] = Field(None, max_length=500, description="첨부 사진 URL")
class TodoItemSchedule(BaseModel):
"""할일 일정 설정"""
start_date: datetime
estimated_minutes: int = Field(..., ge=1, le=120) # 1분~2시간
class TodoItemUpdate(BaseModel):
"""할일 수정"""
content: Optional[str] = Field(None, min_length=1, max_length=2000)
photo_url: Optional[str] = Field(None, max_length=500, description="첨부 사진 URL")
status: Optional[str] = Field(None, pattern="^(draft|scheduled|active|completed|delayed)$")
start_date: Optional[datetime] = None
estimated_minutes: Optional[int] = Field(None, ge=1, le=120)
delayed_until: Optional[datetime] = None
class TodoItemDelay(BaseModel):
"""할일 지연"""
delayed_until: datetime
class TodoItemSplit(BaseModel):
"""할일 분할"""
subtasks: List[str] = Field(..., min_items=2, max_items=10)
estimated_minutes_per_task: List[int] = Field(..., min_items=2, max_items=10)
class TodoItemResponse(TodoItemBase):
id: UUID
user_id: UUID
photo_url: Optional[str] = None
status: str
created_at: datetime
start_date: Optional[datetime]
estimated_minutes: Optional[int]
completed_at: Optional[datetime]
delayed_until: Optional[datetime]
parent_id: Optional[UUID]
split_order: Optional[int]
# 댓글 수
comment_count: int = 0
class Config:
from_attributes = True
class TodoItemWithComments(TodoItemResponse):
"""댓글이 포함된 할일 응답"""
comments: List[TodoCommentResponse] = []
class TodoStats(BaseModel):
"""할일 통계"""
total_count: int
draft_count: int
scheduled_count: int
active_count: int
completed_count: int
delayed_count: int
completion_rate: float # 완료율 (%)
class TodoDashboard(BaseModel):
"""할일 대시보드"""
stats: TodoStats
today_todos: List[TodoItemResponse]
overdue_todos: List[TodoItemResponse]
upcoming_todos: List[TodoItemResponse]

View File

@@ -0,0 +1,12 @@
"""
서비스 레이어 모듈
"""
from .todo_service import TodoService
from .calendar_sync_service import CalendarSyncService, get_calendar_sync_service
__all__ = [
"TodoService",
"CalendarSyncService",
"get_calendar_sync_service"
]

View File

@@ -0,0 +1,74 @@
"""
캘린더 동기화 서비스
- 서비스 클래스 기준: 최대 350줄
- 간결함 원칙: Todo ↔ 캘린더 동기화만 담당
"""
import logging
from typing import Dict, Any, Optional
from ..models.todo import TodoItem
from ..integrations.calendar import get_calendar_router
logger = logging.getLogger(__name__)
class CalendarSyncService:
"""Todo와 캘린더 간 동기화 서비스"""
def __init__(self):
self.calendar_router = get_calendar_router()
async def sync_todo_create(self, todo_item: TodoItem) -> Dict[str, Any]:
"""새 할일을 캘린더에 생성"""
try:
result = await self.calendar_router.sync_todo_to_calendars(todo_item)
logger.info(f"할일 {todo_item.id} 캘린더 생성 완료")
return {"success": True, "result": result}
except Exception as e:
logger.error(f"할일 {todo_item.id} 캘린더 생성 실패: {e}")
return {"success": False, "error": str(e)}
async def sync_todo_complete(self, todo_item: TodoItem) -> Dict[str, Any]:
"""완료된 할일의 캘린더 태그 업데이트 (todo → 완료)"""
try:
# 향후 구현: 기존 이벤트를 찾아서 태그 업데이트
logger.info(f"할일 {todo_item.id} 완료 상태로 캘린더 업데이트")
return {"success": True, "action": "completed"}
except Exception as e:
logger.error(f"할일 {todo_item.id} 완료 상태 캘린더 업데이트 실패: {e}")
return {"success": False, "error": str(e)}
async def sync_todo_delay(self, todo_item: TodoItem) -> Dict[str, Any]:
"""지연된 할일의 캘린더 날짜 수정"""
try:
# 향후 구현: 기존 이벤트를 찾아서 날짜 수정
logger.info(f"할일 {todo_item.id} 지연 날짜로 캘린더 업데이트")
return {"success": True, "action": "delayed"}
except Exception as e:
logger.error(f"할일 {todo_item.id} 지연 날짜 캘린더 업데이트 실패: {e}")
return {"success": False, "error": str(e)}
async def sync_todo_delete(self, todo_item: TodoItem) -> Dict[str, Any]:
"""삭제된 할일의 캘린더 이벤트 제거"""
try:
# 향후 구현: 기존 이벤트를 찾아서 삭제
logger.info(f"할일 {todo_item.id} 캘린더 이벤트 삭제")
return {"success": True, "action": "deleted"}
except Exception as e:
logger.error(f"할일 {todo_item.id} 캘린더 이벤트 삭제 실패: {e}")
return {"success": False, "error": str(e)}
# 전역 인스턴스
_calendar_sync_service: Optional[CalendarSyncService] = None
def get_calendar_sync_service() -> CalendarSyncService:
"""캘린더 동기화 서비스 싱글톤 인스턴스"""
global _calendar_sync_service
if _calendar_sync_service is None:
_calendar_sync_service = CalendarSyncService()
return _calendar_sync_service

View File

@@ -0,0 +1,70 @@
import os
import base64
from datetime import datetime
from typing import Optional
import uuid
from PIL import Image
import io
UPLOAD_DIR = "/app/uploads"
def ensure_upload_dir():
"""업로드 디렉토리 생성"""
if not os.path.exists(UPLOAD_DIR):
os.makedirs(UPLOAD_DIR)
def save_base64_image(base64_string: str) -> Optional[str]:
"""Base64 이미지를 파일로 저장하고 경로 반환"""
try:
ensure_upload_dir()
# Base64 헤더 제거
if "," in base64_string:
base64_string = base64_string.split(",")[1]
# 디코딩
image_data = base64.b64decode(base64_string)
# 이미지 검증 및 형식 확인
image = Image.open(io.BytesIO(image_data))
# iPhone의 .mpo 파일이나 기타 형식을 JPEG로 강제 변환
# RGB 모드로 변환 (RGBA, P 모드 등을 처리)
if image.mode in ('RGBA', 'LA', 'P'):
# 투명도가 있는 이미지는 흰 배경과 합성
background = Image.new('RGB', image.size, (255, 255, 255))
if image.mode == 'P':
image = image.convert('RGBA')
background.paste(image, mask=image.split()[-1] if image.mode == 'RGBA' else None)
image = background
elif image.mode != 'RGB':
image = image.convert('RGB')
# 파일명 생성 (강제로 .jpg)
filename = f"{datetime.now().strftime('%Y%m%d%H%M%S')}_{uuid.uuid4().hex[:8]}.jpg"
filepath = os.path.join(UPLOAD_DIR, filename)
# 이미지 저장 (최대 크기 제한)
max_size = (1920, 1920)
image.thumbnail(max_size, Image.Resampling.LANCZOS)
# 항상 JPEG로 저장
image.save(filepath, 'JPEG', quality=85, optimize=True)
# 웹 경로 반환
return f"/uploads/{filename}"
except Exception as e:
print(f"이미지 저장 실패: {e}")
return None
def delete_file(filepath: str):
"""파일 삭제"""
try:
if filepath and filepath.startswith("/uploads/"):
filename = filepath.replace("/uploads/", "")
full_path = os.path.join(UPLOAD_DIR, filename)
if os.path.exists(full_path):
os.remove(full_path)
except Exception as e:
print(f"파일 삭제 실패: {e}")

View File

@@ -0,0 +1,300 @@
"""
Todo 비즈니스 로직 서비스
- 서비스 클래스 기준: 최대 350줄
- 간결함 원칙: 핵심 Todo 로직만 포함
"""
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func, and_
from typing import List, Optional
from datetime import datetime
from uuid import UUID
import logging
from ..models.user import User
from ..models.todo import TodoItem, TodoComment
from ..schemas.todo import (
TodoItemCreate, TodoItemSchedule, TodoItemSplit, TodoItemDelay,
TodoItemResponse, TodoCommentCreate, TodoCommentResponse
)
logger = logging.getLogger(__name__)
class TodoService:
"""Todo 관련 비즈니스 로직 서비스"""
def __init__(self, db: AsyncSession):
self.db = db
async def create_todo(self, todo_data: TodoItemCreate, user_id: UUID) -> TodoItemResponse:
"""새 할일 생성"""
new_todo = TodoItem(
user_id=user_id,
content=todo_data.content,
status="draft"
)
self.db.add(new_todo)
await self.db.commit()
await self.db.refresh(new_todo)
return await self._build_todo_response(new_todo)
async def schedule_todo(
self,
todo_id: UUID,
schedule_data: TodoItemSchedule,
user_id: UUID
) -> TodoItemResponse:
"""할일 일정 설정"""
todo_item = await self._get_user_todo(todo_id, user_id, "draft")
# 2시간 이상인 경우 분할 제안
if schedule_data.estimated_minutes > 120:
raise ValueError("Tasks longer than 2 hours should be split")
todo_item.start_date = schedule_data.start_date
todo_item.estimated_minutes = schedule_data.estimated_minutes
todo_item.status = "scheduled"
await self.db.commit()
await self.db.refresh(todo_item)
return await self._build_todo_response(todo_item)
async def complete_todo(self, todo_id: UUID, user_id: UUID) -> TodoItemResponse:
"""할일 완료"""
todo_item = await self._get_user_todo(todo_id, user_id, "active")
todo_item.status = "completed"
todo_item.completed_at = datetime.utcnow()
await self.db.commit()
await self.db.refresh(todo_item)
return await self._build_todo_response(todo_item)
async def delay_todo(
self,
todo_id: UUID,
delay_data: TodoItemDelay,
user_id: UUID
) -> TodoItemResponse:
"""할일 지연"""
todo_item = await self._get_user_todo(todo_id, user_id, "active")
todo_item.status = "delayed"
todo_item.delayed_until = delay_data.delayed_until
todo_item.start_date = delay_data.delayed_until
await self.db.commit()
await self.db.refresh(todo_item)
return await self._build_todo_response(todo_item)
async def split_todo(
self,
todo_id: UUID,
split_data: TodoItemSplit,
user_id: UUID
) -> List[TodoItemResponse]:
"""할일 분할"""
original_todo = await self._get_user_todo(todo_id, user_id, "draft")
# 분할된 할일들 생성
subtasks = []
for i, (content, minutes) in enumerate(zip(split_data.subtasks, split_data.estimated_minutes_per_task)):
if minutes > 120:
raise ValueError(f"Subtask {i+1} is longer than 2 hours")
subtask = TodoItem(
user_id=user_id,
content=content,
status="draft",
parent_id=original_todo.id,
split_order=i + 1
)
self.db.add(subtask)
subtasks.append(subtask)
original_todo.status = "split"
await self.db.commit()
# 응답 데이터 구성
response_data = []
for subtask in subtasks:
await self.db.refresh(subtask)
response_data.append(await self._build_todo_response(subtask))
return response_data
async def get_todos(
self,
user_id: UUID,
status_filter: Optional[str] = None
) -> List[TodoItemResponse]:
"""할일 목록 조회"""
query = select(TodoItem).where(TodoItem.user_id == user_id)
if status_filter:
query = query.where(TodoItem.status == status_filter)
query = query.order_by(TodoItem.created_at.desc())
result = await self.db.execute(query)
todo_items = result.scalars().all()
response_data = []
for todo_item in todo_items:
response_data.append(await self._build_todo_response(todo_item))
return response_data
async def get_active_todos(self, user_id: UUID) -> List[TodoItemResponse]:
"""활성 할일 조회 (scheduled → active 자동 변환 포함)"""
now = datetime.utcnow()
# scheduled → active 자동 변환
update_result = await self.db.execute(
select(TodoItem).where(
and_(
TodoItem.user_id == user_id,
TodoItem.status == "scheduled",
TodoItem.start_date <= now
)
)
)
scheduled_items = update_result.scalars().all()
for item in scheduled_items:
item.status = "active"
if scheduled_items:
await self.db.commit()
# active 할일들 조회
result = await self.db.execute(
select(TodoItem).where(
and_(
TodoItem.user_id == user_id,
TodoItem.status == "active"
)
).order_by(TodoItem.start_date.asc())
)
active_todos = result.scalars().all()
response_data = []
for todo_item in active_todos:
response_data.append(await self._build_todo_response(todo_item))
return response_data
async def create_comment(
self,
todo_id: UUID,
comment_data: TodoCommentCreate,
user_id: UUID
) -> TodoCommentResponse:
"""댓글 생성"""
# 할일 존재 확인
await self._get_user_todo(todo_id, user_id)
new_comment = TodoComment(
todo_item_id=todo_id,
user_id=user_id,
content=comment_data.content
)
self.db.add(new_comment)
await self.db.commit()
await self.db.refresh(new_comment)
return TodoCommentResponse(
id=new_comment.id,
todo_item_id=new_comment.todo_item_id,
user_id=new_comment.user_id,
content=new_comment.content,
created_at=new_comment.created_at,
updated_at=new_comment.updated_at
)
async def get_comments(self, todo_id: UUID, user_id: UUID) -> List[TodoCommentResponse]:
"""댓글 목록 조회"""
# 할일 존재 확인
await self._get_user_todo(todo_id, user_id)
result = await self.db.execute(
select(TodoComment).where(TodoComment.todo_item_id == todo_id)
.order_by(TodoComment.created_at.asc())
)
comments = result.scalars().all()
return [
TodoCommentResponse(
id=comment.id,
todo_item_id=comment.todo_item_id,
user_id=comment.user_id,
content=comment.content,
created_at=comment.created_at,
updated_at=comment.updated_at
)
for comment in comments
]
# ========================================================================
# 헬퍼 메서드들
# ========================================================================
async def _get_user_todo(
self,
todo_id: UUID,
user_id: UUID,
required_status: Optional[str] = None
) -> TodoItem:
"""사용자의 할일 조회"""
query = select(TodoItem).where(
and_(
TodoItem.id == todo_id,
TodoItem.user_id == user_id
)
)
if required_status:
query = query.where(TodoItem.status == required_status)
result = await self.db.execute(query)
todo_item = result.scalar_one_or_none()
if not todo_item:
detail = "Todo item not found"
if required_status:
detail += f" or not in {required_status} status"
raise ValueError(detail)
return todo_item
async def _get_comment_count(self, todo_id: UUID) -> int:
"""댓글 수 조회"""
result = await self.db.execute(
select(func.count(TodoComment.id)).where(TodoComment.todo_item_id == todo_id)
)
return result.scalar() or 0
async def _build_todo_response(self, todo_item: TodoItem) -> TodoItemResponse:
"""TodoItem을 TodoItemResponse로 변환"""
comment_count = await self._get_comment_count(todo_item.id)
return TodoItemResponse(
id=todo_item.id,
user_id=todo_item.user_id,
content=todo_item.content,
status=todo_item.status,
created_at=todo_item.created_at,
start_date=todo_item.start_date,
estimated_minutes=todo_item.estimated_minutes,
completed_at=todo_item.completed_at,
delayed_until=todo_item.delayed_until,
parent_id=todo_item.parent_id,
split_order=todo_item.split_order,
comment_count=comment_count
)