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


#==================================================================================================
# 구글 Oauth관련
#==================================================================================================

# DB에서 user 찾기
async def get_user_by_oauth_user_email(user_email, db):
    query = text(f"""
        select 
        user_seq
        from manage_user
        where 
        user_email = :user_email
    """)
    params = {
        "user_email": user_email,
    }
    return db.execute(query, params).fetchall()


# 구글 유저 생성
async def create_oauth_user(oauth_uid, user_email, user_id, user_nickname, db):
    query = text(f"""
        insert into manage_user(user_id, oauth_type, oauth_uid, nickname, profile_img, user_email, department, introduce_myself, last_login_dt, create_dt, update_dt, mandatory_terms_yn, withdraw_yn)
        values(:user_id, 'google', :oauth_uid, :user_nickname, '/user/temp_dir/profile_img/profile.png', :user_email, '', '', now(), now(), now(), 'Y', 'N')
    """)
    params = {
        "user_id": user_id,
        "oauth_uid": oauth_uid,
        "user_nickname": user_nickname,
        "user_email": user_email,
    }
    try:
        db.execute(query, params)
        db.commit()
        return True
    except Exception as e:
        logger.error(f"sql error: {e}")
        return False


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

# 로그인
async 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'
    """)
    params = {
        "user_id": user_id,
        "user_pw": user_pw,
    }
    return db.execute(query, params).fetchall()


# 마지막 로그인 시간 업데이트
async 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'
    """)
    params = {
        "user_seq": user_seq,
    }
    try:
        db.execute(query, params)
        db.commit()
        return True
    except Exception as e:
        logger.error(f"sql error: {e}")
        return False


# 아이디 찾기(닉네임, 이메일)
async 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'
    """)
    params = {
        "nickname": nickname,
        "user_email": user_email,
    }
    return db.execute(query, params).fetchall()


# 아이디 찾기(user_seq)
async 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'
    """)
    params = {
        "user_seq": user_seq,
    }
    return db.execute(query, params).fetchall()


# 이메일 발송 3회 이상 됐는지 확인
async 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'
    """)
    params = {
        "user_email": user_email,
    }
    return db.execute(query, params).fetchall()


# 이메일 발송내역 기록(하루 3회만 발송가능하게 하기 위함)
async 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')
    """)
    params = {
        "user_seq": user_seq,
        "cert_type": cert_type,
        "user_info": user_info,
        "cert_code": cert_code,
    }
    try:
        db.execute(query, params)
        db.commit()
        return True
    except Exception as e:
        logger.error(f"sql error: {e}")
        return False


# 비밀번호 찾기
async 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'
    """)
    params = {
        "user_id": user_id,
        "user_email": user_email,
    }
    return db.execute(query, params).fetchall()


# 신규 비밀번호 업데이트
async 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'
    """)
    params = {
        "new_pw": new_pw,
        "new_solt": new_solt,
        "user_seq": user_seq,
    }
    try:
        db.execute(query, params)
        db.commit()
        return True
    except Exception as e:
        logger.error(f"sql error: {e}")
        return False
    

# 이메일 중복 확인
async def is_valid_user_email_by_user_email(user_email, db):
    query = text(f"""
        select
        user_seq
        from manage_user
        where user_email = :user_email
    """)
    params = {
        "user_email": user_email,
    }
    return db.execute(query, params).fetchall()


# 회원가입 최종 아이디 중복 재검사
async 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
    """)
    params = {
        "user_id": user_id,
    }
    return db.execute(query, params).fetchall()


# 신규 유저 등록
async 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_id,
            (select encode(digest(:user_pw || :user_pw_solt, 'sha256'), 'hex')),
            :user_pw_solt,
            :nickname,
            '/user/temp_dir/profile_img/profile.png',
            :user_email,
            :department,
            :introduce_myself,
            now(),
            now(),
            now(),
            :mandatory_terms_yn,
            'N'
        )
    """)
    params = {
        "user_id": user_info['user_id'],
        "user_pw": user_info['user_pw'],
        "user_pw_solt": user_info['user_pw_solt'],
        "nickname": user_info['nickname'],
        "user_email": user_info['user_email'],
        "department": user_info.get('department', ''),
        "introduce_myself": user_info.get('introduce_myself', ''),
        "mandatory_terms_yn": user_info.get('mandatory_terms_yn', 'N'),
    }
    try:
        db.execute(query, params)
        db.commit()
        return True
    except Exception as e:
        logger.error(f"sql error: {e}")
        return False


# 프로필 이미지 업데이트
async 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'
    """)
    params = {
        "profile_img": profile_img,
        "user_seq": user_seq,
    }
    try:
        db.execute(query, params)
        db.commit()
        return True
    except Exception as e:
        logger.error(f"sql error: {e}")
        return False


# user_seq로 내정보 가져오기
async 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
    """)
    params = {
        "user_seq": user_seq,
    }
    return db.execute(query, params).fetchall()


# 현재 비밀번호 일치 확인
async 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'
            ))
    """)
    params = {
        "user_seq": user_seq,
        "user_pw": user_pw,
    }
    return db.execute(query, params).fetchall()


# 신규 비밀번호 업데이트
async 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(:new_user_pw || :user_pw_solt, 'sha256'), 'hex'),
            user_pw_solt = :user_pw_solt,
            nickname = :nickname,
            user_email = :user_email,
            department = :department,
            profile_img = :profile_img,
            introduce_myself = :introduce_myself
            where 
            user_seq = :user_seq
            and withdraw_yn = 'N'
        """)
    else:
        query = text(f"""
            update manage_user 
            set
            nickname = :nickname,
            user_email = :user_email,
            department = :department,
            profile_img = :profile_img,
            introduce_myself = :introduce_myself
            where 
            user_seq = :user_seq
            and withdraw_yn = 'N'
        """)
    params = {
        "user_seq": user_info['user_seq'],
        "new_user_pw": user_info['new_user_pw'],
        "user_pw_solt": user_info['user_pw_solt'],
        "nickname": user_info['nickname'],
        "user_email": user_info['user_email'],
        "department": user_info.get('department', ''),
        "profile_img": user_info['profile_img'],
        "introduce_myself": user_info.get('introduce_myself', ''),
    }
    try:
        db.execute(query, params)
        db.commit()
        return True
    except Exception as e:
        logger.error(f"sql error: {e}")
        return False


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