Created
January 26, 2026 09:02
-
-
Save ov1d1u/0583f32b9d097bcee42b6348fba17249 to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import requests | |
| import xml.etree.ElementTree as ET | |
| from datetime import datetime | |
| import argparse | |
| import re | |
| import logging | |
| # Configure logging | |
| logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') | |
| def replace_romanian_diacritics(s): | |
| diacritic_map = { | |
| 'Ă': 'A', 'Â': 'A', 'Ș': 'S', 'Ş': 'S', 'Ț': 'T', 'Ţ': 'T', 'Î': 'I', | |
| 'ă': 'a', 'â': 'a', 'ș': 's', 'ş': 's', 'ț': 't', 'ţ': 't', 'î': 'i' | |
| } | |
| return ''.join(diacritic_map.get(c, c) for c in s) | |
| def extract_warnings(data, newer_than): | |
| result = {'lastUpdate': None, 'warnings': {}} | |
| warnings = data.findall('.//avertizare') | |
| latest_modification = None | |
| county_regex = r'Jude[tțțţ]ul (.*?)(?::\s*(.*?))?(?:[;,]|$)' | |
| for warning in warnings: | |
| modified = datetime.fromisoformat(warning.attrib['modificat']) | |
| if modified <= newer_than: | |
| continue | |
| if not latest_modification or modified > latest_modification: | |
| latest_modification = modified | |
| zones = warning.attrib['zona'].split('<br>') | |
| for zone in zones: | |
| if not zone.strip(): | |
| continue | |
| parts = re.match(county_regex, zone.strip()) | |
| if parts: | |
| county = replace_romanian_diacritics(parts.group(1).strip()) | |
| localities = (parts.group(2).split(',') if parts.group(2) else []) | |
| localities = [replace_romanian_diacritics(loc.strip()) for loc in localities] | |
| warning_details = { | |
| 'semnalare': warning.attrib['semnalare'], | |
| 'dataInceput': warning.attrib['dataInceput'], | |
| 'dataSfarsit': warning.attrib['dataSfarsit'], | |
| 'numeCuloare': warning.attrib['numeCuloare'], | |
| 'localitati': localities | |
| } | |
| if county not in result['warnings']: | |
| result['warnings'][county] = [] | |
| result['warnings'][county].append(warning_details) | |
| if latest_modification: | |
| result['lastUpdate'] = latest_modification.isoformat() | |
| return result | |
| def read_latest_alert_date(file_path): | |
| try: | |
| with open(file_path, 'r') as file: | |
| date_str = file.read().strip() | |
| return datetime.fromisoformat(date_str) | |
| except (FileNotFoundError, ValueError) as e: | |
| logging.warning(f"Could not read the latest alert date from file: {e}") | |
| return datetime(1970, 1, 1) | |
| def write_latest_alert_date(file_path, date): | |
| with open(file_path, 'w') as file: | |
| file.write(date) | |
| logging.info(f"Saved latest alert date: {date}") | |
| def main(xml_file_path=None, ignore_latest_alert=False): | |
| # Load XML data | |
| if xml_file_path: | |
| logging.info(f"Reading XML data from file: {xml_file_path}") | |
| try: | |
| with open(xml_file_path, 'r', encoding='utf-8') as file: | |
| xml_content = file.read() | |
| except Exception as e: | |
| logging.error(f"Failed to read XML file: {e}") | |
| return | |
| else: | |
| logging.info("Downloading XML data from the URL") | |
| url = "https://www.meteoromania.ro/avertizari-nowcasting-xml.php" | |
| try: | |
| response = requests.get(url) | |
| response.raise_for_status() | |
| xml_content = response.text | |
| except requests.RequestException as e: | |
| logging.error(f"Failed to download XML data: {e}") | |
| return | |
| root = ET.fromstring(xml_content) | |
| # Determine whether to read or ignore the latest alert date | |
| if ignore_latest_alert: | |
| logging.info("Ignoring the latest alert date. Processing all warnings.") | |
| start_date = datetime(1970, 1, 1) | |
| else: | |
| alert_file = 'latest_nowcasting_alert.txt' | |
| start_date = read_latest_alert_date(alert_file) | |
| # Extract warnings | |
| nowcasting_alerts = extract_warnings(root, start_date) | |
| results = [] | |
| # Update the latest alert date if there is a new update | |
| if not ignore_latest_alert and nowcasting_alerts['lastUpdate']: | |
| alert_file = 'latest_nowcasting_alert.txt' | |
| write_latest_alert_date(alert_file, nowcasting_alerts['lastUpdate']) | |
| for alert in nowcasting_alerts['warnings'].get('Iasi', []): | |
| if not alert['localitati'] or "Iasi" in alert['localitati']: | |
| results.append(alert) | |
| # Output results | |
| if results: | |
| logging.info(f"Found {len(results)} alert(s) for Iași.") | |
| for result in results: | |
| print(result) | |
| else: | |
| logging.info("No alerts found for Iași.") | |
| if __name__ == "__main__": | |
| parser = argparse.ArgumentParser(description='Process nowcasting alerts.') | |
| parser.add_argument('--xml-file', type=str, help='Optional path to a local XML file.') | |
| parser.add_argument('--ignore-latest-alert', action='store_true', help='Ignore the latest alert date and process all warnings.') | |
| args = parser.parse_args() | |
| main(xml_file_path=args.xml_file, ignore_latest_alert=args.ignore_latest_alert) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment