Miscelleanous tools used by OpenERP.
"""
-import os, time, sys
import inspect
-
-from config import config
-
-import zipfile
-import release
-import socket
+import subprocess
+import logging
+import os
import re
-from itertools import islice
+import smtplib
+import socket
+import sys
import threading
+import time
+import warnings
+import zipfile
+from datetime import datetime
+from email.MIMEText import MIMEText
+from email.MIMEBase import MIMEBase
+from email.MIMEMultipart import MIMEMultipart
+from email.Header import Header
+from email.Utils import formatdate, COMMASPACE
+from email import Encoders
+from itertools import islice, izip
+from lxml import etree
from which import which
-
-
if sys.version_info[:2] < (2, 4):
from threadinglocal import local
else:
from threading import local
+try:
+ from html2text import html2text
+except ImportError:
+ html2text = None
-from itertools import izip
+import netsvc
+from config import config
+from lru import LRU
+
+_logger = logging.getLogger('tools')
+
+# List of etree._Element subclasses that we choose to ignore when parsing XML.
+# We include the *Base ones just in case, currently they seem to be subclasses of the _* ones.
+SKIPPED_ELEMENT_TYPES = (etree._Comment, etree._ProcessingInstruction, etree.CommentBase, etree.PIBase)
# initialize a database with base/base.sql
def init_db(cr):
import addons
f = addons.get_module_resource('base', 'base.sql')
- for line in file_open(f).read().split(';'):
- if (len(line)>0) and (not line.isspace()):
- cr.execute(line)
- cr.commit()
+ base_sql_file = file_open(f)
+ try:
+ cr.execute(base_sql_file.read())
+ cr.commit()
+ finally:
+ base_sql_file.close()
for i in addons.get_modules():
mod_path = addons.get_module_path(i)
p_id = None
while categs:
if p_id is not None:
- cr.execute('select id \
- from ir_module_category \
- where name=%s and parent_id=%s', (categs[0], p_id))
+ cr.execute('SELECT id \
+ FROM ir_module_category \
+ WHERE name=%s AND parent_id=%s', (categs[0], p_id))
else:
- cr.execute('select id \
- from ir_module_category \
- where name=%s and parent_id is NULL', (categs[0],))
+ cr.execute('SELECT id \
+ FROM ir_module_category \
+ WHERE name=%s AND parent_id IS NULL', (categs[0],))
c_id = cr.fetchone()
if not c_id:
- cr.execute('select nextval(\'ir_module_category_id_seq\')')
+ cr.execute('INSERT INTO ir_module_category \
+ (name, parent_id) \
+ VALUES (%s, %s) RETURNING id', (categs[0], p_id))
c_id = cr.fetchone()[0]
- cr.execute('insert into ir_module_category \
- (id, name, parent_id) \
- values (%s, %s, %s)', (c_id, categs[0], p_id))
else:
c_id = c_id[0]
p_id = c_id
state = 'uninstalled'
else:
state = 'uninstallable'
- cr.execute('select nextval(\'ir_module_module_id_seq\')')
- id = cr.fetchone()[0]
- cr.execute('insert into ir_module_module \
- (id, author, website, name, shortdesc, description, \
- category_id, state, certificate) \
- values (%s, %s, %s, %s, %s, %s, %s, %s, %s)', (
- id, info.get('author', ''),
+ cr.execute('INSERT INTO ir_module_module \
+ (author, website, name, shortdesc, description, \
+ category_id, state, certificate, web, license) \
+ VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) RETURNING id', (
+ info.get('author', ''),
info.get('website', ''), i, info.get('name', False),
- info.get('description', ''), p_id, state, info.get('certificate') or None))
- cr.execute('insert into ir_model_data \
- (name,model,module, res_id, noupdate) values (%s,%s,%s,%s,%s)', (
+ info.get('description', ''), p_id, state, info.get('certificate') or None,
+ info.get('web') or False,
+ info.get('license') or 'AGPL-3'))
+ id = cr.fetchone()[0]
+ cr.execute('INSERT INTO ir_model_data \
+ (name,model,module, res_id, noupdate) VALUES (%s,%s,%s,%s,%s)', (
'module_meta_information', 'ir.module.module', i, id, True))
dependencies = info.get('depends', [])
for d in dependencies:
- cr.execute('insert into ir_module_module_dependency \
- (module_id,name) values (%s, %s)', (id, d))
+ cr.execute('INSERT INTO ir_module_module_dependency \
+ (module_id,name) VALUES (%s, %s)', (id, d))
cr.commit()
def find_in_path(name):
prog = find_pg_tool(name)
if not prog:
raise Exception('Couldn\'t find %s' % name)
- args2 = (os.path.basename(prog),) + args
- return os.spawnv(os.P_WAIT, prog, args2)
+ args2 = (prog,) + args
+
+ return subprocess.call(args2)
def exec_pg_command_pipe(name, *args):
prog = find_pg_tool(name)
if not prog:
raise Exception('Couldn\'t find %s' % name)
- if os.name == "nt":
- cmd = '"' + prog + '" ' + ' '.join(args)
- else:
- cmd = prog + ' ' + ' '.join(args)
- return os.popen2(cmd, 'b')
+ # on win32, passing close_fds=True is not compatible
+ # with redirecting std[in/err/out]
+ pop = subprocess.Popen((prog,) + args, bufsize= -1,
+ stdin=subprocess.PIPE, stdout=subprocess.PIPE,
+ close_fds=(os.name=="posix"))
+ return (pop.stdin, pop.stdout)
def exec_command_pipe(name, *args):
prog = find_in_path(name)
if not prog:
raise Exception('Couldn\'t find %s' % name)
- if os.name == "nt":
- cmd = '"'+prog+'" '+' '.join(args)
- else:
- cmd = prog+' '+' '.join(args)
- return os.popen2(cmd, 'b')
+ # on win32, passing close_fds=True is not compatible
+ # with redirecting std[in/err/out]
+ pop = subprocess.Popen((prog,) + args, bufsize= -1,
+ stdin=subprocess.PIPE, stdout=subprocess.PIPE,
+ close_fds=(os.name=="posix"))
+ return (pop.stdin, pop.stdout)
#----------------------------------------------------------
# File paths
subdir2 = (subdir2 != 'addons' or None) and subdir2
for adp in adps:
- try:
- if subdir2:
- fn = os.path.join(adp, subdir2, name)
- else:
- fn = os.path.join(adp, name)
- fn = os.path.normpath(fn)
- fo = file_open(fn, mode=mode, subdir=None, pathinfo=pathinfo)
- if pathinfo:
- return fo, fn
- return fo
- except IOError, e:
- pass
+ try:
+ if subdir2:
+ fn = os.path.join(adp, subdir2, name)
+ else:
+ fn = os.path.join(adp, name)
+ fn = os.path.normpath(fn)
+ fo = file_open(fn, mode=mode, subdir=None, pathinfo=pathinfo)
+ if pathinfo:
+ return fo, fn
+ return fo
+ except IOError:
+ pass
if subdir:
name = os.path.join(rtp, subdir, name)
if pathinfo:
return fo, name
return fo
- except:
+ except Exception:
name2 = os.path.normpath(os.path.join(head + '.zip', zipname))
pass
for i in (name2, name):
return fo
if os.path.splitext(name)[1] == '.rml':
raise IOError, 'Report %s doesn\'t exist or deleted : ' %str(name)
- raise IOError, 'File not found : '+str(name)
+ raise IOError, 'File not found : %s' % name
#----------------------------------------------------------
html = ustr(html)
+ from lxml.etree import tostring
try:
- from BeautifulSoup import BeautifulSoup, SoupStrainer, Comment
- except:
- return html
+ from lxml.html.soupparser import fromstring
+ kwargs = {}
+ except ImportError:
+ _logger.debug('tools.misc.html2plaintext: cannot use BeautifulSoup, fallback to lxml.etree.HTMLParser')
+ from lxml.etree import fromstring, HTMLParser
+ kwargs = dict(parser=HTMLParser())
+
+ tree = fromstring(html, **kwargs)
- urls = []
if body_id is not None:
- strainer = SoupStrainer(id=body_id)
+ source = tree.xpath('//*[@id=%s]'%(body_id,))
else:
- strainer = SoupStrainer('body')
-
- soup = BeautifulSoup(html, parseOnlyThese=strainer, fromEncoding=encoding)
- for link in soup.findAll('a'):
- title = link.renderContents()
- for url in [x[1] for x in link.attrs if x[0]=='href']:
- urls.append(dict(url=ustr(url), tag=ustr(link), title=ustr(title)))
-
- html = ustr(soup.__str__())
+ source = tree.xpath('//body')
+ if len(source):
+ tree = source[0]
url_index = []
i = 0
- for d in urls:
- if d['title'] == d['url'] or 'http://'+d['title'] == d['url']:
- html = html.replace(d['tag'], d['url'])
- else:
+ for link in tree.findall('.//a'):
+ url = link.get('href')
+ if url:
i += 1
- html = html.replace(d['tag'], '%s [%s]' % (d['title'], i))
- url_index.append(d['url'])
+ link.tag = 'span'
+ link.text = '%s [%s]' % (link.text, i)
+ url_index.append(url)
+
+ html = ustr(tostring(tree, encoding=encoding))
html = html.replace('<strong>','*').replace('</strong>','*')
html = html.replace('<b>','*').replace('</b>','*')
html = html.replace('<h2>','**').replace('</h2>','**')
html = html.replace('<h1>','**').replace('</h1>','**')
html = html.replace('<em>','/').replace('</em>','/')
-
-
- # the only line breaks we respect is those of ending tags and
- # breaks
-
- html = html.replace('\n',' ')
- html = html.replace('<br>', '\n')
html = html.replace('<tr>', '\n')
- html = html.replace('</p>', '\n\n')
- html = re.sub('<br\s*/>', '\n', html)
+ html = html.replace('</p>', '\n')
+ html = re.sub('<br\s*/?>', '\n', html)
+ html = re.sub('<.*?>', ' ', html)
html = html.replace(' ' * 2, ' ')
-
- # for all other tags we failed to clean up, just remove then and
- # complain about them on the stderr
- def desperate_fixer(g):
- #print >>sys.stderr, "failed to clean up %s" % str(g.group())
- return ' '
-
- html = re.sub('<.*?>', desperate_fixer, html)
-
- # lstrip all lines
- html = '\n'.join([x.lstrip() for x in html.splitlines()])
+ # strip all lines
+ html = '\n'.join([x.strip() for x in html.splitlines()])
+ html = html.replace('\n' * 2, '\n')
for i, url in enumerate(url_index):
if i == 0:
html += '\n\n'
html += ustr('[%s] %s\n') % (i+1, url)
+
return html
+def generate_tracking_message_id(openobject_id):
+ """Returns a string that can be used in the Message-ID RFC822 header field so we
+ can track the replies related to a given object thanks to the "In-Reply-To" or
+ "References" fields that Mail User Agents will set.
+ """
+ return "<%s-openobject-%s@%s>" % (time.time(), openobject_id, socket.gethostname())
+
+def _email_send(smtp_from, smtp_to_list, message, openobject_id=None, ssl=False, debug=False):
+ """Low-level method to send directly a Message through the configured smtp server.
+ :param smtp_from: RFC-822 envelope FROM (not displayed to recipient)
+ :param smtp_to_list: RFC-822 envelope RCPT_TOs (not displayed to recipient)
+ :param message: an email.message.Message to send
+ :param debug: True if messages should be output to stderr before being sent,
+ and smtplib.SMTP put into debug mode.
+ :return: True if the mail was delivered successfully to the smtp,
+ else False (+ exception logged)
+ """
+ class WriteToLogger(object):
+ def __init__(self):
+ self.logger = netsvc.Logger()
+
+ def write(self, s):
+ self.logger.notifyChannel('email_send', netsvc.LOG_DEBUG, s)
+
+ if openobject_id:
+ message['Message-Id'] = generate_tracking_message_id(openobject_id)
+
+ try:
+ smtp_server = config['smtp_server']
+
+ if smtp_server.startswith('maildir:/'):
+ from mailbox import Maildir
+ maildir_path = smtp_server[8:]
+ mdir = Maildir(maildir_path,factory=None, create = True)
+ mdir.add(message.as_string(True))
+ return True
+
+ oldstderr = smtplib.stderr
+ if not ssl: ssl = config.get('smtp_ssl', False)
+ s = smtplib.SMTP()
+ try:
+ # in case of debug, the messages are printed to stderr.
+ if debug:
+ smtplib.stderr = WriteToLogger()
+
+ s.set_debuglevel(int(bool(debug))) # 0 or 1
+ s.connect(smtp_server, config['smtp_port'])
+ if ssl:
+ s.ehlo()
+ s.starttls()
+ s.ehlo()
+
+ if config['smtp_user'] or config['smtp_password']:
+ s.login(config['smtp_user'], config['smtp_password'])
+
+ s.sendmail(smtp_from, smtp_to_list, message.as_string())
+ finally:
+ try:
+ s.quit()
+ if debug:
+ smtplib.stderr = oldstderr
+ except Exception:
+ # ignored, just a consequence of the previous exception
+ pass
+
+ except Exception:
+ _logger.error('could not deliver email', exc_info=True)
+ return False
+
+ return True
+
+
def email_send(email_from, email_to, subject, body, email_cc=None, email_bcc=None, reply_to=False,
attach=None, openobject_id=False, ssl=False, debug=False, subtype='plain', x_headers=None, priority='3'):
`email_to`: a sequence of addresses to send the mail to.
"""
- import smtplib
- from email.MIMEText import MIMEText
- from email.MIMEBase import MIMEBase
- from email.MIMEMultipart import MIMEMultipart
- from email.Header import Header
- from email.Utils import formatdate, COMMASPACE
- from email.Utils import formatdate, COMMASPACE
- from email import Encoders
- import netsvc
-
if x_headers is None:
x_headers = {}
- if not ssl: ssl = config.get('smtp_ssl', False)
if not (email_from or config['email_from']):
raise ValueError("Sending an email requires either providing a sender "
"address or having configured one")
if not email_from: email_from = config.get('email_from', False)
+ email_from = ustr(email_from).encode('utf-8')
if not email_cc: email_cc = []
if not email_bcc: email_bcc = []
if not body: body = u''
- try: email_body = body.encode('utf-8')
- except (UnicodeEncodeError, UnicodeDecodeError):
- email_body = body
- try:
- email_text = MIMEText(email_body.encode('utf8') or '',_subtype=subtype,_charset='utf-8')
- except:
- email_text = MIMEText(email_body or '',_subtype=subtype,_charset='utf-8')
+ email_body = ustr(body).encode('utf-8')
+ email_text = MIMEText(email_body or '',_subtype=subtype,_charset='utf-8')
- if attach: msg = MIMEMultipart()
- else: msg = email_text
+ msg = MIMEMultipart()
msg['Subject'] = Header(ustr(subject), 'utf-8')
msg['From'] = email_from
msg['To'] = COMMASPACE.join(email_to)
if email_cc:
msg['Cc'] = COMMASPACE.join(email_cc)
- if email_bcc:
- msg['Bcc'] = COMMASPACE.join(email_bcc)
msg['Date'] = formatdate(localtime=True)
- # Add OpenERP Server information
- msg['X-Generated-By'] = 'OpenERP (http://www.openerp.com)'
- msg['X-OpenERP-Server-Host'] = socket.gethostname()
- msg['X-OpenERP-Server-Version'] = release.version
msg['X-Priority'] = priorities.get(priority, '3 (Normal)')
# Add dynamic X Header
for key, value in x_headers.iteritems():
- msg['X-OpenERP-%s' % key] = str(value)
msg['%s' % key] = str(value)
- if openobject_id:
- msg['Message-Id'] = "<%s-openobject-%s@%s>" % (time.time(), openobject_id, socket.gethostname())
+ if html2text and subtype == 'html':
+ text = html2text(email_body.decode('utf-8')).encode('utf-8')
+ alternative_part = MIMEMultipart(_subtype="alternative")
+ alternative_part.attach(MIMEText(text, _charset='utf-8', _subtype='plain'))
+ alternative_part.attach(email_text)
+ msg.attach(alternative_part)
+ else:
+ msg.attach(email_text)
if attach:
- msg.attach(email_text)
for (fname,fcontent) in attach:
part = MIMEBase('application', "octet-stream")
part.set_payload( fcontent )
part.add_header('Content-Disposition', 'attachment; filename="%s"' % (fname,))
msg.attach(part)
- class WriteToLogger(object):
- def __init__(self):
- self.logger = netsvc.Logger()
-
- def write(self, s):
- self.logger.notifyChannel('email_send', netsvc.LOG_DEBUG, s)
-
- smtp_server = config['smtp_server']
- if smtp_server.startswith('maildir:/'):
- from mailbox import Maildir
- maildir_path = smtp_server[8:]
- try:
- mdir = Maildir(maildir_path,factory=None, create = True)
- mdir.add(msg.as_string(True))
- return True
- except Exception,e:
- netsvc.Logger().notifyChannel('email_send (maildir)', netsvc.LOG_ERROR, e)
- return False
-
- try:
- oldstderr = smtplib.stderr
- s = smtplib.SMTP()
- try:
- # in case of debug, the messages are printed to stderr.
- if debug:
- smtplib.stderr = WriteToLogger()
-
- s.set_debuglevel(int(bool(debug))) # 0 or 1
- s.connect(smtp_server, config['smtp_port'])
- if ssl:
- s.ehlo()
- s.starttls()
- s.ehlo()
-
- if config['smtp_user'] or config['smtp_password']:
- s.login(config['smtp_user'], config['smtp_password'])
- s.sendmail(email_from,
- flatten([email_to, email_cc, email_bcc]),
- msg.as_string()
- )
- finally:
- s.quit()
- if debug:
- smtplib.stderr = oldstderr
-
- except Exception, e:
- netsvc.Logger().notifyChannel('email_send', netsvc.LOG_ERROR, e)
- return False
-
- return True
+ return _email_send(email_from, flatten([email_to, email_cc, email_bcc]), msg, openobject_id=openobject_id, ssl=ssl, debug=debug)
#----------------------------------------------------------
# SMS
url = "http://api.urlsms.com/SendSMS.aspx"
#url = "http://196.7.150.220/http/sendmsg"
params = urllib.urlencode({'UserID': user, 'Password': password, 'SenderID': api_id, 'MsgText': text, 'RecipientMobileNo':to})
- f = urllib.urlopen(url+"?"+params)
+ urllib.urlopen(url+"?"+params)
# FIXME: Use the logger if there is an error
return True
__caches = []
- def __init__(self, timeout=None, skiparg=2, multi=None):
+ def __init__(self, timeout=None, skiparg=2, multi=None, size=8192):
assert skiparg >= 2 # at least self and cr
if timeout is None:
self.timeout = config['cache_timeout']
self.skiparg = skiparg
self.multi = multi
self.lasttime = time.time()
- self.cache = {}
+ self.cache = LRU(size) # TODO take size from config
self.fun = None
cache.__caches.append(self)
def to_xml(s):
return s.replace('&','&').replace('<','<').replace('>','>')
-def get_encodings():
- yield 'utf8'
+def get_encodings(hint_encoding='utf-8'):
+ fallbacks = {
+ 'latin1': 'latin9',
+ 'iso-8859-1': 'iso8859-15',
+ 'cp1252': '1252',
+ }
+ if hint_encoding:
+ yield hint_encoding
+ if hint_encoding.lower() in fallbacks:
+ yield fallbacks[hint_encoding.lower()]
+
+ # some defaults (also taking care of pure ASCII)
+ for charset in ['utf8','latin1']:
+ if not (hint_encoding) or (charset.lower() != hint_encoding.lower()):
+ yield charset
+
from locale import getpreferredencoding
prefenc = getpreferredencoding()
- if prefenc:
+ if prefenc and prefenc.lower() != 'utf-8':
yield prefenc
-
- prefenc = {
- 'latin1': 'latin9',
- 'iso-8859-1': 'iso8859-15',
- 'cp1252': '1252',
- }.get(prefenc.lower())
+ prefenc = fallbacks.get(prefenc.lower())
if prefenc:
yield prefenc
-def ustr(value):
+def ustr(value, hint_encoding='utf-8'):
"""This method is similar to the builtin `str` method, except
- it will return Unicode string.
+ it will return unicode() string.
@param value: the value to convert
+ @param hint_encoding: an optional encoding that was detected
+ upstream and should be tried first to
+ decode ``value``.
@rtype: unicode
@return: unicode string
"""
- orig = value
if isinstance(value, Exception):
return exception_to_unicode(value)
if isinstance(value, unicode):
return value
- try:
- return unicode(value)
- except:
- pass
+ if not isinstance(value, basestring):
+ try:
+ return unicode(value)
+ except Exception:
+ raise UnicodeError('unable to convert %r' % (value,))
- for ln in get_encodings():
+ for ln in get_encodings(hint_encoding):
try:
return unicode(value, ln)
- except:
+ except Exception:
pass
- raise UnicodeError('unable de to convert %r' % (orig,))
+ raise UnicodeError('unable to convert %r' % (value,))
def exception_to_unicode(e):
return "\n".join((ustr(a) for a in e.args))
try:
return ustr(e)
- except:
+ except Exception:
return u"Unknown message"
return lang
def get_languages():
+ # The codes below are those from Launchpad's Rosetta, with the exception
+ # of some trivial codes where the Launchpad code is xx and we have xx_XX.
languages={
- 'ab_RU': u'Abkhazian (RU)',
+ 'ab_RU': u'Abkhazian / аҧсуа',
'ar_AR': u'Arabic / الْعَرَبيّة',
- 'bg_BG': u'Bulgarian / български',
+ 'bg_BG': u'Bulgarian / български език',
'bs_BS': u'Bosnian / bosanski jezik',
'ca_ES': u'Catalan / Català',
'cs_CZ': u'Czech / Čeština',
'en_GB': u'English (UK)',
'en_US': u'English (US)',
'es_AR': u'Spanish (AR) / Español (AR)',
+ 'es_BO': u'Spanish (BO) / Español (BO)',
+ 'es_CL': u'Spanish (CL) / Español (CL)',
+ 'es_CO': u'Spanish (CO) / Español (CO)',
+ 'es_CR': u'Spanish (CR) / Español (CR)',
+ 'es_DO': u'Spanish (DO) / Español (DO)',
+ 'es_EC': u'Spanish (EC) / Español (EC)',
'es_ES': u'Spanish / Español',
+ 'es_GT': u'Spanish (GT) / Español (GT)',
+ 'es_HN': u'Spanish (HN) / Español (HN)',
+ 'es_MX': u'Spanish (MX) / Español (MX)',
+ 'es_NI': u'Spanish (NI) / Español (NI)',
+ 'es_PA': u'Spanish (PA) / Español (PA)',
+ 'es_PE': u'Spanish (PE) / Español (PE)',
+ 'es_PR': u'Spanish (PR) / Español (PR)',
+ 'es_PY': u'Spanish (PY) / Español (PY)',
+ 'es_SV': u'Spanish (SV) / Español (SV)',
+ 'es_UY': u'Spanish (UY) / Español (UY)',
+ 'es_VE': u'Spanish (VE) / Español (VE)',
'et_EE': u'Estonian / Eesti keel',
'fa_IR': u'Persian / فارس',
- 'fi_FI': u'Finland / Suomi',
+ 'fi_FI': u'Finnish / Suomi',
'fr_BE': u'French (BE) / Français (BE)',
'fr_CH': u'French (CH) / Français (CH)',
'fr_FR': u'French / Français',
'gl_ES': u'Galician / Galego',
- 'gu_IN': u'Gujarati / India',
- 'hi_IN': u'Hindi / India',
+ 'gu_IN': u'Gujarati / ગુજરાતી',
+ 'he_IL': u'Hebrew / עִבְרִי',
+ 'hi_IN': u'Hindi / हिंदी',
'hr_HR': u'Croatian / hrvatski jezik',
'hu_HU': u'Hungarian / Magyar',
'id_ID': u'Indonesian / Bahasa Indonesia',
'it_IT': u'Italian / Italiano',
- 'iu_CA': u'Inuktitut / Canada',
- 'ja_JP': u'Japanese / Japan',
- 'ko_KP': u'Korean / Korea, Democratic Peoples Republic of',
- 'ko_KR': u'Korean / Korea, Republic of',
+ 'iu_CA': u'Inuktitut / ᐃᓄᒃᑎᑐᑦ',
+ 'ja_JP': u'Japanese / 日本語',
+ 'ko_KP': u'Korean (KP) / 한국어 (KP)',
+ 'ko_KR': u'Korean (KR) / 한국어 (KR)',
'lt_LT': u'Lithuanian / Lietuvių kalba',
- 'lv_LV': u'Latvian / Latvia',
- 'ml_IN': u'Malayalam / India',
- 'mn_MN': u'Mongolian / Mongolia',
- 'nb_NO': u'Norwegian Bokmål / Norway',
+ 'lv_LV': u'Latvian / latviešu valoda',
+ 'ml_IN': u'Malayalam / മലയാളം',
+ 'mn_MN': u'Mongolian / монгол',
+ 'nb_NO': u'Norwegian Bokmål / Norsk bokmål',
'nl_NL': u'Dutch / Nederlands',
- 'nl_BE': u'Dutch (Belgium) / Nederlands (Belgïe)',
- 'oc_FR': u'Occitan (post 1500) / France',
+ 'nl_BE': u'Flemish (BE) / Vlaams (BE)',
+ 'oc_FR': u'Occitan (FR, post 1500) / Occitan',
'pl_PL': u'Polish / Język polski',
- 'pt_BR': u'Portugese (BR) / português (BR)',
- 'pt_PT': u'Portugese / português',
- 'ro_RO': u'Romanian / limba română',
+ 'pt_BR': u'Portugese (BR) / Português (BR)',
+ 'pt_PT': u'Portugese / Português',
+ 'ro_RO': u'Romanian / română',
'ru_RU': u'Russian / русский язык',
- 'si_LK': u'Sinhalese / Sri Lanka',
+ 'si_LK': u'Sinhalese / සිංහල',
'sl_SI': u'Slovenian / slovenščina',
'sk_SK': u'Slovak / Slovenský jazyk',
- 'sq_AL': u'Albanian / Shqipëri',
- 'sr_RS': u'Serbian / Serbia',
+ 'sq_AL': u'Albanian / Shqip',
+ 'sr_RS': u'Serbian (Cyrillic) / српски',
+ 'sr@latin': u'Serbian (Latin) / srpski',
'sv_SE': u'Swedish / svenska',
- 'te_IN': u'Telugu / India',
+ 'te_IN': u'Telugu / తెలుగు',
'tr_TR': u'Turkish / Türkçe',
- 'vi_VN': u'Vietnam / Cộng hòa xã hội chủ nghĩa Việt Nam',
- 'uk_UA': u'Ukrainian / украї́нська мо́ва',
- 'ur_PK': u'Urdu / Pakistan',
+ 'vi_VN': u'Vietnamese / Tiếng Việt',
+ 'uk_UA': u'Ukrainian / українська',
+ 'ur_PK': u'Urdu / اردو',
'zh_CN': u'Chinese (CN) / 简体中文',
'zh_HK': u'Chinese (HK)',
'zh_TW': u'Chinese (TW) / 正體字',
return languages
def scan_languages():
-# import glob
-# file_list = [os.path.splitext(os.path.basename(f))[0] for f in glob.glob(os.path.join(config['root_path'],'addons', 'base', 'i18n', '*.po'))]
-# ret = [(lang, lang_dict.get(lang, lang)) for lang in file_list]
# Now it will take all languages from get languages function without filter it with base module languages
lang_dict = get_languages()
ret = [(lang, lang_dict.get(lang, lang)) for lang in list(lang_dict)]
@wraps(f)
def wrapper(*args, **kwargs):
- import netsvc
from pprint import pformat
vector = ['Call -> function: %r' % f]
--log-level=debug
"""
- import netsvc
+ warnings.warn("The tools.debug() method is deprecated, please use logging.",
+ DeprecationWarning, stacklevel=2)
from inspect import stack
- import re
from pprint import pformat
st = stack()[1]
param = re.split("debug *\((.+)\)", st[4][0].strip())[1].strip()
what = pformat(what)
if param != what:
what = "%s = %s" % (param, what)
- netsvc.Logger().notifyChannel(st[3], netsvc.LOG_DEBUG, what)
+ logging.getLogger(st[3]).debug(what)
-icons = map(lambda x: (x,x), ['STOCK_ABOUT', 'STOCK_ADD', 'STOCK_APPLY', 'STOCK_BOLD',
+__icons_list = ['STOCK_ABOUT', 'STOCK_ADD', 'STOCK_APPLY', 'STOCK_BOLD',
'STOCK_CANCEL', 'STOCK_CDROM', 'STOCK_CLEAR', 'STOCK_CLOSE', 'STOCK_COLOR_PICKER',
'STOCK_CONNECT', 'STOCK_CONVERT', 'STOCK_COPY', 'STOCK_CUT', 'STOCK_DELETE',
'STOCK_DIALOG_AUTHENTICATION', 'STOCK_DIALOG_ERROR', 'STOCK_DIALOG_INFO',
'terp-gdu-smart-failing','terp-go-week','terp-gtk-select-all','terp-locked','terp-mail-forward',
'terp-mail-message-new','terp-mail-replied','terp-rating-rated','terp-stage','terp-stock_format-scientific',
'terp-dolar_ok!','terp-idea','terp-stock_format-default','terp-mail-','terp-mail_delete'
-])
+]
-def extract_zip_file(zip_file, outdirectory):
- import zipfile
- import os
+def icons(*a, **kw):
+ global __icons_list
+ return [(x, x) for x in __icons_list ]
+def extract_zip_file(zip_file, outdirectory):
zf = zipfile.ZipFile(zip_file, 'r')
out = outdirectory
for path in zf.namelist():
"""
def _detect_ip_addr():
from array import array
- import socket
from struct import pack, unpack
try:
try:
ip_addr = _detect_ip_addr()
- except:
+ except Exception:
ip_addr = 'localhost'
return ip_addr
res = str(_winreg.QueryValueEx(current_tz_key,"StandardName")[0]) # [0] is value, [1] is type code
_winreg.CloseKey(current_tz_key)
_winreg.CloseKey(hklm)
- except:
+ except Exception:
pass
return res
Defaults to UTC if no working timezone can be found.
@return: the timezone identifier as expected by pytz.timezone.
"""
- import time
- import netsvc
try:
import pytz
- except:
+ except Exception:
netsvc.Logger().notifyChannel("detect_server_timezone", netsvc.LOG_WARNING,
"Python pytz module is not available. Timezone will be set to UTC by default.")
return 'UTC'
try:
f = open("/etc/timezone")
tz_value = f.read(128).strip()
- except:
+ except Exception:
pass
finally:
f.close()
"No valid timezone could be detected, using default UTC timezone. You can specify it explicitly with option 'timezone' in the server configuration.")
return 'UTC'
+def get_server_timezone():
+ # timezone detection is safe in multithread, so lazy init is ok here
+ if (not config['timezone']):
+ config['timezone'] = detect_server_timezone()
+ return config['timezone']
+
+
+DEFAULT_SERVER_DATE_FORMAT = "%Y-%m-%d"
+DEFAULT_SERVER_TIME_FORMAT = "%H:%M:%S"
+DEFAULT_SERVER_DATETIME_FORMAT = "%s %s" % (
+ DEFAULT_SERVER_DATE_FORMAT,
+ DEFAULT_SERVER_TIME_FORMAT)
+
+# Python's strftime supports only the format directives
+# that are available on the platform's libc, so in order to
+# be cross-platform we map to the directives required by
+# the C standard (1989 version), always available on platforms
+# with a C standard implementation.
+DATETIME_FORMATS_MAP = {
+ '%C': '', # century
+ '%D': '%m/%d/%Y', # modified %y->%Y
+ '%e': '%d',
+ '%E': '', # special modifier
+ '%F': '%Y-%m-%d',
+ '%g': '%Y', # modified %y->%Y
+ '%G': '%Y',
+ '%h': '%b',
+ '%k': '%H',
+ '%l': '%I',
+ '%n': '\n',
+ '%O': '', # special modifier
+ '%P': '%p',
+ '%R': '%H:%M',
+ '%r': '%I:%M:%S %p',
+ '%s': '', #num of seconds since epoch
+ '%T': '%H:%M:%S',
+ '%t': ' ', # tab
+ '%u': ' %w',
+ '%V': '%W',
+ '%y': '%Y', # Even if %y works, it's ambiguous, so we should use %Y
+ '%+': '%Y-%m-%d %H:%M:%S',
+
+ # %Z is a special case that causes 2 problems at least:
+ # - the timezone names we use (in res_user.context_tz) come
+ # from pytz, but not all these names are recognized by
+ # strptime(), so we cannot convert in both directions
+ # when such a timezone is selected and %Z is in the format
+ # - %Z is replaced by an empty string in strftime() when
+ # there is not tzinfo in a datetime value (e.g when the user
+ # did not pick a context_tz). The resulting string does not
+ # parse back if the format requires %Z.
+ # As a consequence, we strip it completely from format strings.
+ # The user can always have a look at the context_tz in
+ # preferences to check the timezone.
+ '%z': '',
+ '%Z': '',
+}
+
+def server_to_local_timestamp(src_tstamp_str, src_format, dst_format, dst_tz_name,
+ tz_offset=True, ignore_unparsable_time=True):
+ """
+ Convert a source timestamp string into a destination timestamp string, attempting to apply the
+ correct offset if both the server and local timezone are recognized, or no
+ offset at all if they aren't or if tz_offset is false (i.e. assuming they are both in the same TZ).
+
+ WARNING: This method is here to allow formatting dates correctly for inclusion in strings where
+ the client would not be able to format/offset it correctly. DO NOT use it for returning
+ date fields directly, these are supposed to be handled by the client!!
+
+ @param src_tstamp_str: the str value containing the timestamp in the server timezone.
+ @param src_format: the format to use when parsing the server timestamp.
+ @param dst_format: the format to use when formatting the resulting timestamp for the local/client timezone.
+ @param dst_tz_name: name of the destination timezone (such as the 'tz' value of the client context)
+ @param ignore_unparsable_time: if True, return False if src_tstamp_str cannot be parsed
+ using src_format or formatted using dst_format.
+
+ @return: local/client formatted timestamp, expressed in the local/client timezone if possible
+ and if tz_offset is true, or src_tstamp_str if timezone offset could not be determined.
+ """
+ if not src_tstamp_str:
+ return False
+
+ res = src_tstamp_str
+ if src_format and dst_format:
+ # find out server timezone
+ server_tz = get_server_timezone()
+ try:
+ # dt_value needs to be a datetime.datetime object (so no time.struct_time or mx.DateTime.DateTime here!)
+ dt_value = datetime.strptime(src_tstamp_str, src_format)
+ if tz_offset and dst_tz_name:
+ try:
+ import pytz
+ src_tz = pytz.timezone(server_tz)
+ dst_tz = pytz.timezone(dst_tz_name)
+ src_dt = src_tz.localize(dt_value, is_dst=True)
+ dt_value = src_dt.astimezone(dst_tz)
+ except Exception:
+ pass
+ res = dt_value.strftime(dst_format)
+ except Exception:
+ # Normal ways to end up here are if strptime or strftime failed
+ if not ignore_unparsable_time:
+ return False
+ return res
+
def split_every(n, iterable, piece_maker=tuple):
"""Splits an iterable into length-n pieces. The last piece will be shorter
fp = urllib.urlopen('http://www.openerp.com/scripts/survey.php', args)
fp.read()
fp.close()
- except:
+ except Exception:
pass
def upload_data(email, data, type='SURVEY'):
a = upload_data_thread(email, data, type)
a.start()
return True
+
+
+# port of python 2.6's attrgetter with support for dotted notation
+def resolve_attr(obj, attr):
+ for name in attr.split("."):
+ obj = getattr(obj, name)
+ return obj
+
+def attrgetter(*items):
+ if len(items) == 1:
+ attr = items[0]
+ def g(obj):
+ return resolve_attr(obj, attr)
+ else:
+ def g(obj):
+ return tuple(resolve_attr(obj, attr) for attr in items)
+ return g
+
+
# vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4:
projects / odoo / odoo.git / blobdiff