chore: add newline

dzgui/app_init.py

@@ -121,7 +121,7 @@ def load_steam_icon() -> None:
         logger.critical(e)
 
 
-def load_gui(version: str, is_debug: bool) -> None:
+def build_gui(version: str, is_debug: bool) -> App | None:
     lock = lock_acquire()  # noqa
 
     set_locale()
@@ -151,7 +151,7 @@ def load_gui(version: str, is_debug: bool) -> None:
         wizard = SetupWizard(_is_steam_deck, XDG, xdg_paths)
 
         if not wizard.is_setup_complete():
-            return
+            return None
 
     setup_logger(XDG.debug)
     with open(XDG.debug, "w") as f:
@@ -186,4 +186,4 @@ def load_gui(version: str, is_debug: bool) -> None:
     print(boot.all_ok)
     load_steam_icon()
     app = App(prefs)
-    app.run(None)
+    return app

dzgui/config/ipdb.py

@@ -3,9 +3,6 @@ import logging
 import re
 import requests
 import shutil
-import socket
-import sqlite3
-import time
 
 from pathlib import Path
 
@@ -25,16 +22,15 @@ def find_download_url(url: str) -> str | None:
     return str(urls[0])
 
 
-def get_ipdb(db_path: Path) -> None:
+def get_ipdb(ips_path: Path) -> None:
     try:
         url = find_download_url(DB_IP)
         if url is None:
             return
 
-        ips_path = db_path.with_suffix(".csv")
-        month_file = db_path.parent / ".month"
         date = find_date(url)
-        if db_path.exists() and month_file.exists():
+        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}")
@@ -49,9 +45,6 @@ def get_ipdb(db_path: Path) -> None:
         logger.info("Stripping IPv6 records")
         strip_ipv6(ips_path)
 
-        create_table(db_path, ips_path)
-        ips_path.unlink()
-
         with open(month_file, "w") as f:
             f.write(date)
         logger.info(f"Wrote {date} to {month_file}")
@@ -115,95 +108,3 @@ def strip_ipv6(path: Path) -> None:
             out.write(merged)
     path.unlink()
     alt_path.rename(path)
-
-
-def ip_to_int(ip: str) -> int:
-    return int.from_bytes(socket.inet_aton(ip), "big")
-
-
-def marshal_rows(ip_file: Path) -> list[tuple[int, int, str, str]]:
-    YIELD = 0.003
-    SLEEP = 0.001
-    with open(ip_file, "r") as f:
-        rows: list[tuple[int, int, str, str]] = []
-        # NOTE: yield to GIL periodically to reduce load on spinner
-        last_yield = time.monotonic()
-        for row in f:
-            start, end, lat, lon = row.rstrip("\n").split(",")
-            int_start = ip_to_int(start)
-            int_end = ip_to_int(end)
-            rows.append((int_start, int_end, lat, lon))
-
-            now = time.monotonic()
-            if (now - last_yield) >= YIELD:
-                time.sleep(SLEEP)
-                last_yield = time.monotonic()
-    return rows
-
-
-def create_table(db_file: Path, ip_file: Path) -> None:
-    create = """
-    CREATE TABLE IF NOT EXISTS ips (
-    start_range INTEGER PRIMARY KEY,
-    end_range INTEGER,
-    latitude REAL,
-    longitude REAL
-    ) WITHOUT ROWID;
-    """
-    insert = """
-        INSERT INTO ips
-        (start_range, end_range, latitude, longitude)
-        values(?, ?, ?, ?)
-    """
-
-    if db_file.exists():
-        db_file.unlink()
-
-    conn = sqlite3.connect(":memory:")
-    c = conn.cursor()
-
-    try:
-        c.execute(create)
-        rows = marshal_rows(ip_file)
-        c.executemany(insert, rows)
-        logger.info(f"Inserted {len(rows)} IP records into database")
-        conn.commit()
-        with sqlite3.connect(db_file) as dest:
-            conn.backup(dest)
-        logger.info(f"Serialized IP records to database: '{db_file}'")
-        conn.close()
-    except Exception as e:
-        logger.critical(e)
-
-
-def read_from_table(db_path: Path, ip: str) -> tuple[float, float]:
-    if not db_path.exists():
-        raise FileNotFoundError(f"No file found at the location '{db_path}")
-
-    conn = sqlite3.connect(db_path)
-    c = conn.cursor()
-
-    int_ip = ip_to_int(ip)
-
-    query = """
-    SELECT end_range, latitude, longitude
-    FROM ips
-    WHERE start_range <= ?
-    ORDER BY start_range DESC
-    LIMIT 1
-    """
-    index = """
-    CREATE INDEX IF NOT EXISTS start_index
-    ON ips(start_range);
-    """
-    try:
-        c.execute(index)
-        conn.commit()
-        c.execute(query, (int_ip,))
-        row = c.fetchone()
-        end, lat, lon = row
-        conn.close()
-        return lat, lon
-    except Exception as e:
-        logger.critical(e)
-        raise e

