97 lines
3.4 KiB
Python
97 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
测试井数据管理器功能的脚本
|
|
"""
|
|
|
|
import sys
|
|
import logging
|
|
from ddms_compliance_suite.utils.well_data_manager import WellDataManager
|
|
|
|
def test_well_data_manager():
|
|
"""测试井数据管理器的基本功能"""
|
|
|
|
# 设置日志
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# 初始化井数据管理器
|
|
base_url = "https://www.dev.ideas.cnpc"
|
|
well_manager = WellDataManager(
|
|
base_url=base_url,
|
|
ignore_ssl=True,
|
|
logger=logger
|
|
)
|
|
|
|
logger.info("开始测试井数据管理器...")
|
|
|
|
# 测试初始化井数据
|
|
logger.info("1. 测试井数据初始化...")
|
|
success = well_manager.initialize_well_data()
|
|
|
|
if success:
|
|
logger.info("✅ 井数据初始化成功")
|
|
|
|
# 获取数据摘要
|
|
summary = well_manager.get_well_data_summary()
|
|
logger.info(f"📊 井数据摘要: {summary}")
|
|
|
|
# 测试获取随机井数据
|
|
logger.info("2. 测试获取随机井数据...")
|
|
well_data = well_manager.get_random_well_data()
|
|
if well_data:
|
|
logger.info(f"✅ 获取到井数据: wellId={well_data.get('wellId')}, wellCommonName={well_data.get('wellCommonName')}")
|
|
else:
|
|
logger.warning("❌ 未获取到井数据")
|
|
|
|
# 测试获取随机井筒数据
|
|
logger.info("3. 测试获取随机井筒数据...")
|
|
wellbore_data = well_manager.get_random_wellbore_data()
|
|
if wellbore_data:
|
|
logger.info(f"✅ 获取到井筒数据: wellboreId={wellbore_data.get('wellboreId')}, wellboreCommonName={wellbore_data.get('wellboreCommonName')}")
|
|
else:
|
|
logger.warning("❌ 未获取到井筒数据")
|
|
|
|
# 测试字段值获取
|
|
logger.info("4. 测试字段值获取...")
|
|
test_fields = ['wellId', 'wellboreId', 'wellCommonName', 'wellboreCommonName']
|
|
for field in test_fields:
|
|
value = well_manager.get_well_value_for_field(field)
|
|
if value:
|
|
logger.info(f"✅ {field}: {value}")
|
|
else:
|
|
logger.warning(f"❌ {field}: 未获取到值")
|
|
|
|
# 测试数据增强
|
|
logger.info("5. 测试数据增强...")
|
|
test_data = {
|
|
'wellId': 'mock_well_id',
|
|
'wellboreId': 'mock_wellbore_id',
|
|
'wellCommonName': 'mock_well_name',
|
|
'otherField': 'other_value'
|
|
}
|
|
enhanced_data = well_manager.enhance_data_with_well_values(test_data)
|
|
logger.info(f"原始数据: {test_data}")
|
|
logger.info(f"增强数据: {enhanced_data}")
|
|
|
|
# 检查是否有真实数据替换了模拟数据
|
|
changes = []
|
|
for key in test_data:
|
|
if test_data[key] != enhanced_data.get(key):
|
|
changes.append(f"{key}: {test_data[key]} -> {enhanced_data.get(key)}")
|
|
|
|
if changes:
|
|
logger.info(f"✅ 数据增强成功，替换了以下字段: {', '.join(changes)}")
|
|
else:
|
|
logger.warning("❌ 数据增强未发生变化")
|
|
|
|
else:
|
|
logger.error("❌ 井数据初始化失败")
|
|
return False
|
|
|
|
logger.info("井数据管理器测试完成")
|
|
return True
|
|
|
|
if __name__ == "__main__":
|
|
success = test_well_data_manager()
|
|
sys.exit(0 if success else 1)
|