from flask_sqlalchemy import SQLAlchemy
from flask_mail import Mail
from flask_caching import Cache
from flask_wtf import CSRFProtect
from flask_avatars import Avatars
from flask_jwt_extended import JWTManager
from flask_cors import CORSdb = SQLAlchemy( )
mail = Mail( )
cache = Cache( )
csrf = CSRFProtect( )
avatars = Avatars( )
jwt = JWTManager( )
cors = CORS( ) from gevent import pywsgifrom geventwebsocket. handler import WebSocketHandlerserver = pywsgi. WSGIServer( ( '0.0.0.0' , 8099 ) , app, handler_class= WebSocketHandler) server. serve_forever( )
import os
BASE_DIR = os. path. dirname( os. path. abspath( __file__) )
SECRET_KEY = "dfasdfsdflasdjfl"
from utils. configUtils import get_section_dictcfg_msyql = get_section_dict( file_path= os. path. join( BASE_DIR, "config.ini" ) , section= "mysql"
) DB_URI = 'mysql+pymysql://%s:%s@%s:%s/%s?charset=utf8mb4' % ( cfg_msyql. get( "user" , "root" ) , cfg_msyql. get( "password" , "123456" ) , cfg_msyql. get( "host" , "localhost" ) , cfg_msyql. get( "port" , "3306" ) , cfg_msyql. get( "database" , "employee_face_recognition_and_behavior_detection" ) , ) SQLALCHEMY_DATABASE_URI = DB_URI
import redis
cfg_redis = get_section_dict( file_path= os. path. join( BASE_DIR, "config.ini" ) , section= "redis"
) redis_con = redis. Redis( host= cfg_redis. get( "host" , "localhost" ) , port= cfg_redis. get( "port" , 6379 ) , db= cfg_redis. get( "db" , 1 ) , decode_responses= cfg_redis. get( "decode_responses" , False )
) cfg_flask = get_section_dict( file_path= os. path. join( BASE_DIR, "config.ini" ) , section= "flask"
) from flask import Flask
from flask_migrate import Migrate
import config
import pymysql
from exts import db, login_manager, cors
import redis
def create_app ( config_class= config) : app = Flask( __name__) app. config. from_object( config_class) db. init_app( app) migrate = Migrate( app, db) cors. init_app( app= app, resources= { r"/api/*" : { "origins" : "*" } } , supports_credentials= True ) login_manager. init_app( app) login_manager. login_view = 'api.login' with app. app_context( ) : db. create_all( ) pymysql. install_as_MySQLdb( ) return app
from apps import create_appapp = create_app( ) if __name__ == '__main__' : app. run( debug= True ) [mysql]
# 连接数据库基本参数
host = localhost
port = 3306
user = root
password = 123456
database = databaseName
charset = utf8mb4# 可选：连接池／超时配置
connect_timeout = 10
pool_size = 5[redis]
host = localhost
port = 6379
db = 1
decode_responses = False[flask]
APP_NAME = 'flask_app'
import configparser
from typing import Dict
from typing import Dict, Any
import configparser
import re, random
from app. models. employee import BehaviorTypeEnum, Behavior
from datetime import datetime, timedelta
import pytzdef get_section_dict ( file_path: str , section: str , encoding: str = 'utf-8'
) - > Dict[ str , Any] : """从指定的 ini 文件中读取一个区块，将该区块的 key/value 封装为字典返回，并自动将字符串形式的布尔值、整数和浮点数转换为相应的 Python 类型。:param file_path: ini 文件路径:param section: 要读取的区块名称:param encoding: 文件编码，默认为 'utf-8':return: 区块键值对字典，value 类型可能为 str, int, float, 或 bool:raises KeyError: 如果指定区块在文件中不存在""" config = configparser. ConfigParser( ) config. optionxform = str config. read( file_path, encoding= encoding) if section not in config: raise KeyError( f"Section ' { section} ' not found in ' { file_path} '." ) raw = config[ section] parsed: Dict[ str , Any] = { } for key, value in raw. items( ) : v = value. strip( ) low = v. lower( ) if low in ( 'true' , 'yes' , 'on' ) : parsed_val: Any = True elif low in ( 'false' , 'no' , 'off' ) : parsed_val = False elif re. fullmatch( r'-?\d+' , v) : parsed_val = int ( v) elif re. fullmatch( r'-?\d+\.\d+' , v) : parsed_val = float ( v) else : parsed_val = vparsed[ key] = parsed_valreturn parsed
from flask import jsonifyclass HttpCode ( object ) : ok = 200 unloginerror = 401 permissionerror = 403 paramserror = 400 servererror = 500 def _restful_result ( code, message, data) : return jsonify( { "message" : message or "" , "data" : data or { } , "code" : code} ) def ok ( message= None , data= None ) : return _restful_result( code= HttpCode. ok, message= message, data= data) def unlogin_error ( message= "没有登录！" ) : return _restful_result( code= HttpCode. unloginerror, message= message, data= None ) def permission_error ( message= "没有权限访问！" ) : return _restful_result( code= HttpCode. paramserror, message= message, data= None ) def params_error ( message= "参数错误！" ) : return _restful_result( code= HttpCode. paramserror, message= message, data= None ) def server_error ( message= "服务器开小差啦！" ) : return _restful_result( code= HttpCode. servererror, message= message or '服务器内部错误' , data= None ) 
)
