Spaces:
Runtime error
Runtime error
| #!/usr/bin/env python3 | |
| """ | |
| Test script to verify that valori_stazioni functionality still works | |
| after adding massimi precipitazioni to the table scraper. | |
| """ | |
| import asyncio | |
| import sys | |
| from pathlib import Path | |
| # Add parent directories to path for imports | |
| sys.path.insert(0, str(Path(__file__).parent.parent.parent)) | |
| from services.web.compat import fetch_omirl_stations | |
| async def test_valori_stazioni(): | |
| """Test the existing valori_stazioni functionality""" | |
| print("🧪 Testing OMIRL Valori Stazioni (existing functionality)...") | |
| print("=" * 60) | |
| try: | |
| # Test 1: Basic extraction without sensor filter | |
| print("\n📋 Test 1: Basic station data extraction (no filter)") | |
| stations_all = await fetch_omirl_stations() | |
| print(f"✅ Successfully extracted {len(stations_all)} stations (all sensors)") | |
| if stations_all: | |
| sample_station = stations_all[0] | |
| print(f" Sample station: {sample_station.get('Nome', '')} ({sample_station.get('Codice', '')})") | |
| print(f" Location: {sample_station.get('Comune', '')}, {sample_station.get('Provincia', '')}") | |
| print(f" Available fields: {list(sample_station.keys())}") | |
| # Test 2: Precipitation sensor filter | |
| print("\n🌧️ Test 2: Precipitation sensor filter") | |
| stations_precip = await fetch_omirl_stations("Precipitazione") | |
| print(f"✅ Successfully extracted {len(stations_precip)} precipitation stations") | |
| if stations_precip: | |
| sample_precip = stations_precip[0] | |
| print(f" Sample precipitation station: {sample_precip.get('Nome', '')} ({sample_precip.get('Codice', '')})") | |
| # Show measurement fields (ultimo, Max, Min if available) | |
| measurement_fields = {k: v for k, v in sample_precip.items() | |
| if k not in ['Nome', 'Codice', 'Comune', 'Provincia', 'Area', 'Bacino', 'Sottobacino', 'UM']} | |
| if measurement_fields: | |
| print(f" Measurement data: {measurement_fields}") | |
| # Test 3: Temperature sensor filter | |
| print("\n🌡️ Test 3: Temperature sensor filter") | |
| stations_temp = await fetch_omirl_stations("Temperatura") | |
| print(f"✅ Successfully extracted {len(stations_temp)} temperature stations") | |
| # Test 4: Verify different sensor types work | |
| print("\n🔍 Test 4: Testing different sensor types") | |
| sensor_tests = [ | |
| ("Vento", "wind"), | |
| ("Livelli Idrometrici", "water levels"), | |
| ("Umidità dell'aria", "humidity") | |
| ] | |
| for sensor_name, description in sensor_tests: | |
| try: | |
| stations = await fetch_omirl_stations(sensor_name) | |
| print(f" {sensor_name} ({description}): {len(stations)} stations ✅") | |
| except Exception as e: | |
| print(f" {sensor_name} ({description}): FAILED - {e} ❌") | |
| # Summary | |
| print(f"\n📊 Summary:") | |
| print(f" Total stations (all sensors): {len(stations_all)}") | |
| print(f" Precipitation stations: {len(stations_precip)}") | |
| print(f" Temperature stations: {len(stations_temp)}") | |
| # Validate basic structure | |
| if stations_all: | |
| required_fields = ['Nome', 'Codice', 'Comune', 'Provincia'] | |
| missing_fields = [field for field in required_fields | |
| if field not in stations_all[0]] | |
| if missing_fields: | |
| print(f" ❌ Missing required fields: {missing_fields}") | |
| return False | |
| else: | |
| print(f" ✅ All required fields present: {required_fields}") | |
| print(f" Test: ✅ PASSED") | |
| return True | |
| except Exception as e: | |
| print(f"\n❌ Test failed: {e}") | |
| import traceback | |
| traceback.print_exc() | |
| return False | |
| if __name__ == "__main__": | |
| success = asyncio.run(test_valori_stazioni()) | |
| sys.exit(0 if success else 1) | |