mechanical/main.py

# fastapi
from fastapi import FastAPI, Request, Response, HTTPException, status, Depends , Form
from fastapi import templating
from fastapi.templating import Jinja2Templates
from fastapi.responses import HTMLResponse, RedirectResponse, JSONResponse
from fastapi.middleware.cors import CORSMiddleware

from fastapi.staticfiles import StaticFiles

# fastapi view function parameters
from typing import List, Optional
import json
# path
import sys

from sqlalchemy.sql.elements import False_  

# time
# import datetime
from datetime import timedelta, datetime
# db
import dataset
from passlib import context
import models
from random import randint
# authorize
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
import jwt
from fastapi_jwt_auth import AuthJWT
from fastapi_jwt_auth.exceptions import AuthJWTException
from fastapi.security import OAuth2AuthorizationCodeBearer, OAuth2PasswordRequestForm
import numpy as np
import pymysql
pymysql.install_as_MySQLdb()
db_settings = {
    "host": "db.ptt.cx",
    "port": 3306,
    "user": "choozmo",
    "password": "pAssw0rd",
    "db": "Water_tower",
    "charset": "utf8mb4"
}

# app
app = FastAPI()
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)



SECRET_KEY = "df2f77bd544240801a048bd4293afd8eeb7fff3cb7050e42c791db4b83ebadcd"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 3000
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# 
app.mount(path='/templates', app=StaticFiles(directory='templates'), name='templates')
app.mount(path='/static', app=StaticFiles(directory='static'), name='static ')



# 
templates = Jinja2Templates(directory='templates')


@AuthJWT.load_config
def get_config():
    return models.Settings()


# view
@app.get('/', response_class=HTMLResponse)
async def index(request: Request):
    print(request)
    return templates.TemplateResponse(name='index.html', context={'request': request})
     

@app.get('/login', response_class=HTMLResponse)
async def login(request: Request):
    return templates.TemplateResponse(name='login_test.html', context={'request': request})


@app.post("/login")
async def login_for_access_token(request: Request, form_data: OAuth2PasswordRequestForm = Depends(), Authorize: AuthJWT = Depends()):
    db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/Water_tower?charset=utf8mb4')
    
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    table = db['users']
    user.token = access_token
    print(user)
    table.update(dict(user), ['username'],['password'])
    access_token = Authorize.create_access_token(subject=user.username)
    refresh_token = Authorize.create_refresh_token(subject=user.username)
    Authorize.set_access_cookies(access_token)
    Authorize.set_refresh_cookies(refresh_token)
    #return templates.TemplateResponse("home.html", {"request": request, "msg": 'Login'})
    return {"access_token": access_token, "token_type": "bearer"}       # 回傳token給前端


@app.get('/register', response_class=HTMLResponse)
async def login(request: Request):
    return templates.TemplateResponse(name='rigister_test.html', context={'request': request})


@app.post('/register')
async def register(request: Request, form_data: OAuth2PasswordRequestForm = Depends()):
    user = models.User(**await request.form())
    print(form_data.username, form_data.password, user)
    user.id = randint(1000, 9999)
    user.isAdmin = 0 #預設為非管理者
    user.roleType = 0 #預設為employee
    # 密碼加密
    #user.password = get_password_hash(user.password)
    
    # 存入DB
    db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/Water_tower?charset=utf8mb4')
    user_table = db['users']
    user_table.insert(dict(user))
    
    # 跳轉頁面至登入
    return templates.TemplateResponse(name='login.html', context={'request': request})


@app.get('/home', response_class=HTMLResponse)
async def home(request: Request):
    return templates.TemplateResponse(name='home.html', context={'request': request})


@app.get('/tower', response_class=HTMLResponse)
async def tower(request: Request, Authorize: AuthJWT = Depends()):
    try:
        Authorize.jwt_required()
    except Exception as e:
        print(e)
        return RedirectResponse('/login')
    # current_user = Authorize.get_jwt_subject()
    return templates.TemplateResponse(name='tower.html', context={'request': request})
    

