| Server IP : 62.109.13.187 / Your IP : 216.73.216.184 [ Web Server : Apache/2.4.6 (CentOS) mpm-itk/2.4.7-04 OpenSSL/1.0.2k-fips PHP/8.2.28 System : Linux robothost.ru 3.10.0-1160.119.1.el7.x86_64 #1 SMP Tue Jun 4 14:43:51 UTC 2024 x86_64 User : mosrembit ( 6064) PHP Version : 8.2.28 Disable Function : NONE Domains : 0 Domains MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : ON | Sudo : ON | Pkexec : ON Directory : /usr/local/mgr5/lib/pkgpython/ |
Upload File : |
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Модуль python для выполнения простых функций
import os
import sys
import subprocess
from xml.etree import cElementTree as etree
MGRPATH = '/usr/local/mgr5'
MGRCTL = os.path.join(MGRPATH, 'sbin/mgrctl')
def CmdRun(args, input=None):
''' Выполнить команду '''
sp = subprocess.Popen(args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
(out, err) = sp.communicate(input)
if sys.version_info[0] < 3:
return (out, err, sp.returncode)
else:
return (out.decode(), err.decode(), sp.returncode)
def ReloadMgr(mgrname):
''' Manager Restart '''
CmdRun([MGRCTL, '-m', mgrname, 'exit'])
def CreateDbMysql(dbname, user_root=False):
'''
Создать базу данных и пользователя к ней.
Eсли второй аргумент=True, то создаваемый пользователь - root.
TODO: Сейчас делается через шеловый скрипт. В теории надо наоборот.
'''
dbcreate_scriptname = os.path.join(MGRPATH, 'etc/scripts/create_db')
cmdargs = [dbcreate_scriptname, dbname]
# Если пользователь должен быть рут, добавляем к аргументам цифру "1". Нужна для шелового скрипта.
if user_root:
cmdargs.append('1')
# Зовём команду
out, err, code = CmdRun(cmdargs)
# Если код выхода не 0, генерируем исключение
if code == 0:
# Убираем перенос строки в конце строки
return out.strip()
else:
raise Exception('Can`t create database {0}\n{1}\n{2}'.format(dbname, out, err))
class BaseService(object):
''' Базовый виртуальный класс для работы с сервисами '''
def __init__(self, name):
''' Конструктор '''
self.name = name
self.exist = None
@staticmethod
def if_exist(method):
def check_is_exist(self, *args, **kwargs):
if self.exist:
return method(self, *args, **kwargs)
return check_is_exist
def reload(self, hard=False):
if not self.exist:
return None
CmdRun(['service', self.name, 'reload'])
def enable(self):
if not self.exist:
return None
CmdRun(['service', self.name, 'start'])
def disable(self):
if not self.exist:
return None
CmdRun(['service', self.name, 'stop'])
class CentosService(BaseService):
''' Класс для сервисов centos '''
def __init__(self, name):
BaseService.__init__(self, name)
self.script = os.path.join('/etc/init.d', name)
self.systemd_unit = os.path.join('/usr/lib/systemd/system/', '{0}.service'.format(name))
self.exist = os.path.exists(self.script) or os.path.exists(self.systemd_unit)
@BaseService.if_exist
def enable(self):
if super(CentosService, self).enable():
CmdRun(['chkconfig', self.name, 'on'])
@BaseService.if_exist
def disable(self):
if super(CentosService, self).disable():
CmdRun(['chkconfig', self.name, 'off'])
class DebianService(BaseService):
''' Класс для сервисов centos '''
def __init__(self, name):
BaseService.__init__(self, name)
self.script = os.path.join('/etc/init.d', name)
self.exist = os.path.exists(self.script)
@BaseService.if_exist
def enable(self):
if super(DebianService, self).enable():
CmdRun(['update-rc.d', self.name, 'enable'])
@BaseService.if_exist
def disable(self):
if super(DebianService, self).disable():
CmdRun(['update-rc.d', self.name, 'disable'])
class BsdService(BaseService):
''' Класс для сервисов FreeBSD '''
def __init__(self, name):
BaseService.__init__(self, name)
self.script_base = os.path.join('/etc/rc.d', name)
self.script_ext = os.path.join('/usr/local/etc/rc.d', name)
self.exist = os.path.exists(self.script_base) or os.path.exists(self.script_ext)
# TODO: Реализовать эти методы
def enable(self):
return
def disable(self):
return
class Service(BaseService):
''' Прокси класс '''
def __new__(self, name):
''' Вернуть нужный объект '''
if os.path.isfile('/sbin/chkconfig'):
return CentosService(name)
elif os.path.isfile('/etc/debian_version'):
return DebianService(name)
else:
uname, _, _ = CmdRun(['uname'])
if 'FreeBSD' in uname:
return BsdService(name)
class Mgr(object):
''' Класс для работы с манагером '''
def __init__(self, mgrname='core'):
self.mgrname = mgrname
self.mgrctl_list = [MGRCTL, '-m', self.mgrname]
def exit(self):
''' Перезагрузить манагер. '''
self.query({'func': 'exit'})
def query(self, params, restart=False):
''' Сделать запрос. Вернуть XML ответ '''
if not 'out' in params:
params['out'] = 'xml'
query_list = self.mgrctl_list[:] # Копируем, во избежание
query_list.append(params.pop('func'))
param_list = ['%s=%s' % (k, params.get(k)) for k in params]
query_list.extend(param_list)
if restart:
query_list.append('-R')
out, err, errcode = CmdRun(query_list)
if errcode == 0:
return out
else:
raise Exception(out + err)
def path_get(self, path_name):
''' Получить path '''
path_value_xml = etree.fromstring(self.query({
'func': 'pathlist.edit',
'elid': path_name
},
restart=True)
)
elem = path_value_xml.find('value')
if elem is not None and elem.text:
return elem.text
def param_get(self, param_name):
''' Получить параметр '''
param_value_xml = etree.fromstring(self.query({
'func': 'paramlist.edit',
'elid': param_name
},
restart=True)
)
elem = param_value_xml.find('value')
if elem is not None and elem.text:
return elem.text
def param_get_list(self, param_name):
param_value = self.param_get(param_name)
if param_value:
return param_value.split()
else:
return []
def param_add(self, param_name, param_value):
''' Добавить параметр '''
cur_values = self.param_get_list(param_name)
if param_value not in cur_values:
cur_values.append(param_value)
self.query({
'func': 'paramlist.edit',
'elid': param_name,
'value': ' '.join(cur_values),
'sok': 'ok'
}
)
self.exit()
return True
# Если не добавили, то возвращает false
return False
def param_remove(self, param_name, param_value):
''' Добавить параметр '''
cur_values = self.param_get_list(param_name)
if param_value in cur_values:
cur_values.remove(param_value)
self.query({
'func': 'paramlist.edit',
'elid': param_name,
'value': ' '.join(cur_values),
'sok': 'ok'
}
)
self.exit()
return True
# Если не добавили, то возвращает false
return False