3.0.0

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

@@ -47,10 +47,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 +62,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 +92,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 +206,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 +217,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,29 +240,20 @@ 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
-
     fields = {}
     tables = []
     is_table = False
     is_table_heading = False
     table_heading_raw = ''
     is_notes = False
     table_fields = []
+    col_positions = []
 
     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:])
+    print(raw)
     for line in raw.splitlines():
         if '#' in line: # prompt lines
             continue
@@ -291,41 +284,45 @@ def _find_all(string: str, finding: str) -> list[int]:
             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()))})
+            first_value_match = search(r'\S+', line)
+            first_value_pos = first_value_match.start() if first_value_match else 0
+
+            start_col = 0
+            for i, col_pos in enumerate(col_positions):
+                if first_value_pos >= col_pos:
+                    start_col = i
+                else:
+                    break
+
+            row_data = {}
+            line_stripped = line.rstrip()
+            for i, field in enumerate(table_fields):
+                if i < start_col:
+                    row_data[field] = None
+                elif i < len(col_positions):
+                    start = col_positions[i]
+                    end = col_positions[i + 1] if i + 1 < len(col_positions) else len(line_stripped)
+                    value = line_stripped[start:end].strip() if start < len(line_stripped) else ''
+                    row_data[field] = _parse_value(value) if value else None
+                else:
+                    row_data[field] = None
+            tables[-1].append(row_data)
             continue
 
-        if not is_table and len(split(r'\s+', line)) > 1: # table start heading line
+        if not is_table and len(split(r'\s+', line)) > 1 and search(r'\s{3,}', line): # 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([])
+            # calculate column pos by heading
+            col_positions = []
+            for field in table_fields:
+                pos = table_heading_raw.find(field)
+                if pos != -1:
+                    col_positions.append(pos)
             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)
-
     return fields, [table for table in tables if table]
 
 def _parse_basic_info(raw: str) -> dict:
@@ -355,7 +352,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'),
@@ -408,16 +405,23 @@ 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/
+    ) # 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 +477,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

routers/attach.py

@@ -11,8 +11,7 @@ def api_get_attachs(id: int, include_task: bool = False):
     if include_task:
         tasks = api_call('task', 'get_list', f'customer_id={id}')['list'].split(',')
         for task in tasks:
-            task_attachs = normalize_items(api_call('attach', 'get',
-                f'object_id={task}&object_type=task'))
+            task_attachs = normalize_items(api_call('attach', 'get', f'object_id={task}&object_type=task'))
             if isinstance(task_attachs, dict):
                 for attach in task_attachs: attach['source'] = 'task'
                 attachs.extend(task_attachs)
@@ -23,8 +22,7 @@ def api_get_attachs(id: int, include_task: bool = False):
             # 'url': api_call('attach', 'get_file_temporary_link', f'uuid={attach["id"]}'),
             'url': get_attach_url(attach['id']),
             'name': attach['internal_filepath'],
-            'extension': attach['internal_filepath'].split('.')[1].lower()
-                if '.' in attach['internal_filepath'] else None,
+            'extension': attach['internal_filepath'].split('.')[1].lower() if '.' in attach['internal_filepath'] else None,
             'created_at': attach['date_add'],
             'source': attach.get('source', 'customer'),
             'source_id': attach.get('object_id'),

routers/box.py

@@ -1,27 +1,25 @@
+from json import loads
+from json.decoder import JSONDecodeError
+
 from fastapi import APIRouter
 from fastapi.responses import JSONResponse
+from fastapi.requests import Request
 
 from api import api_call
-from utils import extract_sn, normalize_items, remove_sn, status_to_str, list_to_str, str_to_list, get_coordinates, get_box_map_link
+from routers.customer import _process_customer
+from utils import normalize_items, list_to_str, str_to_list, get_coordinates, get_box_map_link
 
 router = APIRouter(prefix='/box')
 
 @router.get('/{id}')
 def api_get_box(
+    request: Request,
     id: int,
-    get_onu_level: bool = False,
+    get_olt_data: bool = False,
     get_tasks: bool = False,
     limit: int | None = None,
-    exclude_customer_ids: list[int] = []
+    exclude_customer_ids: str = '[]'
 ):
-    def _get_onu_level(name) -> float | None:
-        if extract_sn(name) is None:
-            return
-        data = api_call('device', 'get_ont_data', f'id={extract_sn(name)}').get('data')
-        if not isinstance(data, dict):
-            return
-        return data.get('level_onu_rx')
-
     def _get_tasks(entity: str, entity_id: int) -> list[int]:
         res = api_call('task', 'get_list', f'{entity}_id={entity_id}&state_id=18,3,17,11,1,16,19')
         return list(map(int, str_to_list(res.get('list', ''))))
@@ -30,23 +28,33 @@ def _build_customer(customer: dict) -> dict | None:
         name = customer.get('full_name')
         if name is None:
             return None
-        return {
-            'id': customer['id'],
-            'name': remove_sn(name),
-            'last_activity': customer.get('date_activity'),
-            'status': status_to_str(customer['state_id']),
-            'sn': extract_sn(name),
-            'onu_level': _get_onu_level(name) if get_onu_level else None,
-            'tasks': _get_tasks('customer', customer['id']) if get_tasks else None
-        }
+        return _process_customer(request.app.state.tariffs, request.app.state.customer_groups, customer, get_olt_data)
+        # {
+        #     'id': customer['id'],
+        #     'name': remove_sn(name),
+        #     'last_activity': customer.get('date_activity'),
+        #     'status': status_to_str(customer['state_id']),
+        #     'sn': extract_sn(name),
+        #     'onu_level': _get_onu_level(name) if get_onu_level else None,
+        #     'tasks': _get_tasks('customer', customer['id']) if get_tasks else None
+        # }
+
+    exclude_ids = []
+    if exclude_customer_ids:
+        try:
+            exclude_ids: list[int] = loads(exclude_customer_ids)
+            if not (isinstance(exclude_ids, list) and all(isinstance(customer, int) for customer in exclude_ids)):
+                return JSONResponse({'status': 'fail', 'detail': 'incorrect type of exclude_customer_ids param'}, 422)
+        except JSONDecodeError:
+            return JSONResponse({'status': 'fail', 'detail': 'unable to parse exclude_customer_ids param'}, 422)
 
     house_data = api_call('address', 'get_house', f'building_id={id}').get('data')
     if not house_data:
         return JSONResponse({'status': 'fail', 'detail': 'box not found'}, 404)
 
     house = list(house_data.values())[0]
     customer_ids: list = api_call('customer', 'get_customers_id', f'house_id={id}').get('data', [])
-    for customer in exclude_customer_ids:
+    for customer in exclude_ids:
         if customer in customer_ids:
             customer_ids.remove(customer)
     customers_count = len(customer_ids)