====== OverAll ====== ===== Модули/Настройки окружения ===== # Установка модуля для конкретного python python3.8 -m pip install pyTelegramBotAPI # Модуль для телеграм бота бота pyTelegramBotAPI # Модуль для json логов python-json-logger # Модуль для виндовых хостов /usr/bin/python3.9 -m pip install pywinrm (узнать на каком именно питоне работает ансибл ansible --version) # pip3 install --upgrade pip pip3 install --upgrade setuptools ===== Аргументы командной строки ===== Параметры с префиксом тире необязательные, без префикса обязательные они видимо и являются обязательными\\
* **name or flags** - или имя, или список строк опций, например foo или -f, --foo, * **action** - основной тип действия, которое необходимо выполнить, * **nargs** - количество параметров командной строки, которые следует использовать, * **const** - постоянное значение, требуемое некоторыми action и nargs, * **default** - значение, полученное, если опция отсутствует в командной строке, * **type** - тип, в который должен быть преобразован параметр/опция командной строки, * **choices** - контейнер допустимых значений для параметра/опции, * **required** - можно ли опустить параметр командной строки (только для опций), * **help** - краткое описание того, что делает параметр/опция, * **metavar** - имя параметра/опции в сообщениях об использовании, * **dest** - имя параметра/опции, которое будет добавлено к объекту.
:!: Аргументы командной строки # file.py --a='string' import argparse parser= argparse.ArgumentParser(description='fvfv') parser.add_argument('--a', default='def-a', type=str, help='var - a') parser.add_argument('--b', default='def-b', type=str, help='var - b') args= parser.parse_args() print(args.a) print(args.b) print(args) ###################### import argparse argParser= argparse.ArgumentParser() argParser.add_argument('-f', type= str, help= 'Файл для создания хеш-строки') args= argParser.parse_args() if not args.f: print("Укажите файл для создания хеш-строки") exit()
===== Zabbix Web API =====
:!: Обращение к Zabbix, работа с JSON результатом import requests import json import os str1= """{ "jsonrpc": "2.0", "method": "template.get", "params": { "selectItems": "extend", "filter": { "host": [ "ServerApp" ] } }, "auth": "///////40a3593////", "id": 1 }""" query= str1.replace(' ', '').replace('\n', '') result= requests.post('http://10.10.10.30/zabbix/api_jsonrpc.php', json=json.loads(query)) items= result.json() outfile= open('ServApp-items.txt', 'w') outfile.write(str(items['result'][0]['items'])) print(str(items['result'][0]['items']))
:!: Обращение к Zabbix, токен авторизации query= """{"jsonrpc": "2.0", "method": "user.login", "params": { "user": "Admin", "password": "Admin" }, "id": 1, "auth": null }""".replace('\n', '') answer= requests.post('http://10.10.10.30/zabbix/api_jsonrpc.php', json=json.loads(query))
===== SQLite =====
:!: Работа с SQLite import sqlite3 con= sqlite3.connect('zbxtg-local.db'); # con.text_factory = bytes # Чтобы данные сохранялись как есть, без изменения кодировки и т.д. # cur= con.cursor(); cur.execute('create table if not exists `IdMessages` (`Id` INTEGER PRIMARY KEY AUTOINCREMENT, `MessageId` VARCHAR(45), `AlertId` VARCHAR(45));'); con.commit(); #cur.execute('insert into `IdMessages` (`MessageId`, `AlertId`) values ("1", "11"); '); #cur.execute('insert into `IdMessages` (`MessageId`, `AlertId`) values ("2", "22"); '); #cur.execute('insert into `IdMessages` (`MessageId`, `AlertId`) values ("3", "33"); '); #cur.execute('insert into `IdMessages` (`MessageId`, `AlertId`) values ("4", "44"); '); #cur.execute('insert into `IdMessages` (`MessageId`, `AlertId`) values ("5", "55"); '); #con.commit(); cur.execute('select `MessageId`, `AlertId` from `IdMessages`; '); #con.commit; cur.fetchall(); print("Count rows = ", cur.rowcount); print(cur.fetchall()) #for row in cur.fetchall(): # print(row) # # Доступ к данным # con.row_factory= sqlite3.Row; r= cur.fetchall(); print(r[0][0]+ ' -==- '+ r[0][1]); #for member in r: # print(member[0]+ ' -==- '+ member[1])
:!: Работа по телеграм-боту import requests import sqlite3 import json import sys class LocalDB: def __init__(self): self.con= None; self.cur= None; def Connect(self): self.con= sqlite3.connect('zbxtg-local.db'); self.cur= self.con.cursor(); # Создаем таблицу если нету self.cur.execute('create table if not exists `IdMessages` (`Id` INTEGER PRIMARY KEY AUTOINCREMENT, `MessageId` VARCHAR(45), `AlertId` VARCHAR(45));'); self.con.commit(); def Dissconnect(self): self.con.close(); def SaveMessageId(self, MessageId, AlertId): self.cur.execute('insert into `IdMessages` (`MessageId`, `AlertId`) values("'+str(MessageId)+'", "'+AlertId+'"); '); self.con.commit(); print("-= write is done =-"); def GetMessageId(self, AlertId): self.cur.execute('select `MessageId` from `IdMessages` where `AlertId`= "'+AlertId+'"; '); return self.cur.fetchall(); def main(): if(len(sys.argv) < 3): print("You must input number alert"); quit(); elif(sys.argv[1]== '-edit' and len(sys.argv) < 4): print("For edit message, input text"); quit(); if(sys.argv[1]== '-add'): textMessage= 'Alert #'+ sys.argv[2]+ '\nBody message \nMessage boddy'; mydata = {'chat_id': '-751016983', 'text': textMessage}; answer = requests.post('https://api.telegram.org/bot1074186011:AAHu8aVvPXU0GinMYkra6HnR5l53sqAZEcM/sendMessage', params=mydata); MessageId= answer.json()["result"]["message_id"]; if(MessageId): # Если получили в ответ ID сообщения, записываем его в БД db= LocalDB(); db.Connect(); db.SaveMessageId(MessageId, sys.argv[2]); db.Dissconnect(); print('Message sended! - '+ str(MessageId)); elif(sys.argv[1]== '-edit'): textMessage= 'Старое содержание полученное из алерта\n--------------------------------\n--==Новое содержание==--\nАлерт решен!!!'; db= LocalDB(); db.Connect(); MessId= db.GetMessageId(sys.argv[2]); db.Dissconnect(); for row in MessId: mydata = {'chat_id': '-751016983', 'message_id': row, 'text': textMessage}; answer = requests.post('https://api.telegram.org/bot1074186011:AAHu8aVvPXU0GinMYkra6HnR5l53sqAZEcM/editMessageText', params=mydata); print('Message is edited'); print(answer.json()); print('***************************************'); #print(answer.json()); if __name__ == "__main__": main()
===== Работа с текстом =====
:!: Разбор auth.log """./py.py > file.csv""" import re import os def main(): infile= open("auth.log", 'r') data= infile.readlines() # checkLogin(data) checkHosts(data) def checkHosts(inData): i= 1 lstry= [] lstry2= [] for line in inData: if line.find('Failed')== -1: continue line= line[line.find(' for '):].replace(' for ', '') if line.split()[0]== 'invalid': line= line.replace('invalid user ', '') lstry.append(line.replace('\n','')) lstry.sort() for line in lstry: lsline= line.split() trying= {'number': i, 'user': lsline[0], 'host': lsline[2], 'protocol': lsline[5]} lstry2.append(trying) i += 1 for item in lstry2: print(item['user']+ ','+ item['host']) def checkLogin(inData): isFailed= re.compile(r"Failed") logCorr= '\n' logIncorr= '\n' for i in inData: if isFailed.search(i): #i= i.replace('\n','') isLogin= re.compile(r"for invalid") if isLogin.search(i): logIncorr += i else: logCorr += i print("Login correct:", logCorr) #print(type(logCorr)) print("\n\nLogin Incorrect:", logIncorr) if __name__== '__main__': main()
===== e-mail =====
:!: Отправка письма В обоих случаях, в аккаунте нужно дополнительно разрешить аутентификацию такого скрипта (приложения)\\ В гугле- https://myaccount.google.com/lesssecureapps\\ В яндексе "портальный пароль" нужно включить\\ #!/usr/bin/python3 import smtplib srv= 'smtp.yandex.ru' / 'smtp.gmail.com' port= 587 login= '*******@yandex.ru' passwd= '********' mailto= '******@gmail.com' message = 'Subject: {}\n\n{}'.format('Warning Notification', 'This is warning notification and sending from my-client') s= smtplib.SMTP(srv, port) s.starttls() s.login(login, passwd) s.sendmail(login, mailto, message) s.quit()
:!: Тут вроде пример отправки вложения (не проверял) import smtplib from os.path import basename from email.mime.application import MIMEApplication from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from email.utils import COMMASPACE, formatdate def send_mail(send_from, send_to, subject, text, files=None, server="127.0.0.1"): assert isinstance(send_to, list) msg = MIMEMultipart() msg['From'] = send_from msg['To'] = COMMASPACE.join(send_to) msg['Date'] = formatdate(localtime=True) msg['Subject'] = subject msg.attach(MIMEText(text)) for f in files or []: with open(f, "rb") as fil: part = MIMEApplication( fil.read(), Name=basename(f) ) # After the file is closed part['Content-Disposition'] = 'attachment; filename="%s"' % basename(f) msg.attach(part) smtp = smtplib.SMTP(server) smtp.sendmail(send_from, send_to, msg.as_string()) smtp.close()
===== Файлы ===== * **read()** - загружает **весь файл в ОЗУ** * **readline()** - читает одну строчку * **for line in file:** - цикл обработки всего файла построчно
:!: Последовательная обработка файла Весь файл не будет загружаться в память\\ import os file= open('bigdict.txt', 'r') outfile= open('outfile.txt', 'w') for line in file: outfile.write(line)
===== WSGI Server ===== Сервер может обрабатывать веб запросы, передавать их приложениям\\ Переменная **environ** содержит служебную информацию о клиенте\\ В **start_response** собсна помещаем ответ
:!: Простой ответ на веб запросы к хосту #!/usr/bin/python3 import wsgiserver from wsgiref.simple_server import make_server, demo_app def my_app(environ, start_response): status= '200 OK' headers= [('Content-type', 'text/plain')] start_response(status, headers) return ['HelloW'.encode()] with make_server('', 8000, my_app) as httpd: print('Serving HTTP on port 8000..') httpd.serve_forever() httpd.handle_request()
:!: Содержимое environ {'SHELL': '/bin/bash', 'PWD': '/home/dn/pywh', 'LOGNAME': 'dn', 'XDG_SESSION_TYPE': 'tty', 'MOTD_SHOWN': 'pam', 'HOME': '/home/dn', 'LANG': 'ru_RU.UTF-8', 'LS_COLORS': 'rs=0:di=01;34:ln=01;36:mh=00:pi=40;33:so=01;35:do=01;35:bd=40;33;01:cd=40;33;01:or=40;31;01:mi=00:su=37;41:sg=30;43:ca=30;41:tw=30;42:ow=34;42:st=37;44:ex=01;32:*.tar=01;31:*.tgz=01;31:*.arc=01;31:*.arj=01;31:*.taz=01;31:*.lha=01;31:*.lz4=01;31:*.lzh=01;31:*.lzma=01;31:*.tlz=01;31:*.txz=01;31:*.tzo=01;31:*.t7z=01;31:*.zip=01;31:*.z=01;31:*.dz=01;31:*.gz=01;31:*.lrz=01;31:*.lz=01;31:*.lzo=01;31:*.xz=01;31:*.zst=01;31:*.tzst=01;31:*.bz2=01;31:*.bz=01;31:*.tbz=01;31:*.tbz2=01;31:*.tz=01;31:*.deb=01;31:*.rpm=01;31:*.jar=01;31:*.war=01;31:*.ear=01;31:*.sar=01;31:*.rar=01;31:*.alz=01;31:*.ace=01;31:*.zoo=01;31:*.cpio=01;31:*.7z=01;31:*.rz=01;31:*.cab=01;31:*.wim=01;31:*.swm=01;31:*.dwm=01;31:*.esd=01;31:*.jpg=01;35:*.jpeg=01;35:*.mjpg=01;35:*.mjpeg=01;35:*.gif=01;35:*.bmp=01;35:*.pbm=01;35:*.pgm=01;35:*.ppm=01;35:*.tga=01;35:*.xbm=01;35:*.xpm=01;35:*.tif=01;35:*.tiff=01;35:*.png=01;35:*.svg=01;35:*.svgz=01;35:*.mng=01;35:*.pcx=01;35:*.mov=01;35:*.mpg=01;35:*.mpeg=01;35:*.m2v=01;35:*.mkv=01;35:*.webm=01;35:*.webp=01;35:*.ogm=01;35:*.mp4=01;35:*.m4v=01;35:*.mp4v=01;35:*.vob=01;35:*.qt=01;35:*.nuv=01;35:*.wmv=01;35:*.asf=01;35:*.rm=01;35:*.rmvb=01;35:*.flc=01;35:*.avi=01;35:*.fli=01;35:*.flv=01;35:*.gl=01;35:*.dl=01;35:*.xcf=01;35:*.xwd=01;35:*.yuv=01;35:*.cgm=01;35:*.emf=01;35:*.ogv=01;35:*.ogx=01;35:*.aac=00;36:*.au=00;36:*.flac=00;36:*.m4a=00;36:*.mid=00;36:*.midi=00;36:*.mka=00;36:*.mp3=00;36:*.mpc=00;36:*.ogg=00;36:*.ra=00;36:*.wav=00;36:*.oga=00;36:*.opus=00;36:*.spx=00;36:*.xspf=00;36:', 'SSH_CONNECTION': '192.168.0.10 59664 192.168.0.64 22', 'XDG_SESSION_CLASS': 'user', 'TERM': 'xterm', 'USER': 'dn', 'SHLVL': '1', 'XDG_SESSION_ID': '1', 'XDG_RUNTIME_DIR': '/run/user/1000', 'SSH_CLIENT': '192.168.0.10 59664 22', 'PATH': '/usr/local/bin:/usr/bin:/bin:/usr/local/games:/usr/games', 'SSH_TTY': '/dev/pts/0', '_': './whsrv.py', 'OLDPWD': '/home/dn', 'SERVER_NAME': 'deb64', 'GATEWAY_INTERFACE': 'CGI/1.1', 'SERVER_PORT': '8000', 'REMOTE_HOST': '', 'CONTENT_LENGTH': '26', 'SCRIPT_NAME': '', 'SERVER_PROTOCOL': 'HTTP/1.1', 'SERVER_SOFTWARE': 'WSGIServer/0.2', 'REQUEST_METHOD': 'POST', 'PATH_INFO': '/', 'QUERY_STRING': '', 'REMOTE_ADDR': '192.168.0.10', 'CONTENT_TYPE': 'application/json', 'HTTP_HOST': '192.168.0.64:8000', 'HTTP_USER_AGENT': 'curl/7.79.1', 'HTTP_ACCEPT': '*/*', 'wsgi.input': <_io.BufferedReader name=4>, 'wsgi.errors': <_io.TextIOWrapper name='' mode='w' encoding='utf-8'>, 'wsgi.version': (1, 0), 'wsgi.run_once': False, 'wsgi.url_scheme': 'http', 'wsgi.multithread': False, 'wsgi.multiprocess': False, 'wsgi.file_wrapper': }
:!: Парсим, сортируем список треков из ВК **raw** получаем вручную\\ **pip install beautifulsoup4**\\ from bs4 import BeautifulSoup import re def main(): ParseFromRaw() def ParseFromRaw(): infile= open('vk_raw.txt', 'r', encoding= 'utf-8') rawdata= infile.read() # Парсим содержимое как html parsdata= BeautifulSoup(rawdata, 'html.parser') # Извлекаем только то что "BeautifulSoup" определяет как текст textls= parsdata.get_text().split('\n') # К каждому треку относится >= 3 строки инфы, здесь она идет сплошником, отдельными строками # последней строкой каждого трека идет его длительность, разбиваем треки по ней # для результата reslistInlist= [] # тут накапливается инфа по одному треку buffer= [] # регулярка для определения строки с длительностью timeline= re.compile(r'^\d{1,2}:\d{1,2}.*') for line in textls: # есть пустые строки, пропускаем if line == '': continue buffer.append(line.strip()) # Делаем обрезку после строки с длительностью if timeline.search(line): reslistInlist.append(buffer.copy()) buffer.clear() # каждый трек записан в строку, в типе list, список всех треков ListInList reslistInlist.sort() count= 0 outfile= open('vk_res_sort.txt', 'w', encoding= 'utf-8') for line in reslistInlist: # Одна строка один трек, записи одного трека разделяем- `;` + tab outfile.write(';\t'.join(line)+ '\n') count += 1 print("Всего записей - "+ str(count)) if __name__ == "__main__": main()
===== Yandex Disk API =====
:!: Yandex Disk API Скрипт для работы с диском, создание каталога, загрузка файлов\\ Неплохие примеры обхода дерева каталогов\\ [[https://yandex.ru/dev/disk/poligon/|Неплохая дока]]\\ [[https://ramziv.com/article/8|Собсна источник примера]]\\ #!/bin/python3 import os import requests from datetime import datetime from progress.bar import Bar URL = 'https://cloud-api.yandex.net/v1/disk/resources' TOKEN = '' headers = {'Content-Type': 'application/json', 'Accept': 'application/json', 'Authorization': f'OAuth {TOKEN}'} def create_folder(path): """Создание папки. \n path: Путь к создаваемой папке.""" requests.put(f'{URL}?path={path}', headers=headers) def upload_file(loadfile, savefile, replace=False): """Загрузка файла. savefile: Путь к файлу на Диске loadfile: Путь к загружаемому файлу replace: true or false Замена файла на Диске""" res = requests.get(f'{URL}/upload?path={savefile}&overwrite={replace}', headers=headers, verify=False).json() with open(loadfile, 'rb') as f: try: requests.put(res['href'], files={'file':f}, verify=False) except KeyError: print(res) def backup(savepath, loadpath): """Загрузка папки на Диск. \n savepath: Путь к папке на Диске для сохранения \n loadpath: Путь к загружаемой папке""" # Формируем название папки на диске, название целевой папки (из переданного пути) + дата-время # `loadpath.split('/')[-1]` разбиваем путь на list, и возвращаем последний элемент т.к. папку date_folder = 'currDate'#'{0}_{1}'.format(loadpath.split('/')[-1], datetime.now().strftime("%Y-%m-%d_%H:%M")) create_folder(savepath) # ф-я `os.walk` возвращает массив файлов из дерева каталогов (в параметрах можно указать направление обхода) # возвращает тройной кортеж - "dirpath (тек путь), dirnames (папки), filenames (файлы)" # по дефолту вложенность вся, идет по порядку от указанного пути и далее for address, _, files in os.walk(loadpath): # тут создаем подпапки, из путей, что приходят в цикле обхода # `address.replace(loadpath, "")[1:]` - здесь вычитаем исходный путь из вложенного create_folder('{0}/{1}/{2}'.format(savepath, date_folder, address.replace(loadpath, "")[1:])) bar = Bar('Loading', fill='X', max=len(files)) # Циклом проходим по всем файлам что есть в данном каталоге for file in files: bar.next() # в первой части путь в ФС, во второй путь на диске, исходный снова вырезан upload_file('{0}/{1}'.format(address, file),\ '{0}/{1}{2}/{3}'.format(savepath, date_folder, address.replace(loadpath, ""), file)) bar.finish() if __name__ == '__main__': backup('Backup', r'/home/ddn/bcps') #backup('Backup', os.getcwd()) {{:develop:pyton:screenshot_4.jpg?direct&800|}}
:!: Yandex Disk API. += На скорую руку, загрузка указанного файла #!/bin/python3 import os, argparse import requests from datetime import datetime from progress.bar import Bar argParser= argparse.ArgumentParser() argParser.add_argument('upFilename', type= str, help= 'Файл для загрузки') argParser.add_argument('saveFolder', type= str, help= 'Папка дял сохранения') args= argParser.parse_args() URL = 'https://cloud-api.yandex.net/v1/disk/resources' TOKEN = 'AQAAAAA5-bYIAADLW5YbQ0L1yUakkN5VrOYQtto' headers = {'Content-Type': 'application/json', 'Accept': 'application/json', 'Authorization': f'OAuth {TOKEN}'} def create_folder(path): """Создание папки. \n path: Путь к создаваемой папке.""" requests.put(f'{URL}?path={path}', headers=headers) def upload_file(loadfile, savefile, replace=False): """Загрузка файла. savefile: Путь к файлу на Диске loadfile: Путь к загружаемому файлу replace: true or false Замена файла на Диске""" res = requests.get(f'{URL}/upload?path={savefile}&overwrite={replace}', headers=headers, verify=False).json() with open(loadfile, 'rb') as f: try: requests.put(res['href'], files={'file':f}, verify=False) except KeyError: print(res) def backup(savepath, loadpath): """Загрузка папки на Диск. \n savepath: Путь к папке на Диске для сохранения \n loadpath: Путь к загружаемой папке""" # Формируем название папки на диске, название целевой папки (из переданного пути) + дата-время # `loadpath.split('/')[-1]` разбиваем путь на list, и возвращаем последний элемент т.е. папку date_folder = 'currDate'#'{0}_{1}'.format(loadpath.split('/')[-1], datetime.now().strftime("%Y-%m-%d_%H:%M")) create_folder(savepath) # ф-я `os.walk` возвращает массив файлов из дерева каталогов (в параметрах можно указать направление обхода) # возвращает тройной кортеж - "dirpath (тек путь), dirnames (папки), filenames (файлы)" # по дефолту вложенность вся, идет по порядку от указанного пути и далее for address, _, files in os.walk(loadpath): # тут создаем подпапки, из путей, что приходят в цикле обхода # `address.replace(loadpath, "")[1:]` - здесь вычитаем исходный путь из вложенного create_folder('{0}/{1}/{2}'.format(savepath, date_folder, address.replace(loadpath, "")[1:])) bar = Bar('Loading', fill='X', max=len(files)) # Циклом проходим по всем файлам что есть в данном каталоге for file in files: bar.next() # в первой части путь в ФС, во второй путь на диске, исходный снова вырезан upload_file('{0}/{1}'.format(address, file),\ '{0}/{1}{2}/{3}'.format(savepath, date_folder, address.replace(loadpath, ""), file+ "this_my_file")) bar.finish() def backupOnce(diskFolder, uploadFilePath): """Загрузка папки на Диск. \n diskFolder: Путь к папке на Диске для сохранения \n uploadFilePath: Путь к загружаемой папке""" # Формируем название папки на диске, название целевой папки (из переданного пути) + дата-время # `uploadFilePath.split('/')[-1]` разбиваем путь на list, и возвращаем последний элемент т.е. папку currDateTime = datetime.now().strftime("%Y-%m-%d_%H-%M") create_folder(diskFolder) upload_file(uploadFilePath, '{0}/{1}'.format(diskFolder, currDateTime+'_'+ uploadFilePath.split('/')[-1] )) if __name__ == '__main__': backupOnce(args.saveFolder, args.upFilename) #backup('Backup', os.getcwd())
===== Хеш функции. Hashlib ===== Читать файл лучше всего по блокам, функцией **update()** данные помещаются в объект, можно делать это поблочно\\ Собсна каждый алгоритм имеет свою функцию, с соответствующим названием\\ **digest()** - хеш в виде байтов, **hexdigest()** в виде строки\\
:!: Хеш строка из файла import hashlib infile= open('my-file', 'rb') hsh= hashlib.md5() while True: fileData= infile.read(2048) if not fileData: break hsh.update(fileData) result= hsh.hexdigest() print(result)
:!: Перебор папки #!/usr/bin/python3 import hashlib, argparse, os argParser= argparse.ArgumentParser() argParser.add_argument('filename', type= str, help= 'Проверяемый файл') argParser.add_argument('folder', type= str, help= 'Путь для проверки') args= argParser.parse_args() def main(): # Хеш из проверяемого файла #checkFileHash= GetHash(args.filename) # Перебираем целевую папку for file in os.listdir(args.folder): currentHash= GetHash(args.folder +'/'+ file) print(file +' - '+ currentHash) def GetHash(fullFileName): """ Сгенерировать хеш строку из указанного файла """ openedFile= open(fullFileName, 'rb') md5= hashlib.md5() while True: fileData= openedFile.read(2048) if not fileData: break md5.update(fileData) return md5.hexdigest() if __name__ == "__main__": main()
===== Работа с оболочкой =====
:!: Пару вариантов Вроде более рекомендованный способ, как то заморочено вроде import subprocess var= subprocess.run("ls /home/admin", shell=True, stdout=subprocess.PIPE, encoding='utf-8') print(var.stdout) Вывод сразу на консоль, возвращает код возврата\\ import os os.system("ls /home/admin >/dev/null")
===== Цвет текста =====
:!: ANSI последовательности, работает для *nix консолей {{:develop:pyton:py_color.png?direct&400|}} class bcolors: HEADER = '\033[95m' OKBLUE = '\033[94m' OKCYAN = '\033[96m' OKGREEN = '\033[92m' WARNING = '\033[93m' FAIL = '\033[91m' ENDC = '\033[0m' BOLD = '\033[1m' UNDERLINE = '\033[4m' print(bcolors.HEADER+ 'This is test TeXt in color'+ bcolors.ENDC) print(bcolors.OKBLUE+ 'This is test TeXt in color'+ bcolors.ENDC) print(bcolors.OKCYAN+ 'This is test TeXt in color'+ bcolors.ENDC) print(bcolors.OKGREEN+ 'This is test TeXt in color'+ bcolors.ENDC) print(bcolors.WARNING+ 'This is test TeXt in color'+ bcolors.ENDC) print(bcolors.FAIL+ 'This is test TeXt in color'+ bcolors.ENDC) print(bcolors.BOLD+ 'This is test TeXt in color'+ bcolors.ENDC) print(bcolors.UNDERLINE+ 'This is test TeXt in color'+ bcolors.ENDC)
:!: Обширный пример #!/usr/bin/python import subprocess, os glNameNewImport= "" glSelectedMultip= "" glDictFreePorts= {} glHostIP= "" class bcolors: HEADER = '\033[95m' OKBLUE = '\033[94m' OKCYAN = '\033[96m' OKGREEN = '\033[92m' WARNING = '\033[93m' FAIL = '\033[91m' ENDC = '\033[0m' BOLD = '\033[1m' UNDERLINE = '\033[4m' def main(): global glNameNewImport # # -= IP адрес хоста =- getHostIP() # # -= Имя нового инстанса =- getNameNewImport() # # -= Ищем порты для нового инстанса (все) =- GetFreePorts() # # -= Выбор мультипликатора =- if SelectMultipl(): # -= Ветка создания нового мультипл =- createNewMultip() # Создаем входящую директорию CreateFolder("/tank/IN/"+ glSelectedMultip.replace('FileMultiplicator_', '')) # Шарим ее по самбе addSambaShared("/tank/IN/"+ glSelectedMultip.replace('FileMultiplicator_', '')) else: # -= Использование существующего =- useExistMultip() # # Переходим к импорту # -= Создаем входящую папку импорта и папки для ошибок =- CreateFolder("/tank/MGA_IN/IN_"+ glNameNewImport) CreateFolder("/tank_storage/dberrors/DBERROR_"+ glNameNewImport) CreateFolder("/tank_storage/septics/SEPTIC_"+ glNameNewImport) # # -= Создаем копированием новый импорт, пишем ему параметры =- createNewImport() print("""\n\033[92mНовый импорт успешно создан!\033[0m Параметры следующие: Имя: \033[92m'"""+ glNameNewImport+"""'\033[0m Расположение: \033[92m'/tank/BIN/MgaImport/MgaImport_"""+ glNameNewImport+"""'\033[0m Входящая директория: \033[92m'/tank/MGA_IN/IN_"""+ glNameNewImport+"""'\033[0m Jmx-порт: \033[92m'"""+ glDictFreePorts['ImportJmxPort']+"""'\033[0m Командный порт: \033[92m'"""+ glDictFreePorts['ImportCommandPort']+"""'\033[0m Конфигурация импорта: \033[92m'/tank/BIN/MgaImport/MgaImport_"""+ glNameNewImport+"""/properties/mgaimport.toml'\033[0m Общая конфигурация всех импортов: \033[92m'/tank/BIN/MgaImport/app.conf'\033[0m Файл мультипликатор: \033[92m'/tank/BIN/FileMultiplicators/"""+ glSelectedMultip+"""'\033[0m Укажите \033[96mпараметры импорта в БД\033[0m в файле конфигурации импорта Используя Jmx-порт, создайте самостоятельно \033[96mмониторинг в Zabbix\033[0m, при необходимости\n""") def getNameNewImport(): """ Ввод имени импорта пользователем """ global glNameNewImport glNameNewImport= input("\nВведите имя для нового импорта: ").replace(' ', '') if len(glNameNewImport)== 0: print(bcolors.FAIL+ "Вы не указали имя новому сервису, не надо так :("+ bcolors.ENDC) exit(1) # Проверим уникальность имени existImports= subprocess.run('cat /tank/BIN/MgaImport/app.conf | grep -Poi "WORKSPACE=.*MgaImport_\K.*"', shell=True, stdout=subprocess.PIPE, encoding='utf-8').stdout.split() if glNameNewImport in existImports: print(bcolors.FAIL+ "Импорт с таким именем уже существует"+ bcolors.ENDC) exit(1) def createNewImport(): """ Создание нового импорта """ global glNameNewImport, glDictFreePorts, glHostIP # Копированием создаем сам новый мультип subprocess.run('cp -r ClearImport /tank/BIN/MgaImport/MgaImport_'+ glNameNewImport, shell=True, stdout=subprocess.PIPE, encoding='utf-8') # Меняем права доступа юзеру mgaimport subprocess.run('chown -R mgaimport:mgaimport /tank/BIN/MgaImport/MgaImport_'+ glNameNewImport, shell=True, stdout=subprocess.PIPE, encoding='utf-8') # Пишем параметры в "cmd.properties" Configuration= subprocess.run("cat /tank/BIN/MgaImport/MgaImport_"+glNameNewImport +"/cmd.properties", shell=True, stdout=subprocess.PIPE, encoding='utf-8').stdout Configuration= Configuration.replace('%JmxPort%', glDictFreePorts['ImportJmxPort']) Configuration= Configuration.replace('%CommandPort%', glDictFreePorts['ImportCommandPort']) Configuration= Configuration.replace('%ImportName%', glNameNewImport) multipConfig= open("/tank/BIN/MgaImport/MgaImport_"+glNameNewImport +"/cmd.properties", "w", encoding='utf-8') multipConfig.write(Configuration) multipConfig.close() # Пишем параметры в "./properties/mgaimport.toml" Configuration= subprocess.run("cat /tank/BIN/MgaImport/MgaImport_"+glNameNewImport +"/properties/mgaimport.toml", shell=True, stdout=subprocess.PIPE, encoding='utf-8').stdout Configuration= Configuration.replace('%InputDir%', "/tank/MGA_IN/"+ glNameNewImport) Configuration= Configuration.replace('%SepticDir%', "/tank_storage/septics/SEPTIC_"+ glNameNewImport) Configuration= Configuration.replace('%DBErrorDir%', "/tank_storage/dberrors/DBERROR_"+ glNameNewImport) Configuration= Configuration.replace('%CommandPort%', glDictFreePorts['ImportCommandPort']) multipConfig= open("/tank/BIN/MgaImport/MgaImport_"+glNameNewImport +"/properties/mgaimport.toml", "w", encoding='utf-8') multipConfig.write(Configuration) multipConfig.close() # Делаем запись о новом мультипе в общем конфиге newText= """\n["""+glNameNewImport +"""] #---------------------------------------- 1 ENABLE="true" JAVA_HOME=/usr/lib/jvm/bellsoft-java14-full.x86_64/bin START_USER=mgaimport START_TIME=30 STOP_TIME=90 COMMAND_PORT="""+glDictFreePorts['ImportCommandPort'] +""" WORKSPACE=/tank/BIN/MgaImport/MgaImport_"""+glNameNewImport +""" CMD='java -XX:+UnlockExperimentalVMOptions -XX:+UseG1GC -XX:+DisableExplicitGC -Xmx16G -Xms3G -XX:InitiatingHeapOccupancyPercent=75 -XX:G1NewSizePercent=1 -XX:G1MaxNewSizePercent=15 -XX:G1MixedGCLiveThresholdPercent=90 -XX:G1HeapWastePercent=5 -XX:G1ReservePercent=1 -XX:+ExplicitGCInvokesConcurrent -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath="/tank/BIN/MgaImport/MgaImport_"""+glNameNewImport +"""/out_of_memory_dump/dump.hprof" -XX:-OmitStackTraceInFastThrow -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.port="""+glDictFreePorts['ImportJmxPort'] +""" -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.rmi.server.hostname="""+glHostIP +""" -Djava.net.preferIPv4Stack=true -jar mgaimport.jar'\n""" configFile= open("/tank/BIN/MgaImport/app.conf", "a", encoding='utf-8') configFile.write(newText) configFile.close() def getHostIP(): """ Получаем IP адрес хоста """ global glHostIP # Получаем перечень ИПшников в системе haveList= [] haveList= subprocess.run('ip a | grep -Po "inet \K[^/]*" | grep -v "127.0.0.1"', shell=True, stdout=subprocess.PIPE, encoding='utf-8').stdout.split() # Если не удалось, просим ввести вручную, если несколько, просим выбрать if len(haveList) == 0: glHostIP= input('\nНе найден IP адрес хоста, укажите вручную: ').replace(' ', '') if len(glHostIP)== 0: print(bcolors.FAIL+ "Вы не указали IP адрес хоста, не надо так :("+ bcolors.ENDC) exit(1) elif len(haveList) == 1: glHostIP= haveList[0] else: print("\nНайдено несколько IP адресов:") showList= {} i= 1 for itemIP in haveList: showList[i] = itemIP i += 1 for x, y in showList.items(): print(str(x) +" - "+ y) SelectItem= input("\nВыберите актуальный: ") if not SelectItem.isdigit(): print(bcolors.FAIL+ "Указано некорректное значение :("+ bcolors.ENDC) exit(1) glHostIP= showList[int(SelectItem)] print(bcolors.OKGREEN+ "\nIP адрес хоста - "+ glHostIP+ bcolors.ENDC) def addSambaShared(fullPath): """ Добавляем шаринг папки в конфиг самбы Параметры дефолтные (для импортов) """ # Добавляем запись о новой шаре в конфиг самбы newText= """\n["""+fullPath.split('/')[-1] +"""] path = """+fullPath +""" guest ok = no browsable = no writeable = yes create mask = 0770 directory mask = 0770 write list = xlogssupplier valid users = xlogssupplier force user = mgaimport force group = mgaimport\n""" configFile= open("/etc/samba/smb.conf", "a", encoding='utf-8') configFile.write(newText) configFile.close() # Перезапускаем самбу subprocess.run('systemctl restart smb', shell=True) # Напоминаем пользователю что нужно примонтировать ее на источнике источника print("""\033[91m Внимание!\033[0m Создана входящая папка для файлмультипликатора \033[96m'"""+ fullPath+ """'\033[0m К ней \033[96mоткрыт сетевой доступ\033[0m через "smb", \033[96mее необходимо примонтировать\033[0m на сервере "лог-дистрибьютора" (либо какой то другой источник) для получения входящих логов""") def createNewMultip(): """ Создание нового мультипа """ global glNameNewImport, glSelectedMultip, glDictFreePorts, glHostIP # Копированием создаем сам новый мультип subprocess.run('cp -r ClearFileMultiplicator /tank/BIN/FileMultiplicators/'+ glSelectedMultip, shell=True, stdout=subprocess.PIPE, encoding='utf-8') # Меняем права доступа юзеру mgaimport subprocess.run('chown -R mgaimport:mgaimport /tank/BIN/FileMultiplicators/'+ glSelectedMultip, shell=True, stdout=subprocess.PIPE, encoding='utf-8') # Пишем параметры в конфиг нового мультипа Configuration= subprocess.run("cat /tank/BIN/FileMultiplicators/"+glSelectedMultip +"/FileMultiplicator.toml", shell=True, stdout=subprocess.PIPE, encoding='utf-8').stdout Configuration= Configuration.replace('%InputDir%', '"/tank/IN/'+ glSelectedMultip.replace('FileMultiplicator_', '') +'"') Configuration= Configuration.replace('%CommandPort%', glDictFreePorts['MultipCommandPorts']) Configuration= Configuration.replace('%ImportName%', glNameNewImport) Configuration= Configuration.replace('%ImportPath%', '"/tank/MGA_IN/in_'+ glNameNewImport +'"') multipConfig= open("/tank/BIN/FileMultiplicators/"+glSelectedMultip +"/FileMultiplicator.toml", "w", encoding='utf-8') multipConfig.write(Configuration) multipConfig.close() # Делаем запись о новом мультипе в общем конфиге newText= """\n["""+glSelectedMultip +"""] # ----------------------------------- ENABLE="true" JAVA_HOME=/usr/lib/jvm/bellsoft-java14-full.x86_64/bin START_USER=mgaimport START_TIME=5 STOP_TIME=10 WORKSPACE=/tank/BIN/FileMultiplicators/"""+glSelectedMultip +""" CMD='java -XX:+UnlockExperimentalVMOptions -XX:+UseG1GC -XX:+DisableExplicitGC -Xmx2G -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.port="""+glDictFreePorts['MultipJmxPort'] +""" -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.rmi.server.hostname="""+glHostIP +""" -Djava.net.preferIPv4Stack=true -jar filemultiplicator-all.jar'\n""" configFile= open("/tank/BIN/FileMultiplicators/app.conf", "a", encoding='utf-8') configFile.write(newText) configFile.close() def useExistMultip(): """ Использование существующего мультипа, дописываем в его конфиг блок по новому импорту """ global glNameNewImport, glSelectedMultip newText= """\n[[DestDir]] name = \""""+glNameNewImport +"""\" destDir = "/tank/MGA_IN/in_"""+glNameNewImport +"""" pattern = "^.*$"\n""" configFile= open("/tank/BIN/FileMultiplicators/"+glSelectedMultip +"/FileMultiplicator.toml", "a", encoding='utf-8') configFile.write(newText) configFile.close() # Перезапуск сервиса, для принятия изменений os.system('/bin/bash /tank/BIN/FileMultiplicators/'+glSelectedMultip +'/restart.sh') os.system('/bin/bash /tank/BIN/FileMultiplicators/'+glSelectedMultip +'/status.sh') def SelectMultipl(): """ Выбор мультипликатора (в т.ч. создание нового) """ global glSelectedMultip # Перечень мультипликаторов с путями из общешго конфига listMultiplLocate= subprocess.run('cat /tank/BIN/FileMultiplicators/app.conf | grep "WORKSPACE" | grep -Po "/.*"', shell=True, stdout=subprocess.PIPE, encoding='utf-8').stdout.split() # Проходим циклом, из конфига каждого мультип берем параметр "srcDir", и делаем сопоставление в словаре availableMultiples= {} i= 1 listNamesForCheckUnique= [] for itemLocate in listMultiplLocate: availableMultiples[i] = { 'HomeDir': itemLocate.split('/')[-1], 'SrcDir': subprocess.run('cat '+ itemLocate +'/FileMultiplicator.toml | grep srcDir | grep -Po "\\".*\\""', shell=True, stdout=subprocess.PIPE, encoding='utf-8').stdout.split()[0] } i += 1 listNamesForCheckUnique.append(itemLocate.split('/')[-1]) # Выводм перечеь доступных мультипликаторов пользователю, для выбора print("\nВыберите "+bcolors.OKCYAN +"существующий"+bcolors.ENDC+" файлмультипликатор, либо создайте "+bcolors.OKCYAN+"новый"+bcolors.ENDC+", исходя из источника данных:") print("0: Создать новый файлмультипликатор") for x, y in availableMultiples.items(): print(str(x) +": "+ y['HomeDir'] +'; srcDir: '+ y['SrcDir']) SelectItem= input("\nВыберите действие: ") if not SelectItem.isdigit(): print(bcolors.FAIL+ "Указано некорректное значение :("+ bcolors.ENDC) exit(1) # Выбирается мультип либо создание нового если выбран 0 (или вне диапазона) if int(SelectItem) in availableMultiples: glSelectedMultip= availableMultiples[int(SelectItem)]['HomeDir'] return 0 else: glSelectedMultip= 'FileMultiplicator_'+ input("Укажите название новому файлмультипликатору: "+bcolors.OKCYAN +"FileMultiplicator_").replace(' ', '') print(bcolors.ENDC) if glSelectedMultip== 'FileMultiplicator_': print(bcolors.FAIL+ "Вы не указали имя новому сервису, не надо так :("+ bcolors.ENDC) exit(1) elif glSelectedMultip in listNamesForCheckUnique: print(bcolors.FAIL+ "Сервис с таким именем уже существует"+ bcolors.ENDC) exit(1) return 1 def CreateFolder(fullPath): """ Создаем указанную директорию, сразу назначаем пользователя и права """ result= subprocess.run('install -d -m 775 -o mgaimport -g mgaimport '+ fullPath, shell=True) if result.returncode: if input("\nОшибка при создании директории, можете создать вручную. Важно задать доступ пользователю mgaimport\nПродолжить скрипт? ")!= 'y': exit(1) def GetFreePorts(): """ Подбираем свободные порты (Командный + Jmx) для нового инстанса """ global glDictFreePorts # Мультипликатор # Команный порт MultipCommandPorts= subprocess.run('cat /tank/BIN/FileMultiplicators/*/FileMultiplicator.toml | grep "listeningCommandPort" | grep -Po "\d*"', shell=True, stdout=subprocess.PIPE, encoding='utf-8').stdout.split() glDictFreePorts['MultipCommandPorts'] = FindFreePort(MultipCommandPorts, 5550) # Jmx порт MultipJmxPorts= subprocess.run('cat /tank/BIN/FileMultiplicators/app.conf | grep "jmxremote.port" | grep -Po "\d*"', shell=True, stdout=subprocess.PIPE, encoding='utf-8').stdout.split() glDictFreePorts['MultipJmxPort']= FindFreePort(MultipJmxPorts, 10040) # Импорт # Команный порт ImportCommandPorts= subprocess.run('cat /tank/BIN/MgaImport/app.conf | grep "COMMAND_PORT" | grep -Po "\d*"', shell=True, stdout=subprocess.PIPE, encoding='utf-8').stdout.split() glDictFreePorts['ImportCommandPort']= FindFreePort(ImportCommandPorts, 6660) # Jmx порт ImportJmxPorts= subprocess.run('cat /tank/BIN/MgaImport/app.conf | grep "jmxremote.port" | grep -Po "\d*"', shell=True, stdout=subprocess.PIPE, encoding='utf-8').stdout.split() glDictFreePorts['ImportJmxPort']= FindFreePort(ImportJmxPorts, 10050) def FindFreePort(listExistPorts, defaultNewPort): """ Ищем новый свободный порт. Передаем сюда список используемых, ищем наибольший, увеличиваем его на единицу Если список пуст, используем начальный, выбор начального передается сюда же """ if len(listExistPorts)== 0: return str(defaultNewPort) else: listExistPorts.sort(reverse=True) return str(int(listExistPorts[0])+ 1) if __name__ == "__main__": main()
===== Кодировка ===== [[https://chase-seibert.github.io/blog/2014/01/12/python-unicode-console-output.html|На счет вывода]]\\
:!: Кодировка # При открытии файла with open(my_file, 'r', encoding="utf-8") as raw_input_log: ... # Так же проблема может возникнуть при записи в консоль # Ошибка "UnicodeEncodeError: 'ascii' codec can't encode character" # Смысл в том что python путает какую кодировку принимает консоль, нужно установить ее явно # можно установить через переменную окружения, напирмер перед запуском скрипта set PYTHONIOENCODING=UTF-8 & python make_backup.py
===== Ошибка при отсутствии lxml ===== Была трабла при ее установке, как вариант эти пару команд\\ yum install python-devel pip3 install -U pip setuptools python3 -m pip install lxml ===== Ошибка при отсутствии lxml =====
:!: