Wox.Plugin.Script.60s

Wox.Plugin.Script.60s.py

#!/usr/bin/env python3
# {
#   "Id": "6f920562-1570-4e63-991b-0b39e7d4a2e9",
#   "Name": "每日 60 秒新闻",
#   "Author": "qianlifeng",
#   "Version": "1.0.1",
#   "MinWoxVersion": "2.0.0",
#   "Description": "基于 https://60s.viki.moe/v2/60s 的每日新闻展示",
#   "Icon": "base64:data:image/jpeg;base64,/9j/2wCEAAgGBgcGBQgHBwcJCQgKDBQNDAsLDBkSEw8UHRofHh0aHBwgJC4nICIsIxwcKDcpLDAxNDQ0Hyc5PTgyPC4zNDIBCQkJDAsMGA0NGDIhHCEyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMv/AABEIADAAMAMBIgACEQEDEQH/xAGiAAABBQEBAQEBAQAAAAAAAAAAAQIDBAUGBwgJCgsQAAIBAwMCBAMFBQQEAAABfQECAwAEEQUSITFBBhNRYQcicRQygZGhCCNCscEVUtHwJDNicoIJChYXGBkaJSYnKCkqNDU2Nzg5OkNERUZHSElKU1RVVldYWVpjZGVmZ2hpanN0dXZ3eHl6g4SFhoeIiYqSk5SVlpeYmZqio6Slpqeoqaqys7S1tre4ubrCw8TFxsfIycrS09TV1tfY2drh4uPk5ebn6Onq8fLz9PX29/j5+gEAAwEBAQEBAQEBAQAAAAAAAAECAwQFBgcICQoLEQACAQIEBAMEBwUEBAABAncAAQIDEQQFITEGEkFRB2FxEyIygQgUQpGhscEJIzNS8BVictEKFiQ04SXxFxgZGiYnKCkqNTY3ODk6Q0RFRkdISUpTVFVWV1hZWmNkZWZnaGlqc3R1dnd4eXqCg4SFhoeIiYqSk5SVlpeYmZqio6Slpqeoqaqys7S1tre4ubrCw8TFxsfIycrS09TV1tfY2dri4+Tl5ufo6ery8/T19vf4+fr/2gAMAwEAAhEDEQA/AO886MdWA9zxTwQRkc0pHNM2xpl9qrgctjFfCH0VyB79FlmjVHdolZnCYJCqoZmIzwMEfXoKtA5rlrnSop7uTUrezna8l2lVUboz8oG4naeO2AecdOajsbi8k1yEz3N1lS7Sq4McaoFOcLwMAleTk89a9KWCvT546WWt+vocyre9ZnUiUGZowCdoBJ9PapKx7LWEub4xQwN5DsdszNy59lx046n8q2K4KlKdN2mrG8ZKWwVy+sapqMOvQ26wK2nxmOSYcbpF3AsMHqCAQAOpzk1peI1mk0aWKCcwPKQgkXOV/Ij6de9U/DvhqbUrSWOWb5YZQsUxJ3qCoLYHTb7HuM9ea9HLsPf961fy/U58RUsuTYg+KWna9rcFlNoV/wCRCNkqy+aYlI+bJ3/wn7p5IzxjO2sTWvElzoPh5XuoIdUFpHGqtJGR8xCK5BIyELhjgjGNowMjHsk81npVg007xwWsQ5J4AH/6z+NZ0FnoOu2purWOKRGyDJGCh5xkEccHg4IweDjpX0d+jPLS6nDaCBqfhqHxBaW32HbC1zJb7iRsJdSQWzyNrEDoRt6ZxXS26SJAiSuHcDBIB5/MmuI1lNS1BfIsZrwksRtAlRXXaQOen93lsgAY6cV29usi28azMGlCAOw6E45NeBm3I3Fx3PRwnMlZhcQJcwNE4+Vh+XvTvD9x/Y5ktryNisr7kniBZenRhjK/Xke4p9FcGGxdTDv3djoq0Y1FqR+KxNq3h60lSAsq3AaeKNt5VSjr/D15ZT7de1Ytkh8L+Gbi3AMM984it4ZHJbYB8znJznk/+O1u7cNuBKt/eUkH8xTfKTzjMV3SkYMjcsR6ZPNej/a6cdY6+pzfU3fc5/z9WuDlfNC9kjiCL+bc/rVzT4NQW4Dzs6x4+ZXk3E/zrXorzp4rmi4qKS9DpjStrc//2Q==",
#   "TriggerKeywords": ["60s"],
#   "SupportedOS": [
#     "windows",
#     "linux",
#     "darwin"
#   ]
# }


from __future__ import annotations

import datetime
import json
import os
import sys
from typing import Any, List, TypedDict, Tuple, Dict

