93 lines
3.6 KiB
Python
93 lines
3.6 KiB
Python
import requests
|
||
import re
|
||
import json
|
||
from .models import Transponders
|
||
from mainapp.models import Polarization, Satellite
|
||
|
||
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}")
|
||
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):
|
||
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)
|
||
tran_obj = Transponders.objects.create(
|
||
name=tran["name"],
|
||
frequency=f,
|
||
frequency_range=f_range,
|
||
zone_name=zone,
|
||
polarization=Polarization.objects.get(name=tran["pol"]),
|
||
sat_id=Satellite.objects.get(name__iexact=sat_name)
|
||
)
|
||
tran_obj.save()
|
||
|