@app.get('/optim', response_class=HTMLResponse)
async def optim(request: Request, Authorize: AuthJWT = Depends()):
    try:
        Authorize.jwt_required()
    except Exception as e:
        print(e)
        return RedirectResponse('/login')
    # current_user = Authorize.get_jwt_subject()
    return templates.TemplateResponse(name='optim.html', context={'request': request})


@app.get('/vibration', response_class=HTMLResponse)
async def vibration(request: Request, Authorize: AuthJWT = Depends()):
    try:
        Authorize.jwt_required()
    except Exception as e:
        print(e)
        return RedirectResponse('/login')
    # current_user = Authorize.get_jwt_subject()
    return templates.TemplateResponse(name='vibration.html', context={'request': request})


@app.get('/history', response_class=HTMLResponse)
async def history(request: Request, Authorize: AuthJWT = Depends()):
    try:
        Authorize.jwt_required()
    except Exception as e:
        print(e)
        return RedirectResponse('/login')
    # current_user = Authorize.get_jwt_subject()
    return templates.TemplateResponse(name='history.html', context={'request': request})


@app.get('/device', response_class=HTMLResponse)
async def device(request: Request, Authorize: AuthJWT = Depends()):
    try:
        Authorize.jwt_required()
    except Exception as e:
        print(e)
        return RedirectResponse('/login')
    # current_user = Authorize.get_jwt_subject()
    return templates.TemplateResponse(name='device.html', context={'request': request})


@app.get('/system', response_class=HTMLResponse)
async def system(request: Request, Authorize: AuthJWT = Depends()):
    try:
        Authorize.jwt_required()
    except Exception as e:
        print(e)
        return RedirectResponse('/login')
    # current_user = Authorize.get_jwt_subject()
    return templates.TemplateResponse(name='system.html', context={'request': request})

@app.get('/member', response_class=HTMLResponse)
async def get_member(request: Request, Authorize: AuthJWT = Depends()):
    """獲取所有帳號資訊"""
    try:
        Authorize.jwt_required()
    except Exception as e:
        print(e)
        return RedirectResponse('/login')

    db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/Water_tower?charset=utf8mb4')
    statement = 'SELECT  id,username,isAdmin,roleType FROM users'
    json_dic = {}
    for row in db.query(statement):
        #print(row['id'],row['username'])
        json_dic[row['username']] = {'isAdmin':row['isAdmin'],'roleType':row['roleType']}
    result  = json.dumps(json_dic,ensure_ascii=False)
    current_user = Authorize.get_jwt_subject()
    print(current_user)
    return result

@app.get('/member/edit', response_class=HTMLResponse)
async def login(request: Request, Authorize: AuthJWT = Depends()):
    try:
        Authorize.jwt_required()
    except Exception as e:
        print(e)
        return RedirectResponse('/login')
    return templates.TemplateResponse(name='member_edit_test.html', context={'request': request})

@app.get('/member_delete', response_class=HTMLResponse)
async def login(request: Request, Authorize: AuthJWT = Depends()):
    try:
        Authorize.jwt_required()
    except Exception as e:
        print(e)
        return RedirectResponse('/login')
    return templates.TemplateResponse(name='delete_member_test2.html', context={'request': request})

@app.post('/member_delete')
async def delete_member(request: Request):
    """刪除成員"""
    db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/Water_tower?charset=utf8mb4')
    del_user = models.del_user(**await request.form())
    delete_name = del_user.del_name
    statement = 'SELECT * FROM users'
    current_user = ''
    for row in db.query(statement):
        if row['token'] != None :
            if compare_jwt_token(row['token'],del_user.access_token):
                current_user = row['username']
    if current_user == '':
        return {'msg':'尚未登入'}

    statement = 'SELECT isAdmin FROM users WHERE userName = "'+current_user+'"'
    for row in db.query(statement):
        if row['isAdmin']!=1:
            return {'msg': ' 你沒有權限'}
    
    current_user_roleType = check_role_type(current_user)
    del_user_roleType = check_role_type(delete_name)
    
    if del_user_roleType == None:
        return {'msg':'不存在使用者'}
    elif current_user_roleType>del_user_roleType or current_user_roleType==del_user_roleType:
        return {'msg': ' 你沒有權限'}
    else :
        table = db['users']
        table.delete(username=delete_name)

    return {'msg': ' 成功刪除'}

