624 lines
21 KiB
Python
624 lines
21 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
파일 저장소 서비스 모듈
|
|
첨부파일(사진, PDF, HWP 등)을 관리하는 서비스입니다.
|
|
|
|
storage/
|
|
├── 사진/
|
|
│ └── 2025/
|
|
│ └── (20251202) 7편성 3호차 2위 출입문/
|
|
│ └── photo1.jpg
|
|
└── 보고서/
|
|
├── 01.출입문/
|
|
├── 02.ATCATO/
|
|
├── ...
|
|
└── 17.기타/
|
|
└── (20251202) 1호선 제2107열차 7편성 3호차 2위 출입문 닫힘 고장 동향보고.pdf
|
|
"""
|
|
|
|
import os
|
|
import uuid
|
|
import shutil
|
|
from pathlib import Path
|
|
from typing import List, Dict, Optional, Tuple
|
|
from dataclasses import dataclass, field, asdict
|
|
from datetime import datetime, date
|
|
import json
|
|
|
|
from core.constants import ROOT_DIR
|
|
from core.logger import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
# 저장소 기본 경로
|
|
STORAGE_DIR = ROOT_DIR / "storage"
|
|
IMAGES_DIR = STORAGE_DIR / "사진"
|
|
REPORTS_DIR = STORAGE_DIR / "보고서"
|
|
|
|
# 지원 파일 형식
|
|
SUPPORTED_IMAGE_EXTENSIONS = {'.jpg', '.jpeg', '.png', '.gif', '.bmp', '.webp'}
|
|
SUPPORTED_REPORT_EXTENSIONS = {'.pdf', '.hwp', '.hwpx', '.doc', '.docx'}
|
|
|
|
# 보고서 장치 분류 폴더
|
|
REPORT_CATEGORIES = [
|
|
"01.출입문",
|
|
"02.ATCATO",
|
|
"03.VVVF",
|
|
"04.SIV",
|
|
"05.공조장치",
|
|
"06.제동장치",
|
|
"07.대차",
|
|
"08.집전장치",
|
|
"09.차체",
|
|
"10.조명",
|
|
"11.방송",
|
|
"12.TCMS",
|
|
"13.객실설비",
|
|
"14.VOC",
|
|
"15.상시급전",
|
|
"16.기상상황",
|
|
"17.기타",
|
|
]
|
|
|
|
|
|
@dataclass
|
|
class Attachment:
|
|
"""첨부파일 정보"""
|
|
id: str = "" # UUID
|
|
title: str = "" # 제목 (예: "3호차 2위 출입문 사진")
|
|
filename: str = "" # 원본 파일명
|
|
filepath: str = "" # 저장 경로 (상대경로)
|
|
file_type: str = "" # image, pdf, hwp, doc
|
|
file_size: int = 0 # 바이트
|
|
created_at: str = "" # ISO 형식
|
|
record_type: str = "" # faults, works 등
|
|
record_id: int = 0
|
|
group_id: str = "" # 그룹 ID (여러 파일을 하나의 제목으로 묶을 때)
|
|
order: int = 0 # 그룹 내 순서
|
|
|
|
def to_dict(self) -> Dict:
|
|
return asdict(self)
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: Dict) -> 'Attachment':
|
|
return cls(**data)
|
|
|
|
def get_full_path(self) -> Path:
|
|
"""전체 경로 반환"""
|
|
return STORAGE_DIR / self.filepath
|
|
|
|
def get_thumbnail_path(self) -> Optional[Path]:
|
|
"""썸네일 경로 반환 (이미지만)"""
|
|
if self.file_type != 'image':
|
|
return None
|
|
full_path = self.get_full_path()
|
|
thumb_dir = full_path.parent / "thumbnails"
|
|
return thumb_dir / f"thumb_{full_path.name}"
|
|
|
|
|
|
@dataclass
|
|
class AttachmentGroup:
|
|
"""첨부파일 그룹 (하나의 제목에 여러 파일)"""
|
|
id: str = "" # UUID
|
|
title: str = "" # 그룹 제목
|
|
attachments: List[Attachment] = field(default_factory=list)
|
|
created_at: str = ""
|
|
|
|
def to_dict(self) -> Dict:
|
|
return {
|
|
'id': self.id,
|
|
'title': self.title,
|
|
'attachments': [a.to_dict() for a in self.attachments],
|
|
'created_at': self.created_at
|
|
}
|
|
|
|
|
|
class StorageService:
|
|
"""파일 저장소 서비스"""
|
|
|
|
def __init__(self):
|
|
self._ensure_directories()
|
|
self._attachments_cache: Dict[str, List[Attachment]] = {}
|
|
|
|
def _ensure_directories(self):
|
|
"""필요한 디렉토리 생성"""
|
|
STORAGE_DIR.mkdir(exist_ok=True)
|
|
IMAGES_DIR.mkdir(exist_ok=True)
|
|
REPORTS_DIR.mkdir(exist_ok=True)
|
|
|
|
# 보고서 장치분류별 폴더 생성
|
|
for category in REPORT_CATEGORIES:
|
|
(REPORTS_DIR / category).mkdir(exist_ok=True)
|
|
|
|
def _get_image_dir(self, folder_title: str, folder_date: Optional[date] = None) -> Path:
|
|
"""
|
|
사진 저장 디렉토리 반환
|
|
|
|
구조: storage/사진/2025/(20251202) 7편성 3호차 2위 출입문/
|
|
|
|
Args:
|
|
folder_title: 폴더 제목 (예: "7편성 3호차 2위 출입문")
|
|
folder_date: 날짜 (기본값: 오늘)
|
|
|
|
Returns:
|
|
사진 저장 경로
|
|
"""
|
|
if folder_date is None:
|
|
folder_date = date.today()
|
|
|
|
year = str(folder_date.year)
|
|
date_str = folder_date.strftime("%Y%m%d")
|
|
folder_name = f"({date_str}) {folder_title}"
|
|
|
|
image_dir = IMAGES_DIR / year / folder_name
|
|
image_dir.mkdir(parents=True, exist_ok=True)
|
|
return image_dir
|
|
|
|
def _get_report_dir(self, device_category: str) -> Path:
|
|
"""
|
|
보고서 저장 디렉토리 반환
|
|
|
|
구조: storage/보고서/01.출입문/
|
|
|
|
Args:
|
|
device_category: 장치분류 번호 또는 이름 (예: "01.출입문" 또는 "출입문" 또는 "1")
|
|
|
|
Returns:
|
|
보고서 저장 경로
|
|
"""
|
|
# 번호로 입력된 경우 매칭
|
|
category_folder = self._match_category(device_category)
|
|
|
|
report_dir = REPORTS_DIR / category_folder
|
|
report_dir.mkdir(parents=True, exist_ok=True)
|
|
return report_dir
|
|
|
|
def _match_category(self, device_category: str) -> str:
|
|
"""장치분류 매칭"""
|
|
if not device_category:
|
|
return "17.기타"
|
|
|
|
# 이미 전체 이름인 경우
|
|
for cat in REPORT_CATEGORIES:
|
|
if cat == device_category:
|
|
return cat
|
|
|
|
# 번호만 입력된 경우 (예: "1", "01")
|
|
try:
|
|
num = int(device_category.lstrip('0'))
|
|
for cat in REPORT_CATEGORIES:
|
|
cat_num = int(cat.split('.')[0])
|
|
if cat_num == num:
|
|
return cat
|
|
except ValueError:
|
|
pass
|
|
|
|
# 이름만 입력된 경우 (예: "출입문")
|
|
for cat in REPORT_CATEGORIES:
|
|
if device_category in cat:
|
|
return cat
|
|
|
|
return "17.기타"
|
|
|
|
def _get_file_type(self, filename: str) -> str:
|
|
"""파일 확장자로 타입 판별"""
|
|
ext = Path(filename).suffix.lower()
|
|
if ext in SUPPORTED_IMAGE_EXTENSIONS:
|
|
return 'image'
|
|
elif ext == '.pdf':
|
|
return 'pdf'
|
|
elif ext in {'.hwp', '.hwpx'}:
|
|
return 'hwp'
|
|
elif ext in {'.doc', '.docx'}:
|
|
return 'doc'
|
|
return 'unknown'
|
|
|
|
def _generate_thumbnail(self, image_path: Path) -> Optional[Path]:
|
|
"""이미지 썸네일 생성"""
|
|
try:
|
|
from PIL import Image
|
|
|
|
thumb_dir = image_path.parent / "thumbnails"
|
|
thumb_dir.mkdir(exist_ok=True)
|
|
thumb_path = thumb_dir / f"thumb_{image_path.name}"
|
|
|
|
with Image.open(image_path) as img:
|
|
# EXIF 회전 정보 적용
|
|
try:
|
|
from PIL import ImageOps
|
|
img = ImageOps.exif_transpose(img)
|
|
except Exception:
|
|
pass
|
|
|
|
# 썸네일 크기 (150x150)
|
|
img.thumbnail((150, 150), Image.Resampling.LANCZOS)
|
|
|
|
# RGB로 변환 (PNG 알파 채널 처리)
|
|
if img.mode in ('RGBA', 'P'):
|
|
img = img.convert('RGB')
|
|
|
|
img.save(thumb_path, 'JPEG', quality=85)
|
|
|
|
return thumb_path
|
|
except Exception as e:
|
|
logger.error(f"썸네일 생성 실패: {e}")
|
|
return None
|
|
|
|
def save_image(
|
|
self,
|
|
file_path: str,
|
|
record_type: str,
|
|
record_id: int,
|
|
folder_title: str,
|
|
folder_date: Optional[date] = None,
|
|
group_id: str = "",
|
|
order: int = 0
|
|
) -> Optional[Attachment]:
|
|
"""
|
|
사진 파일 저장
|
|
|
|
구조: storage/사진/2025/(20251202) 7편성 3호차 2위 출입문/photo.jpg
|
|
|
|
Args:
|
|
file_path: 원본 파일 경로
|
|
record_type: 레코드 타입 (faults, works 등)
|
|
record_id: 레코드 ID
|
|
folder_title: 폴더 제목 (예: "7편성 3호차 2위 출입문")
|
|
folder_date: 날짜 (기본값: 오늘)
|
|
group_id: 그룹 ID (여러 파일을 하나의 제목으로 묶을 때)
|
|
order: 그룹 내 순서
|
|
|
|
Returns:
|
|
저장된 첨부파일 정보
|
|
"""
|
|
try:
|
|
source_path = Path(file_path)
|
|
if not source_path.exists():
|
|
logger.error(f"파일이 존재하지 않습니다: {file_path}")
|
|
return None
|
|
|
|
filename = source_path.name
|
|
file_type = self._get_file_type(filename)
|
|
|
|
if file_type != 'image':
|
|
logger.error(f"이미지 파일이 아닙니다: {filename}")
|
|
return None
|
|
|
|
# 사진 저장 디렉토리
|
|
save_dir = self._get_image_dir(folder_title, folder_date)
|
|
|
|
# 파일명 (순서 포함)
|
|
file_id = str(uuid.uuid4())[:8]
|
|
dest_path = save_dir / f"{file_id}_{filename}"
|
|
|
|
# 파일 복사
|
|
shutil.copy2(source_path, dest_path)
|
|
|
|
# 썸네일 생성
|
|
self._generate_thumbnail(dest_path)
|
|
|
|
# 상대 경로 계산
|
|
relative_path = dest_path.relative_to(STORAGE_DIR)
|
|
|
|
# 첨부파일 정보 생성
|
|
attachment = Attachment(
|
|
id=file_id,
|
|
title=folder_title,
|
|
filename=filename,
|
|
filepath=str(relative_path),
|
|
file_type=file_type,
|
|
file_size=dest_path.stat().st_size,
|
|
created_at=datetime.now().isoformat(),
|
|
record_type=record_type,
|
|
record_id=record_id,
|
|
group_id=group_id or file_id,
|
|
order=order
|
|
)
|
|
|
|
# 메타데이터 저장
|
|
self._save_metadata(record_type, record_id, attachment)
|
|
|
|
logger.info(f"사진 저장 완료: {filename} -> {dest_path}")
|
|
return attachment
|
|
|
|
except Exception as e:
|
|
logger.error(f"사진 저장 실패: {e}")
|
|
return None
|
|
|
|
def save_report(
|
|
self,
|
|
file_path: str,
|
|
record_type: str,
|
|
record_id: int,
|
|
device_category: str,
|
|
report_title: str = "",
|
|
report_date: Optional[date] = None
|
|
) -> Optional[Attachment]:
|
|
"""
|
|
보고서 파일 저장
|
|
|
|
구조: storage/보고서/01.출입문/(20251202) 1호선 제2107열차 7편성 3호차 2위 출입문 닫힘 고장 동향보고.pdf
|
|
|
|
Args:
|
|
file_path: 원본 파일 경로
|
|
record_type: 레코드 타입 (faults, works 등)
|
|
record_id: 레코드 ID
|
|
device_category: 장치분류 (예: "출입문", "01.출입문", "1")
|
|
report_title: 보고서 제목 (날짜 제외, 예: "1호선 제2107열차 7편성 3호차 2위 출입문 닫힘 고장 동향보고")
|
|
report_date: 보고서 날짜 (기본값: 오늘)
|
|
|
|
Returns:
|
|
저장된 첨부파일 정보
|
|
"""
|
|
try:
|
|
source_path = Path(file_path)
|
|
if not source_path.exists():
|
|
logger.error(f"파일이 존재하지 않습니다: {file_path}")
|
|
return None
|
|
|
|
filename = source_path.name
|
|
file_type = self._get_file_type(filename)
|
|
|
|
if file_type not in ('pdf', 'hwp', 'doc'):
|
|
logger.error(f"보고서 파일 형식이 아닙니다: {filename}")
|
|
return None
|
|
|
|
# 보고서 저장 디렉토리
|
|
save_dir = self._get_report_dir(device_category)
|
|
|
|
# 파일명 생성: (YYYYMMDD) 제목.확장자
|
|
if report_date is None:
|
|
report_date = date.today()
|
|
|
|
date_str = report_date.strftime("%Y%m%d")
|
|
ext = source_path.suffix
|
|
|
|
# 제목이 없으면 원본 파일명 사용 (확장자 제외)
|
|
if not report_title:
|
|
report_title = source_path.stem
|
|
|
|
# 새 파일명
|
|
new_filename = f"({date_str}) {report_title}{ext}"
|
|
dest_path = save_dir / new_filename
|
|
|
|
# 동일 파일명 존재시 고유 ID 추가
|
|
if dest_path.exists():
|
|
file_id = str(uuid.uuid4())[:4]
|
|
new_filename = f"({date_str}) {report_title}_{file_id}{ext}"
|
|
dest_path = save_dir / new_filename
|
|
|
|
# 파일 복사
|
|
shutil.copy2(source_path, dest_path)
|
|
|
|
# 상대 경로 계산
|
|
relative_path = dest_path.relative_to(STORAGE_DIR)
|
|
file_id = str(uuid.uuid4())[:8]
|
|
|
|
# 첨부파일 정보 생성
|
|
attachment = Attachment(
|
|
id=file_id,
|
|
title=report_title,
|
|
filename=new_filename,
|
|
filepath=str(relative_path),
|
|
file_type=file_type,
|
|
file_size=dest_path.stat().st_size,
|
|
created_at=datetime.now().isoformat(),
|
|
record_type=record_type,
|
|
record_id=record_id,
|
|
group_id=file_id,
|
|
order=0
|
|
)
|
|
|
|
# 메타데이터 저장
|
|
self._save_metadata(record_type, record_id, attachment)
|
|
|
|
logger.info(f"보고서 저장 완료: {new_filename} -> {dest_path}")
|
|
return attachment
|
|
|
|
except Exception as e:
|
|
logger.error(f"보고서 저장 실패: {e}")
|
|
return None
|
|
|
|
def save_images(
|
|
self,
|
|
file_paths: List[str],
|
|
record_type: str,
|
|
record_id: int,
|
|
folder_title: str,
|
|
folder_date: Optional[date] = None
|
|
) -> List[Attachment]:
|
|
"""
|
|
여러 사진 파일 저장 (하나의 폴더/그룹으로)
|
|
|
|
Args:
|
|
file_paths: 파일 경로 리스트
|
|
record_type: 레코드 타입
|
|
record_id: 레코드 ID
|
|
folder_title: 폴더 제목 (예: "7편성 3호차 2위 출입문")
|
|
folder_date: 날짜 (기본값: 오늘)
|
|
|
|
Returns:
|
|
저장된 첨부파일 리스트
|
|
"""
|
|
if not file_paths:
|
|
return []
|
|
|
|
group_id = str(uuid.uuid4())[:8]
|
|
attachments = []
|
|
|
|
for i, file_path in enumerate(file_paths):
|
|
attachment = self.save_image(
|
|
file_path=file_path,
|
|
record_type=record_type,
|
|
record_id=record_id,
|
|
folder_title=folder_title,
|
|
folder_date=folder_date,
|
|
group_id=group_id,
|
|
order=i
|
|
)
|
|
if attachment:
|
|
attachments.append(attachment)
|
|
|
|
return attachments
|
|
|
|
def _get_metadata_path(self, record_type: str, record_id: int) -> Path:
|
|
"""메타데이터 파일 경로"""
|
|
return STORAGE_DIR / f"{record_type}_{record_id}_meta.json"
|
|
|
|
def _save_metadata(self, record_type: str, record_id: int, attachment: Attachment):
|
|
"""메타데이터 저장"""
|
|
meta_path = self._get_metadata_path(record_type, record_id)
|
|
|
|
# 기존 메타데이터 로드
|
|
attachments = self.get_attachments(record_type, record_id)
|
|
|
|
# 중복 제거 후 추가
|
|
attachments = [a for a in attachments if a.id != attachment.id]
|
|
attachments.append(attachment)
|
|
|
|
# 저장
|
|
data = [a.to_dict() for a in attachments]
|
|
with open(meta_path, 'w', encoding='utf-8') as f:
|
|
json.dump(data, f, ensure_ascii=False, indent=2)
|
|
|
|
def get_attachments(self, record_type: str, record_id: int) -> List[Attachment]:
|
|
"""레코드의 첨부파일 목록 조회"""
|
|
meta_path = self._get_metadata_path(record_type, record_id)
|
|
|
|
if not meta_path.exists():
|
|
return []
|
|
|
|
try:
|
|
with open(meta_path, 'r', encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
return [Attachment.from_dict(d) for d in data]
|
|
except Exception as e:
|
|
logger.error(f"메타데이터 로드 실패: {e}")
|
|
return []
|
|
|
|
def get_attachments_by_type(
|
|
self,
|
|
record_type: str,
|
|
record_id: int,
|
|
file_type: str
|
|
) -> List[Attachment]:
|
|
"""파일 타입별 첨부파일 조회"""
|
|
attachments = self.get_attachments(record_type, record_id)
|
|
return [a for a in attachments if a.file_type == file_type]
|
|
|
|
def get_attachment_groups(
|
|
self,
|
|
record_type: str,
|
|
record_id: int
|
|
) -> List[AttachmentGroup]:
|
|
"""그룹별 첨부파일 조회"""
|
|
attachments = self.get_attachments(record_type, record_id)
|
|
|
|
# 그룹별로 분류
|
|
groups: Dict[str, AttachmentGroup] = {}
|
|
for attachment in attachments:
|
|
group_id = attachment.group_id
|
|
if group_id not in groups:
|
|
groups[group_id] = AttachmentGroup(
|
|
id=group_id,
|
|
title=attachment.title,
|
|
attachments=[],
|
|
created_at=attachment.created_at
|
|
)
|
|
groups[group_id].attachments.append(attachment)
|
|
|
|
# 그룹 내 정렬
|
|
for group in groups.values():
|
|
group.attachments.sort(key=lambda a: a.order)
|
|
|
|
return list(groups.values())
|
|
|
|
def delete_attachment(self, record_type: str, record_id: int, attachment_id: str) -> bool:
|
|
"""첨부파일 삭제"""
|
|
attachments = self.get_attachments(record_type, record_id)
|
|
|
|
target = None
|
|
for a in attachments:
|
|
if a.id == attachment_id:
|
|
target = a
|
|
break
|
|
|
|
if not target:
|
|
return False
|
|
|
|
try:
|
|
# 파일 삭제
|
|
full_path = target.get_full_path()
|
|
if full_path.exists():
|
|
full_path.unlink()
|
|
|
|
# 썸네일 삭제
|
|
thumb_path = target.get_thumbnail_path()
|
|
if thumb_path and thumb_path.exists():
|
|
thumb_path.unlink()
|
|
|
|
# 메타데이터 업데이트
|
|
attachments = [a for a in attachments if a.id != attachment_id]
|
|
meta_path = self._get_metadata_path(record_type, record_id)
|
|
data = [a.to_dict() for a in attachments]
|
|
with open(meta_path, 'w', encoding='utf-8') as f:
|
|
json.dump(data, f, ensure_ascii=False, indent=2)
|
|
|
|
logger.info(f"첨부파일 삭제: {attachment_id}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"첨부파일 삭제 실패: {e}")
|
|
return False
|
|
|
|
def get_attachment_count(self, record_type: str, record_id: int) -> int:
|
|
"""첨부파일 개수 조회"""
|
|
return len(self.get_attachments(record_type, record_id))
|
|
|
|
def update_attachment(self, attachment: Attachment) -> bool:
|
|
"""첨부파일 메타데이터 업데이트
|
|
|
|
Args:
|
|
attachment: 업데이트할 첨부파일 객체 (record_type, record_id, id 필요)
|
|
|
|
Returns:
|
|
성공 여부
|
|
"""
|
|
try:
|
|
self._save_metadata(
|
|
attachment.record_type,
|
|
attachment.record_id,
|
|
attachment
|
|
)
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"첨부파일 업데이트 실패: {e}")
|
|
return False
|
|
|
|
def open_file(self, attachment: Attachment) -> bool:
|
|
"""파일 열기 (시스템 기본 프로그램)"""
|
|
try:
|
|
full_path = attachment.get_full_path()
|
|
if not full_path.exists():
|
|
logger.error(f"파일이 존재하지 않습니다: {full_path}")
|
|
return False
|
|
|
|
os.startfile(str(full_path))
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"파일 열기 실패: {e}")
|
|
return False
|
|
|
|
|
|
# 싱글톤 인스턴스
|
|
_storage_service: Optional[StorageService] = None
|
|
|
|
|
|
def get_storage_service() -> StorageService:
|
|
"""StorageService 싱글톤 인스턴스 반환"""
|
|
global _storage_service
|
|
if _storage_service is None:
|
|
_storage_service = StorageService()
|
|
return _storage_service
|
|
|