feat(base): 优化环境变量配置

.env.dev

@@ -16,9 +16,12 @@ REDIS_HOST = 192.168.3.4
 REDIS_PORT = 6379
 REDIS_DB = 0
 
+# 文件上传配置
+ASSETS_BASE_URL  = http://127.0.0.1:8001
+
 # 代理
-OVERSEAS_PROXY = 'http://127.0.0.1:7890'
+OVERSEAS_PROXY = http://127.0.0.1:7890
 
 # 微信
-APPID =
-APP_SECRET =
+APPID = wx3d30771f3987cc92
+APP_SECRET = 8e57b1b5344d1dbd058cd45f3b2c4d3c

config.py

@@ -5,7 +5,7 @@
 # @Author : xingc
 # @Desc :
 import logging
-import os
+from pathlib import Path
 
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
@@ -15,7 +15,7 @@ class Settings(BaseSettings):
     NAME: str = 'card book'
     PREFIX: str = NAME.lower().replace(' ', '_')
     ENVIRONMENT: str = 'dev'
-    BASE_DIR: str = os.path.dirname(os.path.abspath(__file__))
+    BASE_DIR: Path = Path(__file__).resolve().parent
     DEBUG: bool = True
 
     # 应用配置
@@ -40,30 +40,30 @@ class Settings(BaseSettings):
     REDIS_HOST: str
     REDIS_PORT: int = 6379
     REDIS_DB: int = 0
-    REDIS_CACHE_EXPIRATION_SECONDS: int = 60 * 30 * 1
+    REDIS_CACHE_EXPIRATION_SECONDS: int = 60 * 5 * 1
 
     # 身份认证
     ACCESS_SECRET_KEY: str = 'a2f718e8313bf8efcb45ef7364d040c45800f34d10f65776eef0f6584677ecd0'
-    REFRESH_SECRET_KEY: str = 'bb0cbf245de98c09c42c928326f6e07a44d91bc5a81787b37b4eb33428915fa7'
     ALGORITHM: str = 'HS256'
     ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 7
 
     # 文件上传配置
-    UPLOADED_FILES_DIR: str = os.path.join(BASE_DIR, 'uploads')
+    UPLOADED_FILES_DIR: Path = BASE_DIR / 'uploads'
     UPLOADED_ALLOWED_MIME_TYPES: set = {
         "image/jpeg",
         "image/png"
     }
     UPLOADED_MAX_FILE_SIZE: int = 1024 * 1024 * 10  # 10MB
+    ASSETS_BASE_URL: str = 'http://127.0.0.1:8000'
 
     # 代理
     OVERSEAS_PROXY: str = 'http://127.0.0.1:7890'
 
     # 微信
-    APPID: str = '<KEY>'
-    APP_SECRET: str = '<KEY>'
+    APPID: str = ''
+    APP_SECRET: str = ''
 
-    model_config = SettingsConfigDict(env_file=F'.env.{ENVIRONMENT.lower()}', env_file_encoding='utf-8')
+    model_config = SettingsConfigDict(env_file=f'.env.{ENVIRONMENT}', env_file_encoding='utf-8')
 
 
 class ProductionSettings(Settings):
@@ -73,7 +73,6 @@ class ProductionSettings(Settings):
 
     # 身份认证
     ACCESS_SECRET_KEY: str = '1f790219840ccd10b5d8b33d7f595b7fe0eed262403689f6fa1cab80149b00f2'
-    REFRESH_SECRET_KEY: str = 'fb57c85dc4e0256024ae809bc3200adc7b7de222b348da6df35d58e27338e3b8'
     ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 1  # 1days
 
 
@@ -95,6 +94,7 @@ class DevelopmentSettings(Settings):
 
 
 def get_settings():
+    import os
     env = os.environ.get('ENVIRONMENT', 'dev')
     if env == 'dev':
         return DevelopmentSettings()

core/app.py

@@ -35,6 +35,10 @@ def create_app() -> FastAPI:
     for router in routers:
         app.include_router(router)
 
+    if settings.ENVIRONMENT != 'prod':
+        from fastapi.staticfiles import StaticFiles
+        app.mount("/uploads", StaticFiles(directory="uploads"), name="uploads")
+
     @app.on_event('startup')
     async def startup_event():
         logger.info(f'Application <<{settings.NAME}>> is started')

core/base/exceptions.py

@@ -5,15 +5,25 @@
 # @Author : xingc
 # @Desc :
 
-class BaseException(Exception):
+class CardBookException(Exception):
     pass
 
 
-class SearchCardNotFound(Exception):
+class ScraperError(CardBookException):
+    """"采集错误"""
+    pass
+
+
+class SearchCardNotFound(ScraperError):
     """搜索卡片不存在"""
     pass
 
 
-class SearchCardRetry(Exception):
+class SearchCardRetry(ScraperError):
     """搜索重试"""
     pass
+
+
+class WechatApiError(CardBookException):
+    """微信开放服务调用api错误"""
+    pass

core/base/schema.py

@@ -20,9 +20,9 @@ class BaseResponse(BaseModel, Generic[T]):
 
 class Paginated(BaseModel, Generic[T]):
     list: Optional[T] = None
-    total: int
-    page: int
-    page_size: int
+    total: int = 0
+    page: int = 1
+    page_size: int = 15
 
 
 class SupportPlatformType(enum.Enum):

core/utils/helper.py

@@ -6,8 +6,11 @@
 # @Desc :
 import asyncio
 import functools
+from datetime import datetime
 from typing import Callable
 
+from config import settings
+
 
 def async_retry(max_retries: int = 3, exceptions: tuple = (Exception,)):
     """
@@ -32,3 +35,25 @@ def async_retry(max_retries: int = 3, exceptions: tuple = (Exception,)):
         return wrapper
 
     return decorator
+
+
+def generate_keyname(name: str, use_today: bool = False) -> str:
+    """生成统一键名前缀"""
+    if name.startswith(settings.PREFIX) is False:
+        names = [settings.PREFIX, name]
+        if use_today:
+            today = datetime.now().strftime('%Y%m%d')
+            names.insert(1, today)
+
+        return ':'.join(names)
+    else:
+        return name
+
+
+def check_picture_for_empty(picture: dict, or_: bool = False) -> bool:
+    """检查图片是否为空"""
+    results = [
+        picture.get('front_picture') is None,
+        picture.get('reverse_picture') is None
+    ]
+    return any(results) if or_ else all(results)