yaml文件配置

core/config/env.py

@@ -8,11 +8,16 @@ import redis
 from pydantic import BaseSettings
 from qcloud_cos import CosConfig
 from qcloud_cos import CosS3Client
+import yaml
 
 SITE_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 SITE_ROOT_TOO = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 FILE_PATH = os.path.join(SITE_ROOT, "runtime", "images")
 
+SITE_ROOT_YAML = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+PEM_PATH = os.path.join(SITE_ROOT_YAML, "config", "")
+apo = yaml.safe_load(open(PEM_PATH + "apollo.yaml", encoding='utf-8'))
+
 region = 'ap-guangzhou'
 token = None
 scheme = 'https'
@@ -39,31 +44,19 @@ class Env(BaseSettings):
     DATABASE_PWD: str = ''
     LOG_PATH = os.path.join(SITE_ROOT_TOO, "bin", "runtime", "logs", "info", "")
     LOG_ERROR_PATH = os.path.join(SITE_ROOT_TOO, "bin", "runtime", "logs", "error", "")
-    DB_HISTORY: dict = {
-        "database": "3y_history",
-        "host": "106.55.103.148",
-        "port": 3398,
-        "pwd": "c1ea602311a369f6",
-        "user": "root"
-    }
-    DB_3YV2: dict = {
-        "database": "3yakj_v2",
-        "host": "106.55.103.148",
-        "port": 3398,
-        "pwd": "c1ea602311a369f6",
-        "user": "root"
-    }
-    Redis: dict = {
-        "host": "43.138.132.9",
-        "password": "fanjiao2022",
-        "port": 6379,
-        "redis_db": 9
-    }
+    PEM_PATH = os.path.join(SITE_ROOT_YAML, "config", "")
+    YAML_DATA = apo.get('yaml')
+    LOGIN_URL = YAML_DATA.get('login_url')
+    NACOS_URL = YAML_DATA.get('config_url')
+    NACOSCONFIG = "show=all&dataId=fj-finance-test&group=DEFAULT_GROUP&tenant=cw-test&namespaceId=cw-test"
+    NAME = "fj_finance-test"
+    PWD = "Fanjiao@2023"
+    DB_HISTORY = apo.get('history')
+    DB_3YV2 = apo.get('business')
+    Redis = apo.get('redis')
     SECRET_KEY: str = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
     ALGORITHM: str = "HS256"
     PASSWORD: str = "fj123456"
-    # DB_HISTORY = "3yakj_v2"
-    # guild_url=""
     oss_url = 'http://oss.3yakj.com/application_static_data'
     CLEARING_CENTER_URL: str = 'http://106.55.103.148:6464/'
 
@@ -82,27 +75,14 @@ class ProdEnv(BaseSettings):
     DATABASE_PWD: str = ''
     LOG_PATH = os.path.join(SITE_ROOT_TOO, "bin", "runtime", "logs", "info", "")
     LOG_ERROR_PATH = os.path.join(SITE_ROOT_TOO, "bin", "runtime", "logs", "error", "")
-    """"""
-    DB_HISTORY: dict = {
-        "database": "site_history",
-        "host": "rm-uf6353zg8maql4284uo.mysql.rds.aliyuncs.com",
-        "port": 3306,
-        "pwd": "@Finance3yakj@",
-        "user": "sa_finance_3yakj_2021"
-    }
-    DB_3YV2: dict = {
-        "database": "site_3yakj20191108",
-        "host": "pc-uf6y3l45p5d174314public.rwlb.rds.aliyuncs.com",
-        "port": 3306,
-        "pwd": "sa_3yakj@2019@@",
-        "user": "sa_3yakj"
-    }
-    Redis: dict = {
-        "host": "159.75.254.122",
-        "password": "fanjiao2021",
-        "port": 6379,
-        "redis_db": 9
-    }
+    YAML_DATA = apo.get('yaml')
+    LOGIN_URL = YAML_DATA.get('login_url')
+    NACOS_URL = YAML_DATA.get('config_url')
+    NACOSCONFIG = "dataId=fj-finance&group=DEFAULT_GROUP&namespaceId=cw-pro&tenant=cw-pro&show=all&username=fj_finance"
+    PEM_PATH = os.path.join(SITE_ROOT_YAML, "config", "")
+    DB_HISTORY = apo.get('history')
+    DB_3YV2 = apo.get('business')
+    Redis = apo.get('redis')
     SECRET_KEY: str = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
     ALGORITHM: str = "HS256"
     PASSWORD: str = "fj123456"

libs/yaml_file.py

+# 获取yaml配置token
+import requests
+import yaml
+from core.config.env import env
+from libs.log_utils import Logger
+
+
+def get_yaml_token():
+    url = env.LOGIN_URL
+    data = {
+        "username": env.NAME,
+        "password": env.PWD
+    }
+    response = requests.post(url=url, data=data)
+    if response.status_code == 200:
+        token = response.json().get("accessToken")
+        return token
+    return ""
+
+
+# 创建本地yaml文件
+def create_yaml():
+    """
+    自动生成yaml文件
+    """
+    token = get_yaml_token()
+    url = f"{env.NACOS_URL}{env.NACOSCONFIG}&accessToken={token}"
+    try:
+        nacos_config_data = requests.get(url=url).json()
+        content = yaml.safe_load(nacos_config_data.get("content"))
+        with open(env.PEM_PATH + "apollo.yaml", "w", encoding='utf-8') as f:  # 写文件
+            yaml.dump(data=content, stream=f, allow_unicode=True, width=1000)
+            f.close()
+        Logger(20).logger.info("yaml初始化成功")
+    except Exception as e:
+        Logger(40).logger.error(f"生成yaml文件失败,原因:{str(e)}")

main.py

@@ -9,6 +9,7 @@ from fastapi import Request
 from core.config.env import env
 from core.dependencies.auth_dependen import create_access_token
 from libs.functions import time_format
+from libs.yaml_file import create_yaml
 
 app = FastAPI()
 
@@ -46,4 +47,5 @@ app.include_router(api_router, prefix="/api")  # 路由
 
 
 if __name__ == '__main__':
+    create_yaml()
     uvicorn.run(app=app, host="0.0.0.0", port=8009)

requirements.txt

@@ -46,4 +46,4 @@ openpyxl==3.1.2
 redis==4.3.4
 xpinyin==0.7.6
 xlsxwriter==3.1.0
-
+PyYAML==6.0