164 lines
6.2 KiB
Python
164 lines
6.2 KiB
Python
"""NRO delegated stats prefix geography collector.
|
|
|
|
Parses the delegated extended/statistics file and stores coarse registry
|
|
allocation geography as prefix-centric fallback hints.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ipaddress
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from app.services.collectors.base import BaseCollector
|
|
from app.services.collectors.downloads import ResumableFileDownloader
|
|
|
|
|
|
class NRODelegatedPrefixGeoCollector(BaseCollector):
|
|
name = "nro_delegated_prefix_geo"
|
|
priority = "P1"
|
|
module = "L3"
|
|
frequency_hours = 24
|
|
data_type = "prefix_geography"
|
|
fail_on_empty = True
|
|
_downloader = ResumableFileDownloader(
|
|
cache_namespace="nro",
|
|
default_accept="text/plain,*/*",
|
|
)
|
|
|
|
async def fetch(self) -> list[dict[str, Any]]:
|
|
if not self._resolved_url:
|
|
raise RuntimeError("NRO delegated stats URL is not configured")
|
|
|
|
async with httpx.AsyncClient(timeout=180.0, follow_redirects=True) as client:
|
|
remote = await self._downloader.fetch_remote_info(client, self._resolved_url)
|
|
total_expected = remote.get("content_length") or 0
|
|
if total_expected > 0 and self._current_task and self._db_session:
|
|
self._current_task.total_records = total_expected
|
|
self._current_task.records_processed = 0
|
|
self._current_task.progress = 0.0
|
|
self._current_task.phase_progress = 0.0
|
|
self._current_task.phase_message = "正在下载 NRO delegated 数据"
|
|
self._current_task.phase_current = 0
|
|
self._current_task.phase_total = total_expected
|
|
self._current_task.phase_unit = "bytes"
|
|
await self._db_session.commit()
|
|
await self._publish_task_update(force=True)
|
|
|
|
async def on_progress(downloaded: int, total: int | None) -> None:
|
|
if not total or total <= 0:
|
|
return
|
|
await self.update_phase_progress(
|
|
current=min(downloaded, total),
|
|
total=total,
|
|
unit="bytes",
|
|
message="正在下载 NRO delegated 数据",
|
|
)
|
|
await self.update_progress(min(downloaded, total), commit=True)
|
|
|
|
body_path = await self._downloader.download_file(
|
|
client,
|
|
self._resolved_url,
|
|
extension=".txt",
|
|
progress_callback=on_progress,
|
|
)
|
|
body = body_path.read_text(encoding="utf-8", errors="replace")
|
|
|
|
rows: list[dict[str, Any]] = []
|
|
for raw_line in body.splitlines():
|
|
line = raw_line.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
|
|
parts = line.split("|")
|
|
if len(parts) < 7:
|
|
continue
|
|
|
|
rir = (parts[0] or "").strip().lower()
|
|
country_code = (parts[1] or "").strip().upper()
|
|
record_type = (parts[2] or "").strip().lower()
|
|
start = (parts[3] or "").strip()
|
|
value = (parts[4] or "").strip()
|
|
allocated_date = (parts[5] or "").strip()
|
|
status = (parts[6] or "").strip().lower()
|
|
|
|
if record_type not in {"ipv4", "ipv6"}:
|
|
continue
|
|
if not start or not value:
|
|
continue
|
|
|
|
rows.append(
|
|
{
|
|
"rir": rir,
|
|
"country_code": country_code,
|
|
"type": record_type,
|
|
"start": start,
|
|
"value": value,
|
|
"allocated_date": allocated_date,
|
|
"status": status,
|
|
}
|
|
)
|
|
|
|
return rows
|
|
|
|
def transform(self, raw_data: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
reference_date = datetime.now(UTC).isoformat()
|
|
transformed: list[dict[str, Any]] = []
|
|
|
|
for item in raw_data:
|
|
record_type = str(item.get("type") or "").strip().lower()
|
|
start = str(item.get("start") or "").strip()
|
|
value = str(item.get("value") or "").strip()
|
|
country_code = str(item.get("country_code") or "").strip().upper()
|
|
|
|
try:
|
|
if record_type == "ipv4":
|
|
start_ip = ipaddress.ip_address(start)
|
|
count = int(value)
|
|
if count <= 0:
|
|
continue
|
|
end_ip_int = int(start_ip) + count - 1
|
|
end_ip = ipaddress.ip_address(end_ip_int)
|
|
network = list(ipaddress.summarize_address_range(start_ip, end_ip))[0]
|
|
elif record_type == "ipv6":
|
|
prefixlen = int(value)
|
|
network = ipaddress.ip_network(f"{start}/{prefixlen}", strict=False)
|
|
start_ip = network.network_address
|
|
end_ip = network.broadcast_address
|
|
else:
|
|
continue
|
|
except (ValueError, TypeError):
|
|
continue
|
|
|
|
family = f"ipv{network.version}"
|
|
prefix = str(network)
|
|
|
|
transformed.append(
|
|
{
|
|
"source_id": f"{item.get('rir')}:{family}:{prefix}:{country_code}",
|
|
"name": prefix,
|
|
"title": f"{prefix} {country_code}".strip(),
|
|
"country": country_code,
|
|
"city": "",
|
|
"latitude": None,
|
|
"longitude": None,
|
|
"metadata": {
|
|
"family": family,
|
|
"prefix": prefix,
|
|
"range_start": str(start_ip),
|
|
"range_end": str(end_ip),
|
|
"country_code": country_code,
|
|
"rir": item.get("rir"),
|
|
"status": item.get("status"),
|
|
"allocated_date": item.get("allocated_date"),
|
|
"source_dataset": "nro_delegated_stats",
|
|
"confidence": "registry_allocated",
|
|
},
|
|
"reference_date": reference_date,
|
|
}
|
|
)
|
|
|
|
return transformed
|