1
0
mirror of https://github.com/spl0k/supysonic.git synced 2024-12-22 08:56:17 +00:00

Initial commit

This commit is contained in:
Alban 2012-10-13 11:29:48 +02:00
commit 6375266fa8
12 changed files with 471 additions and 0 deletions

2
.gitignore vendored Executable file
View File

@ -0,0 +1,2 @@
*.pyc
*.swp

189
api/__init__.py Executable file
View File

@ -0,0 +1,189 @@
# coding: utf-8
from flask import request
import simplejson
import cgi
from web import app
from db import User
import hashlib
@app.before_request
def set_formatter():
if not request.path.startswith('/rest/'):
return
"""Return a function to create the response."""
(f, callback) = map(request.args.get, ['f', 'callback'])
if f == 'jsonp':
if not callback:
# TODO
# MiniSub has a bug, trying to retrieve jsonp without
# callback in case of getCoverArt.view
# it's not a problem because the getCoverArt should
# return a byte stream
return ResponseHelper.responsize_json({
'error': {
'code': 0,
'message': 'Missing callback'
}
}, error = True), 400
request.formatter = lambda x, **kwargs: ResponseHelper.responsize_jsonp(x, callback, kwargs)
elif f == "json":
request.formatter = ResponseHelper.responsize_json
else:
request.formatter = ResponseHelper.responsize_xml
@app.before_request
def authorize():
if not request.path.startswith('/rest/'):
return
error = request.formatter({
'error': {
'code': 40,
'message': 'Unauthorized'
}
}, error = True), 401
(username, decoded_pass) = map(request.args.get, [ 'u', 'p' ])
if not username or not decoded_pass:
return error
user = User.query.filter(User.name == username).first()
if not user:
return error
if decoded_pass.startswith('enc:'):
decoded_pass = hexdecode(decoded_pass[4:])
crypt = hashlib.sha1(user.salt + decoded_pass).hexdigest()
if crypt != user.password:
return error
@app.after_request
def set_content_type(response):
if not request.path.startswith('/rest/'):
return response
f = request.args.get('f')
response.headers['content-type'] = 'application/json' if f in [ 'jsonp', 'json' ] else 'text/xml'
return response
@app.errorhandler(404)
def not_found(error):
if not request.path.startswith('/rest/'):
return error
return request.formatter({
'error': {
'code': 0,
'message': 'Not implemented'
}
}, error = True), 501
class ResponseHelper:
@staticmethod
def responsize_json(ret, error = False, version = "1.8.0"):
# add headers to response
ret.update({
'status': 'failed' if error else 'ok',
'version': version,
'xmlns': "http://subsonic.org/restapi"
})
return simplejson.dumps({ 'subsonic-response': ret }, indent = True, encoding = 'utf-8')
@staticmethod
def responsize_jsonp(ret, callback, error = False, version = "1.8.0"):
return "%s(%s)" % (callback, ResponseHelper.responsize_json(ret, error, version))
@staticmethod
def responsize_xml(ret, error = False, version = "1.8.0"):
"""Return an xml response from json and replace unsupported characters."""
ret.update({
'status': 'failed' if error else 'ok',
'version': version,
'xmlns': "http://subsonic.org/restapi"
})
return '<?xml version="1.0" encoding="UTF-8" ?>' + ResponseHelper.jsonp2xml({'subsonic-response': ret}).replace("&", "\\&amp;")
@staticmethod
def jsonp2xml(json, indent = 0):
"""Convert a json structure to xml. The game is trivial. Nesting uses the [] parenthesis.
ex. { 'musicFolder': {'id': 1234, 'name': "sss" } }
ex. { 'musicFolder': [{'id': 1234, 'name': "sss" }, {'id': 456, 'name': "aaa" }]}
ex. { 'musicFolders': {'musicFolder' : [{'id': 1234, 'name': "sss" }, {'id': 456, 'name': "aaa" }] } }
ex. { 'index': [{'name': 'A', 'artist': [{'id': '517674445', 'name': 'Antonello Venditti'}] }] }
ex. {"subsonic-response": { "musicFolders": {"musicFolder": [{ "id": 0,"name": "Music"}]},
"status": "ok","version": "1.7.0","xmlns": "http://subsonic.org/restapi"}}
"""
ret = '\n' + '\t' * indent
content = None
for c in [str, int, unicode]:
if isinstance(json, c):
return str(json)
if not isinstance(json, dict):
raise Exception("class type: %s" % json)
# every tag is a dict.
# its value can be a string, a list or a dict
for tag in json.keys():
tag_list = json[tag]
# if tag_list is a list, then it represent a list of elements
# ex. {index: [{ 'a':'1'} , {'a':'2'} ] }
# --> <index a="1" /> <index b="2" />
if isinstance(tag_list, list):
for t in tag_list:
# for every element, get the attributes
# and embed them in the tag named
attributes = ""
content = ""
for (attr, value) in t.iteritems():
# only serializable values are attributes
if value.__class__.__name__ in 'str':
attributes = '%s %s="%s"' % (
attributes,
attr,
cgi.escape(value, quote=None)
)
elif value.__class__.__name__ in ['int', 'unicode', 'bool', 'long']:
attributes = '%s %s="%s"' % (
attributes, attr, value)
# other values are content
elif isinstance(value, dict):
content += ResponseHelper.jsonp2xml(value, indent + 1)
elif isinstance(value, list):
content += ResponseHelper.jsonp2xml({attr: value}, indent + 1)
if content:
ret += "<%s%s>%s\n%s</%s>" % (
tag, attributes, content, '\t' * indent, tag)
else:
ret += "<%s%s />" % (tag, attributes)
if isinstance(tag_list, dict):
attributes = ""
content = ""
for (attr, value) in tag_list.iteritems():
# only string values are attributes
if not isinstance(value, dict) and not isinstance(value, list):
attributes = '%s %s="%s"' % (
attributes, attr, value)
else:
content += ResponseHelper.jsonp2xml({attr: value}, indent + 1)
if content:
ret += "<%s%s>%s\n%s</%s>" % (tag, attributes, content, '\t' * indent, tag)
else:
ret += "<%s%s />" % (tag, attributes)
return ret.replace('"True"', '"true"').replace('"False"', '"false"')
def hexdecode(enc):
ret = ''
while enc:
ret = ret + chr(int(enc[:2], 16))
enc = enc[2:]
return ret

