diff --git a/apps/admin/dependencies.py b/apps/admin/dependencies.py
index b935ce5..e9aa9b3 100644
--- a/apps/admin/dependencies.py
+++ b/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))
diff --git a/apps/admin/schemas.py b/apps/admin/schemas.py
index 5cc0802..5ef4f85 100644
--- a/apps/admin/schemas.py
+++ b/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
diff --git a/apps/admin/services.py b/apps/admin/services.py
index 133c54c..c012772 100644
--- a/apps/admin/services.py
+++ b/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):
diff --git a/apps/admin/views.py b/apps/admin/views.py
index 515f0c1..f84fac1 100644
--- a/apps/admin/views.py
+++ b/apps/admin/views.py
@@ -6,35 +6,37 @@ 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:
raise HTTPException(status_code=401, detail="密码错误")
-
+
# 生成包含管理员身份的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)
diff --git a/apps/base/dependencies.py b/apps/base/dependencies.py
index 614eb5e..577a2dc 100644
--- a/apps/base/dependencies.py
+++ b/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
diff --git a/apps/base/models.py b/apps/base/models.py
index a56dfc3..3224b05 100644
--- a/apps/base/models.py
+++ b/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")
diff --git a/apps/base/utils.py b/apps/base/utils.py
index 7cf2839..c375f97 100644
--- a/apps/base/utils.py
+++ b/apps/base/utils.py
@@ -14,47 +14,57 @@ 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 作为子目录名
base_path = f"share/data/{today.strftime('%Y/%m/%d')}/{file_uuid}"
-
+
# 如果设置了存储路径,将其添加到基础路径中
path = f"{storage_path}/{base_path}" if storage_path else base_path
-
+
prefix, suffix = os.path.splitext(file.filename)
# 保持原始文件名
save_path = f"{path}/{file.filename}"
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),
}
diff --git a/apps/base/views.py b/apps/base/views.py
index 6b0aaed..863ecb6 100644
--- a/apps/base/views.py
+++ b/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)
diff --git a/core/response.py b/core/response.py
index f6b8a38..5769111 100644
--- a/core/response.py
+++ b/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
diff --git a/core/settings.py b/core/settings.py
index 20709db..f05eaf5 100644
--- a/core/settings.py
+++ b/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,本程序开源于 Github ,欢迎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,本程序开源于 Github ,欢迎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
diff --git a/core/storage.py b/core/storage.py
index 960a100..958733e 100644
--- a/core/storage.py
+++ b/core/storage.py
@@ -412,7 +412,7 @@ class WebDAVFileStorage(FileStorageInterface):
async def _is_dir_empty(self, dir_path: str) -> bool:
"""检查目录是否为空"""
url = self._build_url(dir_path)
-
+
async with aiohttp.ClientSession(auth=self.auth) as session:
async with session.request("PROPFIND", url, headers={"Depth": "1"}) as resp:
if resp.status != 207: # 207 是 Multi-Status 响应
@@ -425,16 +425,16 @@ class WebDAVFileStorage(FileStorageInterface):
"""递归删除空目录"""
path_obj = Path(file_path)
current_path = path_obj.parent
-
+
while str(current_path) != ".":
if not await self._is_dir_empty(str(current_path)):
break
-
+
url = self._build_url(str(current_path))
async with session.delete(url) as resp:
if resp.status not in (200, 204, 404):
break
-
+
current_path = current_path.parent
async def save_file(self, file: UploadFile, save_path: str):
@@ -478,10 +478,10 @@ class WebDAVFileStorage(FileStorageInterface):
status_code=resp.status,
detail=f"WebDAV删除失败: {content[:200]}",
)
-
+
# 使用同一个 session 删除空目录
await self._delete_empty_dirs(file_path, session)
-
+
except aiohttp.ClientError as e:
raise HTTPException(status_code=503, detail=f"WebDAV连接异常: {str(e)}")
diff --git a/core/tasks.py b/core/tasks.py
index d3e17ad..726723a 100644
--- a/core/tasks.py
+++ b/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()
diff --git a/core/utils.py b/core/utils.py
index 0a9672d..36661a5 100644
--- a/core/utils.py
+++ b/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
diff --git a/main.py b/main.py
index bca8d2c..c5f3de4 100644
--- a/main.py
+++ b/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
+ )