""" 测试MinIO远程服务器连接和上传功能 使用用户提供的远程服务器配置 """ import os import tempfile from minio import Minio from minio.error import S3Error from datetime import datetime, timedelta import requests # MinIO远程服务器连接配置（用户提供） MINIO_CONFIG = { 'endpoint': '10.100.31.21:9000', # 注意：去掉协议前缀 'access_key': 'minio_PC8dcY', 'secret_key': 'minio_7k7RNJ', 'secure': True # 用户指定为true，但如果连接失败可以尝试false } BUCKET_NAME = 'finyx' TENANT_ID = '615873064429507639' def print_section(title): """打印章节标题""" print("\n" + "="*60) print(f" {title}") print("="*60) def print_result(success, message): """打印测试结果""" status = "[OK]" if success else "[FAIL]" print(f"{status} {message}") def test_connection(): """测试MinIO连接""" print_section("1. 测试MinIO连接") # 先尝试secure=True（用户配置） for secure in [True, False]: try: print(f"\n尝试连接（secure={secure}）...") # 创建MinIO客户端 client = Minio( MINIO_CONFIG['endpoint'], access_key=MINIO_CONFIG['access_key'], secret_key=MINIO_CONFIG['secret_key'], secure=secure ) # 列出所有存储桶（测试连接） buckets = client.list_buckets() print_result(True, f"MinIO连接成功！") print(f"\n 连接信息:") print(f" 端点: {MINIO_CONFIG['endpoint']}") print(f" 使用HTTPS: {secure}") print(f" 访问密钥: {MINIO_CONFIG['access_key']}") print(f"\n 可用存储桶:") for bucket in buckets: print(f" - {bucket.name} (创建时间: {bucket.creation_date})") # 检查目标存储桶是否存在 bucket_exists = client.bucket_exists(BUCKET_NAME) if bucket_exists: print_result(True, f"目标存储桶 '{BUCKET_NAME}' 存在") else: print_result(False, f"目标存储桶 '{BUCKET_NAME}' 不存在") print(f" 建议：需要创建存储桶 '{BUCKET_NAME}'") # 更新配置中的secure值 MINIO_CONFIG['secure'] = secure return client, bucket_exists except Exception as e: error_msg = str(e) if secure == True: print_result(False, f"使用HTTPS连接失败: {error_msg}") print(f" 将尝试使用HTTP连接...") continue else: print_result(False, f"MinIO连接失败: {error_msg}") import traceback traceback.print_exc() return None, False return None, False def create_test_file(): """创建测试文件""" print_section("2. 创建测试文件") # 创建临时测试文件 test_content = f""" 这是一个MinIO远程服务器连接测试文件 创建时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} 服务器地址: {MINIO_CONFIG['endpoint']} 测试内容: 测试MinIO远程服务器上传和下载功能 """ # 创建临时文件 temp_file = tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False, encoding='utf-8') temp_file.write(test_content) temp_file.close() file_size = os.path.getsize(temp_file.name) print_result(True, "测试文件创建成功") print(f"\n 文件路径: {temp_file.name}") print(f" 文件大小: {file_size} 字节") return temp_file.name def upload_file(client, file_path, bucket_exists): """上传文件到MinIO""" print_section("3. 上传文件到MinIO") if not bucket_exists: print("[WARN] 存储桶不存在，尝试创建存储桶...") try: client.make_bucket(BUCKET_NAME) print_result(True, f"存储桶 '{BUCKET_NAME}' 创建成功") bucket_exists = True except Exception as e: print_result(False, f"创建存储桶失败: {str(e)}") print("[WARN] 无法创建存储桶，跳过上传测试") return None, None try: # 生成对象名称（相对路径） timestamp = datetime.now().strftime('%Y%m%d%H%M%S') object_name = f"{TENANT_ID}/TEST/{datetime.now().year}/{datetime.now().month:02d}/test_file_{timestamp}.txt" print(f"\n 上传信息:") print(f" 存储桶: {BUCKET_NAME}") print(f" 对象名称: {object_name}") print(f" 源文件: {file_path}") # 上传文件 print(f"\n 正在上传...") client.fput_object( BUCKET_NAME, object_name, file_path, content_type='text/plain; charset=utf-8' ) print_result(True, "文件上传成功！") # 生成相对路径（用于数据库存储） relative_path = f"/{object_name}" print(f"\n 相对路径（数据库存储）: {relative_path}") return object_name, relative_path except S3Error as e: print_result(False, f"MinIO上传错误: {str(e)}") import traceback traceback.print_exc() return None, None except Exception as e: print_result(False, f"上传失败: {str(e)}") import traceback traceback.print_exc() return None, None def generate_access_url(client, object_name): """生成可访问的URL""" print_section("4. 生成访问URL") if not object_name: print("[WARN] 没有上传文件，跳过URL生成") return None try: # 生成预签名URL（7天有效期） expires = timedelta(days=7) url = client.presigned_get_object( BUCKET_NAME, object_name, expires=expires ) print_result(True, "URL生成成功") print(f"\n 预签名URL（7天有效）:") print(f" {url}") # 生成公共URL（如果存储桶是公共的） protocol = "https" if MINIO_CONFIG['secure'] else "http" public_url = f"{protocol}://{MINIO_CONFIG['endpoint']}/{BUCKET_NAME}/{object_name}" print(f"\n 公共URL（如果存储桶是公共的）:") print(f" {public_url}") return url except Exception as e: print_result(False, f"URL生成失败: {str(e)}") import traceback traceback.print_exc() return None def test_download(url): """测试下载URL""" print_section("5. 测试URL下载") if not url: print("[WARN] 没有URL，跳过下载测试") return False try: print(f" 正在测试下载...") print(f" URL: {url[:100]}...") # 禁用SSL验证（如果是自签名证书） response = requests.get(url, timeout=10, verify=False) if response.status_code == 200: print_result(True, f"下载成功！状态码: {response.status_code}") print(f"\n 响应信息:") print(f" 内容长度: {len(response.content)} 字节") print(f" 内容类型: {response.headers.get('Content-Type', 'N/A')}") # 显示内容预览 try: content_preview = response.text[:200] print(f"\n 内容预览:") print(f" {content_preview}...") except: print(f" (无法显示文本预览)") return True else: print_result(False, f"下载失败！状态码: {response.status_code}") print(f" 响应内容: {response.text[:200]}") return False except requests.exceptions.Timeout: print_result(False, "下载超时") return False except requests.exceptions.SSLError as e: print_result(False, f"SSL错误: {str(e)}") print(" 提示：可能需要验证SSL证书或使用HTTP") return False except Exception as e: print_result(False, f"下载失败: {str(e)}") import traceback traceback.print_exc() return False def test_list_objects(client): """测试列出对象""" print_section("6. 测试列出存储桶中的对象") if not client: print("[WARN] 没有客户端连接，跳过列表测试") return try: # 检查存储桶是否存在 if not client.bucket_exists(BUCKET_NAME): print(f"[WARN] 存储桶 '{BUCKET_NAME}' 不存在") return print(f"\n 列出存储桶 '{BUCKET_NAME}' 中的对象（最多10个）:") objects = client.list_objects(BUCKET_NAME, recursive=True) count = 0 for obj in objects: count += 1 if count <= 10: print(f" {count}. {obj.object_name} ({obj.size} 字节, 修改时间: {obj.last_modified})") else: break if count == 0: print(" (存储桶为空)") elif count > 10: print(f" ... 还有更多对象（总共可能超过10个）") print_result(True, f"成功列出对象（显示前10个）") except Exception as e: print_result(False, f"列出对象失败: {str(e)}") import traceback traceback.print_exc() def cleanup_test_file(local_file_path): """清理测试文件""" try: if local_file_path and os.path.exists(local_file_path): os.unlink(local_file_path) print(f"\n[OK] 已清理测试文件: {local_file_path}") except Exception as e: print(f"\n[WARN] 清理测试文件失败: {e}") def main(): """主函数""" print("\n" + "="*60) print(" MinIO远程服务器连接测试") print("="*60) print(f"\n配置信息:") print(f" 端点: {MINIO_CONFIG['endpoint']}") print(f" 存储桶: {BUCKET_NAME}") print(f" 访问密钥: {MINIO_CONFIG['access_key']}") print(f" 初始secure设置: {MINIO_CONFIG['secure']}") test_file_path = None object_name = None try: # 1. 测试连接 client, bucket_exists = test_connection() if not client: print("\n[FAIL] 连接失败，无法继续测试") print("\n可能的原因:") print(" 1. 网络连接问题（无法访问 10.100.31.21:9000）") print(" 2. 防火墙阻止连接") print(" 3. MinIO服务器未运行") print(" 4. 访问密钥或密钥错误") print(" 5. SSL/TLS配置问题（尝试使用HTTP而不是HTTPS）") return # 2. 列出对象 test_list_objects(client) # 3. 创建测试文件 test_file_path = create_test_file() # 4. 上传文件 object_name, relative_path = upload_file(client, test_file_path, bucket_exists) # 5. 生成访问URL access_url = generate_access_url(client, object_name) # 6. 测试下载 download_success = test_download(access_url) # 总结 print_section("测试总结") results = { '连接': client is not None, '存储桶存在': bucket_exists, '文件上传': object_name is not None, 'URL生成': access_url is not None, 'URL下载': download_success } print("\n测试结果：") for test_name, success in results.items(): status = "[OK] 通过" if success else "[FAIL] 失败" print(f" {test_name}: {status}") passed = sum(1 for v in results.values() if v) total = len(results) print(f"\n通过率: {passed}/{total} ({passed*100//total if total > 0 else 0}%)") if passed == total: print("\n[OK] 所有测试通过！MinIO远程服务器配置正确，可以正常使用。") else: print("\n[WARN] 部分测试失败，请检查配置和网络连接。") # 显示使用建议 if object_name: print("\n" + "="*60) print(" 使用建议") print("="*60) print(f"\n上传的文件路径（用于数据库存储）:") print(f" {relative_path}") print(f"\n访问URL（预签名，7天有效）:") print(f" {access_url}") print(f"\n最终使用的连接配置:") print(f" MINIO_ENDPOINT={MINIO_CONFIG['endpoint']}") print(f" MINIO_ACCESS_KEY={MINIO_CONFIG['access_key']}") print(f" MINIO_SECRET_KEY={MINIO_CONFIG['secret_key']}") print(f" MINIO_BUCKET={BUCKET_NAME}") print(f" MINIO_SECURE={MINIO_CONFIG['secure']}") except KeyboardInterrupt: print("\n\n测试已中断") except Exception as e: print(f"\n[FAIL] 测试过程中发生错误: {e}") import traceback traceback.print_exc() finally: # 清理测试文件 if test_file_path: cleanup_test_file(test_file_path) if __name__ == '__main__': # 禁用SSL警告（如果使用自签名证书） import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) main()