mirror of
https://github.com/Screenly/Anthias.git
synced 2026-03-05 07:21:06 -05:00
102 lines
2.7 KiB
Python
102 lines
2.7 KiB
Python
import requests
|
|
import json
|
|
import re
|
|
from netifaces import ifaddresses
|
|
from sh import grep, netstat
|
|
from urlparse import urlparse
|
|
from datetime import timedelta
|
|
from settings import settings
|
|
|
|
# This will only work on the Raspberry Pi,
|
|
# so let's wrap it in a try/except so that
|
|
# Travis can run.
|
|
try:
|
|
from sh import omxplayer
|
|
except:
|
|
pass
|
|
|
|
|
|
def validate_url(string):
|
|
"""Simple URL verification.
|
|
|
|
>>> validate_url("hello")
|
|
False
|
|
>>> validate_url("ftp://example.com")
|
|
False
|
|
>>> validate_url("http://")
|
|
False
|
|
>>> validate_url("http://wireload.net/logo.png")
|
|
True
|
|
>>> validate_url("https://wireload.net/logo.png")
|
|
True
|
|
|
|
"""
|
|
|
|
checker = urlparse(string)
|
|
return bool(checker.scheme in ('http', 'https') and checker.netloc)
|
|
|
|
|
|
def get_node_ip():
|
|
"""Returns the node's IP, for the interface
|
|
that is being used as the default gateway.
|
|
This shuld work on both MacOS X and Linux."""
|
|
|
|
try:
|
|
default_interface = grep(netstat('-nr'), '-e', '^default', '-e' '^0.0.0.0').split()[-1]
|
|
my_ip = ifaddresses(default_interface)[2][0]['addr']
|
|
return my_ip
|
|
except:
|
|
pass
|
|
|
|
return None
|
|
|
|
|
|
def get_video_duration(file):
|
|
"""
|
|
Returns the duration of a video file in timedelta.
|
|
"""
|
|
time = None
|
|
try:
|
|
run_omxplayer = omxplayer(file, info=True, _err_to_out=True)
|
|
for line in run_omxplayer.split('\n'):
|
|
if 'Duration' in line:
|
|
match = re.search(r'[0-9]+:[0-9]+:[0-9]+\.[0-9]+', line)
|
|
if match:
|
|
time_input = match.group()
|
|
time_split = time_input.split(':')
|
|
hours = int(time_split[0])
|
|
minutes = int(time_split[1])
|
|
seconds = float(time_split[2])
|
|
time = timedelta(hours=hours, minutes=minutes, seconds=seconds)
|
|
break
|
|
except:
|
|
pass
|
|
|
|
return time
|
|
|
|
|
|
def handler(obj):
|
|
if hasattr(obj, 'isoformat'):
|
|
return obj.isoformat()
|
|
else:
|
|
raise TypeError('Object of type %s with value of %s is not JSON serializable' % (type(obj), repr(obj)))
|
|
|
|
|
|
def json_dump(obj):
|
|
return json.dumps(obj, default=handler)
|
|
|
|
|
|
def url_fails(url):
|
|
"""
|
|
Accept 200 and 405 as 'OK' statuses for URLs.
|
|
Some hosting providers (like Google App Engine) throws a 405 at `requests`.
|
|
"""
|
|
try:
|
|
if validate_url(url):
|
|
obj = requests.head(url, allow_redirects=True, timeout=10, verify=settings['verify_ssl'])
|
|
assert obj.status_code in (200, 405)
|
|
except (requests.ConnectionError, requests.exceptions.Timeout, AssertionError):
|
|
return True
|
|
else:
|
|
return False
|