add file list func

Author: ehanson8

MIT-LCS-TR-834.pdf

@@ -12,6 +12,7 @@ requests = "*"
 structlog = "*"
 attr = "*"
 click = "*"
+lxml = "*"
 
 [requires]
 python_version = "3.7"

Pipfile

@@ -1,10 +1,12 @@
 import datetime
 from functools import partial
+import os
 import operator
 import requests
 import time
 
 import attr
+from lxml import html
 import structlog
 
 op = operator.attrgetter('name')
@@ -123,6 +125,28 @@ class MetadataEntry(BaseRecord):
     language = Field()
 
 
+def build_file_list_local(directory, file_extension):
+    """Build list of files in local directory."""
+    file_list = {}
+    for root, dirs, files in os.walk(directory, topdown=True):
+        for file in files:
+            if file.endswith(file_extension):
+                full_file_path = os.path.join(root, file)
+                file_list[file] = full_file_path
+    return file_list
+
+
+def build_file_list_remote(directory_url, file_extension):
+    """Build list of files in local directory."""
+    file_list = {}
+    response = requests.get(directory_url)
+    links = html.fromstring(response.content).iterlinks()
+    for link in links:
+        if link[2].endswith(file_extension):
+            file_list[link[2]] = f'{directory_url}{link[2]}'
+    return file_list
+
+
 def elapsed_time(start_time, label):
     """Calculate elapsed time."""
     td = datetime.timedelta(seconds=time.time() - start_time)

Pipfile.lock

@@ -0,0 +1,128 @@
+import datetime
+import json
+import requests
+
+import structlog
+
+logger = structlog.get_logger()
+
+user_full_name = 'Test Name'
+email = 'test@test.com'
+
+base_url = 'http://publications.csail.mit.edu/lcs/pubs/pdf/'
+id = 'MIT-LCS-TR-834'
+
+
+def download_remote_file(base_url, id):
+    """Download file from remote location."""
+    url = f'{base_url}{id}.??????????????????????????'
+    logger.info(url)
+    response = requests.get(url)
+    if response.status_code != 200:
+        logger.info('File not found')
+    else:
+        return response.content
+
+
+metadata_json = [{'file_identifier': '123',
+                 'metadata': [{'key': 'dc.title', 'value': 'Test title',
+                               'language': 'en_US'}]}
+                 ]
+
+
+def post_items_to_coll(self, base_url, coll_id, metadata_json, file_list):
+    """Post items to specified collection."""
+    collectionMetadata = json.load(open(metadata_json))
+    for item_metadata in collectionMetadata:
+        file_identifier = item_metadata['file_identifier']
+        file_exists = ''
+        data = json.dumps(item_metadata['metadata'])
+        for k in file_list:
+            if file_identifier in k:
+                file_exists = True
+        if file_exists is True:
+            logger.info(file_identifier)
+            post = requests.post(f'{base_url}{coll_id}/items',
+                                 headers=self.header, cookies=self.cookies,
+                                 data=data).json()
+            logger.info(json.dumps(post))
+            item_metadata['item_url'] = post['link']
+            yield item_metadata
+
+
+def post_bitstreams_to_item(self, item_metadata, file_list):
+    """Post bitstreams to specified item."""
+    file_identifier = item_metadata['file_identifier']
+    item_url = item_metadata['item_url']
+    for k, v in file_list.items():
+        if k.startswith(file_identifier):
+            bitstream = file_list[k]
+            file_name = bitstream[bitstream.rfind('/') + 1:]
+            data = open(bitstream, 'rb')
+            post = requests.post(f'{item_url}/bitstreams?name='
+                                 + file_name, headers=self.header_file_upload,
+                                 cookies=self.cookies, data=data).json()
+            logger.info(json.dumps(post))
+            yield post
+
+
+def create_prov_note(self, item_url, prov_note_prefix):
+    """Create provenance note"""
+    prov_note = {}
+    prov_note['key'] = 'dc.description.provenance'
+    prov_note['language'] = 'en_US'
+    utcTime = datetime.datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%SZ')
+    prov_note_value = f'{prov_note_prefix} {utcTime} '
+    bitstreams = requests.get(f'{item_url}/bitstreams', headers=self.header,
+                              cookies=self.cookies).json()
+    prov_note_value += f'(GMT). No. of bitstreams: {len(bitstreams)}'
+    for bitstream in bitstreams:
+        prov_note_value = retrieve_bitstream_metadata(bitstream,
+                                                      prov_note_value)
+    prov_note['value'] = prov_note_value
+    return prov_note
+
+
+def retrieve_bitstream_metadata(bitstream, prov_note_value):
+    """Retrieve count of bitstreams attached to items."""
+    fileName = bitstream['name']
+    size = bitstream['sizeBytes']
+    checksum = bitstream['checkSum']['value']
+    algorithm = bitstream['checkSum']['checkSumAlgorithm']
+    prov_note_value += f' {fileName}: {size} bytes, checkSum: {checksum}'
+    prov_note_value += f' ({algorithm})'
+    return prov_note_value
+
+
+def add_prov_note(self, item_url, avail_prov_note, subm_prov_note):
+    """Add provenance note to item metadata."""
+    provNote = json.dumps([avail_prov_note, subm_prov_note])
+    post = requests.put(f'{item_url}/metadata', headers=self.header,
+                        cookies=self.cookies, data=provNote)
+    logger.info(post)
+
+
+# content = build_file_list_remote(directory_url, file_extension)
+# print(content)
+# file = download_remote_file(base_url, id)
+# if type(file) is not None:
+#     with open(f'{id}.pdf', 'wb') as f:
+#         f.write(file)
+#         logger.info(f'{f.name} downloaded')
+
+
+# directory_url = 'http://publications.csail.mit.edu/lcs/pubs/pdf/'
+# file_extension = 'pdf'
+# comm_handle = '1721.1/121959'
+# coll_name = 'Test Collection'
+# file_list = build_file_list_remote(directory_url, file_extension)
+# coll_id = post_coll_to_comm(base_url, comm_handle, coll_name)
+items = post_items_to_coll(base_url, coll_id, metadata_json, file_list)
+for item in items:
+    item = post_bitstreams_to_item(item, file_list)
+    avail_prov_note = create_prov_note(item['item_url'],
+                                       'Made available in DSpace on')
+    subm_prov_note = create_prov_note(item['item_url'],
+                                      f'Submitted by {user_full_name} '
+                                      + f'({email}) on')
+    post = add_prov_note(item['item_url'], avail_prov_note, subm_prov_note)

