style: code format

apps/admin/dependencies.py

@@ -18,16 +18,15 @@ def create_token(data: dict, expires_in: int = 3600 * 24) -> str:
     :param data: 数据负载
     :param expires_in: 过期时间(秒)
     """
-    header = base64.b64encode(json.dumps({"alg": "HS256", "typ": "JWT"}).encode()).decode()
-    payload = base64.b64encode(json.dumps({
-        **data,
-        "exp": int(time.time()) + expires_in
-    }).encode()).decode()
+    header = base64.b64encode(
+        json.dumps({"alg": "HS256", "typ": "JWT"}).encode()
+    ).decode()
+    payload = base64.b64encode(
+        json.dumps({**data, "exp": int(time.time()) + expires_in}).encode()
+    ).decode()
 
     signature = hmac.new(
-        settings.admin_token.encode(),
-        f"{header}.{payload}".encode(),
-        'sha256'
+        settings.admin_token.encode(), f"{header}.{payload}".encode(), "sha256"
     ).digest()
     signature = base64.b64encode(signature).decode()
 
@@ -41,13 +40,13 @@ def verify_token(token: str) -> dict:
     :return: 解码后的数据
     """
     try:
-        header_b64, payload_b64, signature_b64 = token.split('.')
+        header_b64, payload_b64, signature_b64 = token.split(".")
 
         # 验证签名
         expected_signature = hmac.new(
             settings.admin_token.encode(),
             f"{header_b64}.{payload_b64}".encode(),
-            'sha256'
+            "sha256",
         ).digest()
         expected_signature_b64 = base64.b64encode(expected_signature).decode()
 
@@ -66,7 +65,9 @@ def verify_token(token: str) -> dict:
         raise ValueError(f"token验证失败: {str(e)}")
 
 
-async def admin_required(authorization: str = Header(default=None), request: Request = None):
+async def admin_required(
+    authorization: str = Header(default=None), request: Request = None
+):
     """
     验证管理员权限
     """
@@ -81,12 +82,14 @@ async def admin_required(authorization: str = Header(default=None), request: Req
             except ValueError as e:
                 is_admin = False
 
-        if request.url.path.startswith('/share/'):
+        if request.url.path.startswith("/share/"):
             if not settings.openUpload and not is_admin:
-                raise HTTPException(status_code=403, detail='本站未开启游客上传，如需上传请先登录后台')
+                raise HTTPException(
+                    status_code=403, detail="本站未开启游客上传，如需上传请先登录后台"
+                )
         else:
             if not is_admin:
-                raise HTTPException(status_code=401, detail='未授权或授权校验失败')
+                raise HTTPException(status_code=401, detail="未授权或授权校验失败")
         return is_admin
     except ValueError as e:
         raise HTTPException(status_code=401, detail=str(e))

apps/admin/schemas.py

@@ -7,7 +7,7 @@ class IDData(BaseModel):
 
 class ShareItem(BaseModel):
     expire_value: int
-    expire_style: str = 'day'
+    expire_style: str = "day"
     filename: str
 
 

apps/admin/services.py

@@ -19,16 +19,18 @@ class FileService:
         await self.file_storage.delete_file(file_code)
         await file_code.delete()
 
-    async def list_files(self, page: int, size: int, keyword: str = ''):
+    async def list_files(self, page: int, size: int, keyword: str = ""):
         offset = (page - 1) * size
-        files = await FileCodes.filter(prefix__icontains=keyword).limit(size).offset(offset)
+        files = (
+            await FileCodes.filter(prefix__icontains=keyword).limit(size).offset(offset)
+        )
         total = await FileCodes.filter(prefix__icontains=keyword).count()
         return files, total
 
     async def download_file(self, file_id: int):
         file_code = await FileCodes.filter(id=file_id).first()
         if not file_code:
-            raise HTTPException(status_code=404, detail='文件不存在')
+            raise HTTPException(status_code=404, detail="文件不存在")
         if file_code.text:
             return APIResponse(detail=file_code.text)
         else:
@@ -37,10 +39,12 @@ class FileService:
     async def share_local_file(self, item):
         local_file = LocalFileClass(item.filename)
         if not await local_file.exists():
-            raise HTTPException(status_code=404, detail='文件不存在')
+            raise HTTPException(status_code=404, detail="文件不存在")
 
         text = await local_file.read()
-        expired_at, expired_count, used_count, code = await get_expire_info(item.expire_value, item.expire_style)
+        expired_at, expired_count, used_count, code = await get_expire_info(
+            item.expire_value, item.expire_style
+        )
         path, suffix, prefix, uuid_file_name, save_path = await get_file_path_name(item)
 
         await self.file_storage.save_file(text, save_path)
@@ -58,8 +62,8 @@ class FileService:
         )
 
         return {
-            'code': code,
-            'name': local_file.file,
+            "code": code,
+            "name": local_file.file,
         }
 
 
@@ -68,21 +72,32 @@ class ConfigService:
         return settings.items()
 
     async def update_config(self, data: dict):
-        admin_token = data.get('admin_token')
-        if admin_token is None or admin_token == '':
-            raise HTTPException(status_code=400, detail='管理员密码不能为空')
+        admin_token = data.get("admin_token")
+        if admin_token is None or admin_token == "":
+            raise HTTPException(status_code=400, detail="管理员密码不能为空")
 
         for key, value in data.items():
             if key not in settings.default_config:
                 continue
-            if key in ['errorCount', 'errorMinute', 'max_save_seconds', 'onedrive_proxy', 'openUpload', 'port', 's3_proxy', 'uploadCount', 'uploadMinute', 'uploadSize']:
+            if key in [
+                "errorCount",
+                "errorMinute",
+                "max_save_seconds",
+                "onedrive_proxy",
+                "openUpload",
+                "port",
+                "s3_proxy",
+                "uploadCount",
+                "uploadMinute",
+                "uploadSize",
+            ]:
                 data[key] = int(value)
