# Standard library imports import json import re from io import BytesIO # Third-party imports import requests # Local imports from mainapp.models import Polarization, Satellite from .models import Transponders 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 файла. Args: filepath: путь к JSON файлу user: пользователь для установки created_by и updated_by (optional) """ with open(filepath, encoding="utf-8") as jf: data = json.load(jf) for sat_name, trans_zone in data["satellites"].items(): 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 = Satellite.objects.get(name__iexact=sat_name) 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 файла. 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 = sat.xpath('./ns:name/text()', namespaces=ns)[0] if name == 'X' or 'DONT USE' in name: continue 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] # p = tr_data['pol'][0] if tr_data['pol'] else '-' 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, _ = Satellite.objects.get_or_create( name=name, defaults={ "norad": int(norad[0]) if norad else -1, "international_code": intl_code[0] if intl_code else "", "undersat_point": sub_sat_point[0 if sub_sat_point else ""] }) 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()