dzgui/config/xdg.py

@@ -22,7 +22,7 @@ class Xdg:
     version: Path
     system: Path
     debug: Path
-    db: Path
+    ips: Path
     filters: Path
     shortcut: Path
 
@@ -88,7 +88,7 @@ def parse_filepaths(xdg: dict) -> Xdg:
     resolution = state / "dzg.res.json"
     history = state / "dzg.history"
     versions = state / "dzg.versions"
-    db = state / "ips.db"
+    ips = state / "ips.csv"
     filters = state / "dzg.filters.json"
 
     shortcut = share / APP_NAME_LOWER
@@ -102,7 +102,7 @@ def parse_filepaths(xdg: dict) -> Xdg:
         versions,
         system,
         debug,
-        db,
+        ips,
         filters,
         shortcut,
     )

dzgui/init/coords.py

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

dzgui/main.py

@@ -42,6 +42,9 @@ def main() -> None:
     test_libgi_missing()
 
     # NOTE: only import ligbirepository modules after the above check
-    from dzgui.app_init import load_gui
+    from dzgui.app_init import build_gui
 
-    load_gui(version, args.debug)
+    app = build_gui(version, args.debug)
+    if app is None:
+        sys.exit(0)
+    app.run(None)

dzgui/util/dist.py

@@ -80,9 +80,8 @@ class CalcDist:
         if local is None:
             return None
         try:
-            remote = get_coords(prefs.paths.db, remote_ip)
-        except Exception as e:
-            logger.critical(e)
+            remote = get_coords(prefs.paths.ips, remote_ip)
+        except Exception:
             return None
 
         haversine = Haversine(local.lat, local.lon, remote.lat, remote.lon)

dzgui/util/ip.py

@@ -1,5 +1,4 @@
 import a2s
-from ipaddress import ip_address
 import requests
 import shutil
 import subprocess
@@ -8,7 +7,6 @@ from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
 from dzgui.api.servers import Record
-from dzgui.config.ipdb import read_from_table
 from dzgui.const.constants import REQUEST_TIMEOUT
 from dzgui.const.endpoints import IP_ECHO
 
@@ -35,21 +33,54 @@ def ip2list(ip: str) -> list:
     return [int(digit) for digit in ip.split(".")]
 
 
-def is_usable(ip: str) -> bool:
-    address = ip_address(ip)
-    if address.is_reserved or address.is_private:
-        return False
-    return True
+# TODO: tests
+def is_reserved(digits: list) -> bool:
+    reserved = [0, 10, 127]
+
+    if digits[0] in reserved:
+        return True
+    if digits[0] == 192 and digits[1] == 168:
+        return True
+    return False
 
 
-def get_coords(db_path: "Path", ip: str) -> Coords:
-    if not is_usable(ip):
+def get_coords(ips: "Path", ip: str) -> Coords:
+    split = ip2list(ip)
+
+    if is_reserved(split):
         raise ReservedIPError(f"{ip} is a reserved IP address.")
-    try:
-        lat, lon = read_from_table(db_path, ip)
-        return Coords(lat, lon)
-    except Exception:
-        raise GeolocationError("No matching records found")
+
+    prefix = f"^{split[0]}.{split[1]}."
+
+    if shutil.which("rg") is not None:
+        args = ["/usr/bin/rg", prefix, str(ips)]
+    else:
+        args = ["/usr/bin/grep", "-E", prefix, str(ips)]
+    proc = subprocess.run(args, capture_output=True, text=True)
+
+    if proc.returncode != 0:
+        raise GeolocationError("Failed to split records")
+
+    ip_list = proc.stdout.splitlines()
+
+    for address in ip_list:
+        fields = address.split(",")
+        upper = fields[1]
+
+        upper_digits = ip2list(upper)
+        if ip == upper:
+            return Coords(float(fields[-2]), float(fields[-1]))
+
+        if split[2] > upper_digits[2]:
+            continue
+
+        if split[2] == upper_digits[2]:
+            if split[3] > upper_digits[3]:
+                continue
+
+        if split[2] <= upper_digits[2]:
+            return Coords(float(fields[-2]), float(fields[-1]))
+    raise GeolocationError("No matching records found")
 
 
 def get_local_ip() -> str:

