Shell

讀取 stdin 上的路徑並為每一行生成一個新的互動式 shell

  • May 28, 2020

考慮一個在整個主目錄中搜尋具有錯誤權限的文件或目錄的命令：

$ find $HOME -perm 777

這只是一個例子；該命令可能會列出損壞的符號連結：

$ find $HOME -xtype l

或列出冗長的符號連結：

$ symlinks -s -r $HOME

或任何數量的其他昂貴的命令，將換行符分隔的路徑發送到stdout.

現在，我可以在這樣的尋呼機中收集結果：

$ find $HOME -perm 777 | less

然後cd到不同虛擬終端中的相關目錄。但我寧願有一個腳本為每一行輸出打開一個新的互動式 shell，如下所示：

$ find $HOME -perm 777 | visit-paths.sh

這樣我可以檢查每個文件或目錄，檢查時間戳，決定是否需要更改權限或刪除文件等。

使用從文件或 stdin 讀取路徑的 bash 腳本是可行的，如下所示：

#! /usr/bin/env bash

set -e

declare -A ALREADY_SEEN
while IFS='' read -u 10 -r line || test -n "$line"
do
   if test -d "$line"
   then
       VISIT_DIR="$line"
   elif test -f "$line"
   then
       VISIT_DIR="$(dirname "$line")"
   else
       printf "Warning: path does not exist: '%s'\n" "$line" >&2
       continue
   fi
   if test "${ALREADY_SEEN[$VISIT_DIR]}" != '1'
   then
       ( cd "$VISIT_DIR" && $SHELL -i </dev/tty )
       ALREADY_SEEN[${VISIT_DIR}]=1
       continue
   else
       # Same as last time, skip it.
       continue
   fi
done 10< "${*:-/dev/stdin}"

這有一些好處，例如：

  • 一旦出現新的輸出行，腳本就會打開一個新的 shell stdin。這意味著在我開始做事之前，我不必等待慢速命令完全完成。
  • 當我在新生成的 shell 中執行操作時，slow 命令會一直在後台執行，因此當我完成時，下一條路徑可能已經準備好訪問。
  • 如有必要，我可以提前退出循環，例如使用false; exitCtrl-C Ctrl-D。
  • 該腳本處理文件名和目錄。
  • 該腳本避免連續兩次導航到同一目錄。（感謝@MichaelHomer 解釋瞭如何使用關聯數組來做到這一點。）

但是，這個腳本有一個問題：

  • 如果最後一條命令具有非零狀態，則整個管道退出，這對於提前退出很有用，但通常需要$?每次檢查以防止意外提前退出。

為了嘗試解決這個問題，我編寫了一個 Python 腳本：

#! /usr/bin/env python3

import argparse
import logging
import os
import subprocess
import sys

if __name__ == '__main__':
   parser = argparse.ArgumentParser(
       description='Visit files from file or stdin.'
   )
   parser.add_argument(
       '-v',
       '--verbose',
       help='More verbose logging',
       dest="loglevel",
       default=logging.WARNING,
       action="store_const",
       const=logging.INFO,
   )
   parser.add_argument(
       '-d',
       '--debug',
       help='Enable debugging logs',
       action="store_const",
       dest="loglevel",
       const=logging.DEBUG,
   )
   parser.add_argument(
       'infile',
       nargs='?',
       type=argparse.FileType('r'),
       default=sys.stdin,
       help='Input file (or stdin)',
   )
   args = parser.parse_args()
   logging.basicConfig(level=args.loglevel)
   shell_bin = os.environ['SHELL']
   logging.debug("SHELL = '{}'".format(shell_bin))
   already_visited = set()
   n_visits = 0
   n_skipped = 0
   for i, line in enumerate(args.infile):
       visit_dir = None
       candidate = line.rstrip()
       logging.debug("candidate = '{}'".format(candidate))
       if os.path.isdir(candidate):
           visit_dir = candidate
       elif os.path.isfile(candidate):
           visit_dir = os.path.dirname(candidate)
       else:
           logging.warning("does not exist: '{}'".format(candidate))
           n_skipped +=1
           continue
       if visit_dir is not None:
           real_dir = os.path.realpath(visit_dir)
       else:
           # Should not happen.
           logging.warning("could not determine directory for path: '{}'".format(candidate))
           n_skipped +=1
           continue
       if visit_dir in already_visited:
           logging.info("already visited: '{}'".format(visit_dir))
           n_skipped +=1
           continue
       elif real_dir in already_visited:
           logging.info("already visited: '{}' -> '{}'".format(visit_dir, real_dir))
           n_skipped +=1
           continue
       if i != 0:
           try :
               response = input("#{}. Continue? (y/n) ".format(n_visits + 1))
           except EOFError:
               sys.stdout.write('\n')
               break
           if response in ["n", "no"]:
               break
       logging.info("spawning '{}' in '{}'".format(shell_bin, visit_dir))
       run_args = [shell_bin, "-i"]
       subprocess.call(run_args, cwd=visit_dir, stdin=open('/dev/tty'))
       already_visited.add(visit_dir)
       already_visited.add(real_dir)
       n_visits +=1

   logging.info("# paths received: {}".format(i + 1))
   logging.info("distinct directories visited: {}".format(n_visits))
   logging.info("paths skipped: {}".format(n_skipped))

