From a98b39196bb1ddb6f595f84ddc22d2f15e36d006 Mon Sep 17 00:00:00 2001 From: cfdaily Date: Thu, 26 Mar 2026 20:27:55 +0800 Subject: [PATCH] auto-sync: 2026-03-26 20:27:55 --- .../nas_deployment/nas_config_check.py | 220 ++++++++++++++++++ 1 file changed, 220 insertions(+) create mode 100644 zhaoyun-data/scripts/nas_deployment/nas_config_check.py diff --git a/zhaoyun-data/scripts/nas_deployment/nas_config_check.py b/zhaoyun-data/scripts/nas_deployment/nas_config_check.py new file mode 100644 index 000000000..23a7b4515 --- /dev/null +++ b/zhaoyun-data/scripts/nas_deployment/nas_config_check.py @@ -0,0 +1,220 @@ +#!/usr/bin/env python3 +""" +NAS配置验证脚本 +验证姜维将军提供的NAS存储配置和VPN可用性 +""" +import os +import sys +import json +import time +import subprocess +from datetime import datetime +from typing import Dict, Optional, Tuple +import logging +import warnings + +warnings.filterwarnings('ignore') + +# 配置日志 +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' +) +logger = logging.getLogger(__name__) + + +class NASConfigValidator: + """NAS配置验证器""" + + def __init__(self): + """初始化验证器""" + logger.info("NAS配置验证器初始化") + + # 配置目录 + self.config_dir = "/Users/chufeng/.openclaw/sanguo_projects/sanguo_quant_live/zhaoyun-data/data/running_data/nas_config" + os.makedirs(self.config_dir, exist_ok=True) + + # 验证结果 + self.validation_results = { + "timestamp": datetime.now().isoformat(), + "nas_config": {}, + "validation_results": {}, + "recommendations": [] + } + + # 期望的NAS配置 + self.expected_config = { + "storage_path": "NAS存储路径（如：/mnt/nas/minute_kline）", + "access_type": "访问方式（SMB/NFS/FTP/SSH）", + "network": { + "ip_address": "NAS IP地址", + "port": "端口号", + "credentials": { + "username": "用户名", + "password": "密码" + } + }, + "vpn_config": { + "enabled": True, + "vpn_type": "VPN类型", + "config_file": "VPN配置文件路径", + "credentials": "VPN凭证" + }, + "performance": { + "max_concurrent": 10, + "bandwidth_limit": "无", + "storage_capacity": "30GB+" + } + } + + def create_config_template(self) -> str: + """创建NAS配置模板""" + logger.info("创建NAS配置模板") + + config_template = { + "nas_configuration_required": { + "timestamp": datetime.now().isoformat(), + "requested_by": "赵云（数据工程将军）", + "purpose": "A股分钟K线数据存储", + "data_requirements": { + "total_records": "约2.04亿条", + "storage_needed": "15.5-19GB", + "time_period": "2021年至今（5年历史）", + "data_granularity": "1分钟、5分钟、15分钟K线" + }, + "required_configuration": self.expected_config, + "deployment_timeline": { + "config_validation": "立即", + "vpn_testing": "2小时", + "data_download": "11-15天", + "full_deployment": "2-3周" + } + } + } + + template_file = os.path.join(self.config_dir, "nas_config_template.json") + with open(template_file, 'w', encoding='utf-8') as f: + json.dump(config_template, f, ensure_ascii=False, indent=2) + + logger.info(f"NAS配置模板已保存: {template_file}") + + return template_file + + def check_vpn_connectivity(self) -> Dict: + """检查VPN连接性""" + logger.info("检查VPN连接性") + + vpn_results = { + "timestamp": datetime.now().isoformat(), + "test_results": {}, + "recommendations": [] + } + + # 测试网络连接性 + test_targets = [ + {"name": "baidu", "url": "www.baidu.com", "description": "国内网站"}, + {"name": "github", "url": "github.com", "description": "国际网站"}, + {"name": "akshare_source", "url": "push2his.eastmoney.com", "description": "数据源网站"} + ] + + for target in test_targets: + logger.info(f"测试连接: {target['name']} ({target['url']})") + + try: + # 使用ping测试基本连接 + result = subprocess.run( + ["ping", "-c", "3", "-W", "2", target["url"]], + capture_output=True, + text=True, + timeout=10 + ) + + if result.returncode == 0: + vpn_results["test_results"][target["name"]] = { + "status": "connected", + "latency": "正常", + "description": target["description"] + } + logger.info(f" ✅ {target['name']}: 连接正常") + else: + vpn_results["test_results"][target["name"]] = { + "status": "failed", + "error": "ping失败", + "description": target["description"] + } + vpn_results["recommendations"].append(f"检查{target['name']}连接性") + logger.warning(f" ⚠️ {target['name']}: 连接失败") + + except subprocess.TimeoutExpired: + vpn_results["test_results"][target["name"]] = { + "status": "timeout", + "error": "连接超时", + "description": target["description"] + } + vpn_results["recommendations"].append(f"{target['name']}连接超时，检查网络配置") + logger.warning(f" ⚠️ {target['name']}: 连接超时") + + except Exception as e: + vpn_results["test_results"][target["name"]] = { + "status": "error", + "error": str(e), + "description": target["description"] + } + logger.error(f" ❌ {target['name']}: 测试异常 - {e}") + + # VPN连接性评估 + connected_count = sum(1 for result in vpn_results["test_results"].values() + if result.get("status") == "connected") + total_count = len(vpn_results["test_results"]) + + if connected_count == total_count: + vpn_results["overall_status"] = "excellent" + vpn_results["summary"] = "所有目标连接正常，VPN可用性优秀" + logger.info("VPN连接性: 优秀") + elif connected_count >= 2: + vpn_results["overall_status"] = "good" + vpn_results["summary"] = "主要目标连接正常，VPN可用性良好" + logger.info("VPN连接性: 良好") + else: + vpn_results["overall_status"] = "poor" + vpn_results["summary"] = "连接性较差，需要优化VPN配置" + logger.warning("VPN连接性: 较差") + + # 保存VPN测试结果 + vpn_file = os.path.join(self.config_dir, "vpn_connectivity_test.json") + with open(vpn_file, 'w', encoding='utf-8') as f: + json.dump(vpn_results, f, ensure_ascii=False, indent=2) + + logger.info(f"VPN连接性测试结果已保存: {vpn_file}") + + self.validation_results["vpn_test"] = vpn_results + + return vpn_results + + def test_akshare_with_vpn(self) -> Dict: + """在VPN环境下测试AKShare数据源""" + logger.info("测试VPN环境下的AKShare数据源") + + test_results = { + "timestamp": datetime.now().isoformat(), + "akshare_tests": {}, + "data_availability": {} + } + + try: + import akshare as ak + + # 测试1分钟数据 + logger.info("测试1分钟数据获取...") + try: + data_1min = ak.stock_zh_a_minute( + symbol='sh000001', + period='1', + adjust='hfq' + ) + + if data_1min is not None and not data_1min.empty: + test_results["akshare_tests"]["1min"] = { + "status": "success", + "record_count": len(data_1min), + "date_range": f"{data \ No newline at end of file