dzgui/util/save_screenshot.py

@@ -0,0 +1,80 @@
+from pathlib import Path
+from typing import cast, TYPE_CHECKING
+
+import gi
+
+gi.require_version("Gtk", "3.0")
+from gi.repository import Gtk, Gdk, GLib  # noqa E402
+
+if TYPE_CHECKING:
+    from dzgui.controllers.mc import Controller
+    from dzgui.views.base import OuterWindow
+    from dzgui.views.mixins.context_mixin import ContextMixin
+    from gi.repository import GdkPixbuf
+
+# TODO: randomize/mock PII distance values
+
+
+class IdleScreenshot:
+    def __init__(
+        self, controller: "Controller", window: "OuterWindow", scrot_path: Path
+    ) -> None:
+        self.window = window
+        self.controller = controller
+        self.scrot_path = scrot_path
+
+        clock = window.get_frame_clock()
+        if clock is None:
+            raise TypeError("Widget not realized, FrameClock not initialized")
+        clock.connect("after-paint", self.callback)
+
+    def callback(self, clock: Gdk.FrameClock) -> None:
+        GLib.timeout_add(1000, self.check_idle)
+
+    def check_idle(self) -> bool:
+        tv = self.controller.get_active_treeview()
+        loaded = tv.is_loaded()
+        if loaded is True:
+            self.emit_key_press(tv)
+            GLib.timeout_add(1000, self.save_and_quit)
+            return True
+        return False
+
+    def emit_key_press(self, widget: "ContextMixin") -> None:
+        window = widget.get_parent_window()
+        if window is None:
+            raise TypeError(f"Failed to find a valid parent window for {widget}")
+        event = Gdk.EventKey.new(Gdk.EventType.KEY_PRESS)
+        # NOTE: GTK introspection returns Gdk.Event type
+        event_key = cast(Gdk.EventKey, event)
+        event_key.keyval = Gdk.KEY_l
+        event_key.state = Gdk.ModifierType.CONTROL_MASK
+        event_key.window = window
+        event_key.time = Gdk.CURRENT_TIME
+        event_key.send_event = True
+        # TODO: suppress errors from synthetic key events
+        widget.present_menu(widget, event)
+
+    def save_and_quit(self) -> None:
+        self.save_pixbuf()
+        self.controller.save_res_and_quit()
+
+    def save_pixbuf(self) -> None:
+        pixbuf = self.capture_application()
+        if pixbuf:
+            pixbuf.savev(str(self.scrot_path), "png", [], [])
+
+    def capture_application(self) -> "GdkPixbuf.Pixbuf":
+        # NOTE: composited context menu over X11 window
+        root = Gdk.get_default_root_window()
+
+        xwindow = self.window.get_window()
+        if xwindow is None:
+            raise ValueError("Failed to get a valid X11 window")
+        index, x, y = xwindow.get_origin()
+        width, height = self.window.get_size()
+
+        application = Gdk.pixbuf_get_from_window(root, x, y, width, height)
+        if application is None:
+            raise ValueError("Failed to get a valid GdkPixbuf")
+        return application

dzgui/views/base.py

@@ -1,7 +1,7 @@
 import logging
 import signal
 
