# Standard library imports import json import re from io import BytesIO # Third-party imports import requests from django.db.models import Q # Local imports from mainapp.models import Polarization, Satellite from .models import Transponders def parse_satellite_name(full_name: str) -> tuple[str, str | None]: """ Парсит полное имя спутника и извлекает основное и альтернативное имя. Альтернативное имя находится в скобках после основного названия. Примеры: "Koreasat 116 (ANASIS-II)" -> ("Koreasat 116", "ANASIS-II") "Thaicom 6 (Africom 1)" -> ("Thaicom 6", "Africom 1") "Express AM6" -> ("Express AM6", None) Args: full_name: Полное имя спутника (может содержать альтернативное имя в скобках) Returns: tuple: (основное_имя, альтернативное_имя или None) """ if not full_name: return (full_name, None) # Ищем текст в скобках в конце строки match = re.match(r'^(.+?)\s*\(([^)]+)\)\s*$$', full_name.strip()) if match: main_name = match.group(1).strip() alt_name = match.group(2).strip() return (main_name, alt_name) return (full_name.strip(), None) def find_satellite_by_name(name: str): """ Ищет спутник по имени или альтернативному имени. Все сравнения выполняются в lowercase. Алгоритм поиска: 1. Точное совпадение по name (lowercase) 2. Точное совпадение по alternative_name (lowercase) 3. Частичное совпадение по name (lowercase) 4. Частичное совпадение по alternative_name (lowercase) Args: name: Имя спутника для поиска Returns: Satellite или None: Найденный спутник или None Raises: Satellite.MultipleObjectsReturned: Если найдено несколько спутников """ if not name: return None name_lower = name.strip().lower() # 1. Точное совпадение по name (lowercase) try: return Satellite.objects.get(name__iexact=name_lower) except Satellite.DoesNotExist: pass except Satellite.MultipleObjectsReturned: raise # 2. Точное совпадение по alternative_name (lowercase) try: return Satellite.objects.get(alternative_name__iexact=name_lower) except Satellite.DoesNotExist: pass except Satellite.MultipleObjectsReturned: raise # 3. Частичное совпадение по name или alternative_name (lowercase) satellites = Satellite.objects.filter( Q(name__icontains=name_lower) | Q(alternative_name__icontains=name_lower) ) if satellites.count() == 1: return satellites.first() elif satellites.count() > 1: raise Satellite.MultipleObjectsReturned( f"Найдено несколько спутников с именем '{name_lower}'" ) return None def search_satellite_on_page(data: dict, satellite_name: str): for pos, value in data.get('page', {}).get('positions').items(): for name in value['satellites']: if name['other_names'] is None: name['other_names'] = '' if satellite_name.lower() in name['name'].lower() or satellite_name.lower() in name['other_names'].lower(): return pos, name['id'] return '', '' def get_footprint_data(position: str = 62) -> dict: """Возвращает словарь с данным по footprint для спутников на выбранной долготе""" response = requests.get(f"https://www.satbeams.com/footprints?position={position}", verify=True) response.raise_for_status() match = re.search(r'var data = ({.*?});', response.text, re.DOTALL) if match: json_str = match.group(1) try: data = json.loads(json_str) return data.get("page", {}).get("footprint_data", {}).get("beams",[]) except json.JSONDecodeError as e: print("Ошибка парсинга JSON:", e) else: print("Нужных данных не найдено") return {} def get_all_page_data(url:str = 'https://www.satbeams.com/footprints') -> dict: """Возвращает словарь с данными по всем спутникам на странице""" response = requests.get(url) response.raise_for_status() match = re.search(r'var data = ({.*?});', response.text, re.DOTALL) if match: json_str = match.group(1) try: data = json.loads(json_str) # Файл json на диске для достоверности with open('data.json', 'w') as jf: json.dump(data, jf, indent=2) return data except json.JSONDecodeError as e: print("Ошибка парсинга JSON:", e) else: print("Нужных данных не найдено") return {} def get_names_footprints_for_satellite(footprint_data: dict, sat_id: str) -> list[str]: names = [] for beam in footprint_data: if 'ku' in beam['band'].lower() and sat_id in beam['satellite_id']: names.append( { "name": beam['name'], "fullname": beam['fullname'][8:] } ) return names def get_band_names(satellite_name: str) -> list[str]: data = get_all_page_data() pos, sat_id = search_satellite_on_page(data, satellite_name) footprints = get_footprint_data(pos) names = get_names_footprints_for_satellite(footprints, sat_id) return names def parse_transponders_from_json(filepath: str, user=None): """ Парсит транспондеры из JSON файла. Если имя спутника содержит альтернативное имя в скобках, оно извлекается и сохраняется в поле alternative_name. Args: filepath: путь к JSON файлу user: пользователь для установки created_by и updated_by (optional) """ with open(filepath, encoding="utf-8") as jf: data = json.load(jf) for sat_name_full, trans_zone in data["satellites"].items(): # Парсим имя спутника и альтернативное имя main_name, alt_name = parse_satellite_name(sat_name_full) for zone, trans in trans_zone.items(): for tran in trans: f_b, f_e = tran["freq"][0].split("-") f = round((float(f_b) + float(f_e))/2, 3) f_range = round(abs(float(f_e) - float(f_b)), 3) pol_obj = Polarization.objects.get(name=tran["pol"]) # Ищем спутник по имени или альтернативному имени sat_obj = find_satellite_by_name(main_name) if not sat_obj: # Если не найден, создаём новый с альтернативным именем sat_obj = Satellite.objects.create( name=main_name, alternative_name=alt_name ) elif alt_name and not sat_obj.alternative_name: # Если найден, но альтернативное имя не установлено - обновляем sat_obj.alternative_name = alt_name sat_obj.save() tran_obj, created = Transponders.objects.get_or_create( name=tran["name"], polarization=pol_obj, sat_id=sat_obj, defaults={ "frequency": f, "frequency_range": f_range, "zone_name": zone, } ) # Устанавливаем пользователя, если передан if user: if created: tran_obj.created_by = user tran_obj.updated_by = user tran_obj.save() # Third-party imports (additional) from lxml import etree def parse_transponders_from_xml(data_in: BytesIO, user=None): """ Парсит транспондеры из XML файла. Если имя спутника содержит альтернативное имя в скобках, оно извлекается и сохраняется в поле alternative_name. Args: data_in: BytesIO объект с XML данными user: пользователь для установки created_by и updated_by (optional) """ tree = etree.parse(data_in) ns = { 'i': 'http://www.w3.org/2001/XMLSchema-instance', 'ns': 'http://schemas.datacontract.org/2004/07/Geolocation.Domain.Utils.Repository.SatellitesSerialization.Memos', 'tr': 'http://schemas.datacontract.org/2004/07/Geolocation.Common.Extensions' } satellites = tree.xpath('//ns:SatelliteMemo', namespaces=ns) for sat in satellites[:]: name_full = sat.xpath('./ns:name/text()', namespaces=ns)[0] if name_full == 'X' or 'DONT USE' in name_full: continue # Парсим имя спутника и альтернативное имя main_name, alt_name = parse_satellite_name(name_full) norad = sat.xpath('./ns:norad/text()', namespaces=ns) beams = sat.xpath('.//ns:BeamMemo', namespaces=ns) intl_code = sat.xpath('.//ns:internationalCode/text()', namespaces=ns) sub_sat_point = sat.xpath('.//ns:subSatellitePoint/text()', namespaces=ns) zones = {} for zone in beams: zone_name = zone.xpath('./ns:name/text()', namespaces=ns)[0] if zone.xpath('./ns:name/text()', namespaces=ns) else '-' zones[zone.xpath('./ns:id/text()', namespaces=ns)[0]] = { "name": zone_name, "pol": zone.xpath('./ns:polarization/text()', namespaces=ns)[0], } transponders = sat.xpath('.//ns:TransponderMemo', namespaces=ns) for transponder in transponders: tr_id = transponder.xpath('./ns:downlinkBeamId/text()', namespaces=ns)[0] downlink_start = float(transponder.xpath('./ns:downlinkFrequency/tr:start/text()', namespaces=ns)[0]) downlink_end = float(transponder.xpath('./ns:downlinkFrequency/tr:end/text()', namespaces=ns)[0]) uplink_start = float(transponder.xpath('./ns:uplinkFrequency/tr:start/text()', namespaces=ns)[0]) uplink_end = float(transponder.xpath('./ns:uplinkFrequency/tr:end/text()', namespaces=ns)[0]) tr_data = zones[tr_id] match tr_data['pol']: case 'Horizontal': pol = 'Горизонтальная' case 'Vertical': pol = 'Вертикальная' case 'CircularRight': pol = 'Правая' case 'CircularLeft': pol = 'Левая' case _: pol = '-' tr_name = transponder.xpath('./ns:name/text()', namespaces=ns)[0] pol_obj, _ = Polarization.objects.get_or_create(name=pol) # Ищем спутник по имени или альтернативному имени sat_obj = find_satellite_by_name(main_name) if not sat_obj: # Если не найден, создаём новый с альтернативным именем sat_obj = Satellite.objects.create( name=main_name, alternative_name=alt_name, norad=int(norad[0]) if norad else -1, international_code=intl_code[0] if intl_code else "", undersat_point=float(sub_sat_point[0]) if sub_sat_point else None ) else: # Если найден, обновляем альтернативное имя если не установлено updated = False if alt_name and not sat_obj.alternative_name: sat_obj.alternative_name = alt_name updated = True if norad and not sat_obj.norad: sat_obj.norad = int(norad[0]) updated = True if intl_code and not sat_obj.international_code: sat_obj.international_code = intl_code[0] updated = True if sub_sat_point and not sat_obj.undersat_point: sat_obj.undersat_point = float(sub_sat_point[0]) updated = True if updated: sat_obj.save() trans_obj, created = Transponders.objects.get_or_create( polarization=pol_obj, downlink=(downlink_start+downlink_end)/2/1000000, uplink=(uplink_start+uplink_end)/2/1000000, frequency_range=abs(downlink_end-downlink_start)/1000000, name=tr_name, defaults={ "zone_name": tr_data['name'], "sat_id": sat_obj, } ) if user: if created: trans_obj.created_by = user trans_obj.updated_by = user trans_obj.save()