kangda-robot-backend/ruoyi-fastapi-backend/config/env.py

368 lines
11 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import argparse
import os
import sys
from dotenv import load_dotenv
from functools import lru_cache
from pydantic import computed_field
from pydantic_settings import BaseSettings
from typing import Literal
class AppSettings(BaseSettings):
"""
应用配置
"""
app_env: str = 'dev'
app_name: str = 'RuoYi-FasAPI'
app_root_path: str = '/dev-api'
app_host: str = '0.0.0.0'
app_port: int = 9099
app_version: str = '1.0.0'
app_reload: bool = False
app_ip_location_query: bool = True
app_same_time_login: bool = True
app_docs_url: str = None
app_redoc_url: str = None
app_openapi_url: str = None
class JwtSettings(BaseSettings):
"""
Jwt配置
"""
jwt_secret_key: str = os.getenv('JWT_SECRET_KEY', '')
jwt_algorithm: str = os.getenv('JWT_ALGORITHM', 'HS256')
jwt_expire_minutes: int = int(os.getenv('JWT_EXPIRE_MINUTES', '1440'))
jwt_redis_expire_minutes: int = int(os.getenv('JWT_REDIS_EXPIRE_MINUTES', '30'))
class DataBaseSettings(BaseSettings):
"""
数据库配置
"""
db_type: Literal['mysql', 'postgresql'] = os.getenv('DB_TYPE', 'mysql')
db_host: str = os.getenv('DB_HOST', '127.0.0.1')
db_port: int = int(os.getenv('DB_PORT', '3306'))
db_username: str = os.getenv('DB_USERNAME', '')
db_password: str = os.getenv('DB_PASSWORD', '')
db_database: str = os.getenv('DB_DATABASE', '')
db_echo: bool = os.getenv('DB_ECHO', 'True').lower() == 'true'
db_max_overflow: int = int(os.getenv('DB_MAX_OVERFLOW', '10'))
db_pool_size: int = int(os.getenv('DB_POOL_SIZE', '50'))
db_pool_recycle: int = int(os.getenv('DB_POOL_RECYCLE', '3600'))
db_pool_timeout: int = int(os.getenv('DB_POOL_TIMEOUT', '30'))
@computed_field
@property
def sqlglot_parse_dialect(self) -> str:
if self.db_type == 'postgresql':
return 'postgres'
return self.db_type
class RedisSettings(BaseSettings):
"""
Redis配置
"""
redis_host: str = os.getenv('REDIS_HOST', '127.0.0.1')
redis_port: int = int(os.getenv('REDIS_PORT', '6379'))
redis_username: str = os.getenv('REDIS_USERNAME', '')
redis_password: str = os.getenv('REDIS_PASSWORD', '')
redis_database: int = int(os.getenv('REDIS_DATABASE', '2'))
class GenSettings:
"""
代码生成配置
"""
author = 'insistence'
package_name = 'module_admin.system'
auto_remove_pre = False
table_prefix = 'sys_'
allow_overwrite = False
GEN_PATH = 'vf_admin/gen_path'
def __init__(self):
if not os.path.exists(self.GEN_PATH):
os.makedirs(self.GEN_PATH)
class UploadSettings:
"""
上传配置
"""
UPLOAD_PREFIX = '/profile'
UPLOAD_PATH = 'vf_admin/upload_path'
UPLOAD_MACHINE = 'A'
DEFAULT_ALLOWED_EXTENSION = [
# 图片
'bmp',
'gif',
'jpg',
'jpeg',
'png',
# word excel powerpoint
'doc',
'docx',
'xls',
'xlsx',
'ppt',
'pptx',
'html',
'htm',
'txt',
# 压缩文件
'rar',
'zip',
'gz',
'bz2',
# 视频格式
'mp4',
'avi',
'rmvb',
# pdf
'pdf',
]
DOWNLOAD_PATH = 'vf_admin/download_path'
def __init__(self):
if not os.path.exists(self.UPLOAD_PATH):
os.makedirs(self.UPLOAD_PATH)
if not os.path.exists(self.DOWNLOAD_PATH):
os.makedirs(self.DOWNLOAD_PATH)
class CachePathConfig:
"""
缓存目录配置
"""
PATH = os.path.join(os.path.abspath(os.getcwd()), 'caches')
PATHSTR = 'caches'
class ComprefaceSettings:
"""
Compreface配置
"""
COMPREFACE_BASE_URL = os.getenv("COMPREFACE_BASE_URL", "http://10.0.0.202")
COMPERFACE_BASE_PORT = os.getenv("COMPERFACE_BASE_PORT", "8000")
COMPREFACE_API_KEY = os.getenv("COMPREFACE_API_KEY", "")
COMPREFACE_API_KEY_DETECTION = os.getenv("COMPREFACE_API_KEY_DETECTION", "")
COMPREFACE_FACE_FILE = os.getenv("COMPREFACE_FACE_FILE", "./image_face")
COMPREFACE_SIMILARITY_THRESHOLD = float(os.getenv("COMPREFACE_SIMILARITY_THRESHOLD", "0.995"))
COMPREFACE_POSE_MAX_ANGLE = float(os.getenv("COMPREFACE_POSE_MAX_ANGLE", "10"))
COMPREFACE_MIN_FACE_WIDTH = int(os.getenv("COMPREFACE_MIN_FACE_WIDTH", "200"))
COMPREFACE_MIN_FACE_HEIGHT = int(os.getenv("COMPREFACE_MIN_FACE_HEIGHT", "200"))
COMPREFACE_MIN_DETECTION_PROBABILITY = float(os.getenv("COMPREFACE_MIN_DETECTION_PROBABILITY", "0.9"))
class HaiKangSettings:
"""
海康平台配置
"""
HAIKANG_URL = os.getenv('HAIKANG_URL', 'https://127.0.0.1')
HAIKANG_PORT = int(os.getenv('HAIKANG_PORT', '443'))
HAIKANG_AK = os.getenv('HAIKANG_AK', '')
HAIKANG_SK = os.getenv('HAIKANG_SK', '')
HAIKANG_ACCESS_TOKEN_URL = os.getenv('HAIKANG_ACCESS_TOKEN_URL', '/api/v1/oauth/token')
# HAIKANG_DOOR_STATES_URL = '/api/v1/door/states'
HAIKANG_DOOR_STATES_URL = os.getenv('HAIKANG_DOOR_STATES_URL', '/api/acs/v1/door/states')
HAIKANG_DOOR_DOCONTROL_URL = os.getenv('HAIKANG_DOOR_DOCONTROL_URL', '/api/acs/v1/door/doControl')
HAIKANG_DOOR_ENVENTS_URL = os.getenv('HAIKANG_DOOR_EVENTS_URL', '/api/acs/v2/door/events')
HAIKANG_DOOR_SEARCH = os.getenv('HAIKANG_DOOR_SEARCH', '/api/resource/v2/door/search')
HAIKANG_DOOR_ONLINE_STATUS = os.getenv('HAIKANG_DOOR_ONLINE_STATUS', '/api/nms/v1/online/acs_device/get')
HAIKANG_APPLICATION_ONETOMANY_URL = os.getenv('HAIKANG_APPLICATION_ONETOMANY_URL', '/api/frs/v1/application/oneToMany')
HAIKANG_PICTURE_CHECK_URL = os.getenv('HAIKANG_PICTURE_CHECK_URL', '/api/frs/v1/face/picture/check')
HAIKANG_FACECAPATURE_SEARCH = os.getenv('HAIKANG_FACE_CAPTURE_SEARCH', '/api/frs/v1/event/face_capture/search')
HAIKANG_FACE_GROUP_URL = os.getenv('HAIKANG_FACE_GROUP_URL', '/api/frs/v1/face/group')
HAIKANG_VISITOR_RECORD_SEARCH = os.getenv('HAIKANG_VISITOR_RECORD_SEARCH', '/api/visitor/v2/appointment/records')
HAIKANG_VISITOR_RECORD_PICTURES = os.getenv('HAIKANG_VISITOR_RECORD_PICTURES', '/api/visitor/v1/record/pictures')
HAIKANG_PERSON_LIST = os.getenv('HAIKANG_PERSON_LIST', '/api/resource/v2/person/personList')
HAIKANG_GET_PERSON_PICTURE = os.getenv('HAIKANG_GET_PERSON_PICTURE', '/api/resource/v1/person/picture')
HAIKANG_FACE_GROUP_ADDITION = os.getenv('HAIKANG_FACE_GROUP_ADDITION', '/api/frs/v1/face/group/single/addition')
HAIKANG_FACE_SIGLE_ADDITION = os.getenv('HAIKANG_FACE_SINGLE_ADDITION', '/api/frs/v1/face/single/addition')
HAIKANG_FACE_DELETE = os.getenv('HAIKANG_FACE_DELETE', '/api/frs/v1/face/deletion')
HAIKANG_FACE_SAVE_PATH = os.getenv('HAIKANG_FACE_SAVE_PATH', "./face_images")
HAIKANG_VISITOR_PICTURES_SAVE_PATH = os.getenv('HAIKANG_VISITOR_PICTURES_SAVE_PATH', "./visitor/face_images/")
HAIKANG_VISITOR_RECORD_TXT = os.getenv('HAIKANG_VISITOR_RECORD_TXT', "./visitor/visitorIds.txt")
class RAGFlowSettings:
"""
RAGFlowSettings
"""
RAGFLOW_BASE_URL = os.getenv("RAGFLOW_BASE_URL", "http://10.0.0.202:82")
RAGFLOW_API_KEY = os.getenv("RAGFLOW_API_KEY", "")
class SearchSettings:
"""搜索服务配置"""
SEARCH_API_BASE = os.getenv("SEARCH_API_BASE", "https://serpapi.com")
SEARCH_API_KEY = os.getenv("SEARCH_API_KEY", "")
SEARCH_ENGINE = os.getenv("SEARCH_ENGINE", "google")
SEARCH_LANG = os.getenv("SEARCH_LANG", "zh-cn")
SEARCH_COUNTRY = os.getenv("SEARCH_COUNTRY", "cn")
SEARCH_NUM_RESULTS = int(os.getenv("SEARCH_NUM_RESULTS", "5"))
SEARCH_CACHE_TTL = int(os.getenv("SEARCH_CACHE_TTL", "1800"))
class DeepSeekSettings:
"""DeepSeek大语言模型配置"""
DEEPSEEK_API_BASE = os.getenv("DEEPSEEK_API_BASE", "https://api.deepseek.com")
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY", "sk-56b608b26a6949e4b09b5bf5f11c8f5b")
DEEPSEEK_MODEL = os.getenv("DEEPSEEK_MODEL", "deepseek-chat")
class GetConfig:
"""
获取配置
"""
def __init__(self):
self.parse_cli_args()
@lru_cache()
def get_compreface_config(self):
"""
获取Compreface配置
"""
# 获取Compreface配置
return ComprefaceSettings()
@lru_cache()
def get_app_config(self):
"""
获取应用配置
"""
# 实例化应用配置模型
return AppSettings()
@lru_cache()
def get_jwt_config(self):
"""
获取Jwt配置
"""
# 实例化Jwt配置模型
return JwtSettings()
@lru_cache()
def get_database_config(self):
"""
获取数据库配置
"""
# 实例化数据库配置模型
return DataBaseSettings()
@lru_cache()
def get_redis_config(self):
"""
获取Redis配置
"""
# 实例化Redis配置模型
return RedisSettings()
@lru_cache()
def get_gen_config(self):
"""
获取代码生成配置
"""
# 实例化代码生成配置
return GenSettings()
@lru_cache()
def get_upload_config(self):
"""
获取数据库配置
"""
# 实例上传配置
return UploadSettings()
@lru_cache()
def get_haikang_config(self):
"""
获取海康平台配置
"""
return HaiKangSettings()
@lru_cache()
def get_ragflow_config(self):
"""
获取RAGFlow配置
"""
return RAGFlowSettings()
@lru_cache()
def get_search_config(self):
"""获取搜索配置"""
return SearchSettings()
@lru_cache()
def get_deepseek_config(self):
"""获取DeepSeek配置"""
return DeepSeekSettings()
@staticmethod
def parse_cli_args():
"""
解析命令行参数
"""
if 'uvicorn' in sys.argv[0]:
# 使用uvicorn启动时命令行参数需要按照uvicorn的文档进行配置无法自定义参数
pass
else:
# 使用argparse定义命令行参数
parser = argparse.ArgumentParser(description='命令行参数')
parser.add_argument('--env', type=str, default='dev', help='运行环境')
# 解析命令行参数
args = parser.parse_args()
# 设置环境变量如果未设置命令行参数默认APP_ENV为dev
os.environ['APP_ENV'] = args.env if args.env else 'dev'
# 读取运行环境
run_env = os.environ.get('APP_ENV', '')
# 运行环境未指定时默认加载.env.dev
env_file = '.env.dev'
# 运行环境不为空时按命令行参数加载对应.env文件
if run_env != '':
env_file = f'.env.{run_env}'
# 加载配置
load_dotenv(env_file)
# 实例化获取配置类
get_config = GetConfig()
# 应用配置
AppConfig = get_config.get_app_config()
# Jwt配置
JwtConfig = get_config.get_jwt_config()
# 数据库配置
DataBaseConfig = get_config.get_database_config()
# Redis配置
RedisConfig = get_config.get_redis_config()
# 代码生成配置
GenConfig = get_config.get_gen_config()
# 上传配置
UploadConfig = get_config.get_upload_config()
# 海康平台配置
HaiKangConfig = get_config.get_haikang_config()
# RAGFlow配置
RAGFlowConfig = get_config.get_ragflow_config()
# compreface配置
ComprefaceConfig = get_config.get_compreface_config()
# 搜索配置
SearchConfig = get_config.get_search_config()
# DeepSeek配置
DeepSeekConfig = get_config.get_deepseek_config()