| 169 | |
| 170 | |
| 171 | def input_with_timeout(prompt_message: str, timeout_seconds: int = 30) -> str: |
| 172 | print(prompt_message, end="", flush=True) |
| 173 | if sys.platform == "win32": |
| 174 | user_input_container: List[Optional[str]] = [None] |
| 175 | |
| 176 | def get_input_in_thread(): |
| 177 | try: |
| 178 | user_input_container[0] = sys.stdin.readline().strip() |
| 179 | except Exception: |
| 180 | user_input_container[0] = "" # Return empty string on error |
| 181 | |
| 182 | input_thread = threading.Thread(target=get_input_in_thread, daemon=True) |
| 183 | input_thread.start() |
| 184 | input_thread.join(timeout=timeout_seconds) |
| 185 | if input_thread.is_alive(): |
| 186 | print("\nInput timed out. Using default value.", flush=True) |
| 187 | return "" |
| 188 | return user_input_container[0] if user_input_container[0] is not None else "" |
| 189 | else: # Linux/macOS |
| 190 | readable_fds, _, _ = select.select([sys.stdin], [], [], timeout_seconds) |
| 191 | if readable_fds: |
| 192 | return sys.stdin.readline().strip() |
| 193 | else: |
| 194 | print("\nInput timed out. Using default value.", flush=True) |
| 195 | return "" |
| 196 | |
| 197 | |
| 198 | def get_proxy_from_gsettings() -> Optional[str]: |