1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
|
#!/usr/bin/env python3
"""
Search the web via the Kagi Search API and print plain-text results.
The API key is read from $KAGI_API_KEY.
Exits 2 on rate-limit (retry later), 1 on other errors.
"""
import argparse
import json
import os
import sys
import urllib.error
import urllib.request
from dataclasses import dataclass
ENDPOINT = "https://kagi.com/api/v1/search"
API_KEY_ENV = "KAGI_API_KEY"
@dataclass
class SearchResult:
title: str
url: str
snippet: str = ""
def fail(message: str, exit_code: int = 1) -> None:
print(f"web_search: {message}", file=sys.stderr)
sys.exit(exit_code)
def error_message(err: urllib.error.HTTPError) -> str:
try:
payload = json.loads(err.read().decode("utf-8", "replace"))
details = payload.get("errors") or payload.get("error") or [{}]
first = details[0]
bits = [b for b in (first.get("code"), first.get("message")) if b]
if bits:
return "; ".join(bits)
except (ValueError, AttributeError, IndexError):
pass
return err.reason or "unknown error"
def fetch(query: str, limit: int) -> dict:
key = os.environ.get(API_KEY_ENV, "").strip()
if not key:
fail(f"{API_KEY_ENV} is not set", exit_code=1)
body = json.dumps({"query": query, "limit": limit}).encode()
request = urllib.request.Request(
ENDPOINT,
data=body,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {key}",
},
)
with urllib.request.urlopen(request, timeout=30) as response:
return json.loads(response.read().decode("utf-8", "replace"))
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
_ = parser.add_argument(
"-n", type=int, default=5, metavar="N", help="max results (default 5)"
)
_ = parser.add_argument("query", nargs="+", help="search terms")
args = parser.parse_args()
query = " ".join(args.query)
try:
payload = fetch(query, args.n)
except urllib.error.HTTPError as err:
if err.code == 429:
fail("rate-limited (HTTP 429); try later", exit_code=2)
fail(f"HTTP {err.code}: {error_message(err)}")
except OSError as err:
fail(f"network error: {err}")
results = [
SearchResult(
title=item.get("title", "").strip(),
url=item.get("url", "").strip(),
snippet=" ".join(item.get("snippet", "").split()),
)
for item in payload.get("data", {}).get("search", [])
]
results = [r for r in results if r.title and r.url]
if not results:
print(f"no results for: {query}")
return
for number, result in enumerate(results[: args.n], 1):
print(f"{number}. {result.title}\n {result.url}")
if result.snippet:
print(f" {result.snippet}")
if __name__ == "__main__":
main()
|
