Upload file

cli/libreant_db.py

@@ -1,6 +1,8 @@
 import click
 import logging
 import json
+import os
+import mimetypes
 
 from archivant import Archivant
 from archivant.exceptions import NotFoundException
@@ -16,7 +18,7 @@
 
 @click.group(name="libreant-db", help="command line program to manage libreant database")
 @click.version_option()
-@click.option('-s', '--settings', type=click.Path(exists=True, readable=True), help='file from wich load settings')
+@click.option('-s', '--settings', type=click.Path(exists=True, readable=True), help='file from which load settings')
 @click.option('-d', '--debug', is_flag=True, help=get_help('DEBUG'))
 @click.option('--fsdb-path', type=click.Path(), metavar="<path>", help=get_help('FSDB_PATH'))
 @click.option('--es-indexname', type=click.STRING, metavar="<name>", help=get_help('ES_INDEXNAME'))
@@ -40,7 +42,7 @@ def libreant_db(debug, settings, fsdb_path, es_indexname, es_hosts):
     try:
         global arc
         arc = Archivant(conf=conf)
-    except Exception, e:
+    except Exception as e:
         if conf.get('DEBUG', False):
             raise
         else:
@@ -95,5 +97,95 @@ def export_all(pretty):
     click.echo(json.dumps(volumes, indent=indent))
 
 
+@libreant_db.command(name='attach', help='adds an attachment to an existing volume')
+@click.argument('volumeid')
+@click.option('-f', 'filepath', type=click.Path(exists=True,resolve_path=True), multiple=True, help='the path of the attachment')
+@click.option('-t', '--notes', type=click.STRING, metavar='<string>', multiple=True, help='notes about the attachment')
+def append_file(volumeid, filepath, notes):
+    attachments = attach_list(filepath, notes)
+    try:
+        arc.insert_attachments(volumeid,attachments)
+    except:
+        click.secho('An upload error occurred in updating an attachment!',fg='yellow', err=True)
+        exit(4)
+
+
+@libreant_db.command(name='insert-volume')
+@click.option('-l', '--language', type=click.STRING, required=True,
+              help='specify the language of the volume')
+@click.option('-f', '--filepath',
+              type=click.Path(exists=True,resolve_path=True),
+              multiple=True, help='path to the attachment to be uploaded')
+@click.option('-t', '--notes', type=click.STRING, multiple=True,
+              help='notes about the attachment '
+              '(ie: "complete version" or "poor quality"')
+@click.argument('metadata', type=click.File('r'), required=False)
+def insert_volume(language, filepath, notes, metadata):
+    '''
+    Add a new volume to libreant.
+
+    The metadata of the volume are taken from a json file whose path must be
+    passed as argument. Passing "-" as argument will read the file from stdin.
+    language is an exception, because it must be set using --language
+
+    For every attachment you must add a --file AND a --notes.
+
+    \b
+    Examples:
+        Adds a volume with no metadata. Yes, it makes no sense but you can
+          libreant-db insert-volume -l en - <<<'{}'
+        Adds a volume with no files attached
+          libreant-db insert-volume -l en - <<EOF
+          {
+            "title": "How to create volumes",
+            "actors": ["libreant devs", "open access conspiration"]
+          }
+          EOF
+        Adds a volume with one attachment but no metadata
+          libreant-db insert-volume -l en -f /path/book.epub --notes 'poor quality'
+        Adds a volume with two attachments but no metadata
+          libreant-db insert-volume -l en -f /path/book.epub --notes 'poor quality' -f /path/someother.epub --notes 'preprint'
+
+    '''
+    meta = {"_language":language}
+    if metadata:
+        meta.update(json.load(metadata))
+    attachments = attach_list(filepath, notes)
+    try:
+        out = arc.insert_volume(meta,attachments)
+    except:
+        click.secho('An upload error have occurred!', fg="yellow", err=True)
+        exit(4)
+    click.echo(out)
+
+
+def attach_list(filepaths, notes):
+    '''
+    all the arguments are lists
+    returns a list of dictionaries; each dictionary "represent" an attachment
+    '''
+    assert type(filepaths) in (list, tuple)
+    assert type(notes) in (list, tuple)
+
+    # this if clause means "if those lists are not of the same length"
+    if len(filepaths) != len(notes):
+        raise click.ClickException('The number of --filepath, and --notes must be the same')
+
+    attach_list = []
+    for fname, note in zip(filepaths, notes):
+        name = os.path.basename(fname)
+        assert os.path.exists(fname)
+        mime = mimetypes.guess_type(fname)[0]
+        if mime is not None and '/' not in mime:
+            mime = None
+        attach_list.append({
+            'file': fname,
+            'name': name,
+            'mime': mime,
+            'note': note
+        })
+    return attach_list
+
+
 if __name__ == '__main__':
     libreant_db()

cli/tests/test_libreant_db.py