@app.get('/member_authority/{edit_one}', response_class=HTMLResponse)
async def member_authority(request:Request,edit_one: str,Authorize: AuthJWT = Depends()):
    """設定成員權限"""
    
    try:
        Authorize.jwt_required()
    except Exception as e:
        print(e)
        return RedirectResponse('/login')
    context  = {'request': request}
    current_user = Authorize.get_jwt_subject()
    db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/Water_tower?charset=utf8mb4')
    statement = check_isAdmin(current_user)
    if statement == "no user":
        return {'msg':statement }
    elif statement == 0:
        return  {'msg':'你沒有權限' }
    current_user_roleType = check_role_type(current_user)
    
    edit_one_roleType = check_role_type(edit_one)  
    if edit_one_roleType == None:
        return {'msg':'不存在使用者'}
    elif current_user_roleType>edit_one_roleType or current_user_roleType==edit_one_roleType:
        return {'msg': ' 你沒有權限'}

    result = check_role_acl(edit_one)
    
    if result == []:
        cmd = 'SELECT id FROM module'
        for row in db.query(cmd):
            dic_tmp = {'id':get_user_id(edit_one),'isView':0,'isAdd':0 ,'isEdit':0,'isDel':0,'role_id' : check_role_type(edit_one)}
            context[get_modul_name(row['id']) ] = dic_tmp
    else:
        for dic in result:
            modul_name = get_modul_name(dic['module_id'])
            del dic['module_id']
            context[modul_name ] = dic
    print(context)
    return templates.TemplateResponse(name='member_authority_test.html', context=context)
    
@app.post('/member_authority')
async def member_authority(request: Request):
    db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/Water_tower?charset=utf8mb4')
    edit_one = models.user_authority(**await request.form())
    statement = 'SELECT * FROM users'
    current_user = ''
    for row in db.query(statement):
        if row['token'] != None :
            if compare_jwt_token(row['token'],edit_one.access_token):
                current_user = row['username']
    if current_user == '':
        return {'msg':'尚未登入'}

    statement = check_isAdmin(current_user)
    if statement == "no user":
        return {'msg':statement }
    elif statement == 0:
        return  {'msg':'你沒有權限' }
    
    current_user_roleType = check_role_type(current_user)
    edit_one_roleType = edit_one.role_id
    
    if edit_one.id == None:
        return {'msg':'不存在使用者'}
    elif current_user_roleType>edit_one_roleType or current_user_roleType==edit_one_roleType:
        return {'msg': ' 你沒有權限'}
    else :
        row = ['ai_prediction' ,'channel' ,'device', 'event', 'index' ,'performance', 'record', 'setting_device' ,'setting_system','tower']
        if check_role_acl(get_user_name(edit_one.id)) == []:
            for module in row :
                new_dict = edit_one.get_acl_from_module_name(module)
                table = db['role_acl']
                table.insert(new_dict)
        else:
            for module in row :
                new_dict = edit_one.get_acl_from_module_name(module)
                table = db['role_acl']
                table.update(new_dict, ['id'],['module_id'])

    return {'msg': ' 成功更改'}
    
    


# 溫度API
@app.get('/temperature')
async def get_temperatures():
    """ 撈DB溫度 """
    return {'hot_water': 30.48, 'cold_water': 28.10, 'wet_ball': 25.14}

@app.post("/example")
async def example(request: Request,Authorize: AuthJWT = Depends()):
    try:
        Authorize.jwt_required()
    except Exception as e:
        print(e)
    current_user = Authorize.get_jwt_subject()
    #form_data = await request.form()
    
    print( current_user)
    return current_user

@app.post('/user')
def user(Authorize: AuthJWT = Depends()):
    Authorize.jwt_required()

    current_user = Authorize.get_jwt_subject()
    return {"user": current_user}

