test/main.py - master - Said/space

test/main.py
217 строк9 KB
Саид Гаджиев
Add parsing
06 ноя 2025, 20:04
06 ноя 2025, 20:04eb30c61
from datetime import datetime
import re
import json

def parse_onu_data(data_text):
    # Инициализация словаря для хранения данных
    onu_data = {
        "interface": None,
        "name": None,
        "type": None,
        "state": None,
        "phase_state": None,
        "serial_number": None,
        "description": None,
        "service_profile": None,
        "distance": None,
        "online_duration": None,
        "history": [],
        "signal_level": {
            "upstream": {
                "olt_rx": None,
                "onu_tx": None,
                "attenuation": None
            },
            "downstream": {
                "olt_tx": None,
                "onu_rx": None,
                "attenuation": None
            }
        },
        "mac_addresses": [],
        "interfaces": [],
        "configuration": []
    }

    # Разделяем данные на секции по заголовкам
    sections = {}
    current_section = None
    section_content = []
    
    for line in data_text.split('\n'):
        line = line.strip()
        if not line:
            continue
            
        # Определяем начало новой секции
        if "УРОВЕНЬ СИГНАЛА" in line:
            current_section = "signal_level"
            section_content = []
        elif "ПРОСМОТР МАКОВ НА ОНУ" in line:
            current_section = "mac_addresses"
            section_content = []
        elif "ПРОСМОТР ЛИНКОВ НА ПОРТУ ОНУШКИ" in line:
            current_section = "interfaces"
            section_content = []
        elif "CONFIGURATION:" in line:
            current_section = "configuration"
            section_content = []
        elif "Authpass Time" in line and "OfflineTime" in line and "Cause" in line:
            current_section = "history"
            section_content = [line]  # Сохраняем заголовок
        elif current_section:
            section_content.append(line)
            
        # Сохраняем предыдущую секцию при обнаружении новой
        if current_section and current_section not in sections and section_content:
            sections[current_section] = section_content
    
    # Обрабатываем основную информацию (все что до первой секции)
    main_info = []
    for line in data_text.split('\n'):
        line = line.strip()
        if not line:
            continue
        if any(header in line for header in ["УРОВЕНЬ СИГНАЛА", "ПРОСМОТР МАКОВ", "ПРОСМОТР ЛИНКОВ", "CONFIGURATION:", "Authpass Time"]):
            break
        if ':' in line:
            main_info.append(line)
    
    for line in main_info:
        key_part, value_part = line.split(':', 1)
        key = key_part.strip()
        value = value_part.strip()
        
        if key == "ONU interface":
            onu_data["interface"] = value
        elif key == "Name":
            onu_data["name"] = value
        elif key == "Type":
            onu_data["type"] = value
        elif key == "State":
            onu_data["state"] = value
        elif key == "Phase state":
            onu_data["phase_state"] = value
        elif key == "Serial number":
            onu_data["serial_number"] = value
        elif key == "Description":
            onu_data["description"] = value
        elif key == "Service Profile":
            onu_data["service_profile"] = value
        elif key == "ONU Distance":
            onu_data["distance"] = value
        elif key == "Online Duration":
            onu_data["online_duration"] = value

    # Обрабатываем историю
    if "history" in sections:
        history_lines = sections["history"]
        header_found = False
        for line in history_lines:
            if "Authpass Time" in line and "OfflineTime" in line and "Cause" in line:
                header_found = True
                continue
                
            if header_found and line:
                parts = line.split()
                if len(parts) >= 6:
                    try:
                        entry_num = parts[0]
                        auth_time = f"{parts[1]} {parts[2]}"
                        offline_time = f"{parts[3]} {parts[4]}"
                        cause = ' '.join(parts[5:])
                        
                        auth_dt = datetime.strptime(auth_time, "%Y-%m-%d %H:%M:%S").isoformat()
                        offline_dt = datetime.strptime(offline_time, "%Y-%m-%d %H:%M:%S").isoformat() if offline_time != "0000-00-00 00:00:00" else None
                        
                        history_entry = {
                            "entry_number": entry_num,
                            "auth_time": auth_dt,
                            "offline_time": offline_dt,
                            "cause": cause
                        }
                        onu_data["history"].append(history_entry)
                    except Exception as e:
                        print(f"Ошибка обработки строки истории: {line}. Ошибка: {e}")

    # Обрабатываем уровень сигнала
    if "signal_level" in sections:
        signal_data = ' '.join(sections["signal_level"])
        
        # Парсим upstream данные
        upstream_match = re.search(r'up\s+Rx\s*:([\d.-]+)\(dbm\)\s+Tx\s*:([\d.-]+)\(dbm\)\s+([\d.-]+)\(dB\)', signal_data)
        if upstream_match:
            onu_data["signal_level"]["upstream"]["olt_rx"] = upstream_match.group(1)
            onu_data["signal_level"]["upstream"]["onu_tx"] = upstream_match.group(2)
            onu_data["signal_level"]["upstream"]["attenuation"] = upstream_match.group(3)
        
        # Парсим downstream данные
        downstream_match = re.search(r'down\s+Tx\s*:([\d.-]+)\(dbm\)\s+Rx\s*:([\d.-]+)\(dbm\)\s+([\d.-]+)\(dB\)', signal_data)
        if downstream_match:
            onu_data["signal_level"]["downstream"]["olt_tx"] = downstream_match.group(1)
            onu_data["signal_level"]["downstream"]["onu_rx"] = downstream_match.group(2)
            onu_data["signal_level"]["downstream"]["attenuation"] = downstream_match.group(3)


    # Обрабатываем MAC-адреса
    if "mac_addresses" in sections:
        mac_lines = sections["mac_addresses"]
        header_found = False
        for line in mac_lines:
            if "Mac address" in line and "Vlan" in line and "Type" in line:
                header_found = True
                continue
            if header_found and line and "---" not in line and "Total mac address" not in line:
                parts = line.split()
                if len(parts) >= 5:
                    mac_entry = {
                        "mac_address": parts[0],
                        "vlan": parts[1],
                        "type": parts[2],
                        "port": parts[3],
                        "vc": ' '.join(parts[4:])
                    }
                    onu_data["mac_addresses"].append(mac_entry)

    # Обрабатываем интерфейсы
    if "interfaces" in sections:
        interface_lines = sections["interfaces"]
        current_interface = None
        for line in interface_lines:
            if "Interface" in line and ":" in line:
                current_interface = line.split(":")[1].strip()
            elif "Speed status" in line and current_interface and ":" in line:
                speed = line.split(":")[1].strip()
                # Добавляем новый интерфейс или обновляем существующий
                if not onu_data["interfaces"] or onu_data["interfaces"][-1]["interface"] != current_interface:
                    onu_data["interfaces"].append({
                        "interface": current_interface,
                        "speed_status": speed
                    })
                else:
                    onu_data["interfaces"][-1]["speed_status"] = speed
            elif "Operate status" in line and current_interface and ":" in line and onu_data["interfaces"]:
                status = line.split(":")[1].strip()
                onu_data["interfaces"][-1]["operate_status"] = status

    # Обрабатываем конфигурацию
    if "configuration" in sections:
        config_lines = sections["configuration"]
        header_found = False
        for line in config_lines:
            if "OnuIndex" in line and "Admin State" in line and "OMCC State" in line:
                header_found = True
                continue
            if header_found and line and "---" not in line and "ONU Number" not in line:
                parts = line.split()
                if len(parts) >= 5:
                    config_entry = {
                        "onu_index": parts[0],
                        "admin_state": parts[1],
                        "omcc_state": parts[2],
                        "phase_state": parts[3],
                        "channel": ' '.join(parts[4:])
                    }
                    onu_data["configuration"].append(config_entry)

    return json.dumps(onu_data, indent=2, ensure_ascii=False)