# Copyright 2010-2011 Florent Le Coz <louiz@louiz.org> # # This file is part of Poezio. # # Poezio is free software: you can redistribute it and/or modify # it under the terms of the zlib license. See the COPYING file. """ Various useful functions. """ from sys import version_info from datetime import datetime, timedelta from sleekxmpp import JID, InvalidJID import base64 import os import mimetypes import hashlib import subprocess import time import string import poezio_shlex as shlex # Needed to avoid datetime.datetime.timestamp() # on python < 3.3. Older versions do not get good dst detection. OLD_PYTHON = (version_info.major + version_info.minor/10) < 3.3 ROOM_STATE_NONE = 11 ROOM_STATE_CURRENT = 10 ROOM_STATE_PRIVATE = 15 ROOM_STATE_MESSAGE = 12 ROOM_STATE_HL = 13 def get_base64_from_file(path): """ Convert the content of a file to base64 :param str path: The path of the file to convert. :return: A tuple of (encoded data, mime type, sha1 hash) if the file exists and does not exceeds the upper size limit of 16384. :return: (None, None, error message) if it fails :rtype: :py:class:`tuple` """ if not os.path.isfile(path): return (None, None, "File does not exist") size = os.path.getsize(path) if size > 16384: return (None, None,"File is too big") fdes = open(path, 'rb') data = fdes.read() encoded = base64.encodestring(data) sha1 = hashlib.sha1(data).hexdigest() mime_type = mimetypes.guess_type(path)[0] return (encoded, mime_type, sha1) def get_output_of_command(command): """ Runs a command and returns its output. :param str command: The command to run. :return: The output or None :rtype: :py:class:`str` """ try: return subprocess.check_output(command.split()).decode('utf-8').split('\n') except subprocess.CalledProcessError: return None def is_in_path(command, return_abs_path=False): """ Check if *command* is in the $PATH or not. :param str command: The command to be checked. :param bool return_abs_path: Return the absolute path of the command instead of True if the command is found. :return: True if the command is found, the command path if the command is found and *return_abs_path* is True, otherwise False. """ for directory in os.getenv('PATH').split(os.pathsep): try: if command in os.listdir(directory): if return_abs_path: return os.path.join(directory, command) else: return True except OSError: # If the user has non directories in his path pass return False DISTRO_INFO = { 'Arch Linux': '/etc/arch-release', 'Aurox Linux': '/etc/aurox-release', 'Conectiva Linux': '/etc/conectiva-release', 'CRUX': '/usr/bin/crux', 'Debian GNU/Linux': '/etc/debian_version', 'Fedora Linux': '/etc/fedora-release', 'Gentoo Linux': '/etc/gentoo-release', 'Linux from Scratch': '/etc/lfs-release', 'Mandrake Linux': '/etc/mandrake-release', 'Slackware Linux': '/etc/slackware-version', 'Solaris/Sparc': '/etc/release', 'Source Mage': '/etc/sourcemage_version', 'SUSE Linux': '/etc/SuSE-release', 'Sun JDS': '/etc/sun-release', 'PLD Linux': '/etc/pld-release', 'Yellow Dog Linux': '/etc/yellowdog-release', # many distros use the /etc/redhat-release for compatibility # so Redhat is the last 'Redhat Linux': '/etc/redhat-release' } def get_os_info(): """ Returns a detailed and well formated string containing informations about the operating system :rtype: str """ if os.name == 'posix': executable = 'lsb_release' params = ' --description --codename --release --short' full_path_to_executable = is_in_path(executable, return_abs_path = True) if full_path_to_executable: command = executable + params process = subprocess.Popen([command], shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE, close_fds=True) process.wait() output = process.stdout.readline().decode('utf-8').strip() # some distros put n/a in places, so remove those output = output.replace('n/a', '').replace('N/A', '') return output # lsb_release executable not available, so parse files for distro_name in DISTRO_INFO: path_to_file = DISTRO_INFO[distro_name] if os.path.exists(path_to_file): if os.access(path_to_file, os.X_OK): # the file is executable (f.e. CRUX) # yes, then run it and get the first line of output. text = get_output_of_command(path_to_file)[0] else: fdes = open(path_to_file, encoding='utf-8') text = fdes.readline().strip() # get only first line fdes.close() if path_to_file.endswith('version'): # sourcemage_version and slackware-version files # have all the info we need (name and version of distro) if not os.path.basename(path_to_file).startswith( 'sourcemage') or not\ os.path.basename(path_to_file).startswith('slackware'): text = distro_name + ' ' + text elif path_to_file.endswith('aurox-release') or \ path_to_file.endswith('arch-release'): # file doesn't have version text = distro_name elif path_to_file.endswith('lfs-release'): # file just has version text = distro_name + ' ' + text os_info = text.replace('\n', '') return os_info # our last chance, ask uname and strip it uname_output = get_output_of_command('uname -sr') if uname_output is not None: os_info = uname_output[0] # only first line return os_info os_info = 'N/A' return os_info def datetime_tuple(timestamp): """ Convert a timestamp using strptime and the format: %Y%m%dT%H:%M:%S. Because various datetime formats are used, the following exceptions are handled: * Optional milliseconds appened to the string are removed * Optional Z (that means UTC) appened to the string are removed * XEP-082 datetime strings have all '-' chars removed to meet the above format. :param str timestamp: The string containing the formatted date. :return: The date. :rtype: :py:class:`datetime.datetime` >>> time.timezone = 0; time.altzone = 0 >>> datetime_tuple('20130226T06:23:12') datetime.datetime(2013, 2, 26, 6, 23, 12) >>> datetime_tuple('2013-02-26T06:23:12+02:00') datetime.datetime(2013, 2, 26, 4, 23, 12) >>> time.timezone = -3600; time.altzone = -3600 >>> datetime_tuple('20130226T07:23:12') datetime.datetime(2013, 2, 26, 8, 23, 12) >>> datetime_tuple('2013-02-26T07:23:12+02:00') datetime.datetime(2013, 2, 26, 6, 23, 12) """ timestamp = timestamp.replace('-', '', 2).replace(':', '') date = timestamp[:15] tz_msg = timestamp[15:] try: ret = datetime.strptime(date, '%Y%m%dT%H%M%S') except Exception: ret = datetime.now() # add the message timezone if any try: if tz_msg and tz_msg != 'Z': tz_mod = -1 if tz_msg[0] == '-' else 1 tz_msg = time.strptime(tz_msg[1:], '%H%M') tz_msg = tz_msg.tm_hour * 3600 + tz_msg.tm_min * 60 tz_msg = timedelta(seconds=tz_mod * tz_msg) ret -= tz_msg except Exception: pass # ignore if we got a badly-formatted offset # convert UTC to local time, with DST etc. if time.daylight and time.localtime().tm_isdst: tz = timedelta(seconds=-time.altzone) else: tz = timedelta(seconds=-time.timezone) ret += tz return ret def get_utc_time(local_time=None): """ Get the current time in UTC :param datetime local_time: The current local time :return: The current UTC time >>> delta = timedelta(seconds=-3600) >>> d = datetime.now() >>> time.timezone = -3600; time.altzone = -3600 >>> get_utc_time(local_time=d) == d + delta True """ if local_time is None: local_time = datetime.now() isdst = time.localtime().tm_isdst else: if OLD_PYTHON: isdst = time.localtime(int(local_time.strftime("%s"))).tm_isdst else: isdst = time.localtime(int(local_time.timestamp())).tm_isdst if time.daylight and isdst: tz = timedelta(seconds=time.altzone) else: tz = timedelta(seconds=time.timezone) utc_time = local_time + tz return utc_time def get_local_time(utc_time): """ Get the local time from an UTC time >>> delta = timedelta(seconds=-3600) >>> d = datetime.now() >>> time.timezone = -3600; time.altzone = -3600 >>> get_local_time(d) == d - delta True """ if OLD_PYTHON: isdst = time.localtime(int(utc_time.strftime("%s"))).tm_isdst else: isdst = time.localtime(int(utc_time.timestamp())).tm_isdst if time.daylight and isdst: tz = timedelta(seconds=time.altzone) else: tz = timedelta(seconds=time.timezone) local_time = utc_time - tz return local_time def find_delayed_tag(message): """ Check if a message is delayed or not. :param sleekxmpp.Message message: The message to check. :return: A tuple containing (True, the datetime) or (False, None) :rtype: :py:class:`tuple` """ delay_tag = message.find('{urn:xmpp:delay}delay') if delay_tag is not None: delayed = True date = datetime_tuple(delay_tag.attrib['stamp']) else: # We support the OLD and deprecated XEP: http://xmpp.org/extensions/xep-0091.html # But it sucks, please, Jabber servers, don't do this :( delay_tag = message.find('{jabber:x:delay}x') if delay_tag is not None: delayed = True date = datetime_tuple(delay_tag.attrib['stamp']) else: delayed = False date = None return (delayed, date) def shell_split(st): """ Split a string correctly according to the quotes around the elements. :param str st: The string to split. :return: A list of the different of the string. :rtype: :py:class:`list` >>> shell_split('"sdf 1" "toto 2"') ['sdf 1', 'toto 2'] >>> shell_split('toto "titi"') ['toto', 'titi'] >>> shell_split('toto ""') ['toto', ''] >>> shell_split('to"to titi "a" b') ['to"to', 'titi', 'a', 'b'] >>> shell_split('"toto titi" toto ""') ['toto titi', 'toto', ''] >>> shell_split('toto "titi') ['toto', 'titi'] """ sh = shlex.shlex(st) ret = [] w = sh.get_token() while w and w[2] is not None: ret.append(w[2]) if w[1] == len(st): return ret w = sh.get_token() return ret def find_argument(pos, text, quoted=True): """ Split an input into a list of arguments, return the number of the argument selected by pos. If the position searched is outside the string, or in a space between words, then it will return the position of an hypothetical new argument. See the doctests of the two methods for example behaviors. :param int pos: The position to search. :param str text: The text to analyze. :param quoted: Whether to take quotes into account or not. :rtype: int """ if quoted: return find_argument_quoted(pos, text) else: return find_argument_unquoted(pos, text) def find_argument_quoted(pos, text): """ >>> find_argument_quoted(4, 'toto titi tata') 3 >>> find_argument_quoted(4, '"toto titi" tata') 0 >>> find_argument_quoted(8, '"toto" "titi tata"') 1 >>> find_argument_quoted(8, '"toto" "titi tata') 1 >>> find_argument_quoted(3, '"toto" "titi tata') 0 >>> find_argument_quoted(18, '"toto" "titi tata" ') 2 """ sh = shlex.shlex(text) count = -1 w = sh.get_token() while w and w[2] is not None: count += 1 if w[0] <= pos < w[1]: return count w = sh.get_token() return count + 1 def find_argument_unquoted(pos, text): """ >>> find_argument_unquoted(2, 'toto titi tata') 0 >>> find_argument_unquoted(3, 'toto titi tata') 0 >>> find_argument_unquoted(6, 'toto titi tata') 1 >>> find_argument_unquoted(4, 'toto titi tata') 3 >>> find_argument_unquoted(25, 'toto titi tata') 3 """ ret = text.split() search = 0 argnum = 0 for i, elem in enumerate(ret): elem_start = text.find(elem, search) elem_end = elem_start + len(elem) search = elem_end if elem_start <= pos < elem_end: return i argnum = i return argnum + 1 def parse_str_to_secs(duration=''): """ Parse a string of with a number of d, h, m, s. :param str duration: The formatted string. :return: The number of seconds represented by the string :rtype: :py:class:`int` >>> parse_str_to_secs("1d3m1h") 90180 """ values = {'s': 1, 'm': 60, 'h': 3600, 'd': 86400} result = 0 tmp = '0' for char in duration: if char in string.digits: tmp += char elif char in values: tmp_i = int(tmp) result += tmp_i * values[char] tmp = '0' else: return 0 if tmp != '0': result += int(tmp) return result def parse_secs_to_str(duration=0): """ Do the reverse operation of :py:func:`parse_str_to_secs`. Parse a number of seconds to a human-readable string. The string has the form XdXhXmXs. 0 units are removed. :param int duration: The duration, in seconds. :return: A formatted string containing the duration. :rtype: :py:class:`str` >>> parse_secs_to_str(3601) '1h1s' """ secs, mins, hours, days = 0, 0, 0, 0 result = '' secs = duration % 60 mins = (duration % 3600) // 60 hours = (duration % 86400) // 3600 days = duration // 86400 result += '%sd' % days if days else '' result += '%sh' % hours if hours else '' result += '%sm' % mins if mins else '' result += '%ss' % secs if secs else '' if not result: result = '0s' return result def format_tune_string(infos): """ Contruct a string from a dict created from an "User tune" event. :param dict infos: The informations :return: The formatted string :rtype: :py:class:`str` """ elems = [] track = infos.get('track') if track: elems.append(track) title = infos.get('title') if title: elems.append(title) else: elems.append('Unknown title') elems.append('-') artist = infos.get('artist') if artist: elems.append(artist) else: elems.append('Unknown artist') rating = infos.get('rating') if rating: elems.append('[ ' + rating + '/10' + ' ]') length = infos.get('length') if length: length = int(length) secs = length % 60 mins = length // 60 secs = str(secs).zfill(2) mins = str(mins).zfill(2) elems.append('[' + mins + ':' + secs + ']') return ' '.join(elems) def format_gaming_string(infos): """ Construct a string from a dict containing the "user gaming" informations. (for now, only use address and name) :param dict infos: The informations :returns: The formatted string :rtype: :py:class:`str` """ name = infos.get('name') if not name: return '' server_address = infos.get('server_address') if server_address: return '%s on %s' % (name, server_address) return name def safeJID(*args, **kwargs): """ Construct a :py:class:`sleekxmpp.JID` object from a string. Used to avoid tracebacks during is stringprep fails (fall back to a JID with an empty string). """ try: return JID(*args, **kwargs) except InvalidJID: return JID('') if __name__ == "__main__": import doctest doctest.testmod()