openmedialibrary/oml/utils.py

225 lines
6.3 KiB
Python
Raw Normal View History

2014-05-04 17:26:43 +00:00
# -*- coding: utf-8 -*-
# vi:si:et:sw=4:sts=4:ts=4
2014-09-02 22:32:44 +00:00
2014-05-04 17:26:43 +00:00
2014-05-18 23:24:04 +00:00
import os
import sys
2014-09-01 10:38:14 +00:00
from PIL import Image
2014-09-02 22:32:44 +00:00
from io import StringIO, BytesIO
2014-05-04 17:26:43 +00:00
import re
import stdnum.isbn
2014-05-17 22:18:32 +00:00
import socket
2014-09-02 22:32:44 +00:00
import io
2014-05-18 23:24:04 +00:00
import gzip
2014-05-21 00:02:21 +00:00
import time
from datetime import datetime
import subprocess
2014-05-04 17:26:43 +00:00
import ox
2014-05-17 22:18:32 +00:00
import ed25519
2014-05-04 17:26:43 +00:00
2014-05-16 08:06:11 +00:00
from meta.utils import normalize_isbn, find_isbns
2014-05-14 09:57:11 +00:00
2014-05-18 23:24:04 +00:00
import logging
logger = logging.getLogger('oml.utils')
2014-05-17 22:18:32 +00:00
ENCODING='base64'
2014-05-24 10:50:27 +00:00
def cleanup_id(key, value):
if key == 'isbn':
value = normalize_isbn(value)
if key in ('lccn', 'olid', 'oclc'):
value = ''.join([v for v in value if v!='-'])
return value
2014-05-04 17:26:43 +00:00
def valid_olid(id):
return id.startswith('OL') and id.endswith('M')
def get_positions(ids, pos):
'''
>>> get_positions([1,2,3,4], [2,4])
{2: 1, 4: 3}
'''
positions = {}
for i in pos:
try:
positions[i] = ids.index(i)
except:
pass
return positions
def get_by_key(objects, key, value):
2014-09-02 22:32:44 +00:00
obj = [o for o in objects if o.get(key) == value]
2014-05-04 17:26:43 +00:00
return obj and obj[0] or None
def get_by_id(objects, id):
return get_by_key(objects, 'id', id)
def resize_image(data, width=None, size=None):
2014-09-02 22:32:44 +00:00
if isinstance(data, bytes):
data = BytesIO(data)
else:
data = StringIO(data)
source = Image.open(data)
2014-05-19 21:15:37 +00:00
if source.mode == 'P':
source = source.convert('RGB')
2014-05-04 17:26:43 +00:00
source_width = source.size[0]
source_height = source.size[1]
if size:
if source_width > source_height:
width = size
height = int(width / (float(source_width) / source_height))
height = height - height % 2
else:
height = size
width = int(height * (float(source_width) / source_height))
width = width - width % 2
else:
height = int(width / (float(source_width) / source_height))
height = height - height % 2
width = max(width, 1)
height = max(height, 1)
if width < source_width:
resize_method = Image.ANTIALIAS
else:
resize_method = Image.BICUBIC
output = source.resize((width, height), resize_method)
2014-09-02 22:32:44 +00:00
o = BytesIO()
2014-05-04 17:26:43 +00:00
output.save(o, format='jpeg')
data = o.getvalue()
o.close()
return data
def sort_title(title):
2014-09-02 22:32:44 +00:00
title = title.replace('Æ', 'Ae')
2014-05-04 17:26:43 +00:00
if isinstance(title, str):
2014-09-02 22:32:44 +00:00
title = str(title)
2014-05-04 17:26:43 +00:00
title = ox.sort_string(title)
#title
2014-09-02 22:32:44 +00:00
title = re.sub('[\'!¿¡,\.;\-"\:\*\[\]]', '', title)
2014-05-04 17:26:43 +00:00
return title.strip()
def get_position_by_id(list, key):
for i in range(0, len(list)):
if list[i]['id'] == key:
return i
return -1
2014-05-17 22:18:32 +00:00
def valid(key, value, sig):
'''
validate that value was signed by key
'''
vk = ed25519.VerifyingKey(str(key), encoding=ENCODING)
try:
vk.verify(str(sig), str(value), encoding=ENCODING)
#except ed25519.BadSignatureError:
except:
return False
return True
def get_public_ipv6():
2014-05-19 18:12:02 +00:00
try:
host = ('2a01:4f8:120:3201::3', 25519)
s = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
s.settimeout(1)
s.connect(host)
ip = s.getsockname()[0]
s.close()
except:
ip = None
2014-05-17 22:18:32 +00:00
return ip
def get_interface():
interface = ''
2014-08-25 17:21:34 +00:00
if sys.platform == 'darwin' or sys.platform.startswith('freebsd'):
#cmd = ['/usr/sbin/netstat', '-rn']
cmd = ['/sbin/route', '-n', 'get', 'default']
2014-08-22 16:49:11 +00:00
p = subprocess.Popen(cmd, stdout=subprocess.PIPE, close_fds=True)
stdout, stderr = p.communicate()
interface = [[p.strip() for p in s.split(':', 1)] for s in stdout.strip().split('\n') if 'interface' in s]
if interface:
interface = '%%%s' % interface[0][1]
else:
interface = ''
return interface
def get_local_ipv4():
2014-05-23 20:10:02 +00:00
ip = None
2014-08-25 17:21:34 +00:00
if sys.platform == 'darwin' or sys.platform.startswith('freebsd'):
cmd = ['/sbin/route', '-n', 'get', 'default']
2014-08-22 16:49:11 +00:00
p = subprocess.Popen(cmd, stdout=subprocess.PIPE, close_fds=True)
stdout, stderr = p.communicate()
2014-09-02 22:32:44 +00:00
stdout = stdout.decode('utf-8')
interface = [[p.strip() for p in s.split(':', 1)]
for s in stdout.strip().split('\n') if 'interface' in s]
if interface:
interface = interface[0][1]
cmd = ['ifconfig', interface]
2014-08-22 16:49:11 +00:00
p = subprocess.Popen(cmd, stdout=subprocess.PIPE, close_fds=True)
2014-05-22 14:40:18 +00:00
stdout, stderr = p.communicate()
2014-09-02 22:32:44 +00:00
stdout = stdout.decode('utf-8')
ips = [l for l in stdout.split('\n') if 'inet ' in l]
if ips:
ip = ips[0].strip().split(' ')[1]
else:
cmd = ['ip', 'route', 'show']
2014-08-22 16:49:11 +00:00
p = subprocess.Popen(cmd, stdout=subprocess.PIPE, close_fds=True)
stdout, stderr = p.communicate()
2014-09-02 22:32:44 +00:00
stdout = stdout.decode('utf-8')
local = [l for l in stdout.split('\n') if 'default' in l]
if local:
dev = local[0].split(' ')[4]
local_ip = [l for l in stdout.split('\n')
2014-05-25 12:40:57 +00:00
if dev in l and not 'default' in l and 'src' in l]
ip = [p for p in local_ip[0].split(' ')[1:] if '.' in p][0]
return ip
2014-05-18 23:24:04 +00:00
def update_dict(root, data):
for key in data:
2014-09-02 22:32:44 +00:00
keys = [part.replace('\0', '\\.') for part in key.replace('\\.', '\0').split('.')]
2014-05-18 23:24:04 +00:00
value = data[key]
p = root
while len(keys)>1:
key = keys.pop(0)
if isinstance(p, list):
p = p[get_position_by_id(p, key)]
else:
if key not in p:
p[key] = {}
p = p[key]
if value == None and keys[0] in p:
del p[keys[0]]
else:
p[keys[0]] = value
def remove_empty_folders(prefix):
empty = []
for root, folders, files in os.walk(prefix):
if not folders and not files:
empty.append(root)
for folder in empty:
remove_empty_tree(folder)
def remove_empty_tree(leaf):
while leaf:
if not os.path.exists(leaf):
leaf = os.path.dirname(leaf)
elif os.path.isdir(leaf) and not os.listdir(leaf):
logger.debug('rmdir %s', leaf)
os.rmdir(leaf)
else:
break
2014-05-21 00:02:21 +00:00
2014-09-02 23:09:42 +00:00
utc_0 = int(time.mktime(datetime(1970, 1, 1).timetuple()))
2014-05-21 00:02:21 +00:00
def datetime2ts(dt):
return int(time.mktime(dt.utctimetuple())) - utc_0
def ts2datetime(ts):
return datetime.utcfromtimestamp(float(ts))