Wox.Plugin.Script.IPGeolocation

Wox.Plugin.Script.IPGeolocation.py

#!/usr/bin/env python3
# {
#   "Id": "8b8a1b35-3d9e-4d7d-9f2e-3b1d0b7f9e10",
#   "Name": "IP Geolocation",
#   "Author": "qianlifeng",
#   "Version": "1.0.0",
#   "MinWoxVersion": "2.0.0",
#   "Description": "Show local/public IPs and query geolocation via ip-api.com",
#   "Icon": "emoji:📍",
#   "TriggerKeywords": ["ip"],
#   "SupportedOS": ["windows", "linux", "darwin"],
#   "I18n": {
#     "en_US": {
#       "plugin.description": "Show local/public IPs and query geolocation via ip-api.com",
#       "query.prompt": "Tips: Enter IP or domain to query",
#       "query.example": "e.g.: ip 8.8.8.8 or ip example.com",
#       "local.ip": "Local IPv4",
#       "public.ip": "Public IP",
#       "public.ip.notfound": "Public IP not found",
#       "public.ip.notfound.subtitle": "Network may be offline or endpoints blocked",
#       "geo.header": "Query",
#       "geo.type": "Type",
#       "geo.location": "Location",
#       "geo.coords": "Coordinates(lat, lon)",
#       "geo.tz.zip": "Timezone / ZIP",
#       "geo.isp": "ISP",
#       "action.copy": "Copy",
#       "action.copy.ip": "Copy IP",
#       "action.copy.location": "Copy Location",
#       "action.copy.coords": "Copy",
#       "action.open.api": "Open API",
#       "action.open.maps": "Open Maps",
#       "error.query.failed": "Query failed"
#     },
#     "zh_CN": {
#       "plugin.description": "显示本地/公网 IP，并通过 ip-api.com 查询地理位置",
#       "query.prompt": "提示: 输入 IP 或域名 进行查询",
#       "query.example": "例如: ip 8.8.8.8 或 ip example.com",
#       "local.ip": "本地 IPv4",
#       "public.ip": "公网 IP",
#       "public.ip.notfound": "未获取到公网 IP",
#       "public.ip.notfound.subtitle": "网络离线或端点被阻止",
#       "geo.header": "查询",
#       "geo.type": "类型",
#       "geo.location": "位置",
#       "geo.coords": "坐标(经度, 纬度)",
#       "geo.tz.zip": "时区 / 邮编",
#       "geo.isp": "运营商",
#       "action.copy": "复制",
#       "action.copy.ip": "复制 IP",
#       "action.copy.location": "复制位置",
#       "action.copy.coords": "复制",
#       "action.open.api": "打开 API",
#       "action.open.maps": "打开地图",
#       "error.query.failed": "查询失败"
#     }
#   }
# }

from __future__ import annotations

import datetime
import json
import os
import socket
import sys
import urllib.parse
import urllib.request
import urllib.error
from typing import Any, Dict, List, TypedDict

# Simple i18n store; can be extended or read from header if needed
I18N: Dict[str, Dict[str, str]] = {
    "en_US": {
        "plugin.description": "Show local/public IPs and query geolocation via ip-api.com",
        "query.prompt": "Enter IP or domain to query",
        "query.example": "e.g.: ip 8.8.8.8 or ip example.com",
        "local.ip": "Local IPv4",
        "public.ip": "Public IP",
        "public.ip.notfound": "Public IP not found",
        "public.ip.notfound.subtitle": "Network may be offline or endpoints blocked",
        "geo.header": "Query",
        "geo.type": "Type",
        "geo.location": "Location",
        "geo.coords": "Coordinates",
        "geo.tz.zip": "Timezone / ZIP",
        "geo.isp": "ISP",
        "action.copy": "Copy",
        "action.copy.ip": "Copy IP",
        "action.copy.location": "Copy Location",
        "action.copy.coords": "Copy",
        "action.open.api": "Open API",
        "action.open.maps": "Open Maps",
        "error.query.failed": "Query failed",
    },
    "zh_CN": {
        "plugin.description": "显示本地/公网 IP，并通过 ip-api.com 查询地理位置",
        "query.prompt": "输入 IP 或域名 进行查询",
        "query.example": "例如: ip 8.8.8.8 或 ip example.com",
        "local.ip": "本地 IPv4",
        "public.ip": "公网 IP",
        "public.ip.notfound": "未获取到公网 IP",
        "public.ip.notfound.subtitle": "网络离线或端点被阻止",
        "geo.header": "查询",
        "geo.type": "类型",
        "geo.location": "位置",
        "geo.coords": "坐标",
        "geo.tz.zip": "时区 / 邮编",
        "geo.isp": "运营商",
        "action.copy": "复制",
        "action.copy.ip": "复制 IP",
        "action.copy.location": "复制位置",
        "action.copy.coords": "复制",
        "action.open.api": "打开 API",
        "action.open.maps": "打开地图",
        "error.query.failed": "查询失败",
    },
}


