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

fix: 修复文本取件问题

Browse Source
This commit is contained in:
lan
2024-05-06 13:02:00 +08:00
parent f4abf130d2
commit 8215ef9e4b
37 changed files with 27 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/__init__.py
+4
View File
@@ -0,0 +1,4 @@
# @Time : 2023/8/13 20:43
# @Author : Lan
# @File : __init__.py.py
# @Software: PyCharm
apps/admin/__init__.py
+4
View File
@@ -0,0 +1,4 @@
# @Time : 2023/8/14 14:38
# @Author : Lan
# @File : __init__.py.py
# @Software: PyCharm
apps/admin/depends.py
+19
View File
@@ -0,0 +1,19 @@
# @Time : 2023/8/15 17:43
# @Author : Lan
# @File : depends.py
# @Software: PyCharm
from typing import Union
from fastapi import Header, HTTPException
from fastapi.requests import Request
from core.settings import settings
async def admin_required(authorization: Union[str, None] = Header(default=None), request: Request = None):
is_admin = authorization == str(settings.admin_token)
if request.url.path.startswith('/share/'):
if not settings.openUpload and not is_admin:
raise HTTPException(status_code=403, detail='本站未开启游客上传，如需上传请先登录后台')
else:
if not is_admin:
raise HTTPException(status_code=401, detail='未授权或授权校验失败')
apps/admin/pydantics.py
+5
View File
@@ -0,0 +1,5 @@
from pydantic import BaseModel
class IDData(BaseModel):
id: int
apps/admin/views.py
+82
View File
@@ -0,0 +1,82 @@
# @Time : 2023/8/14 14:38
# @Author : Lan
# @File : views.py
# @Software: PyCharm
import math
from fastapi import APIRouter, Depends
from apps.admin.depends import admin_required
from apps.admin.pydantics import IDData
from apps.base.models import FileCodes
from core.response import APIResponse
from core.settings import settings
from core.storage import file_storage
admin_api = APIRouter(
prefix='/admin',
tags=['管理'],
)
@admin_api.post('/login', dependencies=[Depends(admin_required)])
async def login():
return APIResponse()
@admin_api.delete('/file/delete', dependencies=[Depends(admin_required)])
async def file_delete(data: IDData):
file_code = await FileCodes.get(id=data.id)
await file_storage.delete_file(file_code)
await file_code.delete()
return APIResponse()
@admin_api.get('/file/list', dependencies=[Depends(admin_required)])
async def file_list(page: float = 1, size: int = 10):
return APIResponse(detail={
'page': page,
'size': size,
'data': await FileCodes.all().limit(size).offset((math.ceil(page) - 1) * size),
'total': await FileCodes.all().count(),
})
@admin_api.get('/config/get', dependencies=[Depends(admin_required)])
async def get_config():
return APIResponse(detail=settings.__dict__)
@admin_api.patch('/config/update', dependencies=[Depends(admin_required)])
async def update_config(data: dict):
admin_token = data.get('admin_token')
if admin_token is None or admin_token == '':
return APIResponse(code=400, detail='管理员密码不能为空')
for k, v in data.items():
settings.__setattr__(k, v)
return APIResponse()
# 根据code获取文件
async def get_file_by_id(id):
# 查询文件
file_code = await FileCodes.filter(id=id).first()
# 检查文件是否存在
if not file_code:
return False, '文件不存在'
return True, file_code
@admin_api.get('/file/download', dependencies=[Depends(admin_required)])
async def file_download(id: int):
has, file_code = await get_file_by_id(id)
# 检查文件是否存在
if not has:
# 返回API响应
return APIResponse(code=404, detail='文件不存在')
# 如果文件是文本，返回文本内容，否则返回文件响应
if file_code.text:
return APIResponse(detail=file_code.text)
else:
return await file_storage.get_file_response(file_code)
apps/base/__init__.py
+4
View File
@@ -0,0 +1,4 @@
# @Time : 2023/8/13 20:43
# @Author : Lan
# @File : __init__.py.py
# @Software: PyCharm
apps/base/depends.py
+45
View File
@@ -0,0 +1,45 @@
# @Time : 2023/8/14 12:20
# @Author : Lan
# @File : depends.py
# @Software: PyCharm
from typing import Union
from datetime import datetime, timedelta
from fastapi import Header, HTTPException, Request
from core.response import APIResponse
class IPRateLimit:
def __init__(self, count, minutes):
self.ips = {}
self.count = count
self.minutes = minutes
def check_ip(self, ip):
# 检查ip是否被禁止
if ip in self.ips:
if self.ips[ip]['count'] >= self.count:
if self.ips[ip]['time'] + timedelta(minutes=self.minutes) > datetime.now():
return False
else:
self.ips.pop(ip)
return True
def add_ip(self, ip):
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']
async def remove_expired_ip(self):
for ip in list(self.ips.keys()):
if self.ips[ip]['time'] + timedelta(minutes=self.minutes) < datetime.now():
self.ips.pop(ip)
def __call__(self, request: Request):
ip = request.headers.get('X-Real-IP', request.headers.get('X-Forwarded-For', request.client.host))
if not self.check_ip(ip):
raise HTTPException(status_code=423, detail=f"请求次数过多，请稍后再试")
return ip
apps/base/models.py
+44
View File
@@ -0,0 +1,44 @@
# @Time : 2023/8/13 20:43
# @Author : Lan
# @File : models.py
# @Software: PyCharm
from datetime import datetime
from typing import Optional
from tortoise import fields
from tortoise.models import Model
from tortoise.contrib.pydantic import pydantic_model_creator
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='创建时间')
async def is_expired(self):
# 按时间
if self.expired_at is None:
return False
if self.expired_at and self.expired_count < 0:
return self.expired_at < await get_now()
# 按次数
else:
return self.expired_count <= 0
async def get_file_path(self):
return f"{self.file_path}/{self.uuid_file_name}"
file_codes_pydantic = pydantic_model_creator(FileCodes, name='FileCodes')
apps/base/pydantics.py
+5
View File
@@ -0,0 +1,5 @@
from pydantic import BaseModel
class SelectFileModel(BaseModel):
code: str
apps/base/utils.py
+92
View File
@@ -0,0 +1,92 @@
# @Time : 2023/8/14 01:10
# @Author : Lan
# @File : utils.py
# @Software: PyCharm
import datetime
import uuid
import os
from fastapi import UploadFile, HTTPException
from apps.base.depends import IPRateLimit
from apps.base.models import FileCodes
from core.settings import settings
from core.utils import get_random_num, get_random_string, max_save_times_desc
async def get_file_path_name(file: UploadFile):
"""
获取文件路径和文件名
:param file:
:return: {
'path': 'share/data/2021/08/13',
'suffix': '.jpg',
'prefix': 'test',
'file_uuid': '44a83bbd70e04c8aa7fd93bfd8c88249',
'uuid_file_name': '44a83bbd70e04c8aa7fd93bfd8c88249.jpg',
'save_path': 'share/data/2021/08/13/44a83bbd70e04c8aa7fd93bfd8c88249.jpg'
}
"""
today = datetime.datetime.now()
path = f"share/data/{today.strftime('%Y/%m/%d')}"
prefix, suffix = os.path.splitext(file.filename)
file_uuid = f"{uuid.uuid4().hex}"
uuid_file_name = f"{file_uuid}{suffix}"
save_path = f"{path}/{uuid_file_name}"
return path, suffix, prefix, uuid_file_name, save_path
async def get_expire_info(expire_value: int, expire_style: str):
"""
获取过期信息
:param expire_value:
:param expire_style:
:return: expired_at 过期时间, expired_count 可用次数, used_count 已用次数, code 随机码
"""
expired_count, used_count, now, code = -1, 0, datetime.datetime.now(), None
if settings.max_save_seconds > 0:
max_timedelta = datetime.timedelta(seconds=settings.max_save_seconds)
detail = await max_save_times_desc(settings.max_save_seconds)
detail = f'保存时间超过限制，{detail[0]}'
else:
max_timedelta = datetime.timedelta(days=7)
detail = '保存时间超过限制，最长保存时间：7天'
if expire_style == 'day':
if datetime.timedelta(days=expire_value) > max_timedelta:
raise HTTPException(status_code=403, detail=detail)
expired_at = now + datetime.timedelta(days=expire_value)
elif expire_style == 'hour':
if datetime.timedelta(hours=expire_value) > max_timedelta:
raise HTTPException(status_code=403, detail=detail)
expired_at = now + datetime.timedelta(hours=expire_value)
elif expire_style == 'minute':
if datetime.timedelta(minutes=expire_value) > max_timedelta:
raise HTTPException(status_code=403, detail=detail)
expired_at = now + datetime.timedelta(minutes=expire_value)
elif expire_style == 'count':
expired_at = now + datetime.timedelta(days=1)
expired_count = expire_value
elif expire_style == 'forever':
expired_at = None
code = await get_random_code(style='string')
else:
expired_at = now + datetime.timedelta(days=1)
if not code:
code = await get_random_code()
return expired_at, expired_count, used_count, code
async def get_random_code(style='num'):
"""
获取随机字符串
:return:
"""
while True:
code = await get_random_num() if style == 'num' else await get_random_string()
if not await FileCodes.filter(code=code).exists():
return code
# 错误IP限制器
error_ip_limit = IPRateLimit(count=settings.errorCount, minutes=settings.errorMinute)
# 上传文件限制器
upload_ip_limit = IPRateLimit(count=settings.uploadCount, minutes=settings.errorMinute)
apps/base/views.py
+157
View File
@@ -0,0 +1,157 @@
# @Time : 2023/8/14 03:59
# @Author : Lan
# @File : views.py
# @Software: PyCharm
# 导入所需的库和模块
from fastapi import APIRouter, Form, UploadFile, File, Depends, HTTPException
from apps.admin.depends import admin_required
from apps.base.models import FileCodes
from apps.base.pydantics import SelectFileModel
from apps.base.utils import get_expire_info, get_file_path_name, error_ip_limit, upload_ip_limit
from core.response import APIResponse
from core.settings import settings
from core.storage import file_storage
from core.utils import get_select_token
# 创建一个API路由
share_api = APIRouter(
prefix='/share', # 路由前缀
tags=['分享'], # 标签
)
# 分享文本的API
@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(upload_ip_limit)):
# 获取过期信息
expired_at, expired_count, used_count, code = await get_expire_info(expire_value, expire_style)
# 创建一个新的FileCodes实例
await FileCodes.create(
code=code,
text=text,
expired_at=expired_at,
expired_count=expired_count,
used_count=used_count,
size=len(text),
prefix='文本分享'
)
# 添加IP到限制列表
upload_ip_limit.add_ip(ip)
# 返回API响应
return APIResponse(detail={
'code': code,
})
# 分享文件的API
@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(upload_ip_limit)):
# 检查文件大小是否超过限制
if file.size > settings.uploadSize:
raise HTTPException(status_code=403, detail=f'文件大小超过限制，最大为{settings.uploadSize}字节')
# 获取过期信息
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)
# 保存文件
await file_storage.save_file(file, save_path)
# 创建一个新的FileCodes实例
await FileCodes.create(
code=code,
prefix=prefix,
suffix=suffix,
uuid_file_name=uuid_file_name,
file_path=path,
size=file.size,
expired_at=expired_at,
expired_count=expired_count,
used_count=used_count,
)
# 添加IP到限制列表
upload_ip_limit.add_ip(ip)
# 返回API响应
return APIResponse(detail={
'code': code,
'name': file.filename,
})
# 根据code获取文件
async def get_code_file_by_code(code, check=True):
# 查询文件
file_code = await FileCodes.filter(code=code).first()
# 检查文件是否存在
if not file_code:
return False, '文件不存在'
# 检查文件是否过期
if await file_code.is_expired() and check:
return False, '文件已过期',
return True, file_code
# 获取文件的API
@share_api.get('/select/')
async def get_code_file(code: str, ip: str = Depends(error_ip_limit)):
# 获取文件
has, file_code = await get_code_file_by_code(code)
# 检查文件是否存在
if not has:
# 添加IP到限制列表
error_ip_limit.add_ip(ip)
# 返回API响应
return APIResponse(code=404, detail=file_code)
# 更新文件的使用次数和过期次数
file_code.used_count += 1
if file_code.expired_count > 0:
file_code.expired_count -= 1
# 保存文件
await file_code.save()
# 返回文件响应
return await file_storage.get_file_response(file_code)
# 选择文件的API
@share_api.post('/select/')
async def select_file(data: SelectFileModel, ip: str = Depends(error_ip_limit)):
# 获取文件
has, file_code = await get_code_file_by_code(data.code)
# 检查文件是否存在
if not has:
# 添加IP到限制列表
error_ip_limit.add_ip(ip)
# 返回API响应
return APIResponse(code=404, detail=file_code)
# 更新文件的使用次数和过期次数
file_code.used_count += 1
if file_code.expired_count > 0:
file_code.expired_count -= 1
# 保存文件
await file_code.save()
# 返回API响应
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),
})
# 下载文件的API
@share_api.get('/download')
async def download_file(key: str, code: str, ip: str = Depends(error_ip_limit)):
# 检查token是否有效
is_valid = await get_select_token(code) == key
if not is_valid:
# 添加IP到限制列表
error_ip_limit.add_ip(ip)
# 获取文件
has, file_code = await get_code_file_by_code(code, False)
# 检查文件是否存在
if not has:
# 返回API响应
return APIResponse(code=404, detail='文件不存在')
# 如果文件是文本，返回文本内容，否则返回文件响应
if file_code.text:
return APIResponse(detail=file_code.text)
else:
return await file_storage.get_file_response(file_code)