请检查static/curl-guide.html文件
", status_code=404) @app.post("/api/upload", response_model=ShareResponse) async def upload_file(file: UploadFile = File(...)): """上传文件并生成分享口令""" # 检查文件大小 if file.size and file.size > MAX_FILE_SIZE: raise HTTPException(status_code=413, detail=f"文件太大，最大支持 {MAX_FILE_SIZE // 1024 // 1024}MB") # 生成分享码和文件路径 share_code = generate_share_code() file_extension = Path(file.filename or "").suffix safe_filename = f"{share_code}{file_extension}" file_path = UPLOAD_DIR / safe_filename try: # 保存文件 content = await file.read() # 再次检查文件大小 if len(content) > MAX_FILE_SIZE: raise HTTPException(status_code=413, detail=f"文件太大，最大支持 {MAX_FILE_SIZE // 1024 // 1024}MB") with open(file_path, "wb") as f: f.write(content) # 保存分享信息 now = datetime.now() expires_at = now + timedelta(minutes=EXPIRE_MINUTES) shares[share_code] = { "filename": file.filename or "unknown", "file_path": str(file_path), "file_type": file.content_type or "application/octet-stream", "size": len(content), "created_at": now, "expires_at": expires_at, "is_text": False } return ShareResponse( code=share_code, expires_at=expires_at.isoformat(), download_url=f"/api/download/{share_code}" ) except Exception as e: # 清理可能创建的文件 if file_path.exists(): file_path.unlink() raise HTTPException(status_code=500, detail=f"上传失败: {str(e)}") @app.post("/api/share-text", response_model=ShareResponse) async def share_text(text_share: TextShare): """分享文本内容（JSON格式）""" return await _create_text_share(text_share.content, text_share.filename) @app.post("/api/text", response_model=ShareResponse) async def share_text_simple(request: Request): """分享文本内容（简化版 - 直接发送文本）""" try: # 读取请求体作为纯文本 content = (await request.body()).decode('utf-8') if not content.strip(): raise HTTPException(status_code=400, detail="文本内容不能为空") return await _create_text_share(content, "shared_text.txt") except UnicodeDecodeError: raise HTTPException(status_code=400, detail="文本编码错误，请使用UTF-8编码") except Exception as e: raise HTTPException(status_code=500, detail=f"分享失败: {str(e)}") @app.post("/api/share-text-form", response_model=ShareResponse) async def share_text_form(content: str = Form(...), filename: str = Form("shared_text.txt")): """分享文本内容（表单格式）""" if not content or not content.strip(): raise HTTPException(status_code=400, detail="文本内容不能为空") return await _create_text_share(content, filename) async def _create_text_share(content: str, filename: str) -> ShareResponse: """创建文本分享的通用函数""" if not content.strip(): raise HTTPException(status_code=400, detail="文本内容不能为空") # 生成分享码 share_code = generate_share_code() file_path = UPLOAD_DIR / f"{share_code}.txt" try: # 保存文本到文件 with open(file_path, "w", encoding="utf-8") as f: f.write(content) # 保存分享信息 now = datetime.now() expires_at = now + timedelta(minutes=EXPIRE_MINUTES) shares[share_code] = { "filename": filename, "file_path": str(file_path), "file_type": "text/plain", "size": len(content.encode("utf-8")), "created_at": now, "expires_at": expires_at, "is_text": True, "content": content } return ShareResponse( code=share_code, expires_at=expires_at.isoformat(), download_url=f"/api/download/{share_code}" ) except Exception as e: # 清理可能创建的文件 if file_path.exists(): file_path.unlink() raise HTTPException(status_code=500, detail=f"分享失败: {str(e)}") @app.get("/api/download/{code}") async def download_file(code: str): """通过分享码下载文件""" share_info = shares.get(code) if not share_info: raise HTTPException(status_code=404, detail="分享码不存在") if is_expired(share_info["expires_at"]): # 清理过期文件 cleanup_expired_files() raise HTTPException(status_code=410, detail="分享已过期") file_path = Path(share_info["file_path"]) if not file_path.exists(): raise HTTPException(status_code=404, detail="文件不存在") # 获取MIME类型 mime_type, _ = mimetypes.guess_type(str(file_path)) if not mime_type: mime_type = share_info["file_type"] return FileResponse( path=file_path, filename=share_info["filename"], media_type=mime_type ) @app.get("/api/info/{code}", response_model=ShareInfo) async def get_share_info(code: str): """获取分享信息""" share_info = shares.get(code) if not share_info: raise HTTPException(status_code=404, detail="分享码不存在") return ShareInfo( code=code, filename=share_info["filename"], file_type=share_info["file_type"], size=share_info["size"], created_at=share_info["created_at"].isoformat(), expires_at=share_info["expires_at"].isoformat(), is_expired=is_expired(share_info["expires_at"]) ) @app.get("/api/shares") async def list_shares(): """列出所有分享（管理用）""" result = [] current_time = datetime.now() for code, share_info in shares.items(): result.append({ "code": code, "filename": share_info["filename"], "file_type": share_info["file_type"], "size": share_info["size"], "created_at": share_info["created_at"].isoformat(), "expires_at": share_info["expires_at"].isoformat(), "is_expired": is_expired(share_info["expires_at"]), "remaining_minutes": max(0, int((share_info["expires_at"] - current_time).total_seconds() / 60)) }) return { "total": len(result), "shares": result } @app.delete("/api/shares/{code}") async def delete_share(code: str): """删除分享""" share_info = shares.get(code) if not share_info: raise HTTPException(status_code=404, detail="分享码不存在") # 删除文件 file_path = Path(share_info["file_path"]) if file_path.exists(): try: file_path.unlink() except Exception as e: print(f"删除文件失败: {e}") # 删除记录 del shares[code] return {"message": "删除成功", "code": code} @app.post("/api/cleanup") async def manual_cleanup(): """手动清理过期文件""" before_count = len(shares) cleanup_expired_files() after_count = len(shares) return { "message": "清理完成", "removed": before_count - after_count, "remaining": after_count } if __name__ == "__main__": print(f"启动文件传输服务...") print(f"地址: http://{HOST}:{PORT}") print(f"上传目录: {UPLOAD_DIR.absolute()}") print(f"最大文件大小: {MAX_FILE_SIZE // 1024 // 1024}MB") print(f"过期时间: {EXPIRE_MINUTES}分钟") uvicorn.run( "app:app", host=HOST, port=PORT, reload=True, log_level="info" )