17
api/browse.py Executable file
View File

@ -0,0 +1,17 @@
# coding: utf-8
from flask import request
from web import app
from db import MusicFolder
@app.route('/rest/getMusicFolders.view')
def list_folders():
return request.formatter({
'musicFolders': {
'musicFolder': [ {
'id': str(f.id),
'name': f.name
} for f in MusicFolder.query.order_by(MusicFolder.name).all() ]
}
})

13
api/system.py Executable file
View File

@ -0,0 +1,13 @@
# coding: utf-8
from flask import request
from web import app
@app.route('/rest/ping.view')
def ping():
return request.formatter({})
@app.route('/rest/getLicense.view')
def license():
return request.formatter({ 'license': { 'valid': False } })

29
config.py Executable file
View File

@ -0,0 +1,29 @@
# coding: utf-8
import os
def try_load_config(path):
if os.path.exists(path):
app.config.from_pyfile(path)
return True
return False
def check():
path = os.path.join(os.path.expanduser('~'), '.supysonic')
if os.path.exists(path):
return path
path = '/etc/supysonic'
if os.path.exists(path):
return path
return False
config_path = check()
config_dict = {}
if config_path:
with open(config_path) as f:
for line in f:
spl = line.split('=')
config_dict[spl[0].strip()] = eval(spl[1])
def get(name):
return config_dict.get(name)

64
db.py Executable file
View File

@ -0,0 +1,64 @@
# coding: utf-8
import config
from sqlalchemy import create_engine, Column, Integer, String, Boolean
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import types
from sqlalchemy import BINARY
from sqlalchemy.schema import Column
import uuid
class UUID(types.TypeDecorator):
impl = BINARY
def __init__(self):
self.impl.length = 16
types.TypeDecorator.__init__(self, length = self.impl.length)
def process_bind_param(self, value, dialect = None):
if value and isinstance(value, uuid.UUID):
return value.bytes
if value and not isinstance(value, uuid.UUID):
raise ValueError, 'value %s is not a valid uuid.UUID' % value
return None
def process_result_value(self, value, dialect = None):
if value:
return uuid.UUID(bytes = value)
return None
def is_mutable(self):
return False
@staticmethod
def gen_id_column():
return Column(UUID, primary_key = True, default = uuid.uuid4)
engine = create_engine(config.get('DATABASE_URI'), convert_unicode = True)
db_session = scoped_session(sessionmaker(autocommit = False, autoflush = False, bind = engine))
Base = declarative_base()
Base.query = db_session.query_property()
class User(Base):
__tablename__ = 'users'
id = UUID.gen_id_column()
name = Column(String, unique = True)
mail = Column(String)
password = Column(String(40))
salt = Column(String(6))
admin = Column(Boolean)
class MusicFolder(Base):
__tablename__ = 'folders'
id = UUID.gen_id_column()
name = Column(String, unique = True)
path = Column(String)
def init_db():
Base.metadata.drop_all(bind = engine)
Base.metadata.create_all(bind = engine)

13
main.py Executable file
View File

@ -0,0 +1,13 @@
# coding: utf-8
from web import app
import db, config
if __name__ == '__main__':
if not config.check():
print >>sys.stderr, "Couldn't find configuration file"
sys.exit(1)
db.init_db()
app.run(debug = True)

