hyungi/Todo-Project
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

🚀 시놀로지 배포 준비 완료

Browse Source 
✨ 주요 변경사항:
- 단일 docker-compose.yml로 통합 (로컬/시놀로지 환경 지원)
- 시놀로지 볼륨 매핑 설정 (volume1: 이미지, volume3: 데이터)
- 통합 배포 가이드 및 자동 배포 스크립트 추가
- 완전한 Memos 스타일 워크플로우 구현

🎯 새로운 기능:
- 📝 메모 작성 (upload.html) - 이미지 업로드 지원
- 📥 수신함 (inbox.html) - 메모 편집 및 Todo/보드 변환
- ✅ Todo 목록 (todo-list.html) - 오늘 할 일 관리
- 📋 보드 (board.html) - 프로젝트 관리, 접기/펼치기, 이미지 지원
- 📚 아카이브 (archive.html) - 완료된 보드 보관
- 🔐 초기 설정 화면 - 관리자 계정 생성

🔧 기술적 개선:
- 이미지 업로드/편집 완전 지원
- 반응형 디자인 및 모바일 최적화
- 보드 완료 후 자동 숨김 처리
- 메모 편집 시 제목 필드 제거
- 테스트 로그인 버튼 제거 (프로덕션 준비)
- 과거 코드 정리 (TodoService, CalendarSyncService 등)

📦 배포 관련:
- env.synology.example - 시놀로지 환경 설정 템플릿
- SYNOLOGY_DEPLOYMENT_GUIDE.md - 상세한 배포 가이드
- deploy-synology.sh - 원클릭 자동 배포 스크립트
- Nginx 정적 파일 서빙 및 이미지 프록시 설정

🗑️ 정리된 파일:
- 사용하지 않는 HTML 페이지들 (dashboard, calendar, checklist 등)
- 복잡한 통합 서비스들 (integrations 폴더)
- 중복된 시놀로지 설정 파일들
This commit is contained in:
Hyungi Ahn
2025-09-24 09:12:39 +09:00
parent 4c7d2d8290
commit 0b967a84fa
42 changed files with 5467 additions and 6691 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
backend/Dockerfile
View File