class WoxPluginBase:
    class ActionItem(TypedDict, total=False):
        id: str
        name: str
        text: str
        url: str
        path: str
        message: str
        data: Any

    class QueryResult(TypedDict, total=False):
        title: str
        subtitle: str
        icon: str
        score: int
        actions: List["WoxPluginBase.ActionItem"]

    class ActionResult(TypedDict, total=False):
        action: str
        message: str

    def __init__(self):
        self.log_file_path = __file__ + ".log"

    def log(self, message: str) -> None:
        ts = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        message = f"[{ts}] {message}"
        if self.is_invoke_from_wox():
            with open(self.log_file_path, "a", encoding="utf-8") as f:
                f.write(f"{message}\n")
        else:
            print(f"LOG: {message}")

    def is_invoke_from_wox(self) -> bool:
        return "WOX_PLUGIN_ID" in os.environ

    def _build_response(self, result: Any, request_id: Any) -> dict:
        return {"jsonrpc": "2.0", "result": result, "id": request_id}

    def _build_error_response(
        self, code: int, message: str, data: Any = None, request_id: Any = None
    ) -> dict:
        error = {"code": code, "message": message}
        if data is not None:
            error["data"] = data
        return {"jsonrpc": "2.0", "error": error, "id": request_id}

    def query(
        self, raw_query: str, trigger_keyword: str, command: str, search: str
    ) -> List["WoxPluginBase.QueryResult"]:
        return [
            {
                "title": "Hello Wox!",
                "subtitle": "Override query() in your plugin class",
                "score": 100,
                "actions": [
                    {"name": "Copy", "id": "copy-to-clipboard", "text": "Hello Wox!"}
                ],
            }
        ]

    def action(self, action_id: str, data: Any) -> None:
        if action_id == "custom-action":
            pass

    def handle_query(self, params, request_id):
        search = params.get("search", "")
        trigger_keyword = params.get("trigger_keyword", "")
        command = params.get("command", "")
        raw_query = params.get("raw_query", params.get("query", ""))
        results = self.query(raw_query, trigger_keyword, command, search)
        return self._build_response({"items": results}, request_id)

    def handle_action(self, params, request_id):
        action_id = params.get("id", "")
        action_data = params.get("data", "")
        self.action(action_id, action_data)
        return self._build_response({}, request_id)

    def run(self):
        if self.is_invoke_from_wox():
            try:
                stdin_text = sys.stdin.read()
            except Exception:
                return 1
        else:
            print("Manual mode - please enter query:")
            query_input = input()
            stdin_text = f'{{"jsonrpc": "2.0", "method": "query", "params": {{"query": "{query_input}"}}, "id": 1}}'
        try:
            request = json.loads(stdin_text)
        except json.JSONDecodeError as e:
            error_response = self._build_error_response(
                -32700, "Parse error", str(e), None
            )
            print(json.dumps(error_response, ensure_ascii=False))
            return 1
        if request.get("jsonrpc") != "2.0":
            error_response = self._build_error_response(
                -32600, "Invalid Request", None, request.get("id")
            )
            print(json.dumps(error_response, ensure_ascii=False))
            return 1
        method = request.get("method")
        params = request.get("params", {})
        request_id = request.get("id")
        if method == "query":
            response = self.handle_query(params, request_id)
        elif method == "action":
            response = self.handle_action(params, request_id)
        else:
            response = self._build_error_response(
                -32601,
                "Method not found",
                f"Method '{method}' not supported",
                request_id,
            )
        print(json.dumps(response, ensure_ascii=False))
        return 0


