allow interactive questions

src/pacroller/main.py

@@ -5,7 +5,7 @@ import subprocess
 import logging
 from re import match
 import json
-from os import environ, getuid
+from os import environ, getuid, isatty
 import traceback
 import pyalpm
 import pycman
@@ -75,7 +75,7 @@ class alpmCallback:
         handle.questioncb = self.noop
         handle.progresscb = self.noop
 
-def upgrade() -> List[str]:
+def upgrade(interactive=False) -> List[str]:
     logger.info('upgrade start')
     pycman.config.cb_log = lambda *_: None
     handle = pycman.config.init_with_config(PACMAN_CONFIG)
@@ -108,11 +108,11 @@ def upgrade() -> List[str]:
             examine_upgrade(t.to_add, t.to_remove)
     finally:
         t.release()
-    pacman_output = execute_with_io(['pacman', '-Su', '--noprogressbar', '--color', 'never'], UPGRADE_TIMEOUT)
+    pacman_output = execute_with_io(['pacman', '-Su', '--noprogressbar', '--color', 'never'], UPGRADE_TIMEOUT, interactive=interactive)
     logger.info('upgrade end')
     return pacman_output
 
-def do_system_upgrade(debug=False) -> checkReport:
+def do_system_upgrade(debug=False, interactive=False) -> checkReport:
     for _ in range(NETWORK_RETRY):
         try:
             sync()
@@ -127,7 +127,7 @@ def do_system_upgrade(debug=False) -> checkReport:
         try:
             with open(PACMAN_LOG, 'r') as pacman_log:
                 log_anchor = pacman_log.seek(0, 2)
-            stdout = upgrade()
+            stdout = upgrade(interactive=interactive)
         except subprocess.CalledProcessError as e:
             if upgrade_err_is_net(e.output):
                 logger.warning('upgrade download failed')
@@ -240,12 +240,15 @@ def main() -> None:
     parser.add_argument('-d', '--debug', action='store_true', help='enable debug mode')
     parser.add_argument('-v', '--verbose', action='store_true', help='show verbose report')
     parser.add_argument('-m', '--max', type=int, default=1, help='Number of upgrades to show')
+    parser.add_argument('-i', '--interactive', choices=['auto', 'on', 'off'], default='auto', help='allow interactive questions')
     args = parser.parse_args()
     if args.debug:
         logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(module)s - %(funcName)s - %(levelname)s - %(message)s')
     else:
         logging.basicConfig(level=logging.INFO, format='%(levelname)s - %(message)s')
     locale_set()
+    interactive = args.interactive == "on" or not (args.interactive == 'off' or not isatty(0))
+    logger.debug(f"interactive questions {'enabled' if interactive else 'disabled'}")
 
     if args.action == 'run':
         if getuid() != 0:
@@ -262,7 +265,7 @@ def main() -> None:
             logger.error(f'Database is locked at {PACMAN_DB_LCK}')
             exit(2)
         try:
-            report = do_system_upgrade(args.debug)
+            report = do_system_upgrade(debug=args.debug, interactive=interactive)
         except NonFatal:
             raise
         except Exception as e:

src/pacroller/utils.py

@@ -6,6 +6,8 @@ from io import DEFAULT_BUFFER_SIZE
 from time import mktime
 from datetime import datetime
 from signal import SIGINT, SIGTERM, Signals
+from select import select
+from sys import stdin
 logger = logging.getLogger()
 
 class UnknownQuestionError(subprocess.SubprocessError):
@@ -15,7 +17,7 @@ class UnknownQuestionError(subprocess.SubprocessError):
     def __str__(self):
         return f"Pacman returned an unknown question {self.question}"
 
-def execute_with_io(command: List[str], timeout: int = 3600) -> List[str]:
+def execute_with_io(command: List[str], timeout: int = 3600, interactive: bool = False) -> List[str]:
     '''
         captures stdout and stderr and
         automatically handles [y/n] questions of pacman
@@ -42,23 +44,50 @@ def execute_with_io(command: List[str], timeout: int = 3600) -> List[str]:
             stderr=subprocess.STDOUT,
             encoding='utf-8'
         )
-    Thread(target=set_timeout, args=(p, timeout), daemon=True).start() # should be configurable
+    try:
-    line = ''
+        Thread(target=set_timeout, args=(p, timeout), daemon=True).start()
-    output = ''
+        line = ''
-    while (r := p.stdout.read(1)) != '':
+        output = ''
-        output += r
+        while (r := p.stdout.read(1)) != '':
-        line += r
+            output += r
-        if r == '\n':
+            line += r
-            logger.debug('STDOUT: %s', line[:-1])
+            if r == '\n':
-            line = ''
+                logger.debug('STDOUT: %s', line[:-1])
-        elif r == ']':
+                line = ''
-            if line == ':: Proceed with installation? [Y/n]':
+            elif r == ']':
-                p.stdin.write('y\n')
+                if line == ':: Proceed with installation? [Y/n]':
-                p.stdin.flush()
+                    p.stdin.write('y\n')
-            elif line.lower().endswith('[y/n]'):
+                    p.stdin.flush()
-                terminate(p, signal=SIGINT)
+                elif line.lower().endswith('[y/n]'):
-                raise UnknownQuestionError(line, output)
+                    if interactive:
-
+                        print(f"Please answer this question in 60 seconds:\n{line}", end=' ', flush=True)
+                        while True:
+                            read_ready, _, _ = select([stdin], list(), list(), 60)
+                            if not read_ready:
+                                terminate(p, signal=SIGINT)
+                                raise UnknownQuestionError(line, output)
+                            choice = read_ready[0].readline().strip()
+                            if choice.lower().startswith('y'):
+                                p.stdin.write('y\n')
+                                p.stdin.flush()
+                                break
+                            elif choice.lower().startswith('n'):
+                                p.stdin.write('n\n')
+                                p.stdin.flush()
+                                break
+                            else:
+                                if choice.lower().startswith('s'):
+                                    print(output)
+                                print("Please give an explicit answer [Y]es [N]o [S]how", end=' ', flush=True)
+                    else:
+                        terminate(p, signal=SIGINT)
+                        raise UnknownQuestionError(line, output)
+    except KeyboardInterrupt:
+        terminate(p, signal=SIGINT)
+        raise
+    except Exception:
+        terminate(p)
+        raise
     if (ret := p.wait()) != 0:
         raise subprocess.CalledProcessError(ret, command, output)
     return output.split('\n')