3.0.0

.gitignore

@@ -1,5 +1,3 @@
-/venv/
-__pycache__/
-/config.py
-/routers/__pycache__/
-/.vscode
+.venv
+config.py
+__pycache__

main.py

@@ -15,12 +15,13 @@
 from routers import attach
 from routers import inventory
 from api import api_call
+from tariff import Tariff
 from config import API_KEY as APIKEY
 
 app = FastAPI(title='SmartLinkAPI')
 
 app.state.tariffs = {
-    tariff['billing_uuid']: unescape(tariff['name'])
+    tariff['billing_uuid']: Tariff(unescape(tariff['name']))
     for tariff in api_call('tariff', 'get')['data'].values()
 }
 app.state.customer_groups = {
@@ -47,8 +48,7 @@
         'host': olt['host'],
         'online': bool(olt['is_online']),
         'location': unescape(olt['location'])
-    } for olt in api_call('device', 'get_data', 'object_type=olt&is_hide_ifaces_data=1')['data']
-        .values()
+    } for olt in api_call('device', 'get_data', 'object_type=olt&is_hide_ifaces_data=1')['data'].values()
 ]
 app.state.divisions = [
     {

ont.py

@@ -4,6 +4,7 @@
 from subprocess import run
 from re import search, fullmatch, split
 
+from asciitable import read as table_read, DictOutputter, FixedWidth
 from paramiko import SSHClient, AutoAddPolicy, Channel
 
 from config import SSH_USER, SSH_PASSWORD
@@ -47,10 +48,10 @@ def search_ont(sn: str, host: str) -> tuple[dict, str | None] | None:
     """Search ONT by serial number and return its basic, optical and catv data"""
     ont_info: dict = {}
     olt_name = None
-    try:
+    if True: #try:
         channel, ssh, olt_name = _connect_ssh(host)
 
-        channel.send(bytes(f"display ont info by-sn {sn}\n", 'utf-8'))
+        channel.send(bytes(f"display ont info by-sn {sn}\n\n", 'utf-8'))
         parsed_ont_info = _parse_basic_info(_read_output(channel))
 
         if 'error' in parsed_ont_info:
@@ -62,14 +63,14 @@ def search_ont(sn: str, host: str) -> tuple[dict, str | None] | None:
         _clear_buffer(channel)
 
         if ont_info.get('online'):
-            channel.send(bytes(f"display ont optical-info {ont_info['interface']['port']} {ont_info['ont_id']}\n", 'utf-8'))
+            channel.send(bytes(f"display ont optical-info {ont_info['interface']['port']} {ont_info['ont_id']}\n\n", 'utf-8'))
             optical_info = _parse_optical_info(_read_output(channel))
             ont_info['optical'] = optical_info
 
         catv_results = []
         for port_num in range(1, (ont_info['_catv_ports'] or 2) + 1):
-            sleep(0.07)
-            channel.send(bytes(f"display ont port attribute {ont_info['interface']['port']} {ont_info['ont_id']} catv {port_num}\n", 'utf-8'))
+            sleep(0.1)
+            channel.send(bytes(f"display ont port attribute {ont_info['interface']['port']} {ont_info['ont_id']} catv {port_num}\n\n", 'utf-8'))
             catv = _parse_port_status(_read_output(channel))
             catv_results.append(catv)
 
@@ -92,25 +93,26 @@ def search_ont(sn: str, host: str) -> tuple[dict, str | None] | None:
         ont_info['service_port'] = _parse_service_port(_read_output(channel), ont_info['interface'])
         if ont_info['service_port']:
             sleep(0.07)
-            channel.send(bytes(f'display mac-address service-port {ont_info["service_port"]}\n', 'utf-8'))
-            ont_info['mac'] = _parse_mac(_read_output(channel))
+            channel.send(bytes(f'display mac-address service-port {ont_info["service_port"]}\n\n', 'utf-8'))
+            ont_info['mac'] = _parse_mac(_read_output(channel), ont_info['interface'])
 
         channel.close()
         ssh.close()
 
         ping_result = _ping(ont_info['ip']) if 'ip' in ont_info else None
         ont_info['ping'] = float(ping_result.split(' ', maxsplit=1)[0]) if ping_result else None
         return ont_info, olt_name
-    except Exception as e:
-        print(f'error search ont: {e.__class__.__name__}: {e}')
-        return {'online': False, 'detail': str(e)}, olt_name
+    # except Exception as e:
+    #     print(f'error search ont: {e.__class__.__name__}: {e}')
+    #     return {'online': False, 'detail': str(e)}, olt_name
 
 
 def reset_ont(host: str, id: int, interface: dict) -> dict:
     """Restart/reset ONT"""
     try:
         channel, ssh, _ = _connect_ssh(host)
 
+        sleep(0.1)
         channel.send(bytes(f"interface gpon {interface['fibre']}/{interface['service']}\n", 'utf-8'))
         sleep(0.1)
         _clear_buffer(channel)
@@ -205,8 +207,8 @@ def _read_output(channel: Channel, force: bool = True):
                     break
                 sleep(0.05)
 
-        if time() - last_data_time > 1.5 and len(output.strip().strip('\n').splitlines()) > 5:
-            print('no new data more than 1.5 seconds')
+        if time() - last_data_time > 2 and len(output.strip().strip('\n').splitlines()) > 5:
+            print('no new data more than 2 seconds')
             break
         if time() - last_data_time > 10 and len(output.strip().strip('\n').splitlines()) <= 5:
             print('no new data more than 10 seconds')
@@ -216,8 +218,9 @@ def _read_output(channel: Channel, force: bool = True):
             print('read output takes more than 5 seconds')
             break
         if time() - start_time > 20:
-            print('read output takes more than 20 sceonds')
+            print('read output takes more than 20 seconds')
             print(output)
+            break
         sleep(0.01)
     return '\n'.join(output.splitlines()[1:]) if output.count('\n') > 1 else output
 
@@ -238,93 +241,121 @@ def _parse_value(value: str) -> str | float | int | bool | None:
             return False
         return value
 
-    def _find_all(string: str, finding: str) -> list[int]:
-        result = []
-        for i, _ in enumerate(string):
-            if string[i:i + len(finding)] == finding:
-                if len(string) > i + len(finding):
-                    if string[i + len(finding)] == ' ':
-                        result.append(i)
-                else:
-                    result.append(i)
-        return result
+    def _is_divider(line):
+        return bool(fullmatch(r'\s*\-{5,}\s*', line.strip()))
 
     fields = {}
     tables = []
-    is_table = False
-    is_table_heading = False
-    table_heading_raw = ''
-    is_notes = False
-    table_fields = []
 
     raw = raw.replace(PAGINATION, '').replace('\x1b[37D', '').replace('x1b[37D', '') # remove stupid pagination
     if "Command:" in raw:
         raw = raw.split("Command:", 1)[1]
         raw = "\n".join(raw.splitlines()[2:])
-    for line in raw.splitlines():
-        if '#' in line: # prompt lines
+    print(raw)
+
+    lines = raw.splitlines()
+    divider_indices = [i for i, line in enumerate(lines) if _is_divider(line)]
+
+    # Identify table sections: divider -> heading -> divider -> data -> [divider]
+    table_line_ranges = set()
+    i = 0
+    while i < len(divider_indices) - 1:
+        first_div = divider_indices[i]
+        second_div = divider_indices[i + 1]
+
+        # Lines between first and second divider - potential heading
+        heading_lines = [lines[j] for j in range(first_div + 1, second_div) if lines[j].strip()]
+
+        # Check if these look like a table heading (multiple words, 3+ spaces, no ":")
+        is_heading = (
+            heading_lines and
+            any(search(r'\s{3,}', line) for line in heading_lines) and
+            not any(':' in line for line in heading_lines)
+        )
+
+        if not is_heading:
+            i += 1
             continue
 
-        if fullmatch(r'\s*\-{5,}\s*', line.strip()): # divider line
-            is_notes = False
-            if is_table_heading:
-                is_table_heading = False
+        # Find data section end (next divider or end of lines)
+        if i + 2 < len(divider_indices):
+            third_div = divider_indices[i + 2]
+        else:
+            third_div = len(lines)
+
+        # Filter data lines (exclude blank, prompts, notes)
+        data_lines = []
+        notes = False
+        for j in range(second_div + 1, third_div):
+            line = lines[j]
+            if not line.strip() or '#' in line:
                 continue
-            if is_table and not is_table_heading:
-                is_table = False
-            continue
-
-        # if line == PAGINATION: # pagination line
-        #     continue
-
-        # if PAGINATION in line: # partially-pagination line
-        #     line = line.strip(PAGINATION).strip('\x1b[37D').strip('x1b[37D')
+            if line.strip().startswith('Note') or notes:
+                notes = True
+                continue
+            data_lines.append(line)
 
-        if line.strip().startswith('Note') or is_notes: # notes line
-            is_notes = True
+        if not data_lines:
+            i += 1
             continue
 
-        if ':' in line: # standalone field line
-            is_table = False
-            pair = list(map(lambda i: i.strip(), line.strip().split(':', maxsplit=1)))
-            fields[pair[0]] = _parse_value(pair[-1])
+        # Build table text for asciitable: divider + heading + divider + data
+        table_text_lines = [lines[first_div]]
+        table_text_lines.extend(heading_lines)
+        table_text_lines.append(lines[second_div])
+        table_text_lines.extend(data_lines)
+        table_text = '\n'.join(table_text_lines)
+
+        # data_start = 1 (first divider) + heading lines + 1 (second divider)
+        data_start_idx = len(heading_lines) + 2
+
+        try:
+            dat = table_read(
+                table_text,
+                reader=FixedWidth,
+                outputter=DictOutputter,
+                header_start=0,
+                data_start=data_start_idx,
+                delimiter=' ',
+                numpy=False
+            )
+
+            table_data = []
+            for row in dat:
+                parsed_row = {}
+                for key, value in row.items():
+                    col_name = key.replace(' ', '-')
+                    parsed_row[col_name] = _parse_value(str(value))
+                table_data.append(parsed_row)
+            if table_data:
+                tables.append(table_data)
+        except Exception as e:
+            print(f'asciitable parse error: {e}')
+
+        # Mark table lines
+        end_mark = (third_div + 1) if i + 2 < len(divider_indices) else third_div
+        for k in range(first_div, min(end_mark, len(lines))):
+            table_line_ranges.add(k)
+
+        i += 2 if i + 2 < len(divider_indices) else len(divider_indices)
+
+    # Parse non-table lines for key-value pairs
+    is_notes = False
+    for i, line in enumerate(lines):
+        if i in table_line_ranges:
+            is_notes = False
             continue
-
-        if is_table and not is_table_heading: # table field line
-            tables[-1].append({key: _parse_value(value.strip()) for key, value in zip(table_fields, split(r'\s+', line.strip()))})
+        if '#' in line:
             continue
-
-        if not is_table and len(split(r'\s+', line)) > 1: # table start heading line
-            is_table = True
-            is_table_heading = True
-            table_heading_raw = line
-            table_fields = [c for c in split(r'\s+', line.strip()) if c]
-            tables.append([])
+        if _is_divider(line):
+            is_notes = False
             continue
-
-        if is_table_heading: # table next heading line
-            line = line[len(table_heading_raw) - len(table_heading_raw.lstrip()):]
-            full_line = line
-            # print('begin table parse; fields:', table_fields, 'appendixes line:', line)
-
-            for i, field in enumerate(table_fields):
-                raw_index = _find_all(table_heading_raw.lstrip(), field)[table_fields[:i].count(field)]
-                # print('found fields:', _find_all(table_heading_raw.lstrip(), field))
-
-                if search(r'\w', full_line[raw_index:raw_index + len(field)]):
-                    # print('found non space appendix:', full_line[raw_index:raw_index + len(field)] + '... for', field)
-                    appendix = line.lstrip().split(' ', maxsplit=1)[0]
-                    # print('cleaned appendix:', appendix)
-                    table_fields[i] += '-' + appendix
-                    # print('invoked to field:', table_fields[i])
-                    line = line[line.index(appendix) + len(appendix):]
-                    # print('line truncated:', line)
-
-                else:
-                    # print('found space appendix for', field)
-                    # spaces += len(table_heading_raw[:raw_index]) - len(table_heading_raw[:raw_index].rstrip()) - 1
-                    line = line[len(field):]
-                    # print('line truncated:', line)
+        if line.strip().startswith('Note') or is_notes:
+            is_notes = True
+            continue
+        if ':' in line:
+            pair = list(map(lambda x: x.strip(), line.strip().split(':', maxsplit=1)))
+            fields[pair[0]] = _parse_value(pair[-1])
 
     return fields, [table for table in tables if table]
 
@@ -355,7 +386,7 @@ def _parse_basic_info(raw: str) -> dict:
         'online': data.get('Run state', False),
         'mem_load': data.get('Memory occupation'),
         'cpu_load': data.get('CPU occupation'),
-        'temp': data['Temperature'],
+        'temp': data.get('Temperature'),
         'ip': data['ONT IP 0 address/mask'].split('/')[0] if data.get('ONT IP 0 address/mask') else None,
         'last_down_cause': data.get('Last down cause'),
         'last_down': data.get('Last down time'),
@@ -407,17 +438,24 @@ def _parse_eth_ports_status(raw: str) -> list[dict]:
 def _parse_service_port(raw: str, interface: dict) -> int | None:
     raw = raw.replace(
         f"{interface['fibre']}/{interface['service']} /{interface['port']}",
-        f"{interface['fibre']}/ {interface['service']}/ {interface['port']}"
-    ) # change F/S /P -> F/ S/ P/
+        f"{interface['fibre']}/{interface['service']}/{interface['port']}"
+    ) # change F/S /P -> F/ S/ P
     raw = raw.replace(' Switch-Oriented Flow List\n', '') # remove extra text
     if 'Failure: No service virtual port can be operated' in raw:
         return
     return _parse_output(raw)[1][0][0].get('INDEX')
 
-def _parse_mac(raw: str) -> str | None:
+def _parse_mac(raw: str, interface: dict) -> str | None:
     if 'Failure: There is not any MAC address record' in raw:
         return
     raw = raw.replace('MAC TYPE', 'MAC-TYPE') # avoid extra spaces for better parsing (prefer "-")
+    raw = raw.replace('It will take some time, please wait...', '') # remove extra text because it is near to table and can perceived as heading
+    raw = raw.replace(
+        f"{interface['fibre']} /{interface['service']}/{interface['port']}",
+        f"{interface['fibre']} /{interface['service']} /{interface['port']}"
+    ) # change F /S/P -> F /S /P
+    raw = raw.replace('VLAN ID', 'VLAN-ID')
+    print(_parse_output(raw))
     return format_mac(_parse_output(raw)[1][0][0].get('MAC'))
 
 def _parse_onts_info(output: str) -> tuple[int, int, list[dict]] | tuple[dict, None, None]:
@@ -473,4 +511,4 @@ def _ping(ip: str) -> None | str:
             return f"{time_match.group(1)} ms" if time_match else "-"
         return None
     except Exception:
-        return None
+        return None

pyproject.toml

@@ -0,0 +1,15 @@
+[project]
+name = "smartlinkapi"
+version = "3.0.0.dev2"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.10"
+dependencies = [
+    "asciitable2==0.1.1",
+    "fastapi[standard]>=0.135.3",
+    "paramiko>=4.0.0",
+    "requests>=2.33.1",
+]
+
+[tool.uv.sources]
+asciitable2 = { git = "https://github.com/firedotguy/asciitable2" }