@app.get("/example", response_class=HTMLResponse)
async def example(request: Request,Authorize: AuthJWT = Depends()):
    try:
        Authorize.jwt_required()
    except Exception as e:
        print(e)
        return RedirectResponse('/login')
    current_user = Authorize.get_jwt_subject()
    print( current_user)
    return  current_user


@app.get('/health')
async def get_health(date: str):
    """ 撈健康指標、預設健康指標 """
    date = str(datetime.strptime(date, "%Y-%m-%d"))[:10]
    print(date)
    print(str(datetime.today()))
    print(str(datetime.today()-timedelta(days=1)))


    fake_data = {
        str(datetime.today())[:10]: {'curr_health': 0.7, 'pred_health': 0.8},
        str(datetime.today()-timedelta(days=1))[:10]: {'curr_health': 0.6, 'pred_health': 0.7},
    }
    
    return fake_data[date]


@app.get('/history_data')
async def get_history(time_end: str):
    """ 透過終點時間，抓取歷史資料。 """
    date = str(datetime.strptime(time_end, "%Y-%m-%d"))[:10]
    print(date)
    print(str(datetime.today()))
    print(str(datetime.today()-timedelta(days=1)))


    fake_data = {
        str(datetime.today())[:10]: {
            'curr_history': {
                'RPM_1X': list(np.random.rand(13)),
                'RPM_2X': list(np.random.rand(13)),
                'RPM_3X': list(np.random.rand(13)),
                'RPM_4X': list(np.random.rand(13)),
                'RPM_5X': list(np.random.rand(13)),
                'RPM_6X': list(np.random.rand(13)),
                'RPM_7X': list(np.random.rand(13)),
                'RPM_8X': list(np.random.rand(13)),
                'Gear_1X': list(np.random.rand(13)),
                'Gear_2X': list(np.random.rand(13)),
                'Gear_3X': list(np.random.rand(13)),
                'Gear_4X': list(np.random.rand(13)),
            },
            'past_history': {
                'RPM_1X': list(np.random.rand(13)),
                'RPM_2X': list(np.random.rand(13)),
                'RPM_3X': list(np.random.rand(13)),
                'RPM_4X': list(np.random.rand(13)),
                'RPM_5X': list(np.random.rand(13)),
                'RPM_6X': list(np.random.rand(13)),
                'RPM_7X': list(np.random.rand(13)),
                'RPM_8X': list(np.random.rand(13)),
                'Gear_1X': list(np.random.rand(13)),
                'Gear_2X': list(np.random.rand(13)),
                'Gear_3X': list(np.random.rand(13)),
                'Gear_4X': list(np.random.rand(13)),
            }
        },
        str(datetime.today()-timedelta(days=1))[:10]: {
            'curr_history': {
                'RPM_1X': list(np.random.rand(13)),
                'RPM_2X': list(np.random.rand(13)),
                'RPM_3X': list(np.random.rand(13)),
                'RPM_4X': list(np.random.rand(13)),
                'RPM_5X': list(np.random.rand(13)),
                'RPM_6X': list(np.random.rand(13)),
                'RPM_7X': list(np.random.rand(13)),
                'RPM_8X': list(np.random.rand(13)),
                'Gear_1X': list(np.random.rand(13)),
                'Gear_2X': list(np.random.rand(13)),
                'Gear_3X': list(np.random.rand(13)),
                'Gear_4X': list(np.random.rand(13)),
            },
            'past_history': {
                'RPM_1X': list(np.random.rand(13)),
                'RPM_2X': list(np.random.rand(13)),
                'RPM_3X': list(np.random.rand(13)),
                'RPM_4X': list(np.random.rand(13)),
                'RPM_5X': list(np.random.rand(13)),
                'RPM_6X': list(np.random.rand(13)),
                'RPM_7X': list(np.random.rand(13)),
                'RPM_8X': list(np.random.rand(13)),
                'Gear_1X': list(np.random.rand(13)),
                'Gear_2X': list(np.random.rand(13)),
                'Gear_3X': list(np.random.rand(13)),
                'Gear_4X': list(np.random.rand(13)),
            }
        },
    }
    return fake_data[date]