Untitled.html

@@ -0,0 +1 @@
+<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 3.2 Final//EN"><html><head><title>Index of /pdf</title></head><body><h1>Index of /pdf</h1><table><tr><th>Name</th><th>Last modified</th><th>Size</th></tr><tr><td><a href="999.pdf">999.pdf</a></td><td>2001-02-16 11:59 </td><td>107K</td></tr></table></body></html>

dsaps/models.py

@@ -1,3 +1,5 @@
+import os
+
 import attr
 import pytest
 import requests_mock
@@ -60,6 +62,7 @@ def test_filtered_item_search(client):
 
 
 def test__pop_inst(client):
+    """Test _pop_inst function."""
     class_type = models.Collection
     rec_obj = {'name': 'Test title', 'type': 'collection', 'items': []}
     rec_obj = client._pop_inst(class_type, rec_obj)
@@ -68,7 +71,36 @@ def test__pop_inst(client):
 
 
 def test__build_uuid_list(client):
+    """Test _build_uuid_list function."""
     rec_obj = {'items': [{'uuid': '1234'}]}
     children = 'items'
     child_list = client._build_uuid_list(rec_obj, children)
     assert '1234' in child_list
+
+
+def test_build_file_list_local():
+    """Test filtered_item_search function."""
+    file_extension = 'pdf'
+    directory = 'test_temp'
+    os.mkdir(directory)
+    open(f'{directory}/999.pdf', 'w')
+    file_list = models.build_file_list_local(directory, file_extension)
+    os.remove(f'{directory}/999.pdf')
+    os.rmdir(directory)
+    assert '999.pdf' in file_list
+
+
+def test_build_file_list_remote():
+    """Test build_file_list_remote function."""
+    content = '<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 3.2 Final//EN"><html>'
+    content += '<head><title>Index of /pdf</title></head><body><h1>Index of /'
+    content += 'pdf</h1><table><tr><th>Name</th><th>Last modified</th><th>'
+    content += 'Size</th></tr><tr><td><a href="999.pdf">999.pdf</a></td><td>'
+    content += '2001-02-16 11:59 </td><td>107K</td></tr></table></body></html>'
+    with requests_mock.Mocker() as m:
+        directory_url = 'mock://test.com/pdfs/'
+        file_extension = 'pdf'
+        m.get(directory_url, text=content)
+        file_list = models.build_file_list_remote(directory_url,
+                                                  file_extension)
+        assert '999.pdf' in file_list