@@ -0,0 +1,145 @@
+import os.path
+from tempfile import mkdtemp
+from unittest import TestCase
+from shutil import rmtree
+import json
+
+from nose.tools import eq_, raises
+import click
+import click.testing
+from elasticsearch import Elasticsearch
+
+from cli.libreant_db import attach_list, libreant_db
+
+
+class TestAttachList(TestCase):
+    def setUp(self):
+        self.tmpdir = mkdtemp(prefix=self.__class__.__name__)
+
+    def tearDown(self):
+        rmtree(self.tmpdir)
+
+    def generate(self, fname):
+        '''helper function: create a temporary file, and return its path'''
+        fpath = os.path.join(self.tmpdir, fname)
+        open(fpath, 'w').close()
+        return fpath
+
+    def test_empty(self):
+        eq_(len(attach_list([], [])), 0)
+
+    @raises(click.ClickException)
+    def test_length_no_notes(self):
+        attach_list([self.generate('foo')], [])
+
+    @raises(click.ClickException)
+    def test_length_too_many_notes(self):
+        attach_list([], ['mynote'])
+
+    def test_no_mime(self):
+        attachments = attach_list([self.generate('foo')], ['mynote'])
+        eq_(len(attachments), 1)
+        eq_(attachments[0]['mime'], None)
+
+    def test_pdf(self):
+        attachments = attach_list([self.generate('a.pdf')], ['mynote'])
+        eq_(len(attachments), 1)
+        eq_(attachments[0]['mime'], 'application/pdf')
+
+
+class TestDedicatedEs(TestCase):
+    def setUp(self):
+        self.fsdbPath = mkdtemp(prefix=self.__class__.__name__ +
+                            '_fsdb_')
+        self.cli = click.testing.CliRunner(env={
+            'LIBREANT_FSDB_PATH': self.fsdbPath,
+            'LIBREANT_ES_INDEXNAME': 'test-book'
+        })
+
+    def tearDown(self):
+        es = Elasticsearch()
+        if es.indices.exists('test-book'):
+            es.indices.delete('test-book')
+        rmtree(self.fsdbPath)
+
+
+class TestSearch(TestDedicatedEs):
+    def test_search_doesnotexist(self):
+        res = self.cli.invoke(libreant_db, ('search', 'notexisting'))
+        assert res.exit_code != 0
+
+
+class TestInsert(TestDedicatedEs):
+    def setUp(self):
+        super(TestInsert, self).setUp()
+        self.tmpdir = mkdtemp(prefix=self.__class__.__name__)
+
+    def tearDown(self):
+        super(TestInsert, self).tearDown()
+        rmtree(self.tmpdir)
+
+    def generate(self, fname, content=None):
+        '''helper function: create a temporary file, and return its path'''
+        fpath = os.path.join(self.tmpdir, fname)
+        buf = open(fpath, 'w')
+        if content is not None:
+            json.dump(content, buf)
+        buf.close()
+        return fpath
+
+    def test_no_metadata(self):
+        res = self.cli.invoke(libreant_db, ('insert-volume', '-l', 'en'))
+        assert res.exit_code == 0
+        vid = [line for line in res.output.split('\n')
+               if line.strip()][-1].strip()
+        export_res = self.cli.invoke(libreant_db, ('export-volume', vid))
+        eq_(export_res.exit_code, 0)
+        volume_data = json.loads(export_res.output)
+        eq_(volume_data['metadata']['_language'], 'en')
+        eq_(len(volume_data['metadata'].keys()), 1)
+
+    def test_no_language(self):
+        '''--language is required'''
+        res = self.cli.invoke(libreant_db, ('insert-volume',
+                                            self.generate('meta.json')))
+        assert res.exit_code != 0
+        assert '--language' in res.output
+
+    def test_empty(self):
+        '''adding empty book'''
+        meta = self.generate('m.json', {})
+        res = self.cli.invoke(libreant_db, ('insert-volume', '-l', 'en',
+                                            meta))
+        eq_(res.exit_code, 0)
+
+    def test_real_metadata(self):
+        meta = self.generate('m.json', dict(
+            title='Some title',
+            actors=['me', 'myself']
+        ))
+        res = self.cli.invoke(libreant_db, ('insert-volume', '-l', 'en',
+                                            meta))
+        eq_(res.exit_code, 0)
+        vid = [line for line in res.output.split('\n')
+               if line.strip()][-1].strip()
+        export_res = self.cli.invoke(libreant_db, ('export-volume', vid))
+        eq_(export_res.exit_code, 0)
+        volume_data = json.loads(export_res.output)
+        eq_(volume_data['metadata']['title'], 'Some title')
+
+    def test_attach(self):
+        meta = self.generate('m.json', {})
+        res = self.cli.invoke(libreant_db, ('insert-volume', '-l', 'en',
+                                            meta))
+        eq_(res.exit_code, 0)
+        vid = [line for line in res.output.split('\n')
+               if line.strip()][-1].strip()
+        attach_res = self.cli.invoke(libreant_db,
+                                     ('attach', '-f', self.generate('empty'),
+                                      '--notes', 'somenote', vid))
+        eq_(attach_res.exit_code, 0)
+        export_res = self.cli.invoke(libreant_db, ('export-volume', vid))
+        eq_(export_res.exit_code, 0)
+        volume_data = json.loads(export_res.output)
+        eq_(len(volume_data['attachments']), 1)
+        eq_(volume_data['attachments'][0]['metadata']['name'], 'empty')