# Login funtion part
def check_user_exists(username):
    db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/Water_tower?charset=utf8mb4')
    if int(next(iter(db.query('SELECT COUNT(*) FROM Water_tower.users WHERE userName = "'+username+'"')))['COUNT(*)']) > 0:
        return True
    else:
        return False


def get_user(username: str):
    """ 取得使用者資訊(Model) """
    db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/Water_tower?charset=utf8mb4')
    if not check_user_exists(username):  # if user don't exist
        return False
    user_dict = next(
        iter(db.query('SELECT * FROM Water_tower.users where userName ="'+username+'"')))
    user = models.User(**user_dict)
    return user
    

def user_register(user):
    db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/Water_tower?charset=utf8mb4')
    table = db['users']
    #user.password = get_password_hash(user.password)
    table.insert(dict(user))


def get_password_hash(password):
    """ 加密密碼 """
    return pwd_context.hash(password)


def verify_password(plain_password, hashed_password):
    """ 驗證密碼(hashed) """
    return pwd_context.verify(plain_password, hashed_password)


def authenticate_user(username: str, password: str):
    """ 連線DB，讀取使用者是否存在。 """
    db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/Water_tower?charset=utf8mb4')
    if not check_user_exists(username):  # if user don't exist
        return False
    user_dict = next(iter(db.query('SELECT * FROM Water_tower.users where userName ="'+username+'"')))
    user = models.User(**user_dict)
    #if not verify_password(password, user.password):
        #return False
    return user


def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    """ 創建token，並設定過期時間。 """
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

def compare_jwt_token(access_token: str, token: str):
    """比對jwt token"""
    if  len(access_token) < len(token):
        if access_token in token:
            return True
        else :
            return False
    elif len(access_token) > len(token):
        if token in access_token:
            return True
        else :
            return False
    else :
        if token == access_token:
            return True
        else :
            return False

def check_isAdmin(user_name:str):
    """查看是否為管理員"""
    db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/Water_tower?charset=utf8mb4')
    isAdmin = None
    cmd =  'SELECT isAdmin FROM users WHERE userName = "'+user_name+'"'
    for row in db.query(cmd) :
        isAdmin = row['isAdmin']
    if isAdmin== None:
        return "no user"
    return isAdmin

def check_role_type(user_name:str):
    """查看使用者權限"""
    db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/Water_tower?charset=utf8mb4')
    cmd = 'SELECT role.id FROM `users` JOIN `role` ON `users`.roleType = `role`.name where `users`.username = "'+user_name+'"'
    role_type = None
    for row in db.query(cmd) :
        role_type = row['id']
    return role_type
            

def check_role_acl(user_name:str):
    """查看權限"""
    db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/Water_tower?charset=utf8mb4')
    cmd = 'SELECT role_acl.* FROM `users` JOIN `role_acl` ON `users`.id = `role_acl`.user_id where `users`.username = "'+user_name+'"'
    
    result = []
    for row in db.query(cmd) :
        dic ={}
        for col_name in db['role_acl'].columns:
            dic[col_name] = row[col_name]
        if dic != {}:
            result.append(dic)
    return result

def get_user_id(user_name:str):
    """獲取user id"""
    db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/Water_tower?charset=utf8mb4')
    cmd = 'SELECT id FROM `users`  where username = "'+user_name+'"'
    id = None
    for row in db.query(cmd) :
        id = row['id']
    return id

def get_user_name(user_id:int):
    """獲取user id"""
    db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/Water_tower?charset=utf8mb4')
    cmd = 'SELECT username FROM `users`  where id = "'+user_id+'"'
    id = None
    for row in db.query(cmd) :
        id = row['username']
    return id

def get_modul_name(modul_id:str):
    """獲取modul名稱"""
    db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/Water_tower?charset=utf8mb4')
    cmd = 'SELECT moduleName FROM  `module` where id = "'+modul_id+'"'    
    modul_name = None
    for row in db.query(cmd) :
        modul_name = row['moduleName']
    return modul_name