“做个shell玩一玩”

shell.py ：Shell 的主程序，负责读取、解析并执行命令
func/ ：自定义模块，所有的自定义命令的实现函数文件都位于这里
init.py ：使 func 文件夹能作为 Python 模块被导入
cd.py ：实现了 Shell 的 cd 命令
constants.py ：定义了各种常量与路径
exit.py ：定义了 exit 命令，用来退出程序
getenv.py ：实现 getenv 命令，获取系统变量的值
history.py：实现 history 命令，展示输入的命令日志

#shell.py

import os
import sys
import shlex
import getpass
import socket
import signal
import subprocess
import platform
from func import *
built_in_cmds = {}
def tokenize(string):
    return shlex.split(string)
def preprocess(tokens):
    processed_token = []
    for token in tokens:
        if token.startswith('$'):
            processed_token.append(os.getenv(token[1:]))
        else:
            processed_token.append(token)
    return processed_token
def handler_kill(signum, frame):
    raise OSError("Killed!")
def execute(cmd_tokens):
    with open(HISTORY_PATH, 'a') as history_file:
        history_file.write(' '.join(cmd_tokens) + os.linesep)
    if cmd_tokens:
        cmd_name = cmd_tokens[0]
        cmd_args = cmd_tokens[1:]
        if cmd_name in built_in_cmds:
            return built_in_cmds[cmd_name](cmd_args)
        signal.signal(signal.SIGINT, handler_kill)
        if platform.system() != "Windows":
            p = subprocess.Popen(cmd_tokens)
            p.communicate()
        else:
            command = ""
            command = ' '.join(cmd_tokens)
            os.system(command)
    return SHELL_STATUS_RUN
def display_cmd_prompt():
    user = getpass.getuser()
    hostname = socket.gethostname()
    cwd = os.getcwd()
    base_dir = os.path.basename(cwd)
    home_dir = os.path.expanduser('~')
    if cwd == home_dir:
        base_dir = '~'
    if platform.system() != 'Windows':
        sys.stdout.write("[\033[1;33m%s\033[0;0m@%s \033[1;36m%s\033[0;0m] $ " % (user, hostname, base_dir))
    else:
        sys.stdout.write("[%s@%s %s]$ " % (user, hostname, base_dir))
    sys.stdout.flush()
def ignore_signals():
    if platform.system() != "Windows":
        signal.signal(signal.SIGTSTP, signal.SIG_IGN)
    signal.signal(signal.SIGINT, signal.SIG_IGN)
def shell_loop():
    status = SHELL_STATUS_RUN
    while status == SHELL_STATUS_RUN:
        display_cmd_prompt()
        ignore_signals()
        try:
            cmd = sys.stdin.readline()
            cmd_tokens = tokenize(cmd)
            cmd_tokens = preprocess(cmd_tokens)
            status = execute(cmd_tokens)
        except:
            _, err, _ = sys.exc_info()
            print(err)
def register_command(name, func):
    built_in_cmds[name] = func
def init():
    register_command("cd", cd)
    register_command("exit", exit)
    register_command("getenv", getenv)
    register_command("history", history)
def main():
    init()
    shell_loop()
if __name__ == "__main__":
    main()

#constants.py

import os
SHELL_STATUS_STOP = 0
SHELL_STATUS_RUN = 1
HISTORY_PATH = os.path.expanduser('~') + os.sep + '.shiyanlou_shell_history'

#exit.py

from .constants import *
def exit(args):
    return SHELL_STATUS_STOP

#getenv.py

from .constants import *
def getenv(args):
    if len(args) > 0:
        print(os.getenv(args[0]))
    return SHELL_STATUS_RUN

#history.py

import sys
from .constants import *
def history(args):
    with open(HISTORY_PATH, 'r') as history_file:
        lines = history_file.readlines()
        limit = len(lines)
        if len(args) > 0:
            limit = int(args[0])
        start = len(lines) - limit
        for line_num, line in enumerate(lines):
            if line_num >= start:
                sys.stdout.write('%d %s' % (line_num + 1, line))
        sys.stdout.flush()
    return SHELL_STATUS_RUN

#init.py

from .cd import cd
from .exit import exit
from .getenv import getenv
from .history import history
from .constants import *

2016-10-16