import urllib.request
import urllib.parse
import urllib.error


class WoxPluginBase:
    """Wox plugin base class for script plugins. Don't modify this class directly."""

    class ActionItem(TypedDict, total=False):
        """Type hint for an action in a result."""

        id: str
        name: str
        text: str
        url: str
        path: str
        message: str
        data: Any

    class QueryResult(TypedDict, total=False):
        """Type hint for a query result item."""

        title: str
        subtitle: str
        icon: str
        """
        support following icon formats:
        - "base64:data:image/png;base64,xxx"
        - "emoji:😀"
        - "svg:<svg>...</svg>"
        - "fileicon:/absolute/path/to/file" (get system file icon)
        - "absolute:/absolute/path/to/image.png"
        - "url:https://example.com/image.png"
        """
        score: int
        actions: List[WoxPluginBase.ActionItem]

    class ActionResult(TypedDict, total=False):
        """Type hint for action method return value."""

        action: str
        message: str

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

    def log(self, message: str) -> None:
        """Log message to log file for debugging."""
        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:
        """Check if the script is invoked from Wox."""
        return "WOX_PLUGIN_ID" in os.environ

    def _build_response(self, result: Any, request_id: Any) -> dict:
        """Build a successful JSON-RPC 2.0 response."""
        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:
        """Build a JSON-RPC 2.0 error response."""
        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]:
        """
        Process user queries and return results.

        Args:
            raw_query: The full raw query string entered by the user
            trigger_keyword: The trigger keyword that activated this plugin
            command: The command part of the query (after trigger keyword)
            search: The search part of the query (after command)

            take `wpm install plugin-name` as an example
            raw_query: wpm install plugin-name
            trigger_keyword: wpm
            command: install
            search: plugin-name

        Returns:
            List of result dictionaries
        """
        # Default implementation - override in your plugin class
        return [
            {
                "title": "Hello Wox!",
                "subtitle": "This is a default result. Override the query method 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:
        """
        Handle action requests (OPTIONAL - only needed for custom actions)

        Built-in actions (copy-to-clipboard, open-url, open-directory, notify) are handled
        automatically by Wox. You only need to implement this method if you have custom actions.

        Note: This method is called as a hook for ALL actions (built-in and custom), so you can
        optionally add additional logic for built-in actions if needed.

        Args:
            action_id: The ID of the action
            data: Additional data for the action
        """
        # Default implementation - override for custom actions
        if action_id == "custom-action":
            # Custom action handling logic can be added here
            pass

    def handle_query(self, params, request_id):
        """Internal method to handle query requests"""

        search = params.get("search", "")
        trigger_keyword = params.get("trigger_keyword", "")
        command = params.get("command", "")
        raw_query = params.get("raw_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):
        """Internal method to handle action requests"""
        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):
        """Main entry point for the script plugin"""
        # Parse input
        if self.is_invoke_from_wox():
            # Running from Wox - read from stdin
            try:
                stdin_text = sys.stdin.read()
            except Exception:
                return 1
        else:
            # Manual testing mode
            print("Manual mode - please enter query:")
            query_input = input()
            stdin_text = f'{{"jsonrpc": "2.0", "method": "query", "params": {{"query": "{query_input}"}}, "id": 1}}'

        # Parse JSON-RPC 2.0 request
        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

        # Validate JSON-RPC 2.0 format
        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")

        # Handle different methods
        if method == "query":
            response = self.handle_query(params, request_id)
        elif method == "action":
            response = self.handle_action(params, request_id)
        else:
            # Method not found
            response = self._build_error_response(
                -32601,
                "Method not found",
                f"Method '{method}' not supported",
                request_id,
            )

        # Output response
        print(json.dumps(response, ensure_ascii=False))
        return 0

class _60sPlugin(WoxPluginBase):
    API_URL = "https://60s.viki.moe/v2/60s"
    ICON = 'base64:data:image/jpeg;base64,/9j/2wCEAAgGBgcGBQgHBwcJCQgKDBQNDAsLDBkSEw8UHRofHh0aHBwgJC4nICIsIxwcKDcpLDAxNDQ0Hyc5PTgyPC4zNDIBCQkJDAsMGA0NGDIhHCEyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMv/AABEIADAAMAMBIgACEQEDEQH/xAGiAAABBQEBAQEBAQAAAAAAAAAAAQIDBAUGBwgJCgsQAAIBAwMCBAMFBQQEAAABfQECAwAEEQUSITFBBhNRYQcicRQygZGhCCNCscEVUtHwJDNicoIJChYXGBkaJSYnKCkqNDU2Nzg5OkNERUZHSElKU1RVVldYWVpjZGVmZ2hpanN0dXZ3eHl6g4SFhoeIiYqSk5SVlpeYmZqio6Slpqeoqaqys7S1tre4ubrCw8TFxsfIycrS09TV1tfY2drh4uPk5ebn6Onq8fLz9PX29/j5+gEAAwEBAQEBAQEBAQAAAAAAAAECAwQFBgcICQoLEQACAQIEBAMEBwUEBAABAncAAQIDEQQFITEGEkFRB2FxEyIygQgUQpGhscEJIzNS8BVictEKFiQ04SXxFxgZGiYnKCkqNTY3ODk6Q0RFRkdISUpTVFVWV1hZWmNkZWZnaGlqc3R1dnd4eXqCg4SFhoeIiYqSk5SVlpeYmZqio6Slpqeoqaqys7S1tre4ubrCw8TFxsfIycrS09TV1tfY2dri4+Tl5ufo6ery8/T19vf4+fr/2gAMAwEAAhEDEQA/AO886MdWA9zxTwQRkc0pHNM2xpl9qrgctjFfCH0VyB79FlmjVHdolZnCYJCqoZmIzwMEfXoKtA5rlrnSop7uTUrezna8l2lVUboz8oG4naeO2AecdOajsbi8k1yEz3N1lS7Sq4McaoFOcLwMAleTk89a9KWCvT546WWt+vocyre9ZnUiUGZowCdoBJ9PapKx7LWEub4xQwN5DsdszNy59lx046n8q2K4KlKdN2mrG8ZKWwVy+sapqMOvQ26wK2nxmOSYcbpF3AsMHqCAQAOpzk1peI1mk0aWKCcwPKQgkXOV/Ij6de9U/DvhqbUrSWOWb5YZQsUxJ3qCoLYHTb7HuM9ea9HLsPf961fy/U58RUsuTYg+KWna9rcFlNoV/wCRCNkqy+aYlI+bJ3/wn7p5IzxjO2sTWvElzoPh5XuoIdUFpHGqtJGR8xCK5BIyELhjgjGNowMjHsk81npVg007xwWsQ5J4AH/6z+NZ0FnoOu2purWOKRGyDJGCh5xkEccHg4IweDjpX0d+jPLS6nDaCBqfhqHxBaW32HbC1zJb7iRsJdSQWzyNrEDoRt6ZxXS26SJAiSuHcDBIB5/MmuI1lNS1BfIsZrwksRtAlRXXaQOen93lsgAY6cV29usi28azMGlCAOw6E45NeBm3I3Fx3PRwnMlZhcQJcwNE4+Vh+XvTvD9x/Y5ktryNisr7kniBZenRhjK/Xke4p9FcGGxdTDv3djoq0Y1FqR+KxNq3h60lSAsq3AaeKNt5VSjr/D15ZT7de1Ytkh8L+Gbi3AMM984it4ZHJbYB8znJznk/+O1u7cNuBKt/eUkH8xTfKTzjMV3SkYMjcsR6ZPNej/a6cdY6+pzfU3fc5/z9WuDlfNC9kjiCL+bc/rVzT4NQW4Dzs6x4+ZXk3E/zrXorzp4rmi4qKS9DpjStrc//2Q=='

    def _fetch_60s(self) -> Tuple[Dict[str, Any], str]:
        """Fetch 60s news JSON. Returns (data, error)."""
        try:
            req = urllib.request.Request(
                self.API_URL,
                method="GET",
            )
            with urllib.request.urlopen(req, timeout=6) as resp:
                text = resp.read().decode("utf-8", errors="ignore")
                obj = json.loads(text)
                if not isinstance(obj, dict):
                    return {}, "Invalid JSON root"
                if obj.get("code") != 200:
                    return {}, f"API error: code={obj.get('code')}"
                data = obj.get("data", {})
                if not isinstance(data, dict):
                    return {}, "Invalid data payload"
                return data, ""
        except urllib.error.URLError as e:
            return {}, f"Network error: {e}"
        except Exception as e:
            return {}, f"Unexpected error: {e}"

    def query(self, raw_query: str, trigger_keyword: str, command: str, search: str) -> List[WoxPluginBase.QueryResult]:
        data, err = self._fetch_60s()
        if err:
            return [{
                "title": "获取 60s 新闻失败",
                "subtitle": err,
                "icon": self.ICON,
            }]

        news_list = data.get("news") or []
        items: List[WoxPluginBase.QueryResult] = []
        for n in news_list:
            if not isinstance(n, str):
                continue
            items.append({
                "title": n.strip(),
                "icon": self.ICON,
                "actions": [
                    {
                        "id": "open-url",
                        "name": "搜索这条新闻",
                        "url": f"https://www.google.com/search?q={urllib.parse.quote(n.strip())}",
                    }
                ],
            })

        return items

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