backend/fastapi/app/router/test_service.py

from fastapi import APIRouter, Depends, HTTPException, Header, Body, status, Request
from sqlalchemy.orm import Session
import json
import datetime
import requests
import jwt
import time
from process.logger import logger

from db.base import get_db
import pandas as pd

from common.config import SECRET_KEY

KST = datetime.timezone(datetime.timedelta(hours=9))

router = APIRouter(
    prefix="/test",
    tags=["test"],
    responses={404: {"description": "Not found"}},
)

# -----------------------------------------------------------------------------------------------

async def get_body(request: Request):
    return await request.body()




# user_seq로 토큰 생성기
@router.post("/make/token/{user_seq}")
async def make_new_token(user_seq: str, body: bytes = Depends(get_body), db: Session = Depends(get_db)):
    try:
        user_token = create_jwt(user_seq=int(user_seq), period=3600)
        return {
            "result": "OK",
            "data": {
                "token": user_token
            }
        }

    except Exception as e:
        return {
            "result": "FAIL",
            "msg": e
        }


# user_seq로 만료된 토큰 생성기
@router.post("/make/token/expired/{user_seq}")
async def make_new_token_expired(user_seq: str, body: bytes = Depends(get_body), db: Session = Depends(get_db)):
    try:
        user_token = create_jwt(user_seq=int(user_seq), period=0)
        return {
            "result": "OK",
            "data": {
                "token": user_token
            }
        }

    except Exception as e:
        return {
            "result": "FAIL",
            "msg": e
        }


# 토큰 정보 확인
@router.post("/check/token")
async def check_token_data(body: bytes = Depends(get_body), db: Session = Depends(get_db)):
    try:
        # body에서 ID 추출
        try:
            body = json.loads(body)
        except json.JSONDecodeError as e:
            return logger.error(f"json.loads error: Invalid JSON format in request body\nerror message: {e}")
        
        token = body['token']
        token_decode = decode_jwt(token=token)

        return {
            "result": "OK",
            "data": {
                "user_seq": token_decode['user_seq'],
                "exp": change_time(token_decode['exp']),
                "iat": change_time(token_decode['iat']),
            }
        }

    except Exception as e:
        return {
            "result": "FAIL",
            "msg": e
        }
    

    

def create_jwt(user_seq: int, period: int):
    # 현재 시간
    now = int(time.time())

    # JWT의 페이로드 (클레임)
    payload = {
        "user_seq": user_seq,
        "exp": now + 60*60*24*30,    # 만료 시간
        "iat": now,  # 토큰 발행 시간
    }

    # JWT 생성 (HS256 알고리즘 사용)
    token = jwt.encode(payload, SECRET_KEY, algorithm="HS256")

    return token


def decode_jwt(token: str):
    try:
        # 토큰 디코드
        decoded_payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return decoded_payload
    except jwt.ExpiredSignatureError:
        return {"error": "Token has expired"}
    except jwt.InvalidTokenError:
        return {"error": "Invalid token"}
    

# 현재 시간 YYYY-MM-DD hh:mm:ss로 변환 함수
def change_time(time_data):
    dt_object = datetime.datetime.fromtimestamp(time_data)
    formatted_time = dt_object.strftime('%Y-%m-%d %H:%M:%S')
    return formatted_time