feat: 3-System 분리 프로젝트 초기 코드 작성
TK-FB(공장관리+신고)와 M-Project(부적합관리)를 3개 독립 시스템으로 분리하기 위한 전체 코드 구조 작성. - SSO 인증 서비스 (bcrypt + pbkdf2 이중 해시 지원) - System 1: 공장관리 (TK-FB 기반, 신고 코드 제거) - System 2: 신고 (TK-FB에서 workIssue 코드 추출) - System 3: 부적합관리 (M-Project 기반) - Gateway 포털 (path-based 라우팅) - 통합 docker-compose.yml 및 배포 스크립트 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
98
system3-nonconformance/api/services/auth_service.py
Normal file
98
system3-nonconformance/api/services/auth_service.py
Normal file
@@ -0,0 +1,98 @@
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Optional
|
||||
from jose import JWTError, jwt
|
||||
from passlib.context import CryptContext
|
||||
from sqlalchemy.orm import Session
|
||||
import os
|
||||
import bcrypt as bcrypt_lib
|
||||
|
||||
from database.models import User, UserRole
|
||||
from database.schemas import TokenData
|
||||
|
||||
# 환경 변수 - SSO 공유 시크릿 사용 (docker-compose에서 SECRET_KEY=SSO_JWT_SECRET)
|
||||
SECRET_KEY = os.getenv("SECRET_KEY", "your-secret-key-here")
|
||||
ALGORITHM = os.getenv("ALGORITHM", "HS256")
|
||||
ACCESS_TOKEN_EXPIRE_MINUTES = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", "10080")) # 7 days
|
||||
|
||||
# 비밀번호 암호화 (pbkdf2_sha256 - 로컬 인증용)
|
||||
pwd_context = CryptContext(
|
||||
schemes=["pbkdf2_sha256"],
|
||||
deprecated="auto"
|
||||
)
|
||||
|
||||
def verify_password(plain_password: str, hashed_password: str) -> bool:
|
||||
"""비밀번호 검증 - bcrypt + pbkdf2_sha256 둘 다 지원 (SSO 호환)"""
|
||||
if not plain_password or not hashed_password:
|
||||
return False
|
||||
|
||||
# bcrypt 형식 ($2b$ 또는 $2a$)
|
||||
if hashed_password.startswith(("$2b$", "$2a$")):
|
||||
try:
|
||||
return bcrypt_lib.checkpw(
|
||||
plain_password.encode('utf-8'),
|
||||
hashed_password.encode('utf-8')
|
||||
)
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
# pbkdf2_sha256 형식 (passlib)
|
||||
try:
|
||||
return pwd_context.verify(plain_password, hashed_password)
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
def get_password_hash(password: str) -> str:
|
||||
"""비밀번호 해시 생성 (pbkdf2_sha256)"""
|
||||
return pwd_context.hash(password)
|
||||
|
||||
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
|
||||
to_encode = data.copy()
|
||||
if expires_delta:
|
||||
expire = datetime.utcnow() + expires_delta
|
||||
else:
|
||||
expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
|
||||
to_encode.update({"exp": expire})
|
||||
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
|
||||
return encoded_jwt
|
||||
|
||||
def verify_token(token: str, credentials_exception):
|
||||
"""JWT 토큰 검증 - SSO 토큰 페이로드 구조 지원"""
|
||||
try:
|
||||
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
||||
# SSO 토큰: "sub" 필드에 username
|
||||
# 기존 M-Project 토큰: "sub" 필드에 username
|
||||
username: str = payload.get("sub")
|
||||
if username is None:
|
||||
raise credentials_exception
|
||||
token_data = TokenData(username=username)
|
||||
return token_data
|
||||
except JWTError:
|
||||
raise credentials_exception
|
||||
|
||||
def authenticate_user(db: Session, username: str, password: str):
|
||||
user = db.query(User).filter(User.username == username).first()
|
||||
if not user:
|
||||
return False
|
||||
if not verify_password(password, user.hashed_password):
|
||||
return False
|
||||
return user
|
||||
|
||||
def create_admin_user(db: Session):
|
||||
"""초기 관리자 계정 생성"""
|
||||
admin_username = os.getenv("ADMIN_USERNAME", "hyungi")
|
||||
admin_password = os.getenv("ADMIN_PASSWORD", "djg3-jj34-X3Q3")
|
||||
|
||||
existing_admin = db.query(User).filter(User.username == admin_username).first()
|
||||
if not existing_admin:
|
||||
admin_user = User(
|
||||
username=admin_username,
|
||||
hashed_password=get_password_hash(admin_password),
|
||||
full_name="관리자",
|
||||
role=UserRole.admin,
|
||||
is_active=True
|
||||
)
|
||||
db.add(admin_user)
|
||||
db.commit()
|
||||
print(f"관리자 계정 생성됨: {admin_username}")
|
||||
else:
|
||||
print(f"관리자 계정이 이미 존재함: {admin_username}")
|
||||
136
system3-nonconformance/api/services/file_service.py
Normal file
136
system3-nonconformance/api/services/file_service.py
Normal file
@@ -0,0 +1,136 @@
|
||||
import os
|
||||
import base64
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
import uuid
|
||||
from PIL import Image
|
||||
import io
|
||||
|
||||
# HEIF/HEIC 지원을 위한 라이브러리
|
||||
try:
|
||||
from pillow_heif import register_heif_opener
|
||||
register_heif_opener()
|
||||
HEIF_SUPPORTED = True
|
||||
except ImportError:
|
||||
HEIF_SUPPORTED = False
|
||||
|
||||
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, prefix: str = "image") -> Optional[str]:
|
||||
"""Base64 이미지를 파일로 저장하고 경로 반환"""
|
||||
try:
|
||||
ensure_upload_dir()
|
||||
|
||||
# Base64 헤더 제거 및 정리
|
||||
if "," in base64_string:
|
||||
base64_string = base64_string.split(",")[1]
|
||||
|
||||
# Base64 문자열 정리 (공백, 개행 제거)
|
||||
base64_string = base64_string.strip().replace('\n', '').replace('\r', '').replace(' ', '')
|
||||
|
||||
print(f"🔍 정리된 Base64 길이: {len(base64_string)}")
|
||||
print(f"🔍 Base64 시작 20자: {base64_string[:20]}")
|
||||
|
||||
# 디코딩
|
||||
try:
|
||||
image_data = base64.b64decode(base64_string)
|
||||
print(f"🔍 디코딩된 데이터 길이: {len(image_data)}")
|
||||
print(f"🔍 바이너리 시작 20바이트: {image_data[:20]}")
|
||||
except Exception as decode_error:
|
||||
print(f"❌ Base64 디코딩 실패: {decode_error}")
|
||||
raise decode_error
|
||||
|
||||
# 파일 시그니처 확인
|
||||
file_signature = image_data[:20]
|
||||
print(f"🔍 파일 시그니처 (hex): {file_signature.hex()}")
|
||||
|
||||
# HEIC 파일 시그니처 확인
|
||||
is_heic = b'ftyp' in image_data[:20] and (b'heic' in image_data[:50] or b'mif1' in image_data[:50])
|
||||
print(f"🔍 HEIC 파일 여부: {is_heic}")
|
||||
|
||||
# 이미지 검증 및 형식 확인
|
||||
image = None
|
||||
try:
|
||||
# HEIC 파일인 경우 pillow_heif를 직접 사용하여 처리
|
||||
if is_heic and HEIF_SUPPORTED:
|
||||
print("🔄 HEIC 파일 감지, pillow_heif로 직접 처리...")
|
||||
try:
|
||||
import pillow_heif
|
||||
heif_file = pillow_heif.open_heif(io.BytesIO(image_data), convert_hdr_to_8bit=False)
|
||||
image = heif_file.to_pillow()
|
||||
print(f"✅ HEIC -> PIL 변환 성공: 모드: {image.mode}, 크기: {image.size}")
|
||||
except Exception as heic_error:
|
||||
print(f"⚠️ pillow_heif 직접 처리 실패: {heic_error}")
|
||||
# PIL Image.open()으로 재시도
|
||||
image = Image.open(io.BytesIO(image_data))
|
||||
print(f"✅ PIL Image.open() 성공: {image.format}, 모드: {image.mode}, 크기: {image.size}")
|
||||
else:
|
||||
# 일반 이미지 처리
|
||||
image = Image.open(io.BytesIO(image_data))
|
||||
print(f"🔍 이미지 형식: {image.format}, 모드: {image.mode}, 크기: {image.size}")
|
||||
except Exception as e:
|
||||
print(f"❌ 이미지 열기 실패: {e}")
|
||||
|
||||
# HEIF 재시도
|
||||
if HEIF_SUPPORTED:
|
||||
print("🔄 HEIF 형식으로 재시도...")
|
||||
try:
|
||||
image = Image.open(io.BytesIO(image_data))
|
||||
print(f"✅ HEIF 재시도 성공: {image.format}, 모드: {image.mode}, 크기: {image.size}")
|
||||
except Exception as heif_e:
|
||||
print(f"❌ HEIF 처리도 실패: {heif_e}")
|
||||
print("❌ 지원되지 않는 이미지 형식")
|
||||
raise e
|
||||
else:
|
||||
print("❌ HEIF 지원 라이브러리가 설치되지 않거나 처리 불가")
|
||||
raise e
|
||||
|
||||
# 이미지가 성공적으로 로드되지 않은 경우
|
||||
if image is None:
|
||||
raise Exception("이미지 로드 실패")
|
||||
|
||||
# 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')
|
||||
|
||||
# 파일명 생성 (prefix 포함)
|
||||
filename = f"{prefix}_{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}")
|
||||
Reference in New Issue
Block a user