"""데이터 저장소 관리"""
import json
from pathlib import Path
from typing import List, Optional, Dict
from datetime import datetime, date
from threading import Lock
from .models import Reservation, Customer, PlatformConfig, Platform
class Storage:
"""인메모리 캐시를 사용하는 로컬 JSON 파일 기반 저장소"""
def __init__(self, data_dir: str = "data"):
self.data_dir = Path(data_dir)
self.data_dir.mkdir(exist_ok=True)
self.reservations_file = self.data_dir / "reservations.json"
self.customers_file = self.data_dir / "customers.json"
self.platforms_file = self.data_dir / "platforms.json"
# 동시성 제어를 위한 Lock
self._lock = Lock()
# 인메모리 캐시
self._reservations_cache: Optional[List[dict]] = None
self._customers_cache: Optional[Dict[str, dict]] = None
self._platforms_cache: Optional[List[dict]] = None
self._init_files()
self._load_all_to_cache()
def _init_files(self):
"""초기 파일 생성"""
if not self.reservations_file.exists():
self._save_json(self.reservations_file, [])
if not self.customers_file.exists():
self._save_json(self.customers_file, {})
if not self.platforms_file.exists():
# 기본 플랫폼 설정
default_platforms = [
{"platform": "airbnb", "fee_rate": 0.15, "enabled": True},
{"platform": "spacecloud", "fee_rate": 0.10, "enabled": True},
{"platform": "naver", "fee_rate": 0.05, "enabled": True},
{"platform": "yanolja", "fee_rate": 0.12, "enabled": True},
{"platform": "kakao", "fee_rate": 0.08, "enabled": True},
]
self._save_json(self.platforms_file, default_platforms)
def _load_json(self, filepath: Path) -> dict | list:
"""JSON 파일 로드"""
with open(filepath, 'r', encoding='utf-8') as f:
return json.load(f)
def _save_json(self, filepath: Path, data: dict | list):
"""JSON 파일 저장"""
with open(filepath, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2, default=str)
def _load_all_to_cache(self):
"""모든 데이터를 메모리에 로드"""
with self._lock:
self._reservations_cache = self._load_json(self.reservations_file)
self._customers_cache = self._load_json(self.customers_file)
self._platforms_cache = self._load_json(self.platforms_file)
# === Reservations ===
def get_all_reservations(self) -> List[Reservation]:
"""모든 예약 조회 (메모리 캐시 사용)"""
with self._lock:
return [Reservation(**item) for item in self._reservations_cache]
def add_reservation(self, reservation: Reservation):
"""예약 추가 (메모리 캐시 + 파일 저장)"""
with self._lock:
self._reservations_cache.append(reservation.model_dump(mode='json'))
self._save_json(self.reservations_file, self._reservations_cache)
def clear_platform_reservations(self, platform: Platform):
"""특정 플랫폼의 모든 예약 삭제 (동기화 전 사용)"""
with self._lock:
self._reservations_cache = [
r for r in self._reservations_cache
if r.get('platform') != platform
]
self._save_json(self.reservations_file, self._reservations_cache)
def sync_platform_reservations(self, platform: Platform, reservations: List[Reservation]):
"""특정 플랫폼의 예약을 새로운 데이터로 완전히 교체 (동기화용)"""
with self._lock:
# 기존 플랫폼 예약 삭제
self._reservations_cache = [
r for r in self._reservations_cache
if r.get('platform') != platform
]
# 새 예약 추가
for reservation in reservations:
self._reservations_cache.append(reservation.model_dump(mode='json'))
self._save_json(self.reservations_file, self._reservations_cache)
def get_reservations_by_date(self, target_date: date) -> List[Reservation]:
"""특정 날짜의 예약 조회"""
all_reservations = self.get_all_reservations()
return [
r for r in all_reservations
if r.start_time.date() == target_date
]
def get_reservations_by_platform(self, platform: Platform) -> List[Reservation]:
"""특정 플랫폼의 예약 조회"""
all_reservations = self.get_all_reservations()
return [r for r in all_reservations if r.platform == platform]
def find_overlapping_reservations(self, reservation: Reservation) -> List[Reservation]:
"""겹치는 예약 찾기"""
all_reservations = self.get_all_reservations()
return [
r for r in all_reservations
if r.id != reservation.id and r.is_overlapping(reservation)
]
# === Customers ===
def get_customer(self, phone: str) -> Optional[Customer]:
"""고객 정보 조회 (메모리 캐시 사용)"""
with self._lock:
if phone in self._customers_cache:
return Customer(**self._customers_cache[phone])
return None
def save_customer(self, customer: Customer):
"""고객 정보 저장 (메모리 캐시 + 파일 저장)"""
with self._lock:
self._customers_cache[customer.phone] = customer.model_dump(mode='json')
self._save_json(self.customers_file, self._customers_cache)
def get_blacklist(self) -> List[Customer]:
"""블랙리스트 조회 (메모리 캐시 사용)"""
with self._lock:
return [
Customer(**data) for data in self._customers_cache.values()
if data.get('is_blacklisted', False)
]
def get_noshow_customers(self, min_count: int = 1) -> List[Customer]:
"""노쇼 고객 조회 (메모리 캐시 사용)"""
with self._lock:
return [
Customer(**data) for data in self._customers_cache.values()
if data.get('noshow_count', 0) >= min_count
]
# === Platforms ===
def get_all_platforms(self) -> List[PlatformConfig]:
"""모든 플랫폼 설정 조회 (메모리 캐시 사용)"""
with self._lock:
return [PlatformConfig(**item) for item in self._platforms_cache]
def get_platform_config(self, platform: Platform) -> Optional[PlatformConfig]:
"""특정 플랫폼 설정 조회 (메모리 캐시 사용)"""
platforms = self.get_all_platforms()
for p in platforms:
if p.platform == platform:
return p
return None
def update_platform_config(self, config: PlatformConfig):
"""플랫폼 설정 업데이트 (메모리 캐시 + 파일 저장)"""
with self._lock:
for i, p in enumerate(self._platforms_cache):
if p['platform'] == config.platform:
self._platforms_cache[i] = config.model_dump(mode='json')
break
self._save_json(self.platforms_file, self._platforms_cache)
