from fastapi import FastAPI, File, UploadFile, HTTPException
from fastapi.responses import JSONResponse, FileResponse, HTMLResponse
import logging
import os
from datetime import datetime
import uvicorn
# 初始化日志
logging.basicConfig(filename='file_server.log', level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
script_dir = os.path.dirname(os.path.abspath(__file__))
pro_dir = os.path.abspath(os.path.join(script_dir, '..'))
dir_upload = os.path.abspath(os.path.join(pro_dir, 'files'))
app = FastAPI()
@app.post("/upload_file/")
async def upload_file(file: UploadFile = File(...)):
"""
接收上传的文件,保存到服务器,并返回文件信息。
"""
# 验证文件格式
if file.content_type != "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet":
raise HTTPException(status_code=400, detail="上传文件格式不符合要求")
try:
# 读取文件的二进制内容
file_content = await file.read()
# 调用save_upload_file函数保存文件
file_info = await save_upload_file(dir_upload, file_content)
if file_info is not None:
# 如果文件保存成功,则返回文件信息
return JSONResponse(content=file_info)
else:
# 文件保存失败,返回错误信息
return JSONResponse(content={"message": "文件保存失败"}, status_code=500)
except Exception as e:
logger.error("文件上传错误: %s", str(e))
return JSONResponse(content={"message": "文件上传失败"}, status_code=500)
# 定义并实现文件保存函数
async def save_upload_file(dir_upload, file_content) -> dict:
try:
# 确保上传目录存在
if not os.path.exists(dir_upload):
os.makedirs(dir_upload)
# 使用当前时间戳生成文件名
timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
file_name = f"{timestamp}.xlsx"
file_path = os.path.join(dir_upload, file_name)
# 保存文件到指定路径
with open(file_path, "wb") as f:
f.write(file_content)
file_size = os.path.getsize(file_path)
return {
"file_name": file_name,
"file_path": file_path,
"file_size": file_size
}
except Exception as e:
logger.error("文件保存失败: %s", str(e))
return None
@app.get("/get_file/", summary="get file", tags=['文件'])
async def get_file(file_name: str):
"""
根据文件名提供文件的 HTTP 服务
"""
try:
file_path = os.path.join(dir_upload, file_name)
if not os.path.exists(file_path):
return JSONResponse(content={"message": "文件未找到"}, status_code=404)
return FileResponse(file_path, media_type="application/octet-stream", filename=file_name)
except Exception as e:
logger.error("文件下载错误: %s", str(e))
return JSONResponse(content={"message": str(e)}, status_code=500)
@app.get("/list_files/", response_class=HTMLResponse)
async def list_files():
"""
提供文件列表的 HTML 页面和文件上传功能
"""
try:
files = sorted(
(f for f in os.listdir(dir_upload) if os.path.isfile(os.path.join(dir_upload, f))),
key=lambda f: os.path.getmtime(os.path.join(dir_upload, f)),
reverse=True
)
if not files:
files_html = "<h2>没有可下载的文件</h2>"
else:
file_links = [f'<li><a href="/get_file/?file_name={file}">{file}</a></li>' for file in files]
files_html = f"<ul>{''.join(file_links)}</ul>"
html_content = f"""
<html>
<head>
<title>文件列表和上传</title>
<script>
async function uploadFile(event) {{
event.preventDefault();
const formData = new FormData();
const fileField = document.querySelector('input[type="file"]');
formData.append('file', fileField.files[0]);
const response = await fetch('/upload_file/', {{
method: 'POST',
body: formData
}});
const result = await response.json();
alert(result.message || '文件上传成功');
window.location.reload();
}}
</script>
</head>
<body>
<h1>文件列表和上传</h1>
<form onsubmit="uploadFile(event)">
<input type="file" name="file" accept=".xlsx" required>
<button type="submit">上传文件</button>
</form>
<h2>文件列表</h2>
{files_html}
</body>
</html>
"""
return HTMLResponse(content=html_content)
except Exception as e:
logger.error("文件列表生成错误: %s", str(e))
return HTMLResponse(content=f"<h2>错误: {str(e)}</h2>", status_code=500)
if __name__ == "__main__":
## 线上模式
# uvicorn.run("file_server:app", host="0.0.0.0", port=1300)
## debug 模式
uvicorn.run("file_server:app", host="0.0.0.0", port=1300, reload=True)
运行上述代码,打开http://127.0.0.1:1300/docs#/
接下来配置nginx文件
user _www;
worker_processes auto;
error_log /opt/homebrew/var/log/nginx/error.log;
pid /System/Volumes/Data/opt/homebrew/var/run/nginx.pid;
# Events
events {
worker_connections 1024;
}
http {
include /opt/homebrew/etc/nginx/mime.types;
default_type application/octet-stream;
log_format main '$remote_addr - $remote_user [$time_local] "$request" '
'$status $body_bytes_sent "$http_referer" '
'"$http_user_agent" "$http_x_forwarded_for"';
access_log /opt/homebrew/var/log/nginx/access.log main;
sendfile on;
keepalive_timeout 65;
types_hash_max_size 2048;
gzip on;
server {
listen 81;
server_name localhost;
location /file/ {
proxy_pass http://127.0.0.1:1300/upload_file/;
}
location /get_file/ {
proxy_pass http://127.0.0.1:1300/get_file/;
}
location /list_files/ {
proxy_pass http://127.0.0.1:1300/list_files/;
}
}
}
接下来:
访问http://127.0.0.1:81/list_files/ http://127.0.0.1:1300/list_files/ 均可以打开对应网站