8
templates/addfolder.html Executable file
View File

@ -0,0 +1,8 @@
{% extends "layout.html" %}
{% block body %}
<form method="post">
<label for="name">Name</label><input type="text" id="name" name="name" value="{{ request.form.name }}" /><br />
<label for="path">Path</label><input type="text" id="path" name="path" value="{{ request.form.path }}" /><br />
<input type="submit" />
</form>
{% endblock %}

10
templates/adduser.html Executable file
View File

@ -0,0 +1,10 @@
{% extends "layout.html" %}
{% block body %}
<form method="post">
<label for="name">Name</label><input type="text" id="name" name="name" value="{{ request.form.name }}" /><br />
<label for="passwd">Password</label><input type="password" id="passwd" name="passwd" /><br />
<label for="passwd_confirm">Confirm</label><input type="password" id="passwd_confirm" name="passwd_confirm" /><br />
<label for="mail">EMail</label><input type="text" id="mail" name="mail" value="{{ request.form.mail }}" /><br />
<input type="submit" />
</form>
{% endblock %}

20
templates/home.html Executable file
View File

@ -0,0 +1,20 @@
{% extends "layout.html" %}
{% block body %}
<h2>Users</h2>
<table>
<tr><th>Name</th><th>EMail</th></tr>
{% for user in users %}
<tr><td>{{ user.name }}</td><td>{{ user.mail }}</td></tr>
{% endfor %}
</table>
<a href="{{ url_for('add_user') }}">Add</a>
<h2>Music folders</h2>
<table>
<tr><th>Name</th><th>Path</th></tr>
{% for folder in folders %}
<tr><td>{{ folder.name }}</td><td>{{ folder.path }}</td></tr>
{% endfor %}
</table>
<a href="{{ url_for('add_folder') }}">Add</a>
{% endblock %}

12
templates/layout.html Executable file
View File

@ -0,0 +1,12 @@
<!DOCTYPE html>
<title>Supysonic</title>
<div class="page">
<h1>Supysonic</h1>
<div class="flash">
{% for message in get_flashed_messages() %}
<p>{{ message }}</p>
{% endfor %}
</div>
{% block body %}{% endblock %}
</div>

94
web.py Executable file
View File

@ -0,0 +1,94 @@
# coding: utf-8
from flask import Flask, request, flash, render_template, redirect, url_for
import string, random, hashlib
import os.path
app = Flask(__name__)
app.secret_key = '?9huDM\\H'
from db import db_session
from db import User, MusicFolder
@app.teardown_request
def teardown(exception):
db_session.remove()
@app.route('/')
def index():
"""
if User.query.count() == 0:
flash('Not configured. Please create the first admin user')
return redirect(url_for('add_user'))
"""
return render_template('home.html', users = User.query.all(), folders = MusicFolder.query.all())
@app.route('/adduser', methods = [ 'GET', 'POST' ])
def add_user():
if request.method == 'GET':
return render_template('adduser.html')
error = False
(name, passwd, passwd_confirm, mail) = map(request.form.get, [ 'name', 'passwd', 'passwd_confirm', 'mail' ])
if name in (None, ''):
flash('The name is required.')
error = True
elif User.query.filter(User.name == name).first():
flash('There is already a user with that name. Please pick another one.')
error = True
if passwd in (None, ''):
flash('Please provide a password.')
error = True
elif passwd != passwd_confirm:
flash("The passwords don't match.")
error = True
if error:
return render_template('adduser.html')
salt = ''.join(random.choice(string.printable.strip()) for i in xrange(6))
crypt = hashlib.sha1(salt + passwd).hexdigest()
user = User(name = name, mail = mail, password = crypt, salt = salt)
db_session.add(user)
db_session.commit()
flash("User '%s' successfully added" % name)
return redirect(url_for('index'))
@app.route('/addfolder', methods = [ 'GET', 'POST' ])
def add_folder():
if request.method == 'GET':
return render_template('addfolder.html')
error = False
(name, path) = map(request.form.get, [ 'name', 'path' ])
if name in (None, ''):
flash('The name is required.')
error = True
elif MusicFolder.query.filter(MusicFolder.name == name).first():
flash('There is already a folder with that name. Please pick another one.')
error = True
if path in (None, ''):
flash('The path is required.')
error = True
else:
path = os.path.abspath(path)
if not os.path.isdir(path):
flash("The path '%s' doesn't exists or isn't a directory" % path)
error = True
folder = MusicFolder.query.filter(MusicFolder.name == name).first()
if folder:
flash("This path is already registered with the name '%s'" % folder.name)
error = True
if error:
return render_template('addfolder.html')
folder = MusicFolder(name = name, path = path)
db_session.add(folder)
db_session.commit()
flash("Folder '%s' created. You should now run a scan" % name)
return redirect(url_for('index'))
import api.system
import api.browse