diff --git a/zhaoyun-data/scripts/nas_deployment/test_nas_vpn.py b/zhaoyun-data/scripts/nas_deployment/test_nas_vpn.py new file mode 100644 index 000000000..b36f7b329 --- /dev/null +++ b/zhaoyun-data/scripts/nas_deployment/test_nas_vpn.py @@ -0,0 +1,186 @@ +#!/usr/bin/env python3 +""" +测试NAS和VPN可用性 +""" +import sys +import os +import json +import time +import subprocess +from datetime import datetime +from typing import Dict, List, Optional +import logging + +logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') +logger = logging.getLogger(__name__) + + +class NASVPNTester: + """NAS和VPN可用性测试""" + + def test_nas_connection(self, nas_path: str) -> Dict: + """测试NAS连接性""" + logger.info(f"测试NAS连接: {nas_path}") + + results = { + "timestamp": datetime.now().isoformat(), + "nas_path": nas_path, + "accessibility": { + "exists": False, + "writable": False, + "free_space": 0 + } + } + + try: + # 检查路径是否存在 + if os.path.exists(nas_path): + results["accessibility"]["exists"] = True + + # 检查可写性 + try: + test_file = os.path.join(nas_path, ".test_write") + with open(test_file, 'w') as f: + f.write("test") + os.remove(test_file) + results["accessibility"]["writable"] = True + + # 检查可用空间 + stat = os.statvfs(nas_path) + free_gb = (stat.f_bavail * stat.f_frsize) / (1024**3) + results["accessibility"]["free_space"] = free_gb + + logger.info(f"✅ NAS连接正常: {nas_path}") + logger.info(f" 可用空间: {free_gb:.2f} GB") + + if free_gb < 20: + logger.warning(f"⚠️ NAS空间不足: 可用 {free_gb:.2f} GB, 建议 30GB+") + else: + logger.info(f"✅ NAS空间充足: {free_gb:.2f} GB") + + else: + logger.warning(f"⚠️ NAS路径不存在: {nas_path}") + + except Exception as e: + logger.error(f"❌ NAS连接测试失败: {e}") + results["error"] = str(e) + + # 保存结果 + test_file = os.path.join(os.path.dirname(__file__), "nas_test_results.json") + with open(test_file, 'w') as f: + json.dump(results, f, indent=2) + + logger.info(f"NAS连接测试完成,结果保存: {test_file}") + + return results + + def test_vpn_connectivity(self) -> Dict: + """测试VPN连接性""" + logger.info("测试VPN连接性") + + vpn_results = { + "timestamp": datetime.now().isoformat(), + "vpn_status": "unknown", + "connection_tests": [] + } + + # 测试目标 + test_targets = [ + {"name": "akshare", "host": "push2his.eastmoney.com"}, + {"name": "tushare", "host": "tushare.org"}, + {"name": "github", "host": "github.com"}, + {"name": "baidu", "host": "baidu.com"} + ] + + for target in test_targets: + logger.info(f"测试VPN连接: {target['name']} ({target['host']})") + + try: + # 使用curl测试连接 + result = subprocess.run( + ["curl", "-I", "-m", "10", f"http://{target['host']}"], + capture_output=True, + text=True, + timeout=15 + ) + + test_result = { + "target": target["name"], + "host": target["host"], + "status": "success" if result.returncode == 0 else "failed", + "response_time": time.time() + } + + if result.returncode == 0: + logger.info(f" ✅ VPN连接正常: {target['name']}") + else: + logger.warning(f" ⚠️ VPN连接失败: {target['name']}") + + vpn_results["connection_tests"].append(test_result) + + except Exception as e: + logger.error(f"❌ VPN连接测试失败: {target['name']} - {e}") + + # 评估VPN状态 + success_count = sum(1 for test in vpn_results["connection_tests"] + if test.get("status") == "success") + total_tests = len(vpn_results["connection_tests"]) + + if success_count == total_tests: + vpn_results["vpn_status"] = "excellent" + vpn_results["summary"] = "VPN连接性优秀,所有目标可访问" + logger.info("VPN连接性: 优秀") + elif success_count >= 2: + vpn_results["vpn_status"] = "good" + vpn_results["summary"] = "VPN连接性良好,主要目标可访问" + logger.info("VPN连接性: 良好") + else: + vpn_results["vpn_status"] = "poor" + vpn_results["summary"] = "VPN连接性较差,需要优化配置" + logger.warning("VPN连接性: 较差") + + # 保存结果 + vpn_file = os.path.join(os.path.dirname(__file__), "vpn_test_results.json") + with open(vpn_file, 'w') as f: + json.dump(vpn_results, f, indent=2) + + logger.info(f"VPN连接性测试完成,结果保存: {vpn_file}") + + return vpn_results + + +def main(): + """主函数""" + logger.info("开始测试NAS和VPN可用性") + + tester = NASVPNTester() + + # 检查VPN连接性 + logger.info("1. 测试VPN连接性...") + vpn_results = tester.test_vpn_connectivity() + + # 测试AKShare数据源 + logger.info("2. 测试AKShare数据源可用性...") + akshare_results = tester.test_akshare_with_vpn() + + logger.info("\n" + "="*70) + logger.info("📋 测试结果摘要") + logger.info("="*70) + + logger.info("✅ VPN连接性:") + for test in vpn_results.get("connection_tests", []): + status = "✅" if test.get("status") == "success" else "⚠️" + logger.info(f" {status} {test.get('target')}: {test.get('status')}") + + logger.info("\n📊 最终评估:") + logger.info(f" VPN状态: {vpn_results.get('vpn_status', 'unknown')}") + logger.info(f" 总结: {vpn_results.get('summary', '无评估')}") + + logger.info("\n" + "="*70) + logger.info("🎯 下一步:等待姜维将军提供NAS配置信息") + logger.info("🚀 配置获取后立即开始分钟数据下载") + logger.info("="*70) + + +if __name__ == "__main__": + main() \ No newline at end of file