class IPGeolocationPlugin(WoxPluginBase):
    ICON = "emoji:📍"
    TIP_ICON = "emoji:💡"
    API_BASE = "http://ip-api.com/json/"

    def _get_public_ip(self) -> str:
        # Use multiple fallback endpoints
        endpoints = [
            "https://api.ipify.org",
            "https://ipv4.icanhazip.com",
            "https://ifconfig.me/ip",
        ]
        for url in endpoints:
            try:
                with urllib.request.urlopen(url, timeout=4) as resp:
                    text = resp.read().decode("utf-8", errors="ignore").strip()
                    if text:
                        return text
            except Exception as e:
                self.log(f"public ip endpoint failed {url}: {e}")
        return ""

    def _get_local_ips(self) -> List[str]:
        ips: List[str] = []
        try:
            hostname = socket.gethostname()
            # primary
            try:
                primary = socket.gethostbyname(hostname)
                if primary and primary not in ips and not primary.startswith("127."):
                    ips.append(primary)
            except Exception:
                pass
            # enumerate via interfaces (basic)
            try:
                for addrinfo in socket.getaddrinfo(hostname, None):
                    ip = addrinfo[4][0]
                    if (
                        ip
                        and isinstance(ip, str)
                        and ip not in ips
                        and not ip.startswith("127.")
                        and ":" not in ip
                    ):
                        ips.append(ip)
            except Exception:
                pass
        except Exception as e:
            self.log(f"local ip error: {e}")
        return ips

    def _geo(self, ip_or_domain: str) -> Dict[str, Any]:
        target = ip_or_domain.strip()
        if not target:
            return {"status": "fail", "message": "empty target"}
        url = self.API_BASE + urllib.parse.quote(target)
        try:
            with urllib.request.urlopen(url, timeout=6) as resp:
                txt = resp.read().decode("utf-8", errors="ignore")
                obj = json.loads(txt)
                if isinstance(obj, dict):
                    return obj
                return {"status": "fail", "message": "invalid json"}
        except urllib.error.HTTPError as e:
            return {"status": "fail", "message": f"http {e.code}"}
        except urllib.error.URLError as e:
            return {"status": "fail", "message": f"network {e.reason}"}
        except Exception as e:
            return {"status": "fail", "message": str(e)}

    def _format_geo_title(self, data: Dict[str, Any]) -> str:
        # Fallback title when needed
        return "Geolocation Result"

    def _build_geo_items(
        self, query: str, data: Dict[str, Any]
    ) -> List[WoxPluginBase.QueryResult]:
        status = data.get("status")
        if status != "success":
            msg = data.get("message") or "Query failed"
            return [
                {
                    "title": f"{query}",
                    "subtitle": str(msg),
                    "icon": self.ICON,
                    "actions": [
                        {
                            "name": "Open API",
                            "id": "open-url",
                            "url": self.API_BASE + urllib.parse.quote(query),
                        },
                        {
                            "name": "Copy Message",
                            "id": "copy-to-clipboard",
                            "text": str(msg),
                        },
                    ],
                }
            ]

        ip = data.get("query") or query
        items: List[WoxPluginBase.QueryResult] = []

        # Header item: the IP queried
        items.append(
            {
                "title": "i18n:geo.header",
                "subtitle": f"{ip} | {data.get('as', '')}",
                "icon": self.ICON,
                "actions": [
                    {
                        "name": "i18n:action.copy.ip",
                        "id": "copy-to-clipboard",
                        "text": str(ip),
                    },
                    {
                        "name": "i18n:action.open.api",
                        "id": "open-url",
                        "url": self.API_BASE + urllib.parse.quote(ip),
                    },
                ],
            }
        )

        # Location details
        country = data.get("country")
        region = data.get("regionName")
        city = data.get("city")
        zip_code = data.get("zip")
        timezone = data.get("timezone")
        lat = data.get("lat")
        lon = data.get("lon")
        isp = data.get("isp")
        org = data.get("org")

        if country or region or city:
            items.append(
                {
                    "title": "i18n:geo.location",
                    "subtitle": " ".join(
                        [str(v) for v in [country, region, city] if v]
                    ),
                    "icon": self.ICON,
                    "actions": [
                        {
                            "name": "i18n:action.copy.location",
                            "id": "copy-to-clipboard",
                            "text": " ".join(
                                [str(v) for v in [country, region, city] if v]
                            ),
                        },
                        {
                            "name": "i18n:action.open.maps",
                            "id": "open-url",
                            "url": f"https://www.google.com/maps/search/?api=1&query={lat},{lon}",
                        }
                        if lat is not None and lon is not None
                        else {
                            "name": "i18n:action.copy.coords",
                            "id": "copy-to-clipboard",
                            "text": f"{lat},{lon}",
                        },
                    ],
                }
            )

        if zip_code or timezone:
            items.append(
                {
                    "title": "i18n:geo.tz.zip",
                    "subtitle": " | ".join(
                        [
                            s
                            for s in [
                                str(zip_code) if zip_code else None,
                                str(timezone) if timezone else None,
                            ]
                            if s
                        ]
                    ),
                    "icon": self.ICON,
                }
            )

        if isp or org:
            items.append(
                {
                    "title": "i18n:geo.isp",
                    "subtitle": " | ".join(
                        [
                            s
                            for s in [
                                str(isp) if isp else None,
                                str(org) if org else None,
                            ]
                            if s
                        ]
                    ),
                    "icon": self.ICON,
                }
            )

        # Coordinates item explicitly
        if lat is not None and lon is not None:
            items.append(
                {
                    "title": "i18n:geo.coords",
                    "subtitle": f"{lat}, {lon}",
                    "icon": self.ICON,
                    "actions": [
                        {
                            "name": "i18n:action.open.maps",
                            "id": "open-url",
                            "url": f"https://www.google.com/maps/search/?api=1&query={lat},{lon}",
                        },
                        {
                            "name": "i18n:action.copy.coords",
                            "id": "copy-to-clipboard",
                            "text": f"{lat},{lon}",
                        },
                    ],
                }
            )

        return items

    def query(
        self, raw_query: str, trigger_keyword: str, command: str, search: str
    ) -> List[WoxPluginBase.QueryResult]:
        items: List[WoxPluginBase.QueryResult] = []

        # If user provides a query, perform geolocation for that IP/domain
        if search:
            data = self._geo(search)
            items.extend(self._build_geo_items(search, data))
            return items

        # No query: show local and public IPs, with actions to geolocate
        local_ips = self._get_local_ips()
        public_ip = self._get_public_ip()

        for lip in local_ips:
            items.append(
                {
                    "title": "i18n:local.ip",
                    "subtitle": lip,
                    "icon": self.ICON,
                    "actions": [
                        {
                            "name": "i18n:action.copy",
                            "id": "copy-to-clipboard",
                            "text": lip,
                        },
                        {
                            "name": "i18n:action.open.api",
                            "id": "open-url",
                            "url": self.API_BASE + urllib.parse.quote(lip),
                        },
                    ],
                }
            )
        if public_ip:
            items.append(
                {
                    "title": "i18n:public.ip",
                    "subtitle": public_ip,
                    "icon": self.ICON,
                    "actions": [
                        {
                            "name": "i18n:action.copy",
                            "id": "copy-to-clipboard",
                            "text": public_ip,
                        },
                        {
                            "name": "i18n:action.open.api",
                            "id": "open-url",
                            "url": self.API_BASE + urllib.parse.quote(public_ip),
                        },
                    ],
                }
            )
        else:
            items.append(
                {
                    "title": "i18n:public.ip.notfound",
                    "subtitle": "i18n:public.ip.notfound.subtitle",
                    "icon": self.ICON,
                }
            )

        # Helper tip
        items.append(
            {
                "title": "i18n:query.prompt",
                "subtitle": "i18n:query.example",
                "icon": self.TIP_ICON,
            }
        )
        return items


if __name__ == "__main__":
    sys.exit(IPGeolocationPlugin().run())