chore: optimize ipv6 strip logic

This commit is contained in:
aclist 2026-05-27 15:49:18 +09:00
parent b5d9c030dd
commit 9266473853
4 changed files with 39 additions and 64 deletions

View File

@ -23,40 +23,37 @@ def find_download_url(url: str) -> str | None:
def get_ipdb(ips_path: Path) -> None:
url = find_download_url(DB_IP)
if url is None:
return
date = find_date(url)
month_file = ips_path.parent / ".month"
if ips_path.exists() and month_file.exists():
old_date = month_file.read_text().rstrip("\n")
if old_date == date:
logger.info(f"IP DB date matches: {date}")
try:
url = find_download_url(DB_IP)
if url is None:
return
# TODO: log additional output
logger.info(f"Fetching IPDB for {date} from {url}")
try:
date = find_date(url)
month_file = ips_path.parent / ".month"
if ips_path.exists() and month_file.exists():
old_date = month_file.read_text().rstrip("\n")
if old_date == date:
logger.info(f"IP DB date matches: {date}")
return
# TODO: log additional output
logger.info(f"Fetching IPDB for {date} from {url}")
tmp = serialize(url)
except Exception as e:
logger.critical(e)
return
logger.info(f"Extracting {tmp}")
unzip(tmp, ips_path)
logger.info(f"Extracting {tmp}")
unzip(tmp, ips_path)
logger.info("Stripping IPv6 records")
try:
logger.info("Stripping IPv6 records")
strip_ipv6(ips_path)
print("here")
with open(month_file, "w") as f:
f.write(date)
logger.info(f"Wrote {date} to {month_file}")
except Exception as e:
# NOTE: in the event of failure, geolocation calc is simply not performed
logger.critical(e)
return
with open(month_file, "w") as f:
f.write(date)
logger.info(f"Wrote {date} to {month_file}")
def find_date(url: str) -> str:
date = re.sub(r"(dbip-city-lite-)(.*)(.csv.gz)", r"\2", url)
@ -81,19 +78,16 @@ def serialize(url: str) -> Path:
return tmp
# TODO: optimize this function
# consider using grep
# cf. grep -vE "^[a-z0-9]{4}:" | grep -v "::" > "$ip_file"
def strip_ipv6(path: Path) -> None:
ips = []
with open(path, "r") as f:
s = f.read()
reg = r"^\d{1,3}\..*"
ips = re.findall(reg, s, re.MULTILINE)
assert ips[0].split(",")[0] == "0.0.0.0"
assert ips[-1].split(",")[0] == "224.0.0.0"
with open(path, "w") as f:
for ip in ips:
f.write(ip + "\n")
alt_path = path.parent.joinpath("stripped.csv")
with open(path, "r") as f, open(alt_path, "w") as out:
for line in f:
reg = r"^\d{1,3}\..*"
m = re.match(reg, line)
if not m:
continue
els = line.split(",")
final = ",".join([els[0], els[1], els[6], els[7]])
out.write(final)
path.unlink()
alt_path.rename(path.name)

View File

@ -13,11 +13,6 @@ logger = logging.getLogger(APP_NAME)
def get_local_coords(path: "Path") -> Coords | None:
try:
my_ip = get_local_ip()
except Exception as e:
logger.warn(e)
return None
try:
return get_coords(path, my_ip)
except Exception as e:
logger.warn(e)

View File

@ -133,9 +133,6 @@ def main() -> None:
# remove_stale_signatures(XDG.config, XDG.version)
## TODO: handle IP DB failure and use coords fallback
## TODO: drop this after dialog is complete
# print("Fetching geolocation data, may take some time...")
# get_ipdb(XDG.ips)
# local_coords = get_local_coords(XDG.ips)
# latest_release = check_updates(version)

View File

@ -15,8 +15,8 @@ from dzgui.const.constants import EXPAND, FILL
from dzgui.managers.threading import call_on_thread, StoredFunc, ThreadingManager
from dzgui.util.strings import dialog_header
from dzgui.util.symlink import rebuild_symlinks
from dzgui.views.components.buttons import ClipboardButton
# from dzgui.views.components.buttons import ClipboardButton
import gi
@ -37,15 +37,9 @@ class Success(Enum):
class BootDialog(Gtk.Dialog):
def __init__(self, parent: "BootWindow", xdg: "Xdg", version: str) -> None:
super().__init__(
# text="TEST",
# buttons=Gtk.ButtonsType.NONE,
# message_type=Gtk.MessageType.INFO,
# secondary_text="BOOTING UP",
# TODO: strings
title=dialog_header,
parent=parent,
modal=True,
# flags=Gtk.DialogFlags.MODAL,
)
self.parent = parent
@ -60,14 +54,12 @@ class BootDialog(Gtk.Dialog):
self.view = Gtk.TreeView(enable_search=False, headers_visible=False)
self.view.set_model(self.store)
# self.view.set_fixed_height_mode(True)
for i, column_title in enumerate(["Task", "State"]): # , "Spinner"]):
for i, column_title in enumerate(["Task", "State"]):
renderer = Gtk.CellRendererText()
column = Gtk.TreeViewColumn(column_title, renderer, text=i)
column.set_sizing(Gtk.TreeViewColumnSizing.FIXED)
column.set_resizable(False)
# column.set_expand(True)
column.set_sort_column_id(i)
column.set_cell_data_func(renderer, self._format_color, func_data=None)
self.view.append_column(column)
@ -108,10 +100,8 @@ class BootDialog(Gtk.Dialog):
)
self.error_label = Gtk.Label()
# TODO: recycle ClipboardButton from generics
# TODO: abstract class
# self.copy_button = ClipboardButton()
self.copy_button = Gtk.Button(label="Copy")
self.copy_button = ClipboardButton(None, lambda: self.error_label.get_text())
self.button_hbox = Gtk.Box(
orientation=Gtk.Orientation.HORIZONTAL, halign=Gtk.Align.CENTER, spacing=10
)
@ -166,7 +156,6 @@ class BootDialog(Gtk.Dialog):
return True
def run(self) -> None:
GLib.timeout_add(100, self.pulse_spinner)
self.iter_step()
def _on_delete(self, widget: Self, event: Gdk.Event) -> Literal[True]:
@ -190,7 +179,8 @@ class BootDialog(Gtk.Dialog):
@call_on_thread("", show_dialog=False)
def background(self, func: StoredFunc, store_output: bool) -> None:
# time.sleep(0.1)
# FIXME: freezes during heavy IO
GLib.timeout_add(100, self.pulse_spinner)
try:
if store_output:
res = func.call()
@ -209,7 +199,6 @@ class BootDialog(Gtk.Dialog):
def update_task(self, task: str) -> None:
self.store.append((task, "Running", True, 0))
# TODO: use enum
def update_status(self, state: Success) -> None:
d = {Success.OK: "OK", Success.FAIL: "FAILED"}
msg = d[state]