-            elif key in ['opacity']:
+            elif key in ["opacity"]:
                 data[key] = float(value)
             else:
                 data[key] = value
 
-        await KeyValue.filter(key='settings').update(value=data)
+        await KeyValue.filter(key="settings").update(value=data)
         for k, v in data.items():
             settings.__setattr__(k, v)
 
@@ -90,9 +105,9 @@ class ConfigService:
 class LocalFileService:
     async def list_files(self):
         files = []
-        if not os.path.exists(data_root / 'local'):
-            os.makedirs(data_root / 'local')
-        for file in os.listdir(data_root / 'local'):
+        if not os.path.exists(data_root / "local"):
+            os.makedirs(data_root / "local")
+        for file in os.listdir(data_root / "local"):
             files.append(LocalFileClass(file))
         return files
 
@@ -100,22 +115,24 @@ class LocalFileService:
         file = LocalFileClass(filename)
         if await file.exists():
             await file.delete()
-            return '删除成功'
-        raise HTTPException(status_code=404, detail='文件不存在')
+            return "删除成功"
+        raise HTTPException(status_code=404, detail="文件不存在")
 
 
 class LocalFileClass:
     def __init__(self, file):
         self.file = file
-        self.path = data_root / 'local' / file
-        self.ctime = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(os.path.getctime(self.path)))
+        self.path = data_root / "local" / file
+        self.ctime = time.strftime(
+            "%Y-%m-%d %H:%M:%S", time.localtime(os.path.getctime(self.path))
+        )
         self.size = os.path.getsize(self.path)
 
     async def read(self):
-        return open(self.path, 'rb')
+        return open(self.path, "rb")
 
     async def write(self, data):
-        with open(self.path, 'w') as f:
+        with open(self.path, "w") as f:
             f.write(data)
 
     async def delete(self):

apps/admin/views.py

@@ -6,17 +6,22 @@ import datetime
 
 from fastapi import APIRouter, Depends, HTTPException
 from apps.admin.services import FileService, ConfigService, LocalFileService
-from apps.admin.dependencies import admin_required, get_file_service, get_config_service, get_local_file_service
+from apps.admin.dependencies import (
+    admin_required,
+    get_file_service,
+    get_config_service,
+    get_local_file_service,
+)
 from apps.admin.schemas import IDData, ShareItem, DeleteItem, LoginData
 from core.response import APIResponse
 from apps.base.models import FileCodes, KeyValue
 from apps.admin.dependencies import create_token
 from core.settings import settings
 
-admin_api = APIRouter(prefix='/admin', tags=['管理'])
+admin_api = APIRouter(prefix="/admin", tags=["管理"])
 
 
-@admin_api.post('/login')
+@admin_api.post("/login")
 async def login(data: LoginData):
     # 验证管理员密码
     if data.password != settings.admin_token:
@@ -24,17 +29,14 @@ async def login(data: LoginData):
 
     # 生成包含管理员身份的token
     token = create_token({"is_admin": True})
-    return APIResponse(detail={
-        "token": token,
-        "token_type": "Bearer"
-    })
+    return APIResponse(detail={"token": token, "token_type": "Bearer"})
 
 
-@admin_api.get('/dashboard')
+@admin_api.get("/dashboard")
 async def dashboard(admin: bool = Depends(admin_required)):
     all_codes = await FileCodes.all()
     all_size = str(sum([code.size for code in all_codes]))
-    sys_start = await KeyValue.filter(key='sys_start').first()
+    sys_start = await KeyValue.filter(key="sys_start").first()
     # 获取当前日期时间
     now = datetime.datetime.now()
     today_start = now.replace(hour=0, minute=0, second=0, microsecond=0)
@@ -42,104 +44,105 @@ async def dashboard(admin: bool = Depends(admin_required)):
     yesterday_end = today_start - datetime.timedelta(microseconds=1)
     # 统计昨天一整天的记录数（从昨天0点到23:59:59）
     yesterday_codes = FileCodes.filter(
-        created_at__gte=yesterday_start,
-        created_at__lte=yesterday_end
+        created_at__gte=yesterday_start, created_at__lte=yesterday_end
     )
     # 统计今天到现在的记录数（从今天0点到现在）