-from typing import TYPE_CHECKING, Literal, Self
+from typing import Any, Literal, Self, TYPE_CHECKING
 
 from dzgui.const.constants import (
     APP_NAME,
@@ -47,7 +47,7 @@ logger = logging.getLogger(APP_NAME)
 
 
 class OuterWindow(Gtk.ApplicationWindow):
-    def __init__(self, app: Gtk.Application) -> None:
+    def __init__(self, app: "App") -> None:
         super().__init__(
             application=app, title=APP_NAME, border_width=10, icon_name=APP_NAME_LOWER
         )
@@ -78,6 +78,15 @@ class OuterWindow(Gtk.ApplicationWindow):
         MainController.loaded = True
         MainController.populate_model(MainController.get_active_treeview())
 
+        try:
+            from dzgui.util.save_screenshot import IdleScreenshot
+
+            scrot = app.get_automation_param("screenshot_path")
+            self.fullscreen()
+            IdleScreenshot(MainController, self, scrot)
+        except KeyError:
+            pass
+
     def _on_delete_event(self, window: "OuterWindow", event: Gdk.EventKey) -> None:
         self.halt_proc_and_quit()
 
@@ -291,6 +300,14 @@ class App(ColorAwareApp, Gtk.Application):  # type: ignore
         self._setup_signals()
         self.connect("activate", self._on_activate)
 
+        self.automation_params: dict[str, Any] = {}
+
+    def get_automation_param(self, key: str) -> Any:
+        return self.automation_params[key]
+
+    def set_automation_param(self, key: str, value: Any) -> None:
+        self.automation_params[key] = value
+
     def _on_activate(self, app: Self) -> None:
         self.win = OuterWindow(app)
 
@@ -331,4 +348,5 @@ class App(ColorAwareApp, Gtk.Application):  # type: ignore
     ) -> None:
         self.win.halt_proc_and_quit()
 
+
 MainController = Controller()

dzgui/views/dialogs/boot.py

@@ -132,9 +132,9 @@ class BootDialog(ColorAwareApp, Gtk.Dialog):  # type: ignore
                 preboot.signatures,
                 False,
             ),
-            (StoredFunc(get_ipdb, self.xdg.db), preboot.geo, False),
+            (StoredFunc(get_ipdb, self.xdg.ips), preboot.geo, False),
             # TODO: handle IP DB failure and use coords fallback
-            (StoredFunc(get_local_coords, self.xdg.db), preboot.coords, True),
+            (StoredFunc(get_local_coords, self.xdg.ips), preboot.coords, True),
             (StoredFunc(check_updates, self.version), preboot.updates, True),
         ]
         self.results: list[Any] = []

dzgui/views/mixins/context_mixin.py

@@ -16,9 +16,7 @@ logger = logging.getLogger(APP_NAME)
 
 class ContextMixin(TreeView):
     def present_menu(
-        self,
-        widget: Gtk.Widget,
-        event: Gdk.EventButton | Gdk.EventKey,
+        self, widget: Gtk.Widget, event: Gdk.Event
     ) -> bool:
 
         if self.menu is None:

scripts/parse_toml.py

@@ -3,6 +3,7 @@ from pathlib import Path
 
 root = Path(__file__).resolve().parents[1]
 
+
 def get_entrypoint() -> None:
     p = root.joinpath("pyproject.toml")
     with open(p, "rb") as f:

tests/test_geo_file.py

@@ -1,40 +1,16 @@
+from dzgui.config.ipdb import get_ipdb
+from pathlib import Path
 import os
-import sqlite3
 import pytest
 import tempfile
 
-from pathlib import Path
-
-from dzgui.const.endpoints import DB_IP
-from dzgui.config.ipdb import (
-    create_table,
-    find_download_url,
-    serialize,
-    strip_ipv6,
-    unzip,
-)
-
-
-def get_db_len(path: Path) -> int:
-    conn = sqlite3.connect(path)
-    c = conn.cursor()
-    c.execute("SELECT COUNT(*) from ips")
-    db_len = c.fetchone()[0]
-    return db_len
-
-
-@pytest.fixture
-def csv() -> Path:
-    csv = Path(tempfile.NamedTemporaryFile().name)
-    url = find_download_url(DB_IP)
-    tmp = serialize(url)
-    unzip(tmp, csv)
-    strip_ipv6(csv)
-    return csv
 
 @pytest.mark.slow
-def test_stripping(csv):
-    with open(csv, "rb") as f:
+def test_stripping():
+    d = tempfile.TemporaryDirectory()
+    path = Path(d.name).joinpath("ips.csv")
+    get_ipdb(path)
+    with open(path, "rb") as f:
         first = f.readline().decode().rstrip()
         try:
             f.seek(-2, os.SEEK_END)
@@ -49,16 +25,3 @@ def test_stripping(csv):
 
         assert first_els[0] == "0.0.0.0"
         assert last_els[0] == "224.0.0.0"
-
-
-@pytest.mark.slow
-def test_row_parity(csv) -> None:
-    rows = csv.read_text().splitlines()
-
-    db = Path(tempfile.NamedTemporaryFile().name)
-    create_table(db, csv)
-
-    csv_len = len(rows)
-    db_len = get_db_len(db)
-
-    assert csv_len == db_len