forked from
lass.pds.lassul.us/superconfig
This repository has no description
kagi-search: move to tools/, drop rofi dep for darwin support
Use pinentry-rofi on linux and pinentry_mac on darwin directly,
removing the now-unused pinentry-auto wrapper.
author
lassulus
date
(Apr 8, 2026, 9:23 AM +0100)
commit
e2dd087c
e2dd087cd11f96a4974bbd05360a8b2835018456
parent
6bf94ab0
6bf94ab03bb04c12ce089da1018dae8ea776e1e4
+617
-627
-585
5pkgs/kagi-search/kagi_search.py
-585
5pkgs/kagi-search/kagi_search.py
···
1
-
#!/usr/bin/env python3
2
-
"""
3
-
Kagi search script using session token authentication.
4
-
Scrapes search results from Kagi web interface.
5
-
"""
6
-
7
-
import json
8
-
import subprocess
9
-
import sys
10
-
import os
11
-
from pathlib import Path
12
-
from typing import List, Dict, Optional, Any
13
-
from urllib.request import Request, HTTPCookieProcessor, build_opener
14
-
from urllib.parse import urlencode
15
-
from urllib.error import URLError, HTTPError
16
-
from http.cookiejar import CookieJar
17
-
import gzip
18
-
from bs4 import BeautifulSoup, Tag
19
-
import time
20
-
from dataclasses import dataclass
21
-
import argparse
22
-
import logging
23
-
24
-
# Module-level logger
25
-
logger = logging.getLogger(__name__)
26
-
27
-
28
-
def colorize(text: str, color: str = "", bold: bool = False, dim: bool = False) -> str:
29
-
"""
30
-
Colorize text if colors are enabled (TTY detected and NO_COLOR not set).
31
-
32
-
Args:
33
-
text: The text to colorize
34
-
color: Color name (red, green, yellow, blue, magenta, cyan, white)
35
-
bold: Whether to make text bold
36
-
dim: Whether to make text dim/faint
37
-
38
-
Returns:
39
-
Colorized text or plain text based on environment
40
-
"""
41
-
# Check if we should use colors
42
-
if not sys.stdout.isatty() or os.environ.get("NO_COLOR"):
43
-
return text
44
-
45
-
# Color codes
46
-
colors = {
47
-
"red": "\033[91m",
48
-
"green": "\033[92m",
49
-
"yellow": "\033[93m",
50
-
"blue": "\033[94m",
51
-
"magenta": "\033[95m",
52
-
"cyan": "\033[96m",
53
-
"white": "\033[97m",
54
-
}
55
-
56
-
# Start with empty escape sequence
57
-
escape = ""
58
-
59
-
# Add color if specified
60
-
if color and color.lower() in colors:
61
-
escape = colors[color.lower()]
62
-
63
-
# Add bold if requested
64
-
if bold:
65
-
escape = "\033[1m" + escape
66
-
67
-
# Add dim if requested
68
-
if dim:
69
-
escape = "\033[2m" + escape
70
-
71
-
# Return plain text if no formatting requested
72
-
if not escape:
73
-
return text
74
-
75
-
# Return formatted text
76
-
return f"{escape}{text}\033[0m"
77
-
78
-
79
-
def hyperlink(url: str, text: str = "") -> str:
80
-
"""
81
-
Create a terminal hyperlink if supported (TTY detected and NO_COLOR not set).
82
-
83
-
Args:
84
-
url: The URL to link to
85
-
text: The text to display (defaults to URL if not provided)
86
-
87
-
Returns:
88
-
Hyperlinked text or plain text based on environment
89
-
"""
90
-
# Check if we should use hyperlinks (same conditions as colors)
91
-
if not sys.stdout.isatty() or os.environ.get("NO_COLOR"):
92
-
return text or url
93
-
94
-
# Use the URL as text if no text provided
95
-
display_text = text or url
96
-
97
-
# OSC 8 hyperlink format: ESC]8;;URL ESC\TEXT ESC]8;; ESC\
98
-
return f"\033]8;;{url}\033\\{display_text}\033]8;;\033\\"
99
-
100
-
101
-
@dataclass
102
-
class SearchResult:
103
-
"""Represents a single search result from Kagi."""
104
-
105
-
title: str
106
-
url: str
107
-
snippet: str
108
-
109
-
110
-
@dataclass
111
-
class QuickAnswer:
112
-
"""Represents a Kagi Quick Answer response."""
113
-
114
-
html: str
115
-
markdown: str
116
-
raw_text: str
117
-
references: List[Dict[str, Any]]
118
-
119
-
120
-
class KagiSearch:
121
-
"""Kagi search client using session token authentication."""
122
-
123
-
def __init__(self, session_token: Optional[str] = None, config_path: Optional[str] = None):
124
-
"""
125
-
Initialize Kagi search client.
126
-
127
-
Args:
128
-
session_token: Kagi session token (if not provided, will load from config)
129
-
config_path: Path to config file (defaults to ~/.config/kagi/config.json)
130
-
"""
131
-
self.base_url = "https://kagi.com"
132
-
self.user_agent = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
133
-
134
-
# Load config and get session token
135
-
if not session_token:
136
-
config = self._load_config(config_path)
137
-
session_token = self._get_session_token(config)
138
-
139
-
self.session_token = session_token
140
-
141
-
# Set up cookie handling
142
-
self.cookie_jar = CookieJar()
143
-
self.opener = build_opener(HTTPCookieProcessor(self.cookie_jar))
144
-
145
-
# Authenticate with token
146
-
self._authenticate()
147
-
148
-
def _load_config(self, config_path: Optional[str] = None) -> Dict[str, Any]:
149
-
"""Load configuration from file."""
150
-
if not config_path:
151
-
config_file = Path.home() / ".config" / "kagi" / "config.json"
152
-
else:
153
-
config_file = Path(config_path)
154
-
155
-
if not config_file.exists():
156
-
# Create default config
157
-
config_file.parent.mkdir(parents=True, exist_ok=True)
158
-
default_config = {
159
-
"password_command": "rbw get kagi-session-link",
160
-
"timeout": 30,
161
-
"max_retries": 5,
162
-
}
163
-
with open(config_file, "w") as f:
164
-
json.dump(default_config, f, indent=2)
165
-
msg = colorize(f"Created default config at {config_file}", color="green")
166
-
print(msg, file=sys.stderr)
167
-
return default_config
168
-
169
-
with open(config_file, "r") as f:
170
-
data: Dict[str, Any] = json.load(f)
171
-
return data
172
-
173
-
def _get_session_token(self, config: Dict[str, Any]) -> str:
174
-
"""Get session token using password command from config."""
175
-
password_command = config.get("password_command", "rbw get kagi-session-link")
176
-
177
-
try:
178
-
# Execute the password command
179
-
result = subprocess.run(
180
-
password_command.split(), capture_output=True, text=True, check=True
181
-
)
182
-
session_link = result.stdout.strip()
183
-
184
-
# Extract token from session link
185
-
if "token=" in session_link:
186
-
return session_link.split("token=")[1].split("&")[0]
187
-
else:
188
-
# Assume the entire output is the token
189
-
return session_link
190
-
191
-
except subprocess.CalledProcessError as e:
192
-
error_msg = colorize(f"Error executing password command: {e}", color="red", bold=True)
193
-
print(error_msg, file=sys.stderr)
194
-
if e.stderr:
195
-
stderr_msg = colorize(f"stderr: {e.stderr}", color="red")
196
-
print(stderr_msg, file=sys.stderr)
197
-
sys.exit(1)
198
-
except Exception as e:
199
-
error_msg = colorize(f"Error getting session token: {e}", color="red", bold=True)
200
-
print(error_msg, file=sys.stderr)
201
-
sys.exit(1)
202
-
203
-
def _authenticate(self) -> None:
204
-
"""Authenticate with Kagi using session token."""
205
-
token_url = f"{self.base_url}/html/search?token={self.session_token}"
206
-
request = Request(token_url)
207
-
request.add_header("User-Agent", self.user_agent)
208
-
209
-
try:
210
-
response = self.opener.open(request, timeout=30)
211
-
final_url = response.geturl()
212
-
213
-
# Check if we were redirected to home page or html search (successful auth)
214
-
if final_url in [f"{self.base_url}/", self.base_url, f"{self.base_url}/html/search"]:
215
-
return # Success
216
-
elif "/signin" in final_url or "/welcome" in final_url:
217
-
raise Exception(f"Authentication failed - redirected to {final_url}")
218
-
except Exception as e:
219
-
raise Exception(f"Failed to authenticate with token: {e}")
220
-
221
-
def search(self, query: str, limit: int = 10) -> List[SearchResult]:
222
-
"""
223
-
Search Kagi and return results.
224
-
225
-
Args:
226
-
query: Search query
227
-
limit: Maximum number of results to return
228
-
229
-
Returns:
230
-
List of SearchResult objects
231
-
"""
232
-
# Construct search URL for HTML version (no token needed, we use cookies)
233
-
params = {"q": query}
234
-
search_url = f"{self.base_url}/html/search?{urlencode(params)}"
235
-
236
-
max_retries = 5
237
-
retry_delay = 1.0
238
-
239
-
for attempt in range(max_retries):
240
-
try:
241
-
# Create request with headers
242
-
request = Request(search_url)
243
-
request.add_header("User-Agent", self.user_agent)
244
-
request.add_header(
245
-
"Accept",
246
-
"text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
247
-
)
248
-
request.add_header("Accept-Language", "en-US,en;q=0.5")
249
-
request.add_header("DNT", "1")
250
-
request.add_header("Connection", "keep-alive")
251
-
request.add_header("Upgrade-Insecure-Requests", "1")
252
-
253
-
# Make request using opener (with cookies)
254
-
response = self.opener.open(request, timeout=30)
255
-
256
-
# Check if we're redirected to sign in
257
-
final_url = response.geturl()
258
-
if "/signin" in final_url or "/welcome" in final_url:
259
-
raise Exception(f"Authentication failed - redirected to {final_url}")
260
-
261
-
# Read response
262
-
content = response.read()
263
-
264
-
# Handle gzip encoding if present
265
-
if response.headers.get("Content-Encoding") == "gzip":
266
-
content = gzip.decompress(content)
267
-
268
-
html_content = content.decode("utf-8")
269
-
270
-
# Parse results
271
-
soup = BeautifulSoup(html_content, "html.parser")
272
-
273
-
# Find results container
274
-
results_box = soup.find(class_="results-box")
275
-
if not results_box or isinstance(results_box, str):
276
-
if attempt < max_retries - 1:
277
-
time.sleep(retry_delay)
278
-
continue
279
-
raise Exception("No results box found on page")
280
-
281
-
# Extract search results
282
-
results = []
283
-
if not isinstance(results_box, Tag):
284
-
raise Exception("Results box is not a Tag element")
285
-
search_results = results_box.find_all(class_="search-result")
286
-
287
-
for result in search_results[:limit]:
288
-
# Extract title
289
-
title_elem = result.find(class_="__sri-title")
290
-
title = ""
291
-
if title_elem:
292
-
# Get only the text content, not the UI elements
293
-
# Remove "More results from this site", "Remove results from this site", etc.
294
-
title_text = title_elem.get_text(separator=" ", strip=True)
295
-
# Split by common Kagi UI elements and take the first part
296
-
for separator in [
297
-
"More results from",
298
-
"Remove results from",
299
-
"Open page in",
300
-
]:
301
-
if separator in title_text:
302
-
title_text = title_text.split(separator)[0]
303
-
title = title_text.strip()
304
-
305
-
# Extract URL
306
-
url_box = result.find(class_="__sri-url-box")
307
-
url = ""
308
-
if url_box:
309
-
link = url_box.find("a", href=True)
310
-
if link:
311
-
href = link.get("href", "")
312
-
url = str(href) if href else ""
313
-
314
-
# Extract snippet
315
-
desc_elem = result.find(class_="__sri-desc")
316
-
snippet = desc_elem.get_text(strip=True) if desc_elem else ""
317
-
318
-
if title and url:
319
-
results.append(SearchResult(title=title, url=url, snippet=snippet))
320
-
321
-
if results:
322
-
return results
323
-
elif attempt < max_retries - 1:
324
-
time.sleep(retry_delay)
325
-
continue
326
-
327
-
except (URLError, HTTPError) as e:
328
-
if attempt < max_retries - 1:
329
-
time.sleep(retry_delay)
330
-
continue
331
-
raise Exception(f"Request failed: {e}")
332
-
333
-
return []
334
-
335
-
def _get_session_cookie(self) -> str:
336
-
"""Extract kagi_session cookie value for X-Kagi-Authorization header."""
337
-
for cookie in self.cookie_jar:
338
-
if cookie.name == "kagi_session":
339
-
return str(cookie.value) if cookie.value else ""
340
-
return ""
341
-
342
-
def get_quick_answer(self, query: str) -> Optional[QuickAnswer]:
343
-
"""
344
-
Get Kagi Quick Answer for a query.
345
-
346
-
Args:
347
-
query: Search query
348
-
349
-
Returns:
350
-
QuickAnswer object or None if no answer available
351
-
"""
352
-
# Construct Quick Answer URL - POST request with query in URL
353
-
params = {"q": query}
354
-
quick_answer_url = f"{self.base_url}/mother/context?{urlencode(params)}"
355
-
logger.debug(f"Quick Answer URL: {quick_answer_url}")
356
-
logger.info("Fetching Quick Answer...")
357
-
358
-
try:
359
-
# Get session cookie for authorization header
360
-
session_cookie = self._get_session_cookie()
361
-
362
-
# Create POST request with headers matching the working curl command
363
-
request = Request(quick_answer_url, data=b"", method="POST")
364
-
request.add_header("User-Agent", self.user_agent)
365
-
request.add_header("Accept", "application/vnd.kagi.stream")
366
-
request.add_header("Accept-Language", "en-US,en;q=0.5")
367
-
request.add_header("Accept-Encoding", "gzip, deflate")
368
-
request.add_header("Referer", f"{self.base_url}/search?{urlencode(params)}")
369
-
request.add_header("Origin", self.base_url)
370
-
request.add_header("Connection", "keep-alive")
371
-
request.add_header("Content-Length", "0")
372
-
if session_cookie:
373
-
request.add_header("X-Kagi-Authorization", session_cookie)
374
-
375
-
# Make request using opener (with cookies)
376
-
logger.debug("Making Quick Answer POST request...")
377
-
response = self.opener.open(request, timeout=30)
378
-
logger.debug(f"Response status: {response.getcode()}")
379
-
380
-
# Check if we're redirected to sign in
381
-
final_url = response.geturl()
382
-
logger.debug(f"Final URL: {final_url}")
383
-
if "/signin" in final_url or "/welcome" in final_url:
384
-
raise Exception(f"Authentication failed - redirected to {final_url}")
385
-
386
-
# Read response
387
-
content = response.read()
388
-
logger.debug(f"Response content length: {len(content)} bytes")
389
-
390
-
# Handle gzip encoding if present
391
-
if response.headers.get("Content-Encoding") == "gzip":
392
-
logger.debug("Decompressing gzip content")
393
-
content = gzip.decompress(content)
394
-
395
-
# Parse streaming response - lines with prefixes like "hi:", "tokens.json:", "new_message.json:"
396
-
content_str = content.decode("utf-8")
397
-
lines = content_str.strip().split("\n")
398
-
logger.debug(f"Response has {len(lines)} lines")
399
-
400
-
final_data = None
401
-
402
-
for line in lines:
403
-
line = line.strip()
404
-
if not line:
405
-
continue
406
-
logger.debug(f"Processing line: {line[:100]}...") # Log first 100 chars
407
-
408
-
# Look for the new_message.json line which contains the final answer
409
-
if line.startswith("new_message.json:"):
410
-
json_str = line[len("new_message.json:") :]
411
-
logger.debug(f"Parsing new_message JSON: {json_str[:200]}...")
412
-
try:
413
-
# Use JSONDecoder to parse only the first JSON object
414
-
# This handles cases where there's trailing data after the JSON
415
-
decoder = json.JSONDecoder()
416
-
final_data, _ = decoder.raw_decode(json_str)
417
-
logger.debug("Successfully parsed new_message data")
418
-
except json.JSONDecodeError as e:
419
-
logger.error(f"Failed to parse new_message JSON: {e}")
420
-
logger.debug(f"Full line: {line}")
421
-
422
-
if not final_data:
423
-
logger.debug("No new_message data found in response")
424
-
return None
425
-
426
-
# New response format has fields: md, reply, references_md, etc.
427
-
markdown = final_data.get("md", "")
428
-
html = final_data.get("reply", "")
429
-
references_md = final_data.get("references_md", "")
430
-
431
-
# Parse references from references_md (format: [^1]: [Title](URL) (percent%))
432
-
references = []
433
-
if references_md:
434
-
import re
435
-
436
-
# Match pattern like: [^1]: [Title](URL) (22%)
437
-
# Use non-greedy match for URL with lookahead to handle URLs containing parentheses
438
-
ref_pattern = r"\[\^\d+\]:\s*\[([^\]]+)\]\((.+?)\)\s*\((\d+)%\)"
439
-
for match in re.finditer(ref_pattern, references_md):
440
-
references.append(
441
-
{
442
-
"title": match.group(1),
443
-
"url": match.group(2),
444
-
"contribution": f"{match.group(3)}%",
445
-
}
446
-
)
447
-
448
-
# If no content, return None
449
-
if not html and not markdown:
450
-
logger.debug("No content found in Quick Answer")
451
-
return None
452
-
453
-
logger.debug(f"Quick Answer found with {len(references)} references")
454
-
455
-
return QuickAnswer(
456
-
html=html, markdown=markdown, raw_text=markdown, references=references
457
-
)
458
-
459
-
except Exception as e:
460
-
# Quick Answer might not be available for all queries
461
-
logger.debug(f"Quick Answer error: {type(e).__name__}: {e}")
462
-
return None
463
-
464
-
465
-
def main() -> None:
466
-
"""Main entry point for command line usage."""
467
-
parser = argparse.ArgumentParser(description="Search Kagi using session token")
468
-
parser.add_argument("query", nargs="?", help="Search query")
469
-
parser.add_argument(
470
-
"-n", "--num-results", type=int, default=3, help="Number of link results (default: 3)"
471
-
)
472
-
parser.add_argument(
473
-
"-l", "--links", action="store_true", help="Show search result links (hidden by default)"
474
-
)
475
-
parser.add_argument("-t", "--token", help="Session token (overrides config)")
476
-
parser.add_argument("-c", "--config", help="Config file path")
477
-
parser.add_argument("-j", "--json", action="store_true", help="Output as JSON")
478
-
parser.add_argument("-d", "--debug", action="store_true", help="Enable debug logging to stderr")
479
-
480
-
args = parser.parse_args()
481
-
482
-
# Configure logging - always to stderr
483
-
log_level = logging.DEBUG if args.debug else logging.WARNING
484
-
logging.basicConfig(
485
-
level=log_level,
486
-
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
487
-
stream=sys.stderr,
488
-
)
489
-
490
-
# If no query provided, read from stdin
491
-
if not args.query:
492
-
if sys.stdin.isatty():
493
-
parser.error("No query provided")
494
-
args.query = sys.stdin.read().strip()
495
-
496
-
# Initialize client
497
-
try:
498
-
client = KagiSearch(session_token=args.token, config_path=args.config)
499
-
except Exception as e:
500
-
error_msg = colorize(f"Error initializing Kagi client: {e}", color="red", bold=True)
501
-
print(error_msg, file=sys.stderr)
502
-
sys.exit(1)
503
-
504
-
# Perform search
505
-
try:
506
-
results = client.search(args.query, limit=args.num_results) if args.links else []
507
-
quick_answer = client.get_quick_answer(args.query)
508
-
except Exception as e:
509
-
error_msg = colorize(f"Search failed: {e}", color="red", bold=True)
510
-
print(error_msg, file=sys.stderr)
511
-
sys.exit(1)
512
-
513
-
# Output results
514
-
if args.json:
515
-
output: Dict[str, Any] = {
516
-
"results": [{"title": r.title, "url": r.url, "snippet": r.snippet} for r in results]
517
-
}
518
-
if quick_answer:
519
-
output["quick_answer"] = {
520
-
"markdown": quick_answer.markdown,
521
-
"raw_text": quick_answer.raw_text,
522
-
"references": quick_answer.references,
523
-
}
524
-
print(json.dumps(output, indent=2))
525
-
else:
526
-
# Display Quick Answer if available
527
-
if quick_answer:
528
-
qa_title = colorize("Quick Answer", color="cyan", bold=True)
529
-
print(f"\n{qa_title}")
530
-
print(colorize("─" * 80, color="cyan", dim=True))
531
-
532
-
# Display raw text for terminal (cleaner than HTML)
533
-
if quick_answer.raw_text:
534
-
print(quick_answer.raw_text)
535
-
elif quick_answer.markdown:
536
-
# Fallback to markdown if no raw text
537
-
print(quick_answer.markdown)
538
-
539
-
# Display references
540
-
if quick_answer.references:
541
-
print()
542
-
refs_title = colorize("References:", color="cyan", dim=True)
543
-
print(refs_title)
544
-
for i, ref in enumerate(quick_answer.references[:5], 1): # Limit to 5 refs
545
-
ref_num = colorize(f"[{i}]", color="cyan", dim=True)
546
-
ref_title = ref.get("title", "")
547
-
ref_url = ref.get("url", "")
548
-
if ref_title and ref_url:
549
-
ref_link = hyperlink(ref_url, colorize(ref_title, color="blue"))
550
-
print(f" {ref_num} {ref_link}")
551
-
url_text = colorize(ref_url, color="green", dim=True)
552
-
url_with_link = hyperlink(ref_url, url_text)
553
-
print(f" {url_with_link}")
554
-
555
-
print(colorize("─" * 80, color="cyan", dim=True))
556
-
557
-
# Display search result links only when --links is passed
558
-
if results:
559
-
if quick_answer:
560
-
print() # Extra newline after Quick Answer
561
-
for i, result in enumerate(results, 1):
562
-
# Result number in yellow/bold
563
-
number = colorize(f"{i}.", color="yellow", bold=True)
564
-
# Title in blue/bold with hyperlink
565
-
title = colorize(result.title, color="blue", bold=True)
566
-
title_with_link = hyperlink(result.url, title)
567
-
print(f"\n{number} {title_with_link}")
568
-
569
-
# URL in green with hyperlink
570
-
url_text = colorize(result.url, color="green")
571
-
url_with_link = hyperlink(result.url, url_text)
572
-
print(f" {url_with_link}")
573
-
574
-
# Snippet in default color but dim
575
-
if result.snippet:
576
-
snippet = colorize(result.snippet, dim=True)
577
-
print(f" {snippet}")
578
-
579
-
if not quick_answer and not results and not args.json:
580
-
error_msg = colorize("No results found", color="red")
581
-
print(error_msg, file=sys.stderr)
582
-
583
-
584
-
if __name__ == "__main__":
585
-
main()
-26
5pkgs/kagi-search/package.nix
-26
5pkgs/kagi-search/package.nix
···
1
-
{
2
-
pkgs,
3
-
pinentry-auto ? pkgs.callPackage ../pinentry-auto/package.nix { },
4
-
...
5
-
}:
6
-
let
7
-
python = pkgs.python3.withPackages (
8
-
ps: with ps; [
9
-
beautifulsoup4
10
-
]
11
-
);
12
-
in
13
-
pkgs.writeShellApplication {
14
-
name = "kagi-search";
15
-
runtimeInputs = [
16
-
python
17
-
pkgs.rbw
18
-
];
19
-
text = ''
20
-
if ! rbw unlocked 2>/dev/null; then
21
-
rbw config set pinentry ${pinentry-auto}/bin/pinentry-auto
22
-
rbw unlock
23
-
fi
24
-
${python}/bin/python ${./kagi_search.py} "$@"
25
-
'';
26
-
}
-9
5pkgs/pinentry-auto/package.nix
-9
5pkgs/pinentry-auto/package.nix
-7
5pkgs/pinentry-auto/pinentry-auto.sh
-7
5pkgs/pinentry-auto/pinentry-auto.sh
+30
tools/kagi-search/flake-module.nix
+30
tools/kagi-search/flake-module.nix
···
1
+
{
2
+
perSystem =
3
+
{ pkgs, self', ... }:
4
+
let
5
+
python = pkgs.python3.withPackages (ps: with ps; [ beautifulsoup4 ]);
6
+
pinentry =
7
+
if pkgs.stdenv.isDarwin then
8
+
"${pkgs.pinentry_mac}/${pkgs.pinentry_mac.binaryPath}"
9
+
else
10
+
"${self'.packages.pinentry-rofi}/bin/pinentry-rofi";
11
+
in
12
+
{
13
+
packages.kagi-search =
14
+
(pkgs.writeShellApplication {
15
+
name = "kagi-search";
16
+
runtimeInputs = [
17
+
python
18
+
pkgs.rbw
19
+
];
20
+
text = ''
21
+
if ! rbw unlocked 2>/dev/null; then
22
+
rbw config set pinentry ${pinentry}
23
+
rbw unlock
24
+
fi
25
+
${python}/bin/python ${./kagi_search.py} "$@"
26
+
'';
27
+
}).overrideAttrs
28
+
{ passthru.usage = builtins.readFile ./usage.kdl; };
29
+
};
30
+
}
+585
tools/kagi-search/kagi_search.py
+585
tools/kagi-search/kagi_search.py
···
1
+
#!/usr/bin/env python3
2
+
"""
3
+
Kagi search script using session token authentication.
4
+
Scrapes search results from Kagi web interface.
5
+
"""
6
+
7
+
import json
8
+
import subprocess
9
+
import sys
10
+
import os
11
+
from pathlib import Path
12
+
from typing import List, Dict, Optional, Any
13
+
from urllib.request import Request, HTTPCookieProcessor, build_opener
14
+
from urllib.parse import urlencode
15
+
from urllib.error import URLError, HTTPError
16
+
from http.cookiejar import CookieJar
17
+
import gzip
18
+
from bs4 import BeautifulSoup, Tag
19
+
import time
20
+
from dataclasses import dataclass
21
+
import argparse
22
+
import logging
23
+
24
+
# Module-level logger
25
+
logger = logging.getLogger(__name__)
26
+
27
+
28
+
def colorize(text: str, color: str = "", bold: bool = False, dim: bool = False) -> str:
29
+
"""
30
+
Colorize text if colors are enabled (TTY detected and NO_COLOR not set).
31
+
32
+
Args:
33
+
text: The text to colorize
34
+
color: Color name (red, green, yellow, blue, magenta, cyan, white)
35
+
bold: Whether to make text bold
36
+
dim: Whether to make text dim/faint
37
+
38
+
Returns:
39
+
Colorized text or plain text based on environment
40
+
"""
41
+
# Check if we should use colors
42
+
if not sys.stdout.isatty() or os.environ.get("NO_COLOR"):
43
+
return text
44
+
45
+
# Color codes
46
+
colors = {
47
+
"red": "\033[91m",
48
+
"green": "\033[92m",
49
+
"yellow": "\033[93m",
50
+
"blue": "\033[94m",
51
+
"magenta": "\033[95m",
52
+
"cyan": "\033[96m",
53
+
"white": "\033[97m",
54
+
}
55
+
56
+
# Start with empty escape sequence
57
+
escape = ""
58
+
59
+
# Add color if specified
60
+
if color and color.lower() in colors:
61
+
escape = colors[color.lower()]
62
+
63
+
# Add bold if requested
64
+
if bold:
65
+
escape = "\033[1m" + escape
66
+
67
+
# Add dim if requested
68
+
if dim:
69
+
escape = "\033[2m" + escape
70
+
71
+
# Return plain text if no formatting requested
72
+
if not escape:
73
+
return text
74
+
75
+
# Return formatted text
76
+
return f"{escape}{text}\033[0m"
77
+
78
+
79
+
def hyperlink(url: str, text: str = "") -> str:
80
+
"""
81
+
Create a terminal hyperlink if supported (TTY detected and NO_COLOR not set).
82
+
83
+
Args:
84
+
url: The URL to link to
85
+
text: The text to display (defaults to URL if not provided)
86
+
87
+
Returns:
88
+
Hyperlinked text or plain text based on environment
89
+
"""
90
+
# Check if we should use hyperlinks (same conditions as colors)
91
+
if not sys.stdout.isatty() or os.environ.get("NO_COLOR"):
92
+
return text or url
93
+
94
+
# Use the URL as text if no text provided
95
+
display_text = text or url
96
+
97
+
# OSC 8 hyperlink format: ESC]8;;URL ESC\TEXT ESC]8;; ESC\
98
+
return f"\033]8;;{url}\033\\{display_text}\033]8;;\033\\"
99
+
100
+
101
+
@dataclass
102
+
class SearchResult:
103
+
"""Represents a single search result from Kagi."""
104
+
105
+
title: str
106
+
url: str
107
+
snippet: str
108
+
109
+
110
+
@dataclass
111
+
class QuickAnswer:
112
+
"""Represents a Kagi Quick Answer response."""
113
+
114
+
html: str
115
+
markdown: str
116
+
raw_text: str
117
+
references: List[Dict[str, Any]]
118
+
119
+
120
+
class KagiSearch:
121
+
"""Kagi search client using session token authentication."""
122
+
123
+
def __init__(self, session_token: Optional[str] = None, config_path: Optional[str] = None):
124
+
"""
125
+
Initialize Kagi search client.
126
+
127
+
Args:
128
+
session_token: Kagi session token (if not provided, will load from config)
129
+
config_path: Path to config file (defaults to ~/.config/kagi/config.json)
130
+
"""
131
+
self.base_url = "https://kagi.com"
132
+
self.user_agent = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
133
+
134
+
# Load config and get session token
135
+
if not session_token:
136
+
config = self._load_config(config_path)
137
+
session_token = self._get_session_token(config)
138
+
139
+
self.session_token = session_token
140
+
141
+
# Set up cookie handling
142
+
self.cookie_jar = CookieJar()
143
+
self.opener = build_opener(HTTPCookieProcessor(self.cookie_jar))
144
+
145
+
# Authenticate with token
146
+
self._authenticate()
147
+
148
+
def _load_config(self, config_path: Optional[str] = None) -> Dict[str, Any]:
149
+
"""Load configuration from file."""
150
+
if not config_path:
151
+
config_file = Path.home() / ".config" / "kagi" / "config.json"
152
+
else:
153
+
config_file = Path(config_path)
154
+
155
+
if not config_file.exists():
156
+
# Create default config
157
+
config_file.parent.mkdir(parents=True, exist_ok=True)
158
+
default_config = {
159
+
"password_command": "rbw get kagi-session-link",
160
+
"timeout": 30,
161
+
"max_retries": 5,
162
+
}
163
+
with open(config_file, "w") as f:
164
+
json.dump(default_config, f, indent=2)
165
+
msg = colorize(f"Created default config at {config_file}", color="green")
166
+
print(msg, file=sys.stderr)
167
+
return default_config
168
+
169
+
with open(config_file, "r") as f:
170
+
data: Dict[str, Any] = json.load(f)
171
+
return data
172
+
173
+
def _get_session_token(self, config: Dict[str, Any]) -> str:
174
+
"""Get session token using password command from config."""
175
+
password_command = config.get("password_command", "rbw get kagi-session-link")
176
+
177
+
try:
178
+
# Execute the password command
179
+
result = subprocess.run(
180
+
password_command.split(), capture_output=True, text=True, check=True
181
+
)
182
+
session_link = result.stdout.strip()
183
+
184
+
# Extract token from session link
185
+
if "token=" in session_link:
186
+
return session_link.split("token=")[1].split("&")[0]
187
+
else:
188
+
# Assume the entire output is the token
189
+
return session_link
190
+
191
+
except subprocess.CalledProcessError as e:
192
+
error_msg = colorize(f"Error executing password command: {e}", color="red", bold=True)
193
+
print(error_msg, file=sys.stderr)
194
+
if e.stderr:
195
+
stderr_msg = colorize(f"stderr: {e.stderr}", color="red")
196
+
print(stderr_msg, file=sys.stderr)
197
+
sys.exit(1)
198
+
except Exception as e:
199
+
error_msg = colorize(f"Error getting session token: {e}", color="red", bold=True)
200
+
print(error_msg, file=sys.stderr)
201
+
sys.exit(1)
202
+
203
+
def _authenticate(self) -> None:
204
+
"""Authenticate with Kagi using session token."""
205
+
token_url = f"{self.base_url}/html/search?token={self.session_token}"
206
+
request = Request(token_url)
207
+
request.add_header("User-Agent", self.user_agent)
208
+
209
+
try:
210
+
response = self.opener.open(request, timeout=30)
211
+
final_url = response.geturl()
212
+
213
+
# Check if we were redirected to home page or html search (successful auth)
214
+
if final_url in [f"{self.base_url}/", self.base_url, f"{self.base_url}/html/search"]:
215
+
return # Success
216
+
elif "/signin" in final_url or "/welcome" in final_url:
217
+
raise Exception(f"Authentication failed - redirected to {final_url}")
218
+
except Exception as e:
219
+
raise Exception(f"Failed to authenticate with token: {e}")
220
+
221
+
def search(self, query: str, limit: int = 10) -> List[SearchResult]:
222
+
"""
223
+
Search Kagi and return results.
224
+
225
+
Args:
226
+
query: Search query
227
+
limit: Maximum number of results to return
228
+
229
+
Returns:
230
+
List of SearchResult objects
231
+
"""
232
+
# Construct search URL for HTML version (no token needed, we use cookies)
233
+
params = {"q": query}
234
+
search_url = f"{self.base_url}/html/search?{urlencode(params)}"
235
+
236
+
max_retries = 5
237
+
retry_delay = 1.0
238
+
239
+
for attempt in range(max_retries):
240
+
try:
241
+
# Create request with headers
242
+
request = Request(search_url)
243
+
request.add_header("User-Agent", self.user_agent)
244
+
request.add_header(
245
+
"Accept",
246
+
"text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
247
+
)
248
+
request.add_header("Accept-Language", "en-US,en;q=0.5")
249
+
request.add_header("DNT", "1")
250
+
request.add_header("Connection", "keep-alive")
251
+
request.add_header("Upgrade-Insecure-Requests", "1")
252
+
253
+
# Make request using opener (with cookies)
254
+
response = self.opener.open(request, timeout=30)
255
+
256
+
# Check if we're redirected to sign in
257
+
final_url = response.geturl()
258
+
if "/signin" in final_url or "/welcome" in final_url:
259
+
raise Exception(f"Authentication failed - redirected to {final_url}")
260
+
261
+
# Read response
262
+
content = response.read()
263
+
264
+
# Handle gzip encoding if present
265
+
if response.headers.get("Content-Encoding") == "gzip":
266
+
content = gzip.decompress(content)
267
+
268
+
html_content = content.decode("utf-8")
269
+
270
+
# Parse results
271
+
soup = BeautifulSoup(html_content, "html.parser")
272
+
273
+
# Find results container
274
+
results_box = soup.find(class_="results-box")
275
+
if not results_box or isinstance(results_box, str):
276
+
if attempt < max_retries - 1:
277
+
time.sleep(retry_delay)
278
+
continue
279
+
raise Exception("No results box found on page")
280
+
281
+
# Extract search results
282
+
results = []
283
+
if not isinstance(results_box, Tag):
284
+
raise Exception("Results box is not a Tag element")
285
+
search_results = results_box.find_all(class_="search-result")
286
+
287
+
for result in search_results[:limit]:
288
+
# Extract title
289
+
title_elem = result.find(class_="__sri-title")
290
+
title = ""
291
+
if title_elem:
292
+
# Get only the text content, not the UI elements
293
+
# Remove "More results from this site", "Remove results from this site", etc.
294
+
title_text = title_elem.get_text(separator=" ", strip=True)
295
+
# Split by common Kagi UI elements and take the first part
296
+
for separator in [
297
+
"More results from",
298
+
"Remove results from",
299
+
"Open page in",
300
+
]:
301
+
if separator in title_text:
302
+
title_text = title_text.split(separator)[0]
303
+
title = title_text.strip()
304
+
305
+
# Extract URL
306
+
url_box = result.find(class_="__sri-url-box")
307
+
url = ""
308
+
if url_box:
309
+
link = url_box.find("a", href=True)
310
+
if link:
311
+
href = link.get("href", "")
312
+
url = str(href) if href else ""
313
+
314
+
# Extract snippet
315
+
desc_elem = result.find(class_="__sri-desc")
316
+
snippet = desc_elem.get_text(strip=True) if desc_elem else ""
317
+
318
+
if title and url:
319
+
results.append(SearchResult(title=title, url=url, snippet=snippet))
320
+
321
+
if results:
322
+
return results
323
+
elif attempt < max_retries - 1:
324
+
time.sleep(retry_delay)
325
+
continue
326
+
327
+
except (URLError, HTTPError) as e:
328
+
if attempt < max_retries - 1:
329
+
time.sleep(retry_delay)
330
+
continue
331
+
raise Exception(f"Request failed: {e}")
332
+
333
+
return []
334
+
335
+
def _get_session_cookie(self) -> str:
336
+
"""Extract kagi_session cookie value for X-Kagi-Authorization header."""
337
+
for cookie in self.cookie_jar:
338
+
if cookie.name == "kagi_session":
339
+
return str(cookie.value) if cookie.value else ""
340
+
return ""
341
+
342
+
def get_quick_answer(self, query: str) -> Optional[QuickAnswer]:
343
+
"""
344
+
Get Kagi Quick Answer for a query.
345
+
346
+
Args:
347
+
query: Search query
348
+
349
+
Returns:
350
+
QuickAnswer object or None if no answer available
351
+
"""
352
+
# Construct Quick Answer URL - POST request with query in URL
353
+
params = {"q": query}
354
+
quick_answer_url = f"{self.base_url}/mother/context?{urlencode(params)}"
355
+
logger.debug(f"Quick Answer URL: {quick_answer_url}")
356
+
logger.info("Fetching Quick Answer...")
357
+
358
+
try:
359
+
# Get session cookie for authorization header
360
+
session_cookie = self._get_session_cookie()
361
+
362
+
# Create POST request with headers matching the working curl command
363
+
request = Request(quick_answer_url, data=b"", method="POST")
364
+
request.add_header("User-Agent", self.user_agent)
365
+
request.add_header("Accept", "application/vnd.kagi.stream")
366
+
request.add_header("Accept-Language", "en-US,en;q=0.5")
367
+
request.add_header("Accept-Encoding", "gzip, deflate")
368
+
request.add_header("Referer", f"{self.base_url}/search?{urlencode(params)}")
369
+
request.add_header("Origin", self.base_url)
370
+
request.add_header("Connection", "keep-alive")
371
+
request.add_header("Content-Length", "0")
372
+
if session_cookie:
373
+
request.add_header("X-Kagi-Authorization", session_cookie)
374
+
375
+
# Make request using opener (with cookies)
376
+
logger.debug("Making Quick Answer POST request...")
377
+
response = self.opener.open(request, timeout=30)
378
+
logger.debug(f"Response status: {response.getcode()}")
379
+
380
+
# Check if we're redirected to sign in
381
+
final_url = response.geturl()
382
+
logger.debug(f"Final URL: {final_url}")
383
+
if "/signin" in final_url or "/welcome" in final_url:
384
+
raise Exception(f"Authentication failed - redirected to {final_url}")
385
+
386
+
# Read response
387
+
content = response.read()
388
+
logger.debug(f"Response content length: {len(content)} bytes")
389
+
390
+
# Handle gzip encoding if present
391
+
if response.headers.get("Content-Encoding") == "gzip":
392
+
logger.debug("Decompressing gzip content")
393
+
content = gzip.decompress(content)
394
+
395
+
# Parse streaming response - lines with prefixes like "hi:", "tokens.json:", "new_message.json:"
396
+
content_str = content.decode("utf-8")
397
+
lines = content_str.strip().split("\n")
398
+
logger.debug(f"Response has {len(lines)} lines")
399
+
400
+
final_data = None
401
+
402
+
for line in lines:
403
+
line = line.strip()
404
+
if not line:
405
+
continue
406
+
logger.debug(f"Processing line: {line[:100]}...") # Log first 100 chars
407
+
408
+
# Look for the new_message.json line which contains the final answer
409
+
if line.startswith("new_message.json:"):
410
+
json_str = line[len("new_message.json:") :]
411
+
logger.debug(f"Parsing new_message JSON: {json_str[:200]}...")
412
+
try:
413
+
# Use JSONDecoder to parse only the first JSON object
414
+
# This handles cases where there's trailing data after the JSON
415
+
decoder = json.JSONDecoder()
416
+
final_data, _ = decoder.raw_decode(json_str)
417
+
logger.debug("Successfully parsed new_message data")
418
+
except json.JSONDecodeError as e:
419
+
logger.error(f"Failed to parse new_message JSON: {e}")
420
+
logger.debug(f"Full line: {line}")
421
+
422
+
if not final_data:
423
+
logger.debug("No new_message data found in response")
424
+
return None
425
+
426
+
# New response format has fields: md, reply, references_md, etc.
427
+
markdown = final_data.get("md", "")
428
+
html = final_data.get("reply", "")
429
+
references_md = final_data.get("references_md", "")
430
+
431
+
# Parse references from references_md (format: [^1]: [Title](URL) (percent%))
432
+
references = []
433
+
if references_md:
434
+
import re
435
+
436
+
# Match pattern like: [^1]: [Title](URL) (22%)
437
+
# Use non-greedy match for URL with lookahead to handle URLs containing parentheses
438
+
ref_pattern = r"\[\^\d+\]:\s*\[([^\]]+)\]\((.+?)\)\s*\((\d+)%\)"
439
+
for match in re.finditer(ref_pattern, references_md):
440
+
references.append(
441
+
{
442
+
"title": match.group(1),
443
+
"url": match.group(2),
444
+
"contribution": f"{match.group(3)}%",
445
+
}
446
+
)
447
+
448
+
# If no content, return None
449
+
if not html and not markdown:
450
+
logger.debug("No content found in Quick Answer")
451
+
return None
452
+
453
+
logger.debug(f"Quick Answer found with {len(references)} references")
454
+
455
+
return QuickAnswer(
456
+
html=html, markdown=markdown, raw_text=markdown, references=references
457
+
)
458
+
459
+
except Exception as e:
460
+
# Quick Answer might not be available for all queries
461
+
logger.debug(f"Quick Answer error: {type(e).__name__}: {e}")
462
+
return None
463
+
464
+
465
+
def main() -> None:
466
+
"""Main entry point for command line usage."""
467
+
parser = argparse.ArgumentParser(description="Search Kagi using session token")
468
+
parser.add_argument("query", nargs="?", help="Search query")
469
+
parser.add_argument(
470
+
"-n", "--num-results", type=int, default=3, help="Number of link results (default: 3)"
471
+
)
472
+
parser.add_argument(
473
+
"-l", "--links", action="store_true", help="Show search result links (hidden by default)"
474
+
)
475
+
parser.add_argument("-t", "--token", help="Session token (overrides config)")
476
+
parser.add_argument("-c", "--config", help="Config file path")
477
+
parser.add_argument("-j", "--json", action="store_true", help="Output as JSON")
478
+
parser.add_argument("-d", "--debug", action="store_true", help="Enable debug logging to stderr")
479
+
480
+
args = parser.parse_args()
481
+
482
+
# Configure logging - always to stderr
483
+
log_level = logging.DEBUG if args.debug else logging.WARNING
484
+
logging.basicConfig(
485
+
level=log_level,
486
+
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
487
+
stream=sys.stderr,
488
+
)
489
+
490
+
# If no query provided, read from stdin
491
+
if not args.query:
492
+
if sys.stdin.isatty():
493
+
parser.error("No query provided")
494
+
args.query = sys.stdin.read().strip()
495
+
496
+
# Initialize client
497
+
try:
498
+
client = KagiSearch(session_token=args.token, config_path=args.config)
499
+
except Exception as e:
500
+
error_msg = colorize(f"Error initializing Kagi client: {e}", color="red", bold=True)
501
+
print(error_msg, file=sys.stderr)
502
+
sys.exit(1)
503
+
504
+
# Perform search
505
+
try:
506
+
results = client.search(args.query, limit=args.num_results) if args.links else []
507
+
quick_answer = client.get_quick_answer(args.query)
508
+
except Exception as e:
509
+
error_msg = colorize(f"Search failed: {e}", color="red", bold=True)
510
+
print(error_msg, file=sys.stderr)
511
+
sys.exit(1)
512
+
513
+
# Output results
514
+
if args.json:
515
+
output: Dict[str, Any] = {
516
+
"results": [{"title": r.title, "url": r.url, "snippet": r.snippet} for r in results]
517
+
}
518
+
if quick_answer:
519
+
output["quick_answer"] = {
520
+
"markdown": quick_answer.markdown,
521
+
"raw_text": quick_answer.raw_text,
522
+
"references": quick_answer.references,
523
+
}
524
+
print(json.dumps(output, indent=2))
525
+
else:
526
+
# Display Quick Answer if available
527
+
if quick_answer:
528
+
qa_title = colorize("Quick Answer", color="cyan", bold=True)
529
+
print(f"\n{qa_title}")
530
+
print(colorize("─" * 80, color="cyan", dim=True))
531
+
532
+
# Display raw text for terminal (cleaner than HTML)
533
+
if quick_answer.raw_text:
534
+
print(quick_answer.raw_text)
535
+
elif quick_answer.markdown:
536
+
# Fallback to markdown if no raw text
537
+
print(quick_answer.markdown)
538
+
539
+
# Display references
540
+
if quick_answer.references:
541
+
print()
542
+
refs_title = colorize("References:", color="cyan", dim=True)
543
+
print(refs_title)
544
+
for i, ref in enumerate(quick_answer.references[:5], 1): # Limit to 5 refs
545
+
ref_num = colorize(f"[{i}]", color="cyan", dim=True)
546
+
ref_title = ref.get("title", "")
547
+
ref_url = ref.get("url", "")
548
+
if ref_title and ref_url:
549
+
ref_link = hyperlink(ref_url, colorize(ref_title, color="blue"))
550
+
print(f" {ref_num} {ref_link}")
551
+
url_text = colorize(ref_url, color="green", dim=True)
552
+
url_with_link = hyperlink(ref_url, url_text)
553
+
print(f" {url_with_link}")
554
+
555
+
print(colorize("─" * 80, color="cyan", dim=True))
556
+
557
+
# Display search result links only when --links is passed
558
+
if results:
559
+
if quick_answer:
560
+
print() # Extra newline after Quick Answer
561
+
for i, result in enumerate(results, 1):
562
+
# Result number in yellow/bold
563
+
number = colorize(f"{i}.", color="yellow", bold=True)
564
+
# Title in blue/bold with hyperlink
565
+
title = colorize(result.title, color="blue", bold=True)
566
+
title_with_link = hyperlink(result.url, title)
567
+
print(f"\n{number} {title_with_link}")
568
+
569
+
# URL in green with hyperlink
570
+
url_text = colorize(result.url, color="green")
571
+
url_with_link = hyperlink(result.url, url_text)
572
+
print(f" {url_with_link}")
573
+
574
+
# Snippet in default color but dim
575
+
if result.snippet:
576
+
snippet = colorize(result.snippet, dim=True)
577
+
print(f" {snippet}")
578
+
579
+
if not quick_answer and not results and not args.json:
580
+
error_msg = colorize("No results found", color="red")
581
+
print(error_msg, file=sys.stderr)
582
+
583
+
584
+
if __name__ == "__main__":
585
+
main()