1.Asciinema 是一款开源免费的终端录制工具，它可以将命令行输入输出的任何内容加上时间保存在文件中，同时还提供方法在终端或者web浏览器中进行回放。Asciinema 的录制和播放都是基于文本的，相比传统的video有很多好处，例如录制文件体积小，在播放的过程中可以暂停复制其中的文本内容等等
2.官网地址：https://asciinema.org/explore/featured
Github地址：https://github.com/asciinema
Asciinema-player地址：https://github.com/asciinema/asciinema-player
Asciinema-player下载地址：https://github.com/asciinema/asciinema-player/releases 基于web页面的播放器，只需要下载asciinema-player.css asciinema-player.js 导入到前端页面即可使用
3.安装Asciinema
python3 install asciinema
4.查看版本
/opt/python36/bin/asciinema --version
asciinema 2.0.2
5.参数说明
1.录制：rec， 2.播放：play， 3.以文件形式查看录制内容：cat， 4.上传文件到asciinema.org网站：upload、 5.asciinema.org账号认证：auth
6.录制，直接在linux命令行中执行
/opt/python36/bin/asciinema rec aaa.cast
参数说明： --stdin 表示启用标准输入录制，比如输入的密码也会被记录，且文件流中会出现 i 表示stdin标准输入 或 o 表示stdout标准输出 --append 添加录制到已存在的文件中 --raw 保存原始STDOUT输出，无需定时信息等 --overwrite 如果文件已存在，则覆盖 -c 要记录的命令，默认为$SHELL -e 要捕获的环境变量列表，默认为SHELL,TERM -t 后跟数字，指定录像的title -i 后跟数字，设置录制时记录的最大空闲时间 -y 所有提示都输入yes -q 静默模式，加了此参数在进入录制或者退出录制时都没有提示 输入exit或按ctrl+D组合键退出录制
7.播放
/opt/python36/bin/asciinema play aaa.cast
参数说明 -s 后边跟数字，表示用几倍的速度来播放录像 -i 后边跟数字，表示在播放录像时空闲时间的最大秒数 在播放的过程中你可以通过空格来控制暂停或播放，也可以通过ctrl+c组合键来退出播放，当你按空格键暂停时，可以通过.号来逐帧显示接下来要播放的内容
8.文件说明
文件头header
{"version": 2, "width": 233, "height": 57, "timestamp": 1589527986.4172204, "env": {"SHELL": "/bin/bash", "TERM": "linux"}, "title": "boamp_webssh_record"} 版本 播放器宽 播放器高 时间 环境 标题
文件内容 [0.5736286640167236, "o", "Last login: Fri May 15 15:29:58 2020 from 11.xx.xx.xx "] [0.6025891304016113, "o", "Hello, u9648u5efau65872 ! "] [2.169491767883301, "o", "u001b[?1034h[root@localhost:11.xx.xx.xx ~]# "]
9.如果有需要，将linux服务器上用户的所有操作过程都记录下来配置
echo "/opt/python36/bin/asciinema rec /tmp/$USER-$(date +%Y%m%d%H%M%S).cast -q" >> /etc/profile source /etc/profile
这样每个用户登陆，都会记录他的所有操作内容
10.Asciinema-player 播放器，web前端的使用
<html> <head> ... <link rel="stylesheet" type="text/css" href="/asciinema-player.css" /> ... </head> <body> ... <asciinema-player src="/demo.cast"></asciinema-player> ... <script src="/asciinema-player.js"></script> </body> </html>
参数说明
cols： 播放终端的列数，默认为80，如果cast文件的header头有设置width，这里无需设置
rows： 播放终端的行数，默认为24，如果cast文件的header头有设置height，这里无需设置
autoplay： 是否自动开始播放，默认不会自动播放
preload： 预加载，如果你想为录像配音，这里可以预加载声音
loop： 是否循环播放，默认不循环
start-at： 从哪个地方开始播放，可以是123这样的秒数或者是1:06这样的时间点
speed： 播放的速度，类似于play命令播放时的-s参数
idle-time-limit： 最大空闲秒数，类似于play命令播放时的-i参数
poster： 播放之前的预览，可以是npt:1:06这样给定时间点的画面，也可以是data:text/plain,ops-coffee.cn这样给定的文字，其中文字支持ANSI编码，例如可以给文字加上颜色data:text/plain,x1b[1;32mops-coffee.cnx1b[1;0m
font-size： 文字大小，可以是small、medium、big或者直接是14px这样的css样式大小
theme： 终端颜色主题，默认是asciinema，也提供有tango、solarized-dark、solarized-light或者monokai可选择，当然你也可以自定义主题
title、author、author-url、author-img-url分别表示了录像的标题、作者、作者的主页、作者的头像，这些配置会在全屏观看录像时显示在标题栏中
11.整合到运维平台中
1.在平台上调用目的主机上的命令执行录制：在主机上添加个脚本，每次连接自动进行录制，但这样不仅要在每台远程主机添加脚本，会很繁琐，而且录制的脚本文件都是放在远程主机上的，后续播放也很麻烦
2.事实上录像文件不是录屏，只是一组组数据流，所以在webssh websocket 交互时记录数据，写到对应的cast文件中，再用Asciinema-player 播放器拿出来播放 【使用方案√】
3.创建记录数据库代码
class RecordWebssh(models.Model): """webssh视频记录表""" user = models.CharField(max_length=128,blank=False,null=False,verbose_name="操作用户") host = models.CharField(max_length=128, blank=False, null=False, verbose_name="操作机器") record_filename = models.CharField(max_length=512, blank=False, null=False, verbose_name="操作记录文件名") create_time = models.DateTimeField(auto_now_add=True, verbose_name="创建时间") update_time = models.DateTimeField(auto_now=True, verbose_name="更新时间") def __str__(self): return self.record_filename class Meta: verbose_name_plural = "webssh视频记录表" db_table = "recordwebssh"
4.核心逻辑代码
import paramiko from threading import Thread from webssh.tools.tools import get_key_obj import socket import json import time import datetime import os from boamp.settings import logger from boamp.settings import BASE_DIR data_list = {} class SSH: def __init__(self, host, user, websocker, message): self.host = host self.user = user self.websocker = websocker self.message = message self.time = time.time() #获取起始时间戳 self.date_time = datetime.datetime.now().strftime('%Y-%m-%d_%H%M') self.msg_list = data_list["%s_%s"%(self.host,self.date_time)] = [] def record_webssh(self, host, user, type, data_list): try: record_dir_path = os.path.join(BASE_DIR,"static/webssh/record_webssh/") if not os.path.exists(record_dir_path): os.makedirs(record_dir_path) record_filename = '%s_%s_%s.cast' % (self.date_time,host, user) #命名录像文件名 record_filename_path = os.path.join(record_dir_path, record_filename) if type == 'header': #是否是头部header内容，只写入一次头部header内容 with open(record_filename_path, 'w') as f: f.write(json.dumps(data_list) + ' ') else: with open(record_filename_path, 'a', buffering=1) as f: #self.msg_list 必须为列表，如果使用字典格式会出现很多问题，已彩坑1天，换回列表就好了 for data in self.msg_list: now_time = data[0] message = data[1] iodata = [now_time - self.time, 'o', message] #生成数据流格式内容 f.write((json.dumps(iodata) + ' ')) #写入执行输出输入的内容数据流 except Exception as e: print(e) print('异常文件：%s ,异常行号：%s' % (e.__traceback__.tb_frame.f_globals['__file__'], e.__traceback__.tb_lineno)) def connect(self, host, user, password, pkey=None, port=22, timeout=30,term='xterm', pty_width=80, pty_height=24): global data_list try: ssh_client = paramiko.SSHClient() ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) if pkey: key = get_key_obj(paramiko.RSAKey, pkey_obj=pkey, password=password) or get_key_obj(paramiko.DSSKey, pkey_obj=pkey, password=password) or get_key_obj(paramiko.ECDSAKey, pkey_obj=pkey, password=password) or get_key_obj(paramiko.Ed25519Key, pkey_obj=pkey, password=password) print("使用密钥登陆") try: ssh_client.connect(username=user, hostname=host, port=port, pkey=key, timeout=timeout) logger.info("以[%s]用户通过[密钥]方式登陆机器[%s]成功"%(user,host)) except Exception as e: print("error：",e) logger.warning("以[%s]用户通过[密钥]方式登陆机器[%s]失败，错误：%s" % (user, host,e)) else: print("使用密码登陆") try: ssh_client.connect(username=user, password=password, hostname=host, port=port, timeout=timeout) logger.info("以[%s]用户通过[密码]方式登陆机器[%s]成功" % (user, host)) except Exception as e: print("error：",e) logger.warning("以[%s]用户通过[密码]方式登陆机器[%s]失败，错误：%s" % (user, host, e)) transport = ssh_client.get_transport() self.channel = transport.open_session() self.channel.get_pty(term=term, width=pty_width, height=pty_height) self.channel.invoke_shell() # 构建录像文件header header_data = { "version": 2, "width": 250, "height": 57, "timestamp": self.time, "env": { "SHELL": "/bin/bash", "TERM": "linux" }, "title": "boamp_webssh_record" } self.record_webssh(host, user,'header', header_data) # 连接建立一次，之后交互数据不会再进入该方法 for i in range(2): recv = self.channel.recv(102400).decode('utf-8') self.message['status'] = 0 self.message['message'] = recv message = json.dumps(self.message) self.websocker.send(message) now_time = time.time() data_list_temp = [now_time,recv] self.msg_list.append(data_list_temp) self.record_webssh(host,user,'iodata',data_list) self.msg_list = [] except socket.timeout as e: self.message['status'] = 1 self.message['message'] = 'ssh connection timed out' message = json.dumps(self.message) self.websocker.send(message) self.websocker.close() now_time = time.time() data_list_temp = [now_time, self.message['message']] self.msg_list.append(data_list_temp) self.record_webssh(host, user, 'iodata', data_list) self.msg_list = [] except Exception as e: print(e) print('异常文件：%s ,异常行号：%s' % (e.__traceback__.tb_frame.f_globals['__file__'], e.__traceback__.tb_lineno)) self.message['status'] = 1 self.message['message'] = str(e) message = json.dumps(self.message) self.websocker.send(message) self.websocker.close() now_time = time.time() data_list_temp = [now_time, self.message['message']] self.msg_list.append(data_list_temp) self.record_webssh(host, user, 'iodata', data_list) self.msg_list = [] def resize_pty(self, cols, rows): self.channel.resize_pty(width=cols, height=rows) def django_to_ssh(self, data): try: self.channel.send(data) return except: self.close() def websocket_to_django(self): global data_list try: while True: data = self.channel.recv(1024).decode('utf-8','ignore') if len(data) != 0: self.message['status'] = 0 self.message['message'] = data message = json.dumps(self.message) self.websocker.send(message) print("===================",len(self.msg_list)) now_time = time.time() data_list_temp = [now_time,data] if len(self.msg_list) < 60: #判断数据列表长度，60个内容就写入一次文件，避免了频繁写入文件，消耗io导致数据丢失的问题 self.msg_list.append(data_list_temp) else: self.msg_list.append(data_list_temp) self.record_webssh(self.host, self.user, 'iodata', data_list) self.msg_list = [] else: return except: self.close() def close(self): global data_list self.message['status'] = 1 self.message['message'] = 'Close connection' message = json.dumps(self.message) now_time = time.time() data_list_temp = [now_time, self.message['message']] self.msg_list.append(data_list_temp) self.record_webssh(self.host, self.user, 'iodata', data_list) self.msg_list = [] self.websocker.send(message) self.channel.close() self.websocker.close() def shell(self, data): Thread(target=self.django_to_ssh, args=(data,)).start() Thread(target=self.websocket_to_django).start()
5.前端代码，实现Asciinema-player播放器
<!DOCTYPE html> <html> <head> <meta charset="UTF-8"> <title>record_webssh_play</title> <meta http-equiv="X-UA-Compatible" content="IE=edge,chrome=1"> <link rel="stylesheet" href="/static/super_cmdb/css/font.css"> <link rel="stylesheet" href="/static/super_cmdb/css/xadmin.css"> <link rel="stylesheet" type="text/css" href="/static/super_cmdb/css/asciinema-player.css" /> </head> <body class="layui-anim layui-anim-up"> <div class="x-nav"> <span class="layui-breadcrumb"> <a href="">首页</a> <a href="">用户列表</a> <a> <cite>导航元素</cite></a> </span> <a class="layui-btn layui-btn-small" style="line-height:1.6em;margin-top:3px;float:right" href="javascript:location.replace(location.href);" title="刷新"> <i class="layui-icon" style="line-height:33px">ဂ</i></a> </div> <div class="x-body"> <asciinema-player title="录像回放" speed="1.5" autoplay author="xxxx" author-img-url="/static/super_cmdb/images/bg.png" src="/static/webssh/record_webssh/{{ record_filename }}"></asciinema-player> </div> <script type="text/javascript" src="/static/super_cmdb/js/jquery.min.js"></script> <script type="text/javascript" src="/static/super_cmdb/lib/layui/layui.js" charset="utf-8"></script> <script type="text/javascript" src="/static/super_cmdb/js/xadmin.js"></script> <script src="/static/super_cmdb/js/asciinema-player.js"></script> </body> </html>
6.效果展示
12.参考链接
https://ops-coffee.cn/s/oqzqgiq3unrnut-ngrh9lg
https://ops-coffee.cn/s/pcstabodjds8d15arwafza
https://www.cnblogs.com/37Y37/p/11909685.html