# -*- coding: utf-8 -*-
"""
Модуль py_cerebro.cclib содержит вспомогательные функции обработки хешей и битовых флагов.
.. rubric:: Функции
* :py:func:`hash16_64() <py_cerebro.cclib.hash16_64>`
* :py:func:`hash64_16() <py_cerebro.cclib.hash64_16>`
* :py:func:`has_flag() <py_cerebro.cclib.has_flag>`
"""
import base64
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.base import MIMEBase
from email.header import Header
import smtplib
import time
import sys
import tempfile
import os
[документация]def hash16_64(b16_str):
"""
:param string b16_str: хеш в формате base16
:returns: хеш в формате base64
:rtype: string
Выполняет конвертирование строки хеша из формата base16 в base64.
"""
if len(b16_str)!=64:
raise Exception('Wrong hash length (muse be 64)');
ba = base64.b16decode(b16_str.encode('ascii'));
return base64.standard_b64encode(ba).decode('ascii').replace('+', '-').replace('/', '_').replace('=', '~')
[документация]def hash64_16(b64_str):
"""
:param string b64_str: хеш в формате base64
:returns: хеш в формате base16
:rtype: string
Выполняет конвертирование строки хеша из формата base64 в base16.
"""
ba = base64.standard_b64decode(b64_str.replace('-', '+').replace('_', '/').replace('~', '=').encode('ascii'))
ret = base64.b16encode(ba).decode('ascii')
if len(ret)!=64:
raise RuntimeError('Wrong hash length (muse be 64): ' + ret);
return ret;
def email(smtpOptions, emailAddr, text, subj, attachments=list(), bcc=list()):
# <smtpOptions> must be a dictianry.
# Mandatory keys are:
# * addr - smtp server
# Optional keys are:
# * from - sender mail address.
# default: to "Cerebro Mail Service <no-reply@cerebrohq.com>"
# * ssl - use SSL.
# default: False
# * login - user name to login on SMTP
# default: None
# * psswd - user password to login on SMTP
# default: None
# * log - log email sent
# default: False
# * skip - skip rea; mail send. Used for debug
# default: False
# * debugEmail - override mail adress
#
# emailAddr - is comma-separated list on emails
#
# attachments must be list of two component entries: [<bytearray> - atachment content, <str> - attachment name]
#
# Example use:
# email({'addr' : 'cerebrohq.com', 'ssl' : False}, 'user@example.com', 'message text', 'message subject')
mailTo = emailAddr;
if 'debugEmail' in smtpOptions:
toAddrs = [ smtpOptions['debugEmail'] ];
print('email adress overriden {0} -> {1}'.format(emailAddr, smtpOptions['debugEmail']));
else:
to_list = mailTo.split(',')
for a in to_list:
try:
bcc.remove(a);
except: pass;
toAddrs = to_list + bcc;
if 'from' in smtpOptions:
mailFrom = smtpOptions['from'];
else:
mailFrom = 'Cerebro Mail Service <no-reply@cerebrohq.com>';
msg = MIMEMultipart();
msg['Subject'] = Header(subj, 'utf-8');
msg['From'] = mailFrom;
msg['To'] = mailTo;
if str('<html>') in text:
part2 = MIMEText(text.encode('utf8'), 'html', 'utf8')
else:
part2 = MIMEText(text.encode('utf8'), 'plain', 'utf8')
msg.attach(part2)
for attach in attachments:
attachment = MIMEBase("application", "octet-stream")
attachment.add_header('Content-Disposition', 'attachment', filename = attach[1])
attachment.add_header('Content-Transfer-Encoding', 'base64')
attachment.set_payload(str(base64.encodebytes(attach[0]),'ascii'))
msg.attach(attachment)
try:
if not(('skip' in smtpOptions) and smtpOptions['skip']):
smtp = smtplib.SMTP(smtpOptions['addr'])
smtp.sendmail(mailFrom, toAddrs, msg.as_string())
if ('log' in smtpOptions) and smtpOptions['log']:
print(time.strftime('%Y-%m-%d %H:%M:%S') + ': email sent to: ', emailAddr);
except Exception as ex:
if ('log' in smtpOptions) and smtpOptions['log']:
msgErr = time.strftime('%Y-%m-%d %H:%M:%S') + ': email FAILED <' + emailAddr + '> : ' + str(ex) + "\n";
sys.stderr.write(msgErr);
sys.stderr.flush();
raise;
def isRunTime(category, timeIntervalSec):
tmp = tempfile.gettempdir();
fn = tmp + '/cerebro.IsTime.' + category + '.signal';
if not os.path.exists(fn):
open(fn, 'w').write(time.ctime());
return True;
ft = os.path.getmtime(fn);
gt = time.time();
if ft + timeIntervalSec <= gt:
open(fn, 'w').write(time.ctime());
return True;
return False;
[документация]def has_flag(flags, flag):
"""
:param int flags: значение флагов.
:param int flag: флаг.
:returns: наличие выставленного флага flag в значении flags.
:rtype: bool
Проверяет, выставлен ли флаг flag в передаваемых флагах flags.
"""
return ((flags & (1 << flag))!=0)