# 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.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() result = get_user_under_organization(current_user) result.append({'Data' : get_tower_info('dev001')}) return templates.TemplateResponse(name='tower.html', context={"request":request}) @app.get('/tower/org', 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() result = get_user_under_organization(current_user) return json.dumps(result,ensure_ascii=False) @app.get('/tower/', response_class=HTMLResponse) async def tower(request: Request,company:str,factory:str,department:str,towerGroup:str, Authorize: AuthJWT = Depends()): try: Authorize.jwt_required() except Exception as e: print(e) return RedirectResponse('/login') #current_user = Authorize.get_jwt_subject() tower_arr = get_tower(company,factory,department,towerGroup) result = [] for tower in tower_arr: result.append({'tower_name': tower,'tower_data': get_tower_info(tower)}) print(result) return json.dumps(result,ensure_ascii=False) @app.get('/optim', response_class=HTMLResponse) async def optim(request: Request, Authorize: AuthJWT = Depends()): db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/Water_tower?charset=utf8mb4') table=db['record_tower'] temp = table.find_one(key = 'hotTemp') statement2 = 'SELECT value FROM record_tower WHERE record_tower.key = "coldTempData2"' for temp2 in db.query(statement2): print(temp2['value']) statement3 = 'SELECT value FROM record_tower WHERE record_tower.key = "wetTemp"' for temp3 in db.query(statement3): print(temp3['value']) try: Authorize.jwt_required() except Exception as e: print(e) return RedirectResponse('/login') return templates.TemplateResponse(name='optim.html',context={'request': request,"temp":temp,"temp2":temp2,"temp3":temp3}) @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 FROM users' json_dic = [] for row in db.query(statement): #print(row['id'],row['username']) json_dic.append({'username':row['username'],'isAdmin':row['isAdmin'],'roleType':check_role_type(row['username']),'role_name' :get_role_name(check_role_type(row['username']))}) result = json.dumps(json_dic,ensure_ascii=False) return result @app.get('/member/edit/', response_class=HTMLResponse) async def login(request: Request, name:str,isAdmin:int,isEnable:int ,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') current_user = Authorize.get_jwt_subject() current_user_roleType = check_role_type(current_user) del_user_roleType = check_role_type(name) statement = 'SELECT isAdmin FROM users WHERE userName = "'+current_user+'"' for row in db.query(statement): if row['isAdmin']!=1: return json.dumps([{'msg':'你沒有權限'}],ensure_ascii=False) if del_user_roleType == None: return json.dumps([{'msg':'不存在使用者'}],ensure_ascii=False) elif current_user_roleType>del_user_roleType or current_user_roleType==del_user_roleType: return json.dumps([{'msg':'你沒有權限'}],ensure_ascii=False) user_dic = get_user(name) print(user_dic) user_dic.isAdmin = isAdmin user_dic.isEnable = isEnable table = db['users'] table.update(dict(user_dic), ['username']) return json.dumps([{'msg':"成功更改"}],ensure_ascii=False) @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: int,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 templates.TemplateResponse(name='notice.html', context={"request":request,'msg':'no user' }) elif statement == 0: return templates.TemplateResponse(name='notice.html', context={"request":request,'msg':"沒有權限" }) current_user_roleType = check_role_type(current_user) if edit_one == None: return templates.TemplateResponse(name='notice.html', context={"request":request,'msg':'no role' }) elif int(current_user_roleType)>int(edit_one) or int(current_user_roleType)==int(edit_one): return templates.TemplateResponse(name='notice.html', context={"request":request,'msg':"沒有權限" }) result = check_role_acl(edit_one) if result == []: cmd = 'SELECT id FROM module' for row in db.query(cmd): dic_tmp = {'id':0,'isView':0,'isAdd':0 ,'isEdit':0,'isDel':0,'role_id' : 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 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 templates.TemplateResponse(name='notice.html', context={"request":request,'msg':'尚未登入'}) statement = check_isAdmin(current_user) if statement == "no user": return templates.TemplateResponse(name='notice.html', context={"request":request,'msg':statement }) elif statement == 0: return templates.TemplateResponse(name='notice.html', context={"request":request,'msg':'你沒有權限' }) current_user_roleType = check_role_type(current_user) edit_one_roleType = edit_one.role_id if current_user_roleType>edit_one_roleType or current_user_roleType==edit_one_roleType: return templates.TemplateResponse(name='notice.html', context={"request":request,'msg': ' 你沒有權限'}) else : row = ['ai_prediction' ,'channel' ,'device', 'event', 'index' ,'performance', 'record', 'setting_device' ,'setting_system','tower'] if check_role_acl(edit_one.role_id) == []: for module in row : new_dict = edit_one.get_acl_from_module_name(module) new_dict["id"]= None 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']) return templates.TemplateResponse(name='notice.html', context={"request":request,'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)->int: """查看使用者權限""" db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/Water_tower?charset=utf8mb4') cmd = 'SELECT user_role.role_id FROM `users` JOIN `user_role` ON `users`.id = `user_role`.user_id where `users`.username = "'+user_name+'"' role_type = None for row in db.query(cmd) : role_type = row['role_id'] return role_type def check_role_acl(role:int): """查看權限""" db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/Water_tower?charset=utf8mb4') cmd = 'SELECT * FROM role_acl where role_id = '+str(role) 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_role_name(role_id:int): db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/Water_tower?charset=utf8mb4') cmd = 'SELECT * FROM role where id = '+str(role_id) role:str for row in db.query(cmd) : role = row['name'] return role def get_user_under_organization(user_name:str): """查看所屬公司""" db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/Water_tower?charset=utf8mb4') user_role = check_role_type(user_name) #print(user_name,user_role) cmd = 'SELECT * FROM organization' result = [] if int(user_role) == 1 : num=0 for row in db.query(cmd) : company = row['Company'] factory = row['Factory'] department = row['Department'] result.append({}) cmd2 = 'SELECT TowerGroupCode FROM device WHERE CompanyCode = "' + company + '" AND FactoryCode = "' + factory + '" AND DepartmentCode = "' + department + '"' group = [] for row2 in db.query(cmd2): if row2['TowerGroupCode'] not in group : group.append(row2['TowerGroupCode']) result[num] = {'company':company,'factory':factory,'department':department,'group':group,'able':1} num = num +1 elif int(user_role) == 2: cmd2 = 'SELECT company FROM user WHERE user.username = '+user_name company_able:str num = 0 for row in db.query(cmd2) : company_able = row['company'] for row in db.query(cmd) : company = row['Company'] factory = row['Factory'] department = row['Department'] cmd3 = 'SELECT TowerGroupCode FROM device WHERE CompanyCode = "' + company + '" AND FactoryCode = "' + factory + '" AND DepartmentCode = "' + department + '"' group = [] for row2 in db.query(cmd3): group.append(row2['TowerGroupCode']) if company == company_able: result[num] = {'company':company,'factory':factory,'department':department,'group':group,'able':1} else: result[num] = {'company':company,'factory':factory,'department':department,'group':group,'able':0} num = num +1 elif int(user_role) == 3: cmd2 = 'SELECT company,factory FROM users WHERE users.username = '+user_name company_able:str factory_able:str num = 0 for row in db.query(cmd2) : company = row['company'] factory = row['factory'] for row in db.query(cmd) : company = row['Company'] factory = row['Factory'] department = row['Department'] cmd3 = 'SELECT TowerGroupCode FROM device WHERE CompanyCode = "' + company + '" AND FactoryCode = "' + factory + '" AND DepartmentCode = "' + department + '"' group = [] for row2 in db.query(cmd3): group.append(row2['TowerGroupCode']) if company == company_able and factory==factory_able: result[num] = {'company':company,'factory':factory,'department':department,'group':group,'able':1} else: result[num] = {'company':company,'factory':factory,'department':department,'group':group,'able':0} num=num+1 elif int(user_role) == 4: cmd2 = 'SELECT company,factory,department FROM users WHERE username = '+user_name company_able:str factory_able:str department_able:str num = 0 for row in db.query(cmd2) : company_able = row['company'] factory_able = row['factory'] department_able = row['Department'] for row in db.query(cmd) : company = row['Company'] factory = row['Factory'] department = row['Department'] cmd3 = 'SELECT TowerGroupCode FROM device WHERE CompanyCode = "' + company + '" AND FactoryCode = "' + factory + '" AND DepartmentCode = "' + department + '"' group = [] for row2 in db.query(cmd3): group.append(row2['TowerGroupCode']) if company == company_able and factory==factory_able and department==department_able: result[num] = {'company':company,'factory':factory,'department':department,'group':group,'able':1} else: result[num] = {'company':company,'factory':factory,'department':department,'group':group,'able':1} num = num +1 else : result =[ {'msg':"error"}] 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 name""" 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 def get_tower_info(tower_id:str): """獲取水塔資料""" db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/Water_tower?charset=utf8mb4') cmd = 'SELECT * FROM `record_dcs` where device_id = "'+tower_id+'"' result ={'DCS':{},'Fan':{},'Moter':{}} for row in db.query(cmd) : result['DCS'][row['key']]=row['value'] cmd = 'SELECT * FROM `record_tower` where device_id = "'+tower_id+'"' for row in db.query(cmd) : result['Fan'][row['key']]=row['value'] result['Moter'] = [] cmd = 'SELECT * FROM `vibration` where device_id = "'+tower_id+'"' for row in db.query(cmd) : result['Moter'].append({'channel_name':row['channelName'],'Vrms':row['DataValue'],'CV' :row['CVIndex'],'threshold':row['threshold']}) return result def get_tower(company:str,factory:str,department:str,towerGroup:str): towergroup_arr =[] db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/Water_tower?charset=utf8mb4') cmd = 'SELECT id FROM `device` where CompanyCode = "'+company+'" AND FactoryCode = "' +factory+'" AND DepartmentCode = "'+department+'" AND TowerGroupCode = "' + towerGroup + '"' for row in db.query(cmd) : towergroup_arr.append(row['id']) return towergroup_arr