读取标准输入上的路径并为每一行生成一个新的交互式 shell

读取标准输入上的路径并为每一行生成一个新的交互式 shell

考虑一个在整个主目录中搜索具有错误权限的文件或目录的命令:

$ 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

这样我可以检查每个文件或目录,检查时间戳,决定是否需要更改权限或删除文件等。

它是可以使用 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 中执行操作时,慢速命令会在后台继续运行,因此当我完成操作时,下一个路径可能已准备好访问。

  • false; exit如有必要,我可以使用例如 Ctrl-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

或者,是否有一种广泛可用的工具可以使这两个脚本变得多余,而我只是没有听说过?

答案1

您的 Bash 脚本似乎正在按预期执行所有操作,它只需要|| break在生成交互式 shell 的子 shell 之后:这样,当您退出该交互式 shell 并出现诱发错误(例如Ctrl+C紧随其后的Ctrl+D, 或exit 1命令)时,您就会退出从整个管道来看。

当然,正如您所指出的,当您从交互式 shell 中使用的最后一个命令因(不需要的)错误退出时,它也会退出,但是您可以通过发出简单的:as最后一个命令在任何之前正常退出,或者也许(作为可能更好的解决方案)通过测试作为Ctrl+C退出整个管道的唯一可接受的方法,即在生成交互式 shell 的子 shell 之后使用|| { [ $? -eq 130 ] && break; }(而不是仅仅)。|| break

作为一种根本不需要关联数组的更简单的方法,您可能只需uniq-ing 的输出,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完成,然后才能看到第一个交互式 shell 生成,这是您似乎不希望的壮举。

接下来让我们看看Python 脚本方法。

您没有说明如何调用它,但如果您通过管道使用它,如下所示:

names-source-cmd | visit-paths.py

那么您将 stdin 用于两个相互冲突的目的:名称输入和 Pythoninput()函数输入。

然后,您可能想调用 Python 脚本,如下所示:

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

请注意上面示例中完成的重定向:我们首先将刚刚创建的管道(将是管道该部分中的 stdin)重定向到任意文件描述符 3,然后将 stdin 重新打开到 tty 上,以便 Python 脚本可以使用是为了它的input()功能。然后,文件描述符 3 通过 Python 脚本的参数用作名称源。

您还可以考虑以下概念验证:

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。

答案2

相关内容