244 lines
12 KiB
Python
244 lines
12 KiB
Python
# /data/gyber/apps/web/gyber/db/resource.py
|
|
import logging
|
|
from django.db import connections, DatabaseError, InterfaceError, OperationalError
|
|
from .base import execute_procedure, dictfetchall
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# --- 자산(Resource) 관련 함수 ---
|
|
|
|
def get_all_resources(page_num, page_size, sort_column, sort_direction, category_id=None, group_id=None, user_id=None):
|
|
"""
|
|
자산 목록을 조회합니다 (페이징, 정렬, 필터링 포함).
|
|
프로시저: sp_get_all_resources (결과셋 2개: 목록, 총 개수)
|
|
반환값: (resource_list: list[dict], total_count: int)
|
|
"""
|
|
params = [page_num, page_size, sort_column, sort_direction, category_id, group_id, user_id]
|
|
logger.debug(f"Calling sp_get_all_resources with params: {params}")
|
|
try:
|
|
with connections['default'].cursor() as cursor:
|
|
cursor.callproc('sp_get_all_resources', params)
|
|
resource_list = dictfetchall(cursor)
|
|
total_count = 0
|
|
if cursor.nextset():
|
|
total_count_result = dictfetchall(cursor)
|
|
total_count = total_count_result[0]['total_count'] if total_count_result else 0
|
|
|
|
logger.debug(f"sp_get_all_resources returned {len(resource_list)} resources, total_count: {total_count}")
|
|
return resource_list, total_count
|
|
except Exception as e:
|
|
logger.error(f"Error in get_all_resources: {e}", exc_info=True)
|
|
return [], 0
|
|
|
|
def get_resources_by_search(search_term, page_num, page_size, sort_column, sort_direction, category_id=None, group_id=None, user_id=None):
|
|
"""
|
|
자산 목록을 검색하여 조회합니다 (페이징, 정렬, 필터링 포함).
|
|
프로시저: sp_get_resources_by_search (결과셋 2개: 목록, 총 개수)
|
|
반환값: (resource_list: list[dict], total_count: int)
|
|
"""
|
|
params = [search_term, page_num, page_size, sort_column, sort_direction, category_id, group_id, user_id]
|
|
logger.debug(f"Calling sp_get_resources_by_search with params: {params}")
|
|
try:
|
|
with connections['default'].cursor() as cursor:
|
|
cursor.callproc('sp_get_resources_by_search', params)
|
|
resource_list = dictfetchall(cursor)
|
|
total_count = 0
|
|
if cursor.nextset():
|
|
total_count_result = dictfetchall(cursor)
|
|
total_count = total_count_result[0]['total_count'] if total_count_result else 0
|
|
|
|
logger.debug(f"sp_get_resources_by_search returned {len(resource_list)} resources, total_count: {total_count}")
|
|
return resource_list, total_count
|
|
except Exception as e:
|
|
logger.error(f"Error in get_resources_by_search: {e}", exc_info=True)
|
|
return [], 0
|
|
|
|
def get_resource_by_id(resource_id):
|
|
"""
|
|
특정 자산 ID로 상세 정보를 조회합니다.
|
|
프로시저: sp_get_resource_by_id
|
|
반환값: dict (자산 정보) 또는 None
|
|
"""
|
|
return execute_procedure('sp_get_resource_by_id', [resource_id], fetch_mode='one_dict')
|
|
|
|
def add_new_resource(admin_user_id, actor_description, category_id, resource_code, manufacturer,
|
|
resource_name, serial_num, spec_value, spec_unit, user_id, comments, purchase_date, is_locked): # register_date는 프로시저에서 NOW() 사용
|
|
"""
|
|
새로운 자산을 추가합니다.
|
|
프로시저: sp_add_resource (OUT: p_new_resource_id - 13번째, index 12)
|
|
Args:
|
|
purchase_date: 구매일 (Nullable)
|
|
Returns:
|
|
tuple: (success: bool, message: str, new_resource_id: int|None)
|
|
"""
|
|
params = [
|
|
admin_user_id, actor_description, category_id, resource_code, manufacturer,
|
|
resource_name, serial_num, spec_value, spec_unit, user_id, comments, purchase_date, is_locked,
|
|
None # OUT p_new_resource_id placeholder
|
|
]
|
|
call_params = params[:-1] # OUT 파라미터 제외하고 로깅
|
|
logger.debug(f"Calling sp_add_resource with params (excluding OUT): {call_params}")
|
|
|
|
try:
|
|
with connections['default'].cursor() as cursor:
|
|
cursor.callproc('sp_add_resource', params)
|
|
# OUT 파라미터 가져오기 (p_new_resource_id 인덱스 12)
|
|
cursor.execute("SELECT @_sp_add_resource_12;")
|
|
result = cursor.fetchone()
|
|
|
|
if result:
|
|
new_id = result[0]
|
|
if new_id is not None and isinstance(new_id, int):
|
|
message = f"자산 (ID: {new_id}) 추가 성공."
|
|
logger.info(message)
|
|
return True, message, new_id
|
|
else:
|
|
# 프로시저 내부 오류 또는 ID 반환 실패 (거의 발생하지 않음)
|
|
message = "자산 추가는 실행되었으나 ID를 가져오지 못했습니다."
|
|
logger.warning(message)
|
|
return False, message, None
|
|
else:
|
|
logger.error("Failed to retrieve OUT parameter p_new_resource_id from sp_add_resource.")
|
|
return False, "프로시저 결과(OUT 파라미터)를 가져오는데 실패했습니다.", None
|
|
|
|
except (DatabaseError, InterfaceError, OperationalError) as e:
|
|
# 프로시저 내부 SIGNAL 로 발생한 오류 처리
|
|
error_message = str(e)
|
|
if '45000' in error_message: # SIGNAL 로 발생한 사용자 정의 오류 코드
|
|
# MySQL/MariaDB 오류 메시지에서 실제 텍스트 추출 (Connector 마다 다를 수 있음)
|
|
# 예: (1644, "이미 등록된 시리얼 번호입니다.") -> "이미 등록된 시리얼 번호입니다."
|
|
try:
|
|
# 정규식 또는 문자열 처리로 메시지 부분만 추출
|
|
import re
|
|
match = re.search(r"'(.*?)'", error_message)
|
|
if match:
|
|
clean_message = match.group(1)
|
|
else: # 다른 형식의 오류 메시지 대비
|
|
clean_message = error_message.split(':', 1)[-1].strip() if ':' in error_message else error_message
|
|
except:
|
|
clean_message = error_message # 실패 시 원본 메시지
|
|
logger.warning(f"sp_add_resource reported an issue: {clean_message}")
|
|
return False, clean_message, None
|
|
else: # 일반 DB 오류
|
|
logger.error(f"Database error calling sp_add_resource: {e}", exc_info=True)
|
|
return False, f"데이터베이스 오류: {e}", None
|
|
except Exception as e:
|
|
logger.error(f"Unexpected error calling sp_add_resource: {e}", exc_info=True)
|
|
return False, f"알 수 없는 오류: {e}", None
|
|
|
|
def update_resource(admin_user_id, actor_description, resource_id, category_id, resource_code, manufacturer,
|
|
resource_name, serial_num, spec_value, spec_unit, user_id, comments, purchase_date, is_locked):
|
|
"""
|
|
자산 정보를 수정합니다. (프로시저에 OUT 파라미터 없음)
|
|
프로시저: sp_update_resource
|
|
Returns:
|
|
tuple: (success: bool, message: str)
|
|
"""
|
|
params = [
|
|
admin_user_id, actor_description, resource_id, category_id, resource_code, manufacturer,
|
|
resource_name, serial_num, spec_value, spec_unit, user_id, comments, purchase_date, is_locked
|
|
]
|
|
logger.debug(f"Calling sp_update_resource for resource_id {resource_id} with params: {params}")
|
|
|
|
try:
|
|
# execute_procedure는 성공 시 True, DB 오류 시 False 반환 (fetch_mode='none')
|
|
success = execute_procedure('sp_update_resource', params, fetch_mode='none')
|
|
|
|
if success:
|
|
message = f"자산 (ID: {resource_id}) 정보 수정 성공."
|
|
logger.info(message)
|
|
return True, message
|
|
else:
|
|
# execute_procedure 내부에서 이미 오류 로깅됨
|
|
# SIGNAL 오류는 execute_procedure 에서 잡히지 않고 예외로 나올 것임
|
|
return False, "자산 정보 수정 중 데이터베이스 오류 발생."
|
|
|
|
except (DatabaseError, InterfaceError, OperationalError) as e:
|
|
# 프로시저 내부 SIGNAL 오류 처리
|
|
error_message = str(e)
|
|
if '45000' in error_message:
|
|
try:
|
|
import re
|
|
match = re.search(r"'(.*?)'", error_message)
|
|
clean_message = match.group(1) if match else error_message
|
|
except:
|
|
clean_message = error_message
|
|
logger.warning(f"sp_update_resource reported an issue: {clean_message}")
|
|
return False, clean_message
|
|
else:
|
|
logger.error(f"Database error calling sp_update_resource: {e}", exc_info=True)
|
|
return False, f"데이터베이스 오류: {e}"
|
|
except Exception as e:
|
|
logger.error(f"Unexpected error calling sp_update_resource: {e}", exc_info=True)
|
|
return False, f"알 수 없는 오류: {e}"
|
|
|
|
def delete_resource(admin_user_id, actor_description, resource_id):
|
|
"""
|
|
자산을 삭제합니다. (프로시저에 OUT 파라미터 없음)
|
|
프로시저: sp_delete_resource
|
|
Returns:
|
|
tuple: (success: bool, message: str)
|
|
"""
|
|
params = [admin_user_id, actor_description, resource_id]
|
|
logger.debug(f"Calling sp_delete_resource for resource_id {resource_id} with params: {params}")
|
|
|
|
try:
|
|
success = execute_procedure('sp_delete_resource', params, fetch_mode='none')
|
|
|
|
if success:
|
|
message = f"자산 (ID: {resource_id}) 삭제 성공."
|
|
logger.info(message)
|
|
return True, message
|
|
else:
|
|
return False, "자산 삭제 중 데이터베이스 오류 발생."
|
|
|
|
except (DatabaseError, InterfaceError, OperationalError) as e:
|
|
# 프로시저 내부 SIGNAL 오류 처리
|
|
error_message = str(e)
|
|
if '45000' in error_message:
|
|
try:
|
|
import re
|
|
match = re.search(r"'(.*?)'", error_message)
|
|
clean_message = match.group(1) if match else error_message
|
|
except:
|
|
clean_message = error_message
|
|
logger.warning(f"sp_delete_resource reported an issue: {clean_message}")
|
|
return False, clean_message
|
|
else:
|
|
logger.error(f"Database error calling sp_delete_resource: {e}", exc_info=True)
|
|
return False, f"데이터베이스 오류: {e}"
|
|
except Exception as e:
|
|
logger.error(f"Unexpected error calling sp_delete_resource: {e}", exc_info=True)
|
|
return False, f"알 수 없는 오류: {e}"
|
|
|
|
def get_resources_by_account(account_name):
|
|
"""
|
|
특정 계정(사용자)에게 할당된 자산 목록을 조회합니다.
|
|
프로시저: sp_get_resources_by_account (account_name 파라미터 사용)
|
|
"""
|
|
params = [account_name]
|
|
logger.debug(f"Calling sp_get_resources_by_account with account_name: {account_name}")
|
|
result = execute_procedure('sp_get_resources_by_account', params, fetch_mode='dict')
|
|
if result is None: return [] # 오류 시 빈 리스트
|
|
# 프로시저가 사용자를 못 찾으면 빈 결과를 반환하므로, 결과가 []인 것은 정상일 수 있음
|
|
logger.debug(f"sp_get_resources_by_account returned {len(result)} resources.")
|
|
return result
|
|
|
|
def get_all_resources_for_export(sort_column, sort_direction, category_id=None, group_id=None, user_id=None):
|
|
"""
|
|
모든 자산 목록을 조회합니다 (내보내기용 - 페이징 없음).
|
|
프로시저: sp_get_all_resources_export
|
|
"""
|
|
params = [sort_column, sort_direction, category_id, group_id, user_id]
|
|
logger.debug(f"Calling sp_get_all_resources_export with params: {params}")
|
|
# execute_procedure가 딕셔너리 리스트를 반환한다고 가정
|
|
return execute_procedure('sp_get_all_resources_export', params, fetch_mode='all_dicts')
|
|
|
|
def get_resources_by_search_for_export(search_term, sort_column, sort_direction, category_id=None, group_id=None, user_id=None):
|
|
"""
|
|
검색된 모든 자산 목록을 조회합니다 (내보내기용 - 페이징 없음).
|
|
프로시저: sp_get_resources_by_search_export
|
|
"""
|
|
params = [search_term, sort_column, sort_direction, category_id, group_id, user_id]
|
|
logger.debug(f"Calling sp_get_resources_by_search_export with params: {params}")
|
|
return execute_procedure('sp_get_resources_by_search_export', params, fetch_mode='all_dicts') |