但是，我在將Continue? (y/n)提示的回复傳遞給生成的 shell 時遇到了一些問題，導致類似y: command not found. 我懷疑問題出在這條線上：

subprocess.call(run_args, cwd=visit_dir, stdin=open('/dev/tty'))

stdin使用時我需要做一些不同的事情subprocess.call嗎？

或者，是否有一個廣泛可用的工具可以使兩個腳本變得多餘，而我只是沒有聽說過？

您的 Bash 腳本似乎正在按預期執行所有操作，它只需要|| break在生成互動式 shell 的子 shell 之後出現：這樣，當您從該互動式 shell 退出並引發錯誤時，例如Ctrl+C緊跟 aCtrl+D或exit 1命令，您退出從整個管道早期。

當然，正如您所指出的，當您從互動式 shell 使用的最後一個命令以（不需要的）錯誤退出時，它也會退出，但您可以通過在任何正常 exit:之前發出簡單的 as last 命令來輕鬆繞過它，或者也許（作為一個可能更好的解決方案）通過測試作為退出整個管道的唯一可接受的方式，即在產生互動式外殼的子外殼之後使用（而不是僅僅）。Ctrl+C``|| { [ $? -eq 130 ] && break; }``|| break

作為一種根本不需要關聯數組的更簡單的方法，您可能只是-uniq輸出find如下所示：

find . -perm 777 -printf '%h\n' | uniq | \
(
while IFS= read -r path ; do
   (cd "${path}" && PS1="[*** REVISE \\w]: " bash --norc -i </dev/tty) || \
       { [ $? -eq 130 ] && break; }
done
)

當然，這需要一個產生連續重複（如果有的話）的名稱源，就像這樣find 做一樣。或者您可以使用sort -u 而不是重新排序它們uniq，但是您必須等待sort完成，然後才能看到第一個互動式外殼生成，這是您似乎不希望的壯舉。

然後讓我們看看 Python 腳本方法。

你沒有說你是如何呼叫它的，但是如果你通過管道使用它，如下所示：

names-source-cmd | visit-paths.py

那麼您將 stdin 用於兩個相互衝突的目的：輸入名稱和輸入 Pythoninput()函式。

然後，您可能想要呼叫您的 Python 腳本，例如：

names-source-cmd | visit-paths.py /dev/fd/3 3<&0 < /dev/tty

請注意上面範例中所做的重定向：我們首先將剛剛創建的管道（在管道的那一部分中將是標準輸入）重定向到任意文件描述符 3，然後將標準輸入重新打開到 tty 以便 Python 腳本可以使用它的input()功能。然後通過 Python 腳本的參數將文件描述符 3 用作名稱的來源。

您還可以考慮以下概念驗證：

find | \
(
while IFS= read -ru 3 name; do
   echo "name is ${name}"
   read -p "Continue ? " && [ "$REPLY" = y ] || break
done 3<&0 < /dev/tty
)

上面的範例使用了相同的重定向技巧。因此，您可以將它用於您自己的 Bash 腳本，該腳本將看到的路徑記憶體在關聯數組中，並在每個新看到的路徑上生成一個互動式 shell。

引用自：https://unix.stackexchange.com/questions/509559

相關問答

Bash

在作為 heredoc 傳遞的 python 腳本中使用管道內容

  • February 18, 2021
檢視問答
Shell

管道如何影響標準輸入？

  • July 7, 2020
檢視問答
Shell

從命名管道獲取標準輸入

  • November 2, 2021
檢視問答
Shell

是否echo -n |... 向管道發送 EOF？

  • December 19, 2020
檢視問答
Text-Processing

如何使 python 程序表現得像適當的 unix 工具？

  • January 14, 2019
檢視問答
Shell

連接 STDOUT 和 STDIN 是什麼意思？

  • August 21, 2018
檢視問答