This repository has no description
2.7 kB
96 lines
1#!/usr/bin/env python3
2"""Native messaging host that bridges browser extension and CLI.
3
4The browser extension communicates via native messaging (stdin/stdout).
5The CLI connects via Unix socket using JSON lines protocol.
6"""
7
8import asyncio
9import contextlib
10import logging
11import logging.handlers
12import os
13import signal
14import sys
15import tempfile
16from pathlib import Path
17
18from browser_cli.bridge import NativeMessagingBridge
19
20logger = logging.getLogger(__name__)
21
22
23def setup_logging() -> None:
24 """Set up logging to both terminal and file."""
25 # Get XDG data home
26 xdg_data_home = os.environ.get("XDG_DATA_HOME", str(Path.home() / ".local" / "share"))
27 log_dir = Path(xdg_data_home) / "browser-cli"
28 log_dir.mkdir(parents=True, exist_ok=True)
29 log_file = log_dir / "server.log"
30
31 # Set up root logger
32 root_logger = logging.getLogger()
33 root_logger.setLevel(logging.INFO)
34
35 # File handler only - stdout is used for native messaging
36 file_handler = logging.handlers.RotatingFileHandler(
37 log_file,
38 maxBytes=10 * 1024 * 1024, # 10MB
39 backupCount=5,
40 encoding="utf-8",
41 )
42 file_handler.setLevel(logging.INFO)
43 file_formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
44 file_handler.setFormatter(file_formatter)
45
46 # Add handler to root logger
47 root_logger.addHandler(file_handler)
48
49 logger.info("Logging initialized. Log file: %s", log_file)
50
51
52async def async_main() -> None:
53 """Run the native messaging bridge."""
54 bridge = NativeMessagingBridge()
55
56 # Handle shutdown gracefully
57 stop: asyncio.Future[None] = asyncio.Future()
58
59 def signal_handler() -> None:
60 stop.set_result(None)
61
62 loop = asyncio.get_event_loop()
63 for sig in (signal.SIGTERM, signal.SIGINT):
64 loop.add_signal_handler(sig, signal_handler)
65
66 # Get runtime directory for Unix socket
67 runtime_dir = os.environ.get("XDG_RUNTIME_DIR")
68 if not runtime_dir:
69 runtime_dir = tempfile.gettempdir()
70
71 socket_path = Path(runtime_dir) / "browser-cli.sock"
72
73 # Run bridge until stopped
74 bridge_task = asyncio.create_task(bridge.start(socket_path))
75
76 await asyncio.wait([bridge_task, stop], return_when=asyncio.FIRST_COMPLETED)
77
78 # Cancel bridge task if still running
79 if not bridge_task.done():
80 bridge_task.cancel()
81 with contextlib.suppress(asyncio.CancelledError):
82 await bridge_task
83
84
85def main() -> None:
86 """Run the native messaging bridge server."""
87 setup_logging()
88 try:
89 asyncio.run(async_main())
90 except KeyboardInterrupt:
91 logger.info("Server stopped by user")
92 sys.exit(0)
93
94
95if __name__ == "__main__":
96 main()