python 实现 shell 实时日志查看

zmx

2021 年 12 月 15 日

522 次浏览

暂无评论

2980字数

Python

# 引言

接口开发是真的恶心，emm，里面考虑不到的exception，直接就报错，所以日志能提供很大的帮助，但是，如果每天都产生一个日志文件，每次查看都需要打开shell客户端，并且输入命令，是不是很麻烦，所以，出了这个小工具。

# 代码

```
import pyperclip
import time
import paramiko
from colorama import init
import threading
import sys
import signal

global terminal
terminal = False

bar = '\r' + '\033[44m' + "=" * 15 + 'CTRL+C to Exit' + "=" * 15 + '\033[0m'

def wait():
    global terminal, shell
    init(autoreset=True)
    try:
        a = input()
        # shell.send(a + '\r\n')
        print('\r' + bar, end="\r")
        threading.Thread(target=wait).start()
    except Exception as e:
        eexit()
        print('\033[1;31;40m[#] 正在结束链接', end='')
        print(e)
        terminal = True
        shell.send('\x03')
        shell.send('exit\r\n')
        sshClient.close()
        print("\033[1;32;40m\t\tDone")

def main():
    init(autoreset=True)
    print('\033[1;34;40m[#] Z-日志监控()')
    print('\033[1;31;40m[#] 正在初始化  ', end='')
    now = time.strftime("%Y-%m-%d", time.localtime())
    global shell
    threading.Thread(target=wait).start()
    daily_log = f"""tail -f /*******/logs/{now}.txt -n 500 | perl -pe 's/(- root - DEBUG - \[.*]-->])|(- root - INFO - \[.*]-->)|(- root - ERROR - \[.*]-->)|(■H■E■A■D■)|(■E■N■D■)/\e[1;34m$1\e[0m\e[1;33m$2\e[0m\e[1;31m$3\e[1;34m$4\e[0m\e[1;35m$5\e[0m/g'"""
    print("\033[1;32;40m\t\tDone")
    print('\033[1;31;40m[#] 正在创建连接', end='')
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    print("\033[1;32;40m\t\tDone")
    print('\033[1;31;40m[#] 正在连接      ', end='')
    sshClient = paramiko.SSHClient()
    sshClient.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    sshClient.connect(hostname='*************', port=***, username='****', password='******')
    shell = sshClient.invoke_shell()
    print("\033[1;32;40m\t\tDone")
    shell.sendall(daily_log + "\n")
    print('\033[1;31;40m[#] 数据接收      ', end='')
    print("\033[1;32;40m\t\tDone")
    try:
        while True:
            sys.stdout.flush()
            try:
                data = shell.recv(10240).decode("utf8", "ignore")
                if terminal:
                    break
                print(data, end="")
                print(bar, end="\r")
            except Exception as e:
                eexit()
                print('\033[1;31;40m[#] 正在结束链接', end='')
                print(e)
                sshClient.close()
                print("\033[1;32;40m\t\tDone")
                break
    except Exception as e:
        print('\033[1;31;40m[#] 正在结束链接', end='')
        print(e)
        sshClient.close()
        print("\033[1;32;40m\t\tDone")

# sshClient.close()

def eexit():
    print("\r\n\033[1;31;40m[#] 退出标识")
    for i in range(0, 3):
        time.sleep(1)

if __name__ == '__main__':
    main()
```

# 附图

![image.png](https://www.zunmx.top/usr/uploads/2021/12/1952043852.png)