-    today_codes = FileCodes.filter(
-        created_at__gte=today_start
+    today_codes = FileCodes.filter(created_at__gte=today_start)
+    return APIResponse(
+        detail={
+            "totalFiles": len(all_codes),
+            "storageUsed": all_size,
+            "sysUptime": sys_start.value,
+            "yesterdayCount": await yesterday_codes.count(),
+            "yesterdaySize": str(sum([code.size for code in await yesterday_codes])),
+            "todayCount": await today_codes.count(),
+            "todaySize": str(sum([code.size for code in await today_codes])),
+        }
     )
-    return APIResponse(detail={
-        'totalFiles': len(all_codes),
-        'storageUsed': all_size,
-        'sysUptime': sys_start.value,
-        'yesterdayCount': await yesterday_codes.count(),
-        'yesterdaySize': str(sum([code.size for code in await yesterday_codes])),
-        'todayCount': await today_codes.count(),
-        'todaySize': str(sum([code.size for code in await today_codes])),
-    })
 
 
-@admin_api.delete('/file/delete')
+@admin_api.delete("/file/delete")
 async def file_delete(
-        data: IDData,
-        file_service: FileService = Depends(get_file_service),
-        admin: bool = Depends(admin_required)
+    data: IDData,
+    file_service: FileService = Depends(get_file_service),
+    admin: bool = Depends(admin_required),
 ):
     await file_service.delete_file(data.id)
     return APIResponse()
 
 
-@admin_api.get('/file/list')
+@admin_api.get("/file/list")
 async def file_list(
-        page: int = 1,
-        size: int = 10,
-        keyword: str = '',
-        file_service: FileService = Depends(get_file_service),
-        admin: bool = Depends(admin_required)
+    page: int = 1,
+    size: int = 10,
+    keyword: str = "",
+    file_service: FileService = Depends(get_file_service),
+    admin: bool = Depends(admin_required),
 ):
     files, total = await file_service.list_files(page, size, keyword)
-    return APIResponse(detail={
-        'page': page,
-        'size': size,
-        'data': files,
-        'total': total,
-    })
+    return APIResponse(
+        detail={
+            "page": page,
+            "size": size,
+            "data": files,
+            "total": total,
+        }
+    )
 
 
-@admin_api.get('/config/get')
+@admin_api.get("/config/get")
 async def get_config(
-        config_service: ConfigService = Depends(get_config_service),
-        admin: bool = Depends(admin_required)
+    config_service: ConfigService = Depends(get_config_service),
+    admin: bool = Depends(admin_required),
 ):
     return APIResponse(detail=config_service.get_config())
 
 
-@admin_api.patch('/config/update')
+@admin_api.patch("/config/update")
 async def update_config(
-        data: dict,
-        config_service: ConfigService = Depends(get_config_service),
-        admin: bool = Depends(admin_required)
+    data: dict,
+    config_service: ConfigService = Depends(get_config_service),
+    admin: bool = Depends(admin_required),
 ):
-    data.pop('themesChoices')
+    data.pop("themesChoices")
     await config_service.update_config(data)
     return APIResponse()
 
 
-@admin_api.get('/file/download')
+@admin_api.get("/file/download")
 async def file_download(
-        id: int,
-        file_service: FileService = Depends(get_file_service),
-        admin: bool = Depends(admin_required)
+    id: int,
+    file_service: FileService = Depends(get_file_service),
+    admin: bool = Depends(admin_required),
 ):
     file_content = await file_service.download_file(id)
     return file_content
 
 
-@admin_api.get('/local/lists')
+@admin_api.get("/local/lists")
 async def get_local_lists(
-        local_file_service: LocalFileService = Depends(get_local_file_service),
-        admin: bool = Depends(admin_required)
+    local_file_service: LocalFileService = Depends(get_local_file_service),
+    admin: bool = Depends(admin_required),
 ):
     files = await local_file_service.list_files()
     return APIResponse(detail=files)
 
 
-@admin_api.delete('/local/delete')
+@admin_api.delete("/local/delete")
 async def delete_local_file(
-        item: DeleteItem,
-        local_file_service: LocalFileService = Depends(get_local_file_service),
-        admin: bool = Depends(admin_required)
+    item: DeleteItem,
+    local_file_service: LocalFileService = Depends(get_local_file_service),
+    admin: bool = Depends(admin_required),
 ):
     result = await local_file_service.delete_file(item.filename)
     return APIResponse(detail=result)
 
 
-@admin_api.post('/local/share')
+@admin_api.post("/local/share")
 async def share_local_file(
-        item: ShareItem,
-        file_service: FileService = Depends(get_file_service),
-        admin: bool = Depends(admin_required)
+    item: ShareItem,
+    file_service: FileService = Depends(get_file_service),
+    admin: bool = Depends(admin_required),
 ):
     share_info = await file_service.share_local_file(item)
     return APIResponse(detail=share_info)

apps/base/dependencies.py

@@ -12,26 +12,34 @@ class IPRateLimit:
     def check_ip(self, ip: str) -> bool:
         if ip in self.ips:
             ip_info = self.ips[ip]
-            if ip_info['count'] >= self.count:
-                if ip_info['time'] + timedelta(minutes=self.minutes) > datetime.now():
+            if ip_info["count"] >= self.count:
+                if ip_info["time"] + timedelta(minutes=self.minutes) > datetime.now():
                     return False
                 self.ips.pop(ip)
         return True
 
     def add_ip(self, ip: str) -> int:
-        ip_info = self.ips.get(ip, {'count': 0, 'time': datetime.now()})
-        ip_info['count'] += 1
-        ip_info['time'] = datetime.now()
+        ip_info = self.ips.get(ip, {"count": 0, "time": datetime.now()})
+        ip_info["count"] += 1
+        ip_info["time"] = datetime.now()
         self.ips[ip] = ip_info
-        return ip_info['count']
+        return ip_info["count"]
 
     async def remove_expired_ip(self) -> None:
         now = datetime.now()
         expiration = timedelta(minutes=self.minutes)
-        self.ips = {ip: info for ip, info in self.ips.items() if info['time'] + expiration >= now}
+        self.ips = {
+            ip: info
+            for ip, info in self.ips.items()
+            if info["time"] + expiration >= now
+        }
 
     def __call__(self, request: Request) -> str:
-        ip = request.headers.get('X-Real-IP') or request.headers.get('X-Forwarded-For') or request.client.host
+        ip = (
+            request.headers.get("X-Real-IP")
+            or request.headers.get("X-Forwarded-For")
+            or request.client.host
+        )
         if not self.check_ip(ip):
             raise HTTPException(status_code=423, detail="请求次数过多，请稍后再试")
         return ip

apps/base/models.py

@@ -14,17 +14,31 @@ from core.utils import get_now
 
 class FileCodes(Model):
     id: Optional[int] = fields.IntField(pk=True)
-    code: Optional[int] = fields.CharField(description='分享码', max_length=255, index=True, unique=True)
-    prefix: Optional[str] = fields.CharField(max_length=255, description='前缀', default='')
-    suffix: Optional[str] = fields.CharField(max_length=255, description='后缀', default='')
-    uuid_file_name: Optional[str] = fields.CharField(max_length=255, description='uuid文件名', null=True)
-    file_path: Optional[str] = fields.CharField(max_length=255, description='文件路径', null=True)
-    size: Optional[int] = fields.IntField(description='文件大小', default=0)
-    text: Optional[str] = fields.TextField(description='文本内容', null=True)
-    expired_at: Optional[datetime] = fields.DatetimeField(null=True, description='过期时间')
-    expired_count: Optional[int] = fields.IntField(description='可用次数', default=0)
-    used_count: Optional[int] = fields.IntField(description='已用次数', default=0)
-    created_at: Optional[datetime] = fields.DatetimeField(auto_now_add=True, description='创建时间')
+    code: Optional[int] = fields.CharField(
+        description="分享码", max_length=255, index=True, unique=True
+    )
+    prefix: Optional[str] = fields.CharField(
+        max_length=255, description="前缀", default=""
+    )
+    suffix: Optional[str] = fields.CharField(
+        max_length=255, description="后缀", default=""
+    )
+    uuid_file_name: Optional[str] = fields.CharField(
+        max_length=255, description="uuid文件名", null=True
+    )
+    file_path: Optional[str] = fields.CharField(
+        max_length=255, description="文件路径", null=True
+    )
+    size: Optional[int] = fields.IntField(description="文件大小", default=0)
+    text: Optional[str] = fields.TextField(description="文本内容", null=True)
+    expired_at: Optional[datetime] = fields.DatetimeField(
+        null=True, description="过期时间"
+    )
+    expired_count: Optional[int] = fields.IntField(description="可用次数", default=0)
+    used_count: Optional[int] = fields.IntField(description="已用次数", default=0)
+    created_at: Optional[datetime] = fields.DatetimeField(
+        auto_now_add=True, description="创建时间"
+    )
 
     async def is_expired(self):
         # 按时间
@@ -42,9 +56,13 @@ class FileCodes(Model):
 
 class KeyValue(Model):
     id: Optional[int] = fields.IntField(pk=True)
-    key: Optional[str] = fields.CharField(max_length=255, description='键', index=True, unique=True)
-    value: Optional[str] = fields.JSONField(description='值', null=True)
-    created_at: Optional[datetime] = fields.DatetimeField(auto_now_add=True, description='创建时间')
+    key: Optional[str] = fields.CharField(
+        max_length=255, description="键", index=True, unique=True
+    )
+    value: Optional[str] = fields.JSONField(description="值", null=True)
+    created_at: Optional[datetime] = fields.DatetimeField(
+        auto_now_add=True, description="创建时间"
+    )
 
 
-file_codes_pydantic = pydantic_model_creator(FileCodes, name='FileCodes')
+file_codes_pydantic = pydantic_model_creator(FileCodes, name="FileCodes")

apps/base/utils.py

@@ -14,7 +14,7 @@ from core.utils import get_random_num, get_random_string, max_save_times_desc
 async def get_file_path_name(file: UploadFile) -> Tuple[str, str, str, str, str]:
     """获取文件路径和文件名"""
     today = datetime.datetime.now()
-    storage_path = settings.storage_path.strip('/')  # 移除开头和结尾的斜杠
+    storage_path = settings.storage_path.strip("/")  # 移除开头和结尾的斜杠
     file_uuid = uuid.uuid4().hex
 
     # 使用 UUID 作为子目录名
@@ -29,32 +29,42 @@ async def get_file_path_name(file: UploadFile) -> Tuple[str, str, str, str, str]
     return path, suffix, prefix, file.filename, save_path
 
 
-async def get_expire_info(expire_value: int, expire_style: str) -> Tuple[Optional[datetime.datetime], int, int, str]:
+async def get_expire_info(
+    expire_value: int, expire_style: str
+) -> Tuple[Optional[datetime.datetime], int, int, str]:
     """获取过期信息"""
     expired_count, used_count = -1, 0
     now = datetime.datetime.now()
     code = None
 
-    max_timedelta = datetime.timedelta(seconds=settings.max_save_seconds) if settings.max_save_seconds > 0 else datetime.timedelta(days=7)
-    detail = await max_save_times_desc(settings.max_save_seconds) if settings.max_save_seconds > 0 else '7天'
-    detail = f'限制最长时间为 {detail[0]}，可换用其他方式'
+    max_timedelta = (
+        datetime.timedelta(seconds=settings.max_save_seconds)
+        if settings.max_save_seconds > 0
+        else datetime.timedelta(days=7)
+    )
+    detail = (
+        await max_save_times_desc(settings.max_save_seconds)
+        if settings.max_save_seconds > 0
+        else "7天"
+    )
+    detail = f"限制最长时间为 {detail[0]}，可换用其他方式"
 
     expire_styles = {
-        'day': lambda: now + datetime.timedelta(days=expire_value),
-        'hour': lambda: now + datetime.timedelta(hours=expire_value),
-        'minute': lambda: now + datetime.timedelta(minutes=expire_value),
-        'count': lambda: (now + datetime.timedelta(days=1), expire_value),
-        'forever': lambda: (None, None),  # 修改这里
+        "day": lambda: now + datetime.timedelta(days=expire_value),
+        "hour": lambda: now + datetime.timedelta(hours=expire_value),
+        "minute": lambda: now + datetime.timedelta(minutes=expire_value),
+        "count": lambda: (now + datetime.timedelta(days=1), expire_value),
+        "forever": lambda: (None, None),  # 修改这里
     }
 
     if expire_style in expire_styles:
         result = expire_styles[expire_style]()
         if isinstance(result, tuple):
             expired_at, extra = result
-            if expire_style == 'count':
+            if expire_style == "count":
                 expired_count = extra
-            elif expire_style == 'forever':
-                code = await get_random_code(style='string')  # 移动到这里
+            elif expire_style == "forever":
+                code = await get_random_code(style="string")  # 移动到这里
         else:
             expired_at = result
         if expired_at and expired_at - now > max_timedelta:
@@ -68,15 +78,15 @@ async def get_expire_info(expire_value: int, expire_style: str) -> Tuple[Optiona
     return expired_at, expired_count, used_count, code
 
 
-async def get_random_code(style='num') -> str:
+async def get_random_code(style="num") -> str:
     """获取随机字符串"""
     while True:
-        code = await get_random_num() if style == 'num' else await get_random_string()
+        code = await get_random_num() if style == "num" else await get_random_string()
         if not await FileCodes.filter(code=code).exists():
             return code
 
 
 ip_limit = {
-    'error': IPRateLimit(count=settings.uploadCount, minutes=settings.errorMinute),
-    'upload': IPRateLimit(count=settings.errorCount, minutes=settings.errorMinute)
+    "error": IPRateLimit(count=settings.uploadCount, minutes=settings.errorMinute),
+    "upload": IPRateLimit(count=settings.errorCount, minutes=settings.errorMinute),
 }

apps/base/views.py

@@ -8,32 +8,36 @@ from core.settings import settings
 from core.storage import storages, FileStorageInterface
 from core.utils import get_select_token
 
-share_api = APIRouter(prefix='/share', tags=['分享'])
+share_api = APIRouter(prefix="/share", tags=["分享"])
 
 
 async def validate_file_size(file: UploadFile, max_size: int):
     if file.size > max_size:
         max_size_mb = max_size / (1024 * 1024)
-        raise HTTPException(status_code=403, detail=f'大小超过限制,最大为{max_size_mb:.2f} MB')
+        raise HTTPException(
+            status_code=403, detail=f"大小超过限制,最大为{max_size_mb:.2f} MB"
+        )
 
 
 async def create_file_code(code, **kwargs):
     return await FileCodes.create(code=code, **kwargs)
 
 
-@share_api.post('/text/', dependencies=[Depends(admin_required)])
+@share_api.post("/text/", dependencies=[Depends(admin_required)])
 async def share_text(
-        text: str = Form(...),
-        expire_value: int = Form(default=1, gt=0),
-        expire_style: str = Form(default='day'),
-        ip: str = Depends(ip_limit['upload'])
+    text: str = Form(...),
+    expire_value: int = Form(default=1, gt=0),
+    expire_style: str = Form(default="day"),
+    ip: str = Depends(ip_limit["upload"]),
 ):
-    text_size = len(text.encode('utf-8'))
+    text_size = len(text.encode("utf-8"))
     max_txt_size = 222 * 1024
     if text_size > max_txt_size:
-        raise HTTPException(status_code=403, detail='内容过多,建议采用文件形式')
+        raise HTTPException(status_code=403, detail="内容过多,建议采用文件形式")
 
-    expired_at, expired_count, used_count, code = await get_expire_info(expire_value, expire_style)
+    expired_at, expired_count, used_count, code = await get_expire_info(
+        expire_value, expire_style
+    )
     await create_file_code(
         code=code,
         text=text,
@@ -41,25 +45,27 @@ async def share_text(
         expired_count=expired_count,
         used_count=used_count,
         size=len(text),
-        prefix='Text'
+        prefix="Text",
     )
-    ip_limit['upload'].add_ip(ip)
-    return APIResponse(detail={'code': code})
+    ip_limit["upload"].add_ip(ip)
+    return APIResponse(detail={"code": code})
 
 
-@share_api.post('/file/', dependencies=[Depends(admin_required)])
+@share_api.post("/file/", dependencies=[Depends(admin_required)])
 async def share_file(
-        expire_value: int = Form(default=1, gt=0),
-        expire_style: str = Form(default='day'),
-        file: UploadFile = File(...),
-        ip: str = Depends(ip_limit['upload'])
+    expire_value: int = Form(default=1, gt=0),
+    expire_style: str = Form(default="day"),
+    file: UploadFile = File(...),
+    ip: str = Depends(ip_limit["upload"]),
 ):
     await validate_file_size(file, settings.uploadSize)
 
     if expire_style not in settings.expireStyle:
-        raise HTTPException(status_code=400, detail='过期时间类型错误')
+        raise HTTPException(status_code=400, detail="过期时间类型错误")
 
-    expired_at, expired_count, used_count, code = await get_expire_info(expire_value, expire_style)
+    expired_at, expired_count, used_count, code = await get_expire_info(
+        expire_value, expire_style
+    )
     path, suffix, prefix, uuid_file_name, save_path = await get_file_path_name(file)
 
     file_storage: FileStorageInterface = storages[settings.file_storage]()
@@ -76,16 +82,16 @@ async def share_file(
         expired_count=expired_count,
         used_count=used_count,
     )
-    ip_limit['upload'].add_ip(ip)
-    return APIResponse(detail={'code': code, 'name': file.filename})
+    ip_limit["upload"].add_ip(ip)
+    return APIResponse(detail={"code": code, "name": file.filename})
 
 
 async def get_code_file_by_code(code, check=True):
     file_code = await FileCodes.filter(code=code).first()
     if not file_code:
-        return False, '文件不存在'
+        return False, "文件不存在"
     if await file_code.is_expired() and check:
-        return False, '文件已过期'
+        return False, "文件已过期"
     return True, file_code
 
 
@@ -96,44 +102,50 @@ async def update_file_usage(file_code):
     await file_code.save()
 
 
-@share_api.get('/select/')
-async def get_code_file(code: str, ip: str = Depends(ip_limit['error'])):
+@share_api.get("/select/")
+async def get_code_file(code: str, ip: str = Depends(ip_limit["error"])):
     file_storage: FileStorageInterface = storages[settings.file_storage]()
     has, file_code = await get_code_file_by_code(code)
     if not has:
-        ip_limit['error'].add_ip(ip)
+        ip_limit["error"].add_ip(ip)
         return APIResponse(code=404, detail=file_code)
 
     await update_file_usage(file_code)
     return await file_storage.get_file_response(file_code)
 
 
-@share_api.post('/select/')
-async def select_file(data: SelectFileModel, ip: str = Depends(ip_limit['error'])):
+@share_api.post("/select/")
+async def select_file(data: SelectFileModel, ip: str = Depends(ip_limit["error"])):
     file_storage: FileStorageInterface = storages[settings.file_storage]()
     has, file_code = await get_code_file_by_code(data.code)
     if not has:
-        ip_limit['error'].add_ip(ip)
+        ip_limit["error"].add_ip(ip)
         return APIResponse(code=404, detail=file_code)
 
     await update_file_usage(file_code)
-    return APIResponse(detail={
-        'code': file_code.code,
-        'name': file_code.prefix + file_code.suffix,
-        'size': file_code.size,
-        'text': file_code.text if file_code.text is not None else await file_storage.get_file_url(file_code),
-    })
+    return APIResponse(
+        detail={
+            "code": file_code.code,
+            "name": file_code.prefix + file_code.suffix,
+            "size": file_code.size,
+            "text": (
+                file_code.text
+                if file_code.text is not None
+                else await file_storage.get_file_url(file_code)
+            ),
+        }
+    )
 
 
-@share_api.get('/download')
-async def download_file(key: str, code: str, ip: str = Depends(ip_limit['error'])):
+@share_api.get("/download")
+async def download_file(key: str, code: str, ip: str = Depends(ip_limit["error"])):
     file_storage: FileStorageInterface = storages[settings.file_storage]()
     if await get_select_token(code) != key:
-        ip_limit['error'].add_ip(ip)
+        ip_limit["error"].add_ip(ip)
 
     has, file_code = await get_code_file_by_code(code, False)
     if not has:
-        return APIResponse(code=404, detail='文件不存在')
+        return APIResponse(code=404, detail="文件不存在")
 
     return (
         APIResponse(detail=file_code.text)

core/response.py

@@ -6,10 +6,10 @@ from typing import Generic, TypeVar
 
 from pydantic.v1.generics import GenericModel
 
-T = TypeVar('T')
+T = TypeVar("T")
 
 
 class APIResponse(GenericModel, Generic[T]):
     code: int = 200
-    message: str = 'ok'
+    message: str = "ok"
     detail: T

core/settings.py

@@ -5,70 +5,70 @@
 from pathlib import Path
 
 BASE_DIR = Path(__file__).resolve().parent.parent
-data_root = BASE_DIR / 'data'
+data_root = BASE_DIR / "data"
 
 if not data_root.exists():
     data_root.mkdir(parents=True, exist_ok=True)
 
 DEFAULT_CONFIG = {
-    'file_storage': 'local',
-    'storage_path': '',
-    'name': '文件快递柜 - FileCodeBox',
-    'description': '开箱即用的文件快传系统',
-    'notify_title': '系统通知',
-    'notify_content': '欢迎使用 FileCodeBox，本程序开源于 <a href="https://github.com/vastsa/FileCodeBox" target="_blank">Github</a> ，欢迎Star和Fork。',
-    'page_explain': '请勿上传或分享违法内容。根据《中华人民共和国网络安全法》、《中华人民共和国刑法》、《中华人民共和国治安管理处罚法》等相关规定。 传播或存储违法、违规内容，会受到相关处罚，严重者将承担刑事责任。本站坚决配合相关部门，确保网络内容的安全，和谐，打造绿色网络环境。',
-    'keywords': 'FileCodeBox, 文件快递柜, 口令传送箱, 匿名口令分享文本, 文件',
-    's3_access_key_id': '',
-    's3_secret_access_key': '',
-    's3_bucket_name': '',
-    's3_endpoint_url': '',
-    's3_region_name': 'auto',
-    's3_signature_version': 's3v2',
-    's3_hostname': '',
-    's3_proxy': 0,
-    'max_save_seconds': 0,
-    'aws_session_token': '',
-    'onedrive_domain': '',
-    'onedrive_client_id': '',
-    'onedrive_username': '',
-    'onedrive_password': '',
-    'onedrive_root_path': 'filebox_storage',
-    'onedrive_proxy': 0,
-    'webdav_hostname': '',
-    'webdav_root_path': 'filebox_storage',
-    'webdav_proxy': 0,
-    'admin_token': 'FileCodeBox2023',
-    'openUpload': 1,
-    'uploadSize': 1024 * 1024 * 10,
-    'expireStyle': ['day', 'hour', 'minute', 'forever', 'count'],
-    'uploadMinute': 1,
-    'webdav_url': '',
-    'webdav_password': '',
-    'webdav_username': '',
-    'opacity': 0.9,
-    'background': '',
-    'uploadCount': 10,
-    'themesChoices': [
+    "file_storage": "local",
+    "storage_path": "",
+    "name": "文件快递柜 - FileCodeBox",
+    "description": "开箱即用的文件快传系统",
+    "notify_title": "系统通知",
+    "notify_content": '欢迎使用 FileCodeBox，本程序开源于 <a href="https://github.com/vastsa/FileCodeBox" target="_blank">Github</a> ，欢迎Star和Fork。',
+    "page_explain": "请勿上传或分享违法内容。根据《中华人民共和国网络安全法》、《中华人民共和国刑法》、《中华人民共和国治安管理处罚法》等相关规定。 传播或存储违法、违规内容，会受到相关处罚，严重者将承担刑事责任。本站坚决配合相关部门，确保网络内容的安全，和谐，打造绿色网络环境。",
+    "keywords": "FileCodeBox, 文件快递柜, 口令传送箱, 匿名口令分享文本, 文件",
+    "s3_access_key_id": "",
+    "s3_secret_access_key": "",
+    "s3_bucket_name": "",
+    "s3_endpoint_url": "",
+    "s3_region_name": "auto",
+    "s3_signature_version": "s3v2",
+    "s3_hostname": "",
+    "s3_proxy": 0,
+    "max_save_seconds": 0,
+    "aws_session_token": "",
+    "onedrive_domain": "",
+    "onedrive_client_id": "",
+    "onedrive_username": "",
+    "onedrive_password": "",
+    "onedrive_root_path": "filebox_storage",
+    "onedrive_proxy": 0,
+    "webdav_hostname": "",
+    "webdav_root_path": "filebox_storage",
+    "webdav_proxy": 0,
+    "admin_token": "FileCodeBox2023",
+    "openUpload": 1,
+    "uploadSize": 1024 * 1024 * 10,
+    "expireStyle": ["day", "hour", "minute", "forever", "count"],
+    "uploadMinute": 1,
+    "webdav_url": "",
+    "webdav_password": "",
+    "webdav_username": "",
+    "opacity": 0.9,
+    "background": "",
+    "uploadCount": 10,
+    "themesChoices": [
         {
-            'name': '2023',
-            'key': 'themes/2023',
-            'author': 'Lan',
-            'version': '1.0',
+            "name": "2023",
+            "key": "themes/2023",
+            "author": "Lan",
+            "version": "1.0",
         },
         {
-            'name': '2024',
-            'key': 'themes/2024',
-            'author': 'Lan',
-            'version': '1.0',
-        }
+            "name": "2024",
+            "key": "themes/2024",
+            "author": "Lan",
+            "version": "1.0",
+        },
     ],
-    'themesSelect': 'themes/2024',
-    'errorMinute': 1,
-    'errorCount': 1,
-    'port': 12345,
-    'showAdminAddr': 0,
-    'robotsText': 'User-agent: *\nDisallow: /',
+    "themesSelect": "themes/2024",
+    "errorMinute": 1,
+    "errorCount": 1,
+    "port": 12345,
+    "showAdminAddr": 0,
+    "robotsText": "User-agent: *\nDisallow: /",
 }
 
 
@@ -82,10 +82,12 @@ class Settings:
             return self.user_config[attr]
         if attr in self.default_config:
             return self.default_config[attr]
-        raise AttributeError(f"'{self.__class__.__name__}' object has no attribute '{attr}'")
+        raise AttributeError(
+            f"'{self.__class__.__name__}' object has no attribute '{attr}'"
+        )
 
     def __setattr__(self, key, value):
-        if key in ['default_config', 'user_config']:
+        if key in ["default_config", "user_config"]:
             super().__setattr__(key, value)
         else:
             self.user_config[key] = value

core/tasks.py

@@ -19,13 +19,15 @@ async def delete_expire_files():
     while True:
         try:
             # 遍历 share目录下的所有文件夹，删除空的文件夹，并判断父目录是否为空，如果为空也删除
-            if settings.file_storage == 'local':
+            if settings.file_storage == "local":
                 for root, dirs, files in os.walk(f"{data_root}/share/data"):
                     if not dirs and not files:
                         os.rmdir(root)
-            await ip_limit['error'].remove_expired_ip()
-            await ip_limit['upload'].remove_expired_ip()
-            expire_data = await FileCodes.filter(Q(expired_at__lt=await get_now()) | Q(expired_count=0)).all()
+            await ip_limit["error"].remove_expired_ip()
+            await ip_limit["upload"].remove_expired_ip()
+            expire_data = await FileCodes.filter(
+                Q(expired_at__lt=await get_now()) | Q(expired_count=0)
+            ).all()
             for exp in expire_data:
                 await file_storage.delete_file(exp)
                 await exp.delete()

core/utils.py

@@ -9,6 +9,7 @@ import string
 import time
 from core.settings import settings
 
+
 async def get_random_num():
     """
     获取随机数
@@ -25,7 +26,7 @@ async def get_random_string():
     获取随机字符串
     :return:
     """
-    return ''.join(random.choice(r_s) for _ in range(5))
+    return "".join(random.choice(r_s) for _ in range(5))
 
 
 async def get_now():
@@ -33,9 +34,7 @@ async def get_now():
     获取当前时间
     :return:
     """
-    return datetime.datetime.now(
-        datetime.timezone(datetime.timedelta(hours=8))
-    )
+    return datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=8)))
 
 
 async def get_select_token(code: str):
@@ -45,7 +44,9 @@ async def get_select_token(code: str):
     :return:
     """
     token = settings.admin_token
-    return hashlib.sha256(f"{code}{int(time.time() / 1000)}000{token}".encode()).hexdigest()
+    return hashlib.sha256(
+        f"{code}{int(time.time() / 1000)}000{token}".encode()
+    ).hexdigest()
 
 
 async def get_file_url(code: str):
@@ -54,7 +55,7 @@ async def get_file_url(code: str):
     :param code:
     :return:
     """
-    return f'/share/download?key={await get_select_token(code)}&code={code}'
+    return f"/share/download?key={await get_select_token(code)}&code={code}"
 
 
 async def max_save_times_desc(max_save_seconds: int):
@@ -66,28 +67,28 @@ async def max_save_times_desc(max_save_seconds: int):
 
     def gen_desc_zh(value: int, desc: str):
         if value > 0:
-            return f'{value}{desc}'
+            return f"{value}{desc}"
         else:
-            return ''
+            return ""
 
     def gen_desc_en(value: int, desc: str):
         if value > 0:
-            ret = f'{value} {desc}'
+            ret = f"{value} {desc}"
             if value > 1:
-                ret += 's'
-            ret += ' '
+                ret += "s"
+            ret += " "
             return ret
         else:
-            return ''
+            return ""
 
     max_timedelta = datetime.timedelta(seconds=max_save_seconds)
-    desc_zh, desc_en = '最长保存时间：', 'Max save time: '
-    desc_zh += gen_desc_zh(max_timedelta.days, '天')
-    desc_en += gen_desc_en(max_timedelta.days, 'day')
-    desc_zh += gen_desc_zh(max_timedelta.seconds // 3600, '小时')
-    desc_en += gen_desc_en(max_timedelta.seconds // 3600, 'hour')
-    desc_zh += gen_desc_zh(max_timedelta.seconds % 3600 // 60, '分钟')
-    desc_en += gen_desc_en(max_timedelta.seconds % 3600 // 60, 'minute')
-    desc_zh += gen_desc_zh(max_timedelta.seconds % 60, '秒')
-    desc_en += gen_desc_en(max_timedelta.seconds % 60, 'second')
+    desc_zh, desc_en = "最长保存时间：", "Max save time: "
+    desc_zh += gen_desc_zh(max_timedelta.days, "天")
+    desc_en += gen_desc_en(max_timedelta.days, "day")
+    desc_zh += gen_desc_zh(max_timedelta.seconds // 3600, "小时")
+    desc_en += gen_desc_en(max_timedelta.seconds // 3600, "hour")
+    desc_zh += gen_desc_zh(max_timedelta.seconds % 3600 // 60, "分钟")
+    desc_en += gen_desc_en(max_timedelta.seconds % 3600 // 60, "minute")
+    desc_zh += gen_desc_zh(max_timedelta.seconds % 60, "秒")
+    desc_en += gen_desc_en(max_timedelta.seconds % 60, "second")
     return desc_zh, desc_en

main.py

@@ -26,10 +26,10 @@ from tortoise import Tortoise
 
 async def init_db():
     await Tortoise.init(
-        db_url=f'sqlite://{data_root}/filecodebox.db',
-        modules={'models': ['apps.base.models']},
+        db_url=f"sqlite://{data_root}/filecodebox.db",
+        modules={"models": ["apps.base.models"]},
         use_tz=False,
-        timezone="Asia/Shanghai"
+        timezone="Asia/Shanghai",
     )
     await Tortoise.generate_schemas()
 
@@ -41,7 +41,11 @@ async def lifespan(app: FastAPI):
 
     # 加载配置
     await load_config()
-    app.mount('/assets', StaticFiles(directory=f'./{settings.themesSelect}/assets'), name="assets")
+    app.mount(
+        "/assets",
+        StaticFiles(directory=f"./{settings.themesSelect}/assets"),
+        name="assets",
+    )
 
     # 启动后台任务
     task = asyncio.create_task(delete_expire_files())
@@ -56,14 +60,18 @@ async def lifespan(app: FastAPI):
 
 
 async def load_config():
-    user_config, _ = await KeyValue.get_or_create(key='settings', defaults={'value': DEFAULT_CONFIG})
-    await KeyValue.update_or_create(key='sys_start', defaults={'value': int(time.time() * 1000)})
+    user_config, _ = await KeyValue.get_or_create(
+        key="settings", defaults={"value": DEFAULT_CONFIG}
+    )
+    await KeyValue.update_or_create(
+        key="sys_start", defaults={"value": int(time.time() * 1000)}
+    )
     settings.user_config = user_config.value
     # 更新 ip_limit 配置
-    ip_limit['error'].minutes = settings.errorMinute
-    ip_limit['error'].count = settings.errorCount
-    ip_limit['upload'].minutes = settings.uploadMinute
-    ip_limit['upload'].count = settings.uploadCount
+    ip_limit["error"].minutes = settings.errorMinute
+    ip_limit["error"].count = settings.errorCount
+    ip_limit["upload"].minutes = settings.uploadMinute
+    ip_limit["upload"].count = settings.uploadCount
 
 
 app = FastAPI(lifespan=lifespan)
@@ -80,11 +88,11 @@ app.add_middleware(
 register_tortoise(
     app,
     config={
-        'connections': {'default': f'sqlite://{data_root}/filecodebox.db'},
-        'apps': {
-            'models': {
-                'models': ['apps.base.models'],
-                'default_connection': 'default',
+        "connections": {"default": f"sqlite://{data_root}/filecodebox.db"},
+        "apps": {
+            "models": {
+                "models": ["apps.base.models"],
+                "default_connection": "default",
             },
         },
     },
@@ -96,39 +104,48 @@ app.include_router(share_api)
 app.include_router(admin_api)
 
 
-@app.get('/')
+@app.get("/")
 async def index():
     return HTMLResponse(
-        content=open(BASE_DIR / f'{settings.themesSelect}/index.html', 'r', encoding='utf-8').read()
-        .replace('{{title}}', str(settings.name))
-        .replace('{{description}}', str(settings.description))
-        .replace('{{keywords}}', str(settings.keywords))
-        .replace('{{opacity}}', str(settings.opacity))
-        .replace('{{background}}', str(settings.background))
-        , media_type='text/html', headers={'Cache-Control': 'no-cache'})
+        content=open(
+            BASE_DIR / f"{settings.themesSelect}/index.html", "r", encoding="utf-8"
+        )
+        .read()
+        .replace("{{title}}", str(settings.name))
+        .replace("{{description}}", str(settings.description))
+        .replace("{{keywords}}", str(settings.keywords))
+        .replace("{{opacity}}", str(settings.opacity))
+        .replace("{{background}}", str(settings.background)),
+        media_type="text/html",
+        headers={"Cache-Control": "no-cache"},
+    )
 
 
-@app.get('/robots.txt')
+@app.get("/robots.txt")
 async def robots():
-    return HTMLResponse(content=settings.robotsText, media_type='text/plain')
+    return HTMLResponse(content=settings.robotsText, media_type="text/plain")
 
 
-@app.post('/')
+@app.post("/")
 async def get_config():
-    return APIResponse(detail={
-        'name': settings.name,
-        'description': settings.description,
-        'explain': settings.page_explain,
-        'uploadSize': settings.uploadSize,
-        'expireStyle': settings.expireStyle,
-        'openUpload': settings.openUpload,
-        'notify_title': settings.notify_title,
-        'notify_content': settings.notify_content,
-        'show_admin_address': settings.showAdminAddr,
-    })
+    return APIResponse(
+        detail={
+            "name": settings.name,
+            "description": settings.description,
+            "explain": settings.page_explain,
+            "uploadSize": settings.uploadSize,
+            "expireStyle": settings.expireStyle,
+            "openUpload": settings.openUpload,
+            "notify_title": settings.notify_title,
+            "notify_content": settings.notify_content,
+            "show_admin_address": settings.showAdminAddr,
+        }
+    )
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     import uvicorn
 
-    uvicorn.run(app='main:app', host="0.0.0.0", port=settings.port, reload=False, workers=1)
+    uvicorn.run(
+        app="main:app", host="0.0.0.0", port=settings.port, reload=False, workers=1
+    )