auto-sync: 2026-03-26 20:27:55

2026-03-26 20:27:55 +08:00
parent 6d65fd8632
commit a98b39196b
1 changed files with 220 additions and 0 deletions
@@ -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