orion/filecodebox
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

style: code format

Browse Source
This commit is contained in:
Lan
2025-02-15 22:00:53 +08:00
parent 7b0ea804cf
commit 10d136b0c5
14 changed files with 404 additions and 311 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/admin/dependencies.py
+17 -14
View File
@@ -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
+1 -1
View File
@@ -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
+39 -22
View File
@@ -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
+65 -62
View File
@@ -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)
apps/base/dependencies.py
+16 -8
View File
@@ -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
+33 -15
View File
@@ -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
+30 -20
View File
@@ -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),
}
apps/base/views.py
+52 -40
View File
@@ -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)