@@ -5,6 +5,7 @@ WORKDIR /app
# 시스템 패키지 업데이트 및 필요한 패키지 설치
RUN apt-get update && apt-get install -y \
gcc \
curl \
&& rm -rf /var/lib/apt/lists/*
# Python 의존성 설치

38
backend/migrations/003_optimize_for_workflow.sql Normal file
View File

@@ -0,0 +1,38 @@
-- 새로운 워크플로우에 맞게 DB 구조 최적화
-- 1. due_date를 start_date로 변경
ALTER TABLE todos RENAME COLUMN due_date TO start_date;
-- 2. tags 컬럼 제거 (사용하지 않음)
ALTER TABLE todos DROP COLUMN IF EXISTS tags;
-- 3. category 기본값 변경 및 기존 데이터 정리
-- 기존 'checklist' 카테고리를 'memo'로 변경
UPDATE todos SET category = 'memo' WHERE category = 'checklist';
-- 기존 'calendar' 카테고리를 'todo'로 변경
UPDATE todos SET category = 'todo' WHERE category = 'calendar';
-- 4. title을 nullable로 변경 (메모의 경우 선택사항)
ALTER TABLE todos ALTER COLUMN title DROP NOT NULL;
-- 5. description을 NOT NULL로 변경 (내용은 필수)
UPDATE todos SET description = COALESCE(title, '내용 없음') WHERE description IS NULL OR description = '';
ALTER TABLE todos ALTER COLUMN description SET NOT NULL;
-- 6. category 기본값을 'memo'로 설정
ALTER TABLE todos ALTER COLUMN category SET DEFAULT 'memo';
-- 7. 불필요한 인덱스 정리 및 새로운 인덱스 추가
-- 기존 due_date 인덱스 제거 (컬럼명이 변경됨)
DROP INDEX IF EXISTS idx_todos_due_date;
-- 새로운 인덱스 생성
CREATE INDEX IF NOT EXISTS idx_todos_start_date ON todos(start_date);
CREATE INDEX IF NOT EXISTS idx_todos_category_status ON todos(category, status);
CREATE INDEX IF NOT EXISTS idx_todos_user_category ON todos(user_id, category);
-- 8. 성능 최적화를 위한 복합 인덱스
CREATE INDEX IF NOT EXISTS idx_todos_workflow ON todos(user_id, category, status, start_date);
COMMIT;

11
backend/src/api/routes/calendar.py
View File

@@ -1,7 +1,8 @@
"""
간단한 캘린더 API 라우터
"""
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi import APIRouter, Depends, HTTPException
from fastapi import status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, and_
from typing import List
@@ -32,10 +33,10 @@ async def get_calendar_todos(
query = select(Todo).where(
and_(
Todo.user_id == current_user.id,
Todo.due_date >= start_date,
Todo.due_date <= end_date
Todo.start_date >= start_date,
Todo.start_date <= end_date
)
).order_by(Todo.due_date.asc())
).order_by(Todo.start_date.asc())
result = await db.execute(query)
todos = result.scalars().all()
@@ -62,7 +63,7 @@ async def get_today_todos(
query = select(Todo).where(
and_(
Todo.user_id == current_user.id,
Todo.due_date == today
Todo.start_date == today
)
).order_by(Todo.created_at.desc())

121
backend/src/api/routes/setup.py Normal file
View File

@@ -0,0 +1,121 @@
"""
초기 설정 관련 API 라우터
- 최초 관리자 계정 설정
- 시스템 초기화 상태 확인
"""
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func
from pydantic import BaseModel, Field
from ...core.database import get_db
from ...core.security import get_password_hash
from ...models.user import User
from ...schemas.auth import CreateUserRequest
router = APIRouter()
class InitialSetupRequest(BaseModel):
"""초기 설정 요청"""
admin_username: str = Field(..., min_length=3, max_length=50, description="관리자 사용자명")
admin_email: str = Field(..., description="관리자 이메일")
admin_password: str = Field(..., min_length=6, description="관리자 비밀번호")
admin_full_name: str = Field(default="Administrator", description="관리자 이름")
class SetupStatusResponse(BaseModel):
"""설정 상태 응답"""
is_setup_required: bool
user_count: int
@router.get("/status", response_model=SetupStatusResponse)
async def get_setup_status(db: AsyncSession = Depends(get_db)):
"""시스템 초기 설정 필요 여부 확인"""
try:
# 사용자 수 확인
result = await db.execute(select(func.count(User.id)))
user_count = result.scalar() or 0
return SetupStatusResponse(
is_setup_required=user_count == 0,
user_count=user_count
)
except Exception as e:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="설정 상태 확인 중 오류가 발생했습니다."
)
@router.post("/initialize")
async def initialize_system(
setup_data: InitialSetupRequest,
db: AsyncSession = Depends(get_db)
):
"""시스템 초기화 및 관리자 계정 생성"""
try:
# 이미 사용자가 있는지 확인
result = await db.execute(select(func.count(User.id)))
user_count = result.scalar() or 0
if user_count > 0:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="시스템이 이미 초기화되었습니다."
)
# 사용자명 중복 확인
existing_user = await db.execute(
select(User).where(User.username == setup_data.admin_username)
)
if existing_user.scalar_one_or_none():
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="이미 존재하는 사용자명입니다."
)
# 이메일 중복 확인
existing_email = await db.execute(
select(User).where(User.email == setup_data.admin_email)
)
if existing_email.scalar_one_or_none():
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="이미 존재하는 이메일입니다."
)
# 관리자 계정 생성
admin_user = User(
username=setup_data.admin_username,
email=setup_data.admin_email,
hashed_password=get_password_hash(setup_data.admin_password),
full_name=setup_data.admin_full_name,
is_active=True,
is_admin=True
)
db.add(admin_user)
await db.commit()
await db.refresh(admin_user)
return {
"message": "시스템이 성공적으로 초기화되었습니다.",
"admin_user": {
"id": str(admin_user.id),
"username": admin_user.username,
"email": admin_user.email,
"full_name": admin_user.full_name
}
}
except HTTPException:
raise
except Exception as e:
await db.rollback()
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="시스템 초기화 중 오류가 발생했습니다."
)

88
backend/src/api/routes/todos.py
View File

@@ -1,7 +1,7 @@
"""
간단한 Todo API 라우터
"""
from fastapi import APIRouter, Depends, HTTPException, status, Query, UploadFile, File
from fastapi import APIRouter, Depends, HTTPException, Query, UploadFile, File, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, and_
from typing import List, Optional
@@ -14,11 +14,54 @@ from ...models.user import User
from ...models.todo import Todo, TodoStatus
from ...schemas.todo import TodoCreate, TodoUpdate, TodoResponse
from ..dependencies import get_current_active_user
from ...services.file_service import save_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="이미지 파일만 업로드 가능합니다."
)
# 파일 크기 검증 (10MB 제한)
content = await image.read()
if len(content) > 10 * 1024 * 1024: # 10MB
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="파일 크기는 10MB를 초과할 수 없습니다."
)
# 이미지 저장
file_url = save_image(content, image.filename)
if not file_url:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="이미지 저장에 실패했습니다."
)
return {"file_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="이미지 업로드 중 오류가 발생했습니다."
)
# ============================================================================
# Todo CRUD API
# ============================================================================
@@ -34,15 +77,17 @@ async def create_todo(
logger.info(f"Todo 생성 요청 - 사용자: {current_user.username}")
logger.info(f"요청 데이터: {todo_data.dict()}")
# 날짜 문자열 파싱 (한국 시간 형식)
parsed_due_date = None
if todo_data.due_date:
# 시작 날짜 문자열 파싱 (Todo일 때만)
parsed_start_date = None
if todo_data.start_date and todo_data.category.value == "todo":
try:
# "2025-09-22T00:00:00+09:00" 형식 파싱
parsed_due_date = datetime.fromisoformat(todo_data.due_date)
logger.info(f"파싱된 날짜: {parsed_due_date}")
# "2025-09-22" 형식 파싱 후 한국 시간으로 변환
from datetime import date
date_obj = datetime.strptime(todo_data.start_date, "%Y-%m-%d").date()
parsed_start_date = datetime.combine(date_obj, datetime.min.time())
logger.info(f"파싱된 시작 날짜: {parsed_start_date}")
except ValueError:
logger.warning(f"Invalid date format: {todo_data.due_date}")
logger.warning(f"Invalid date format: {todo_data.start_date}")
# 이미지 URLs JSON 변환
import json
@@ -56,9 +101,10 @@ async def create_todo(
title=todo_data.title,
description=todo_data.description,
category=todo_data.category,
due_date=parsed_due_date,
start_date=parsed_start_date,
image_urls=image_urls_json,
tags=todo_data.tags
board_id=todo_data.board_id,
is_board_header=todo_data.is_board_header or False
)
db.add(new_todo)
@@ -83,10 +129,11 @@ async def create_todo(
"status": new_todo.status,
"created_at": new_todo.created_at,
"updated_at": new_todo.updated_at,
"due_date": new_todo.due_date.isoformat() if new_todo.due_date else None,
"start_date": new_todo.start_date.strftime("%Y-%m-%d") if new_todo.start_date else None,
"completed_at": new_todo.completed_at,
"image_urls": image_urls_list,
"tags": new_todo.tags
"board_id": str(new_todo.board_id) if new_todo.board_id else None,
"is_board_header": new_todo.is_board_header
}
return response_data
@@ -146,10 +193,11 @@ async def get_todos(
"status": todo.status,
"created_at": todo.created_at,
"updated_at": todo.updated_at,
"due_date": todo.due_date.isoformat() if todo.due_date else None,
"start_date": todo.start_date.isoformat() if todo.start_date else None,
"completed_at": todo.completed_at,
"image_urls": image_urls_list,
"tags": todo.tags
"board_id": str(todo.board_id) if todo.board_id else None,
"is_board_header": todo.is_board_header
})
return response_data
@@ -201,10 +249,11 @@ async def get_todo(
"status": todo.status,
"created_at": todo.created_at,
"updated_at": todo.updated_at,
"due_date": todo.due_date.isoformat() if todo.due_date else None,
"start_date": todo.start_date.isoformat() if todo.start_date else None,
"completed_at": todo.completed_at,
"image_urls": image_urls_list,
"tags": todo.tags,
"board_id": str(todo.board_id) if todo.board_id else None,
"is_board_header": todo.is_board_header
}
return response_data
@@ -246,7 +295,7 @@ async def update_todo(
import json
update_data = todo_data.dict(exclude_unset=True)
for field, value in update_data.items():
if field == 'due_date' and value:
if field == 'start_date' and value:
# 날짜 문자열 파싱 (한국 시간 형식)
try:
parsed_date = datetime.fromisoformat(value)
@@ -285,10 +334,11 @@ async def update_todo(
"status": todo.status,
"created_at": todo.created_at,
"updated_at": todo.updated_at,
"due_date": todo.due_date.isoformat() if todo.due_date else None,
"start_date": todo.start_date.isoformat() if todo.start_date else None,
"completed_at": todo.completed_at,
"image_urls": image_urls_list,
"tags": todo.tags,
"board_id": str(todo.board_id) if todo.board_id else None,
"is_board_header": todo.is_board_header
}
return response_data

4
backend/src/core/config.py
View File

@@ -24,8 +24,8 @@ class Settings(BaseSettings):
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"]
# CORS 설정 (환경변수로 오버라이드 가능)
ALLOWED_HOSTS: List[str] = ["localhost", "127.0.0.1"]
ALLOWED_ORIGINS: List[str] = ["http://localhost:4000", "http://127.0.0.1:4000"]
# 서버 설정

30
backend/src/core/database.py
View File

@@ -68,28 +68,16 @@ async def init_db() -> None:
async def create_admin_user() -> None:
"""관리자 계정 생성 (존재하지 않을 경우)"""
"""관리자 계정 생성 (존재하지 않을 경우) - 초기 설정 API로 대체됨"""
from ..models.user import User
from .security import get_password_hash
from sqlalchemy import select
from sqlalchemy import select, func
async with AsyncSessionLocal() as session:
# 관리자 계정 존재 확인
result = await session.execute(
select(User).where(User.username == settings.ADMIN_USERNAME)
)
admin_user = result.scalar_one_or_none()
# 사용자 수 확인
result = await session.execute(select(func.count(User.id)))
user_count = result.scalar() or 0
if not admin_user:
# 관리자 계정 생성
admin_user = User(
username=settings.ADMIN_USERNAME,
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_USERNAME}")
if user_count == 0:
print("초기 설정이 필요합니다. /api/setup/status 엔드포인트를 확인하세요.")
else:
print(f"시스템에 {user_count}명의 사용자가 등록되어 있습니다.")

0
backend/src/integrations/__init__.py
View File

32
backend/src/integrations/calendar/__init__.py
View File

@@ -1,32 +0,0 @@
"""
캘린더 통합 모듈
- 다중 캘린더 제공자 지원 (시놀로지, 애플, 구글 등)
- 간결한 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"

370
backend/src/integrations/calendar/apple.py
View File

@@ -1,370 +0,0 @@
"""
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", "업무"]
}

332
backend/src/integrations/calendar/base.py
View File

@@ -1,332 +0,0 @@
"""
캘린더 서비스 기본 인터페이스 및 추상화
"""
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 {}

363
backend/src/integrations/calendar/router.py
View File

@@ -1,363 +0,0 @@
"""
캘린더 라우터 - 다중 캘린더 제공자 중앙 관리
- 서비스 클래스 기준: 최대 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

401
backend/src/integrations/calendar/synology.py
View File

@@ -1,401 +0,0 @@
"""
시놀로지 캘린더 서비스 구현
"""
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)}")

68
backend/src/main.py
View File

@@ -6,7 +6,9 @@ from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from fastapi.staticfiles import StaticFiles
import logging
import os
from .core.config import settings
from .api.routes import auth, todos, calendar
@@ -46,15 +48,35 @@ app.add_middleware(
async def validation_exception_handler(request: Request, exc: RequestValidationError):
logger.error(f"Validation 오류 - URL: {request.url}")
logger.error(f"Validation 오류 상세: {exc.errors()}")
# JSON 직렬화 가능한 형태로 에러 변환
serializable_errors = []
for error in exc.errors():
serializable_error = {}
for key, value in error.items():
if isinstance(value, bytes):
serializable_error[key] = value.decode('utf-8')
else:
serializable_error[key] = str(value) if not isinstance(value, (str, int, float, bool, list, dict, type(None))) else value
serializable_errors.append(serializable_error)
return JSONResponse(
status_code=422,
content={
"detail": "요청 데이터 검증 실패",
"errors": exc.errors()
"errors": serializable_errors
}
)
# 정적 파일 서빙 (업로드된 이미지)
UPLOAD_DIR = "/app/uploads"
if not os.path.exists(UPLOAD_DIR):
os.makedirs(UPLOAD_DIR)
app.mount("/uploads", StaticFiles(directory=UPLOAD_DIR), name="uploads")
# 라우터 등록
from .api.routes import setup
app.include_router(setup.router, prefix="/api/setup", tags=["setup"])
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"])
@@ -87,6 +109,47 @@ async def create_sample_data():
return
async def wait_for_database():
"""데이터베이스 연결 대기"""
import asyncio
import asyncpg
from urllib.parse import urlparse
# DATABASE_URL 파싱
parsed_url = urlparse(settings.DATABASE_URL.replace("postgresql+asyncpg://", "postgresql://"))
max_retries = 30 # 최대 30번 시도 (30초)
retry_count = 0
while retry_count < max_retries:
try:
logger.info(f"🔄 데이터베이스 연결 시도 {retry_count + 1}/{max_retries}")
# asyncpg로 직접 연결 테스트
conn = await asyncpg.connect(
host=parsed_url.hostname,
port=parsed_url.port or 5432,
user=parsed_url.username,
password=parsed_url.password,
database=parsed_url.path.lstrip('/'),
timeout=5
)
await conn.close()
logger.info("✅ 데이터베이스 연결 성공!")
return True
except Exception as e:
retry_count += 1
logger.warning(f"❌ 데이터베이스 연결 실패 ({retry_count}/{max_retries}): {e}")
if retry_count < max_retries:
await asyncio.sleep(1)
else:
logger.error("💥 데이터베이스 연결 최대 재시도 횟수 초과")
raise Exception("데이터베이스에 연결할 수 없습니다.")
@app.on_event("startup")
async def startup_event():
"""애플리케이션 시작 시 초기화"""
@@ -94,6 +157,9 @@ async def startup_event():
logger.info(f"📊 환경: {settings.ENVIRONMENT}")
logger.info(f"🔗 데이터베이스: {settings.DATABASE_URL}")
# 데이터베이스 연결 대기
await wait_for_database()
# 데이터베이스 초기화
from .core.database import init_db
await init_db()

19
backend/src/models/todo.py
View File

@@ -13,9 +13,9 @@ from ..core.database import Base
class TodoCategory(str, enum.Enum):
"""Todo 카테고리"""
TODO = "todo"
CALENDAR = "calendar"
CHECKLIST = "checklist"
MEMO = "memo" # 수신함의 메모 (upload.html에서 생성)
TODO = "todo" # Todo 목록 (inbox.html에서 변환된 것)
BOARD = "board" # 보드 (프로젝트 관리)
class TodoStatus(str, enum.Enum):
@@ -35,20 +35,23 @@ class Todo(Base):
user_id = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=False)
# 기본 정보
title = Column(String(200), nullable=False) # 할일 제목
description = Column(Text, nullable=True) # 할일 설명
category = Column(Enum(TodoCategory), nullable=True, default=None)
title = Column(String(200), nullable=True) # 제목 (메모의 경우 선택사항)
description = Column(Text, nullable=False) # 내용 (메모/Todo 모두 필수)
category = Column(Enum(TodoCategory), nullable=False, default=TodoCategory.MEMO)
status = Column(Enum(TodoStatus), nullable=False, default=TodoStatus.PENDING)
# 시간 관리
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)
due_date = Column(DateTime(timezone=True), nullable=True) # 마감일
start_date = Column(DateTime(timezone=True), nullable=True) # Todo 시작일 (category=todo일 때만 사용)
completed_at = Column(DateTime(timezone=True), nullable=True)
# 추가 정보
image_urls = Column(Text, nullable=True) # 첨부 이미지 URLs (JSON 배열 형태, 최대 5개)
tags = Column(String(500), nullable=True) # 태그 (쉼표로 구분)
# 보드 관련 (category=board일 때 사용)
board_id = Column(UUID(as_uuid=True), nullable=True) # 보드 ID (첫 번째 메모가 보드 생성, 나머지는 하위 메모)
is_board_header = Column(Boolean, default=False) # 보드의 헤더(제목) 여부
# 관계
user = relationship("User", back_populates="todos")

57
backend/src/schemas/todo.py
View File

@@ -9,9 +9,9 @@ from enum import Enum
class TodoCategoryEnum(str, Enum):
TODO = "todo"
CALENDAR = "calendar"
CHECKLIST = "checklist"
MEMO = "memo" # 수신함의 메모
TODO = "todo" # Todo 목록
BOARD = "board" # 보드 (프로젝트 관리)
class TodoStatusEnum(str, Enum):
@@ -23,59 +23,48 @@ class TodoStatusEnum(str, Enum):
class TodoBase(BaseModel):
title: str = Field(..., min_length=1, max_length=200)
description: Optional[str] = Field(None, max_length=2000)
category: Optional[TodoCategoryEnum] = None
title: Optional[str] = Field(None, max_length=200) # 메모의 경우 선택사항
description: str = Field(..., min_length=1, max_length=2000) # 내용은 필수
category: TodoCategoryEnum = Field(default=TodoCategoryEnum.MEMO)
class TodoCreate(TodoBase):
"""Todo 생성"""
due_date: Optional[str] = None # 문자열로 변경 (한국 시간 형식)
"""Todo/메모 생성"""
start_date: Optional[str] = None # Todo 시작일 (category=todo일 때만 사용)
image_urls: Optional[List[str]] = Field(None, max_items=5, description="최대 5개의 이미지 URL")
tags: Optional[str] = Field(None, max_length=500)
board_id: Optional[str] = None # 보드 ID (보드 하위 메모일 때)
is_board_header: Optional[bool] = False # 보드 헤더 여부
class TodoUpdate(BaseModel):
"""Todo 수정"""
title: Optional[str] = Field(None, min_length=1, max_length=200)
description: Optional[str] = Field(None, max_length=2000)
"""Todo/메모 수정"""
title: Optional[str] = Field(None, max_length=200)
description: Optional[str] = Field(None, min_length=1, max_length=2000)
category: Optional[TodoCategoryEnum] = None
status: Optional[TodoStatusEnum] = None
due_date: Optional[str] = None # 문자열로 변경 (한국 시간 형식)
start_date: Optional[str] = None # Todo 시작일
image_urls: Optional[List[str]] = Field(None, max_items=5, description="최대 5개의 이미지 URL")
tags: Optional[str] = Field(None, max_length=500)
board_id: Optional[str] = None # 보드 ID
is_board_header: Optional[bool] = None # 보드 헤더 여부
class TodoResponse(BaseModel):
id: UUID
user_id: UUID
title: str
description: Optional[str] = None
category: Optional[TodoCategoryEnum] = None
title: Optional[str] = None
description: str
category: TodoCategoryEnum
status: TodoStatusEnum
created_at: datetime
updated_at: datetime
due_date: Optional[str] = None # 문자열로 변경 (한국 시간 형식)
start_date: Optional[str] = None # Todo 시작일
completed_at: Optional[datetime] = None
image_urls: Optional[List[str]] = None
tags: Optional[str] = None
board_id: Optional[str] = None # 보드 ID
is_board_header: Optional[bool] = None # 보드 헤더 여부
class Config:
from_attributes = True
class TodoStats(BaseModel):
"""Todo 통계"""
total_count: int
pending_count: int
in_progress_count: int
completed_count: int
completion_rate: float # 완료율 (%)
class TodoDashboard(BaseModel):
"""Todo 대시보드"""
stats: TodoStats
today_todos: List[TodoResponse]
overdue_todos: List[TodoResponse]
upcoming_todos: List[TodoResponse]
# 간단한 워크플로우에 맞게 통계 기능 제거

9
backend/src/services/__init__.py
View File

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

74
backend/src/services/calendar_sync_service.py
View File

@@ -1,74 +0,0 @@
"""
캘린더 동기화 서비스
- 서비스 클래스 기준: 최대 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

41
backend/src/services/file_service.py
View File

@@ -13,6 +13,47 @@ def ensure_upload_dir():
if not os.path.exists(UPLOAD_DIR):
os.makedirs(UPLOAD_DIR)
def save_image(image_data: bytes, filename: str = None) -> Optional[str]:
"""이미지 데이터를 파일로 저장하고 경로 반환"""
try:
ensure_upload_dir()
# 파일명 생성
if not filename:
filename = f"{uuid.uuid4()}.jpg"
else:
# 확장자 확인 및 변경
name, ext = os.path.splitext(filename)
filename = f"{name}_{uuid.uuid4()}.jpg"
filepath = os.path.join(UPLOAD_DIR, filename)
# 이미지 처리 및 저장
image = Image.open(io.BytesIO(image_data))
# RGBA를 RGB로 변환 (JPEG는 투명도 지원 안함)
if image.mode in ('RGBA', 'LA'):
background = Image.new('RGB', image.size, (255, 255, 255))
background.paste(image, mask=image.split()[-1] if image.mode == 'RGBA' else None)
image = background
elif image.mode != 'RGB':
image = image.convert('RGB')
# 이미지 크기 조정 (최대 1920px)
max_size = 1920
if image.width > max_size or image.height > max_size:
image.thumbnail((max_size, 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 save_base64_image(base64_string: str) -> Optional[str]:
"""Base64 이미지를 파일로 저장하고 경로 반환"""
try:

300
backend/src/services/todo_service.py
View File

@@ -1,300 +0,0 @@
"""
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
)