backend/fastapi/app/db/crud_user.py

from sqlalchemy.orm import Session
from sqlalchemy import text
from db import models, schemas
from process.logger import logger
from db.base import get_db


# ================================================================
# 회원관리 함수
# ================================================================

# 로그인
def do_login(user_id, user_pw, db):
    query = text(f"""
        select 
        user_seq
        from manage_user
        where
        user_id = '{user_id}'
        and user_pw = (select 
        encode(
            digest(
                '{user_pw}' || (select user_pw_solt from manage_user where user_id = '{user_id}'), 'sha256'
            ), 'hex'
        ))
        and withdraw_yn = 'N'
    """)
    return db.execute(query).fetchall()


# 마지막 로그인 시간 업데이트
def update_last_login_dt(user_seq, db):
    query = text(f"""
        update manage_user
        set last_login_dt = now()
        where user_seq = {user_seq}
        and withdraw_yn = 'N'
    """)
    try:
        db.execute(query)
        db.commit()
        return True
    except Exception as e:
        logger.error(f"sql error: {e}")
        return False


# 아이디 찾기(닉네임, 이메일)
def find_id_by_name_email(nickname, user_email, db):
    query = text(f"""
        select
        user_seq,
        user_id
        from manage_user
        where nickname = '{nickname}'
        and user_email = '{user_email}'
        and withdraw_yn = 'N'
    """)
    return db.execute(query).fetchall()


# 아이디 찾기(user_seq)
def find_id_by_user_seq(user_seq, db):
    query = text(f"""
        select
        user_id,
        user_email
        from manage_user
        where user_seq = {user_seq}
        and withdraw_yn = 'N'
    """)
    return db.execute(query).fetchall()


# 이메일 발송 3회 이상 됐는지 확인
def select_send_email_cnt(user_email, db):
    query = text(f"""
        select
        count(user_cert_seq)
        from manage_user_cert
        where user_info = '{user_email}'
        and cert_expired between current_date and current_date + interval '1 day'
    """)
    return db.execute(query).fetchall()


# 이메일 발송내역 기록(하루 3회만 발송가능하게 하기 위함)
def insert_send_email_info(user_seq, cert_type, user_info, cert_code, db):
    query = text(f"""
        insert into manage_user_cert(user_seq, cert_type, user_info, cert_code, cert_expired)
        values({user_seq}, '{cert_type}', '{user_info}', '{cert_code}', now() + INTERVAL '5 minutes')
    """)
    try:
        db.execute(query)
        db.commit()
        return True
    except Exception as e:
        logger.error(f"sql error: {e}")
        return False


# 비밀번호 찾기
def find_password_by_id_email(user_id, user_email, db):
    query = text(f"""
        select
        user_seq
        from manage_user
        where user_id = '{user_id}'
        and user_email = '{user_email}'
        and withdraw_yn = 'N'
    """)
    return db.execute(query).fetchall()


# 신규 비밀번호 업데이트
def update_new_password(user_seq, new_pw, new_solt, db):
    query = text(f"""
        update manage_user 
        set
        user_pw = encode(digest(encode(digest('{new_pw}', 'sha256'), 'hex') || '{new_solt}', 'sha256'), 'hex'),
        user_pw_solt = '{new_solt}'
        where 
        user_seq = {user_seq}
        and withdraw_yn = 'N'
    """)
    try:
        db.execute(query)
        db.commit()
        return True
    except Exception as e:
        logger.error(f"sql error: {e}")
        return False


# 회원가입 최종 아이디 중복 재검사
def is_valid_user_id_by_user_id(user_id, db):
    query = text(f"""
        select
        user_seq
        from manage_user
        where user_id = '{user_id}'
    """)
    return db.execute(query).fetchall()


# 신규 유저 등록
def insert_new_user(user_info, db):
    query = text(f"""
        insert into 
            manage_user(
                user_id,
                user_pw,
                user_pw_solt,
                nickname,
                profile_img,
                user_email,
                department,
                introduce_myself,
                last_login_dt,
                create_dt,
                update_dt,
                mandatory_terms_yn,
                withdraw_yn
            )
        values(
            '{user_info['user_id']}',
            (select encode(digest('{user_info['user_pw']}{user_info['user_pw_solt']}', 'sha256'), 'hex')),
            '{user_info['user_pw_solt']}',
            '{user_info['nickname']}',
            'images/user/temp_dir/profile_img/basic_{user_info['user_id']}.png',
            '{user_info['user_email']}',
            '{user_info.get('department', '')}',
            '{user_info.get('introduce_myself', '')}',
            now(),
            now(),
            now(),
            '{user_info.get('mandatory_terms_yn', '')}',
            'N'
        )
    """)
    try:
        db.execute(query)
        db.commit()
        return True
    except Exception as e:
        logger.error(f"sql error: {e}")
        return False


# 프로필 이미지 업데이트
def update_profile_img(user_seq, profile_img, db):
    query = text(f"""
        update manage_user 
        set
        profile_img = '{profile_img}'
        where 
        user_seq = {user_seq}
        and withdraw_yn = 'N'
    """)
    try:
        db.execute(query)
        db.commit()
        return True
    except Exception as e:
        logger.error(f"sql error: {e}")
        return False


# user_seq로 내정보 가져오기
def get_my_info_by_user_seq(user_seq, db):
    query = text(f"""
        select
        nickname,
        user_email,
        department,
        profile_img,
        introduce_myself
        from manage_user
        where user_seq = {user_seq}
    """)
    return db.execute(query).fetchall()


# 현재 비밀번호 일치 확인
def check_current_user_pw(user_seq, user_pw, db):
    query = text(f"""
        select
        count(user_seq)
        from manage_user
        where user_seq = {user_seq}
        and user_pw = (select 
            encode(
                digest(
                    '{user_pw}' || (select user_pw_solt from manage_user where user_seq = '{user_seq}'), 'sha256'
                ), 'hex'
            ))
    """)
    return db.execute(query).fetchall()


# 신규 비밀번호 업데이트
def update_user_info(user_info, db):
    if user_info['user_pw_change_yn'] == 'Y':
        query = text(f"""
            update manage_user 
            set
            user_pw = encode(digest('{user_info['new_user_pw']}' || '{user_info['user_pw_solt']}', 'sha256'), 'hex'),
            user_pw_solt = '{user_info['user_pw_solt']}',
            nickname = '{user_info['nickname']}',
            user_email = '{user_info['user_email']}',
            department = '{user_info['department']}',
            profile_img = '{user_info['profile_img']}',
            introduce_myself = '{user_info['introduce_myself']}'
            where 
            user_seq = {user_info['user_seq']}
            and withdraw_yn = 'N'
        """)
    else:
        query = text(f"""
            update manage_user 
            set
            nickname = '{user_info['nickname']}',
            user_email = '{user_info['user_email']}',
            department = '{user_info['department']}',
            profile_img = '{user_info['profile_img']}',
            introduce_myself = '{user_info['introduce_myself']}'
            where 
            user_seq = {user_info['user_seq']}
            and withdraw_yn = 'N'
        """)
    try:
        db.execute(query)
        db.commit()
        return True
    except Exception as e:
        logger.error(f"sql error: {e}")
        return False


# 회원 탈퇴 처리
def user_withdraw(user_seq, db):
    query = text(f"""
        delete from manage_user
        where 
        user_seq = {user_seq}
        and withdraw_yn = 'N'
    """)
    try:
        db.execute(query)
        db.commit()
        return True
    except Exception as e:
        logger.error(f"sql error: {e}")
        return False