diff --git a/modules/weko-search-ui/weko_search_ui/tasks.py b/modules/weko-search-ui/weko_search_ui/tasks.py
index 66133d4dad..fc9a941267 100644
--- a/modules/weko-search-ui/weko_search_ui/tasks.py
+++ b/modules/weko-search-ui/weko_search_ui/tasks.py
@@ -19,6 +19,10 @@
# MA 02111-1307, USA.
"""WEKO3 module docstring."""
+import gc
+import json
+import os
+import pickle
import shutil
from datetime import datetime, timedelta
@@ -27,7 +31,7 @@
from celery.task.control import inspect
from flask import current_app
from weko_admin.api import TempDirInfo
-from weko_admin.utils import get_redis_cache
+from weko_admin.utils import get_redis_cache, reset_redis_cache
from weko_redis.redis import RedisConnection
from invenio_db import db
@@ -35,6 +39,7 @@
check_import_items,
delete_exported,
export_all,
+ write_files,
get_lifetime,
import_items_to_system,
)
@@ -104,37 +109,66 @@ def delete_task_id_cache(task_id, cache_key):
datastore.delete(cache_key)
@shared_task
-def export_all_task(root_url, user_id, data, timezone):
+def export_all_task(root_url, user_id, data, start_time):
"""Export all items."""
- from weko_admin.utils import reset_redis_cache
+ export_all(root_url, user_id, data, start_time)
- _task_config = current_app.config["WEKO_SEARCH_UI_BULK_EXPORT_URI"]
- _expired_time = current_app.config["WEKO_SEARCH_UI_BULK_EXPORT_EXPIRED_TIME"]
- _cache_key = current_app.config["WEKO_ADMIN_CACHE_PREFIX"].format(
- name=_task_config,
+@shared_task
+def write_files_task(export_path, pickle_file_name , user_id):
+ """Write files for export.
+
+ Args:
+ export_path (str): path of files where csv/tsv export to.
+ pickle_file_name (str): pickle file's name
+ user_id (int): a user who processed file output.
+ """
+ _msg_config = current_app.config["WEKO_SEARCH_UI_BULK_EXPORT_MSG"]
+ _msg_key = current_app.config["WEKO_ADMIN_CACHE_PREFIX"].format(
+ name=_msg_config,
user_id=user_id
)
- _task_key_config = current_app.config["WEKO_SEARCH_UI_BULK_EXPORT_TASK"]
- _task_key = current_app.config["WEKO_ADMIN_CACHE_PREFIX"].format(
- name=_task_key_config,
+ _file_create_config = \
+ current_app.config["WEKO_SEARCH_UI_BULK_EXPORT_FILE_CREATE_RUN_MSG"]
+ _file_create_key = current_app.config["WEKO_ADMIN_CACHE_PREFIX"].format(
+ name=_file_create_config,
user_id=user_id
)
- uri = export_all(root_url, user_id, data, timezone)
- reset_redis_cache(_cache_key, uri)
- delete_exported_task.apply_async(
- args=(
- uri,
- _cache_key,
- _task_key
- ),
- countdown=int(_expired_time) * 60,
- )
+ def _update_redis_status(json_data, file_name, status,item_type_id):
+ "Update status in redis cache."
+ part_name = os.path.splitext(file_name)[1]
+ part_index = part_name.find('part')
+ part_number = part_name[part_index + 4:] if part_index != -1 else 1
+ json_data['write_file_status'][item_type_id + '.' + str(part_number)] = status
+ reset_redis_cache(_file_create_key, json.dumps(json_data))
+ del part_name, part_index, part_number
+
+ with open(pickle_file_name, 'rb') as f:
+ import_datas = pickle.load(f)
+ json_data = json.loads(get_redis_cache(_file_create_key))
+ if not json_data['cancel_flg']:
+ _update_redis_status(json_data, import_datas['name'], 'started',import_datas['item_type_id'])
+ with open(pickle_file_name, 'rb') as f:
+ import_datas = pickle.load(f)
+ result = write_files(import_datas, export_path, user_id, 0)
+ json_data = json.loads(get_redis_cache(_file_create_key))
+ if result:
+ _update_redis_status(json_data, import_datas['name'], 'finished',import_datas['item_type_id'])
+ else:
+ reset_redis_cache(_msg_key, "Export failed.")
+ json_data['cancel_flg'] = True
+ _update_redis_status(json_data, import_datas['name'], 'error',import_datas['item_type_id'])
+ else:
+ _update_redis_status(json_data, import_datas['name'], 'canceled',import_datas['item_type_id'])
+ del import_datas,json_data
+ gc.collect()
+ os.remove(pickle_file_name)
@shared_task
-def delete_exported_task(uri, cache_key, task_key):
+def delete_exported_task(uri, cache_key, task_key, export_path):
"""Delete expired exported file."""
+ shutil.rmtree(export_path)
redis_connection = RedisConnection()
datastore = redis_connection.connection(db=current_app.config['CACHE_REDIS_DB'], kv = True)
if datastore.redis.exists(cache_key):
@@ -172,3 +206,8 @@ def check_celery_is_run():
return False
else:
return True
+
+def check_session_lifetime():
+ """Check session lifetime."""
+ lifetime = get_lifetime()
+ return True if lifetime >= 86400 else False
diff --git a/modules/weko-search-ui/weko_search_ui/templates/weko_search_ui/admin/export.html b/modules/weko-search-ui/weko_search_ui/templates/weko_search_ui/admin/export.html
index afe6361251..cd6b5bc448 100644
--- a/modules/weko-search-ui/weko_search_ui/templates/weko_search_ui/admin/export.html
+++ b/modules/weko-search-ui/weko_search_ui/templates/weko_search_ui/admin/export.html
@@ -44,6 +44,8 @@
+
+
@@ -51,6 +53,7 @@
+
{%- endblock body %}
diff --git a/modules/weko-search-ui/weko_search_ui/translations/en/LC_MESSAGES/messages.mo b/modules/weko-search-ui/weko_search_ui/translations/en/LC_MESSAGES/messages.mo
index 8303bfaa2b..01f9f5f84b 100644
Binary files a/modules/weko-search-ui/weko_search_ui/translations/en/LC_MESSAGES/messages.mo and b/modules/weko-search-ui/weko_search_ui/translations/en/LC_MESSAGES/messages.mo differ
diff --git a/modules/weko-search-ui/weko_search_ui/translations/en/LC_MESSAGES/messages.po b/modules/weko-search-ui/weko_search_ui/translations/en/LC_MESSAGES/messages.po
index 6bb1a601ae..eaa121bed6 100644
--- a/modules/weko-search-ui/weko_search_ui/translations/en/LC_MESSAGES/messages.po
+++ b/modules/weko-search-ui/weko_search_ui/translations/en/LC_MESSAGES/messages.po
@@ -828,3 +828,9 @@ msgstr ""
#~ msgid "Execute"
#~ msgstr ""
+msgid "Export Start Time"
+msgstr ""
+
+msgid "Export Finish Time"
+msgstr ""
+
diff --git a/modules/weko-search-ui/weko_search_ui/translations/ja/LC_MESSAGES/messages.mo b/modules/weko-search-ui/weko_search_ui/translations/ja/LC_MESSAGES/messages.mo
index ded7877096..e87f50f884 100644
Binary files a/modules/weko-search-ui/weko_search_ui/translations/ja/LC_MESSAGES/messages.mo and b/modules/weko-search-ui/weko_search_ui/translations/ja/LC_MESSAGES/messages.mo differ
diff --git a/modules/weko-search-ui/weko_search_ui/translations/ja/LC_MESSAGES/messages.po b/modules/weko-search-ui/weko_search_ui/translations/ja/LC_MESSAGES/messages.po
index db13d8b017..1d76bf1717 100644
--- a/modules/weko-search-ui/weko_search_ui/translations/ja/LC_MESSAGES/messages.po
+++ b/modules/weko-search-ui/weko_search_ui/translations/ja/LC_MESSAGES/messages.po
@@ -834,3 +834,10 @@ msgstr "ロールの権限が足りずこのインデックスにアイテム登
#~ msgid "Execute"
#~ msgstr "実行"
+
+msgid "Export Start Time"
+msgstr ""
+
+msgid "Export Finish Time"
+msgstr ""
+
diff --git a/modules/weko-search-ui/weko_search_ui/utils.py b/modules/weko-search-ui/weko_search_ui/utils.py
index 64d0fd164f..23d58be220 100644
--- a/modules/weko-search-ui/weko_search_ui/utils.py
+++ b/modules/weko-search-ui/weko_search_ui/utils.py
@@ -22,16 +22,19 @@
import csv
import json
+import math
import os
import re
import shutil
import sys
import pytz
+import bagit
import tempfile
import traceback
import uuid
import zipfile
import chardet
+import gc
from collections import Callable, OrderedDict
from datetime import datetime
from functools import partial, reduce, wraps
@@ -41,7 +44,6 @@
from time import sleep
import pickle
-import bagit
import redis
from redis import sentinel
from celery.result import AsyncResult
@@ -125,9 +127,9 @@
WEKO_IMPORT_VALIDATE_MESSAGE,
WEKO_REPO_USER,
WEKO_SEARCH_TYPE_DICT,
- WEKO_SEARCH_UI_BULK_EXPORT_LIMIT,
WEKO_SEARCH_UI_BULK_EXPORT_MSG,
WEKO_SEARCH_UI_BULK_EXPORT_RUN_MSG,
+ WEKO_SEARCH_UI_BULK_EXPORT_FILE_CREATE_RUN_MSG,
WEKO_SEARCH_UI_BULK_EXPORT_TASK,
WEKO_SEARCH_UI_BULK_EXPORT_URI,
WEKO_SYS_USER,
@@ -3278,15 +3280,18 @@ def handle_check_duplication_item_id(ids: list):
return list(set(result))
-def export_all(root_url, user_id, data, timezone):
- """Gather all the item data and export and return as a JSON or BIBTEX.
+def export_all(root_url, user_id, data, start_time):
+ """Prepare to gather all the item data and export and return as a JSON or BIBTEX.
Parameter
- path is the path if file temparory
- post_data is the data items
- :return: JSON, BIBTEX
+ root_url (str): this system's root url.
+ user_id (int): a user who processed file output.
+ data (json): export processing's status data.
+ start_time (str): processing start time.
"""
- from weko_items_ui.utils import make_stats_file_with_permission, package_export_file
+ from weko_search_ui.tasks import write_files_task
+
+ current_app.logger.info("Bulk export all start at {}.".format(start_time))
_cache_prefix = current_app.config["WEKO_ADMIN_CACHE_PREFIX"]
_msg_config = current_app.config["WEKO_SEARCH_UI_BULK_EXPORT_MSG"]
@@ -3299,47 +3304,17 @@ def export_all(root_url, user_id, data, timezone):
name=_run_msg_config,
user_id=user_id
)
- _file_format = current_app.config.get('WEKO_ADMIN_OUTPUT_FORMAT', 'tsv').lower()
+ _file_create_config = \
+ current_app.config["WEKO_SEARCH_UI_BULK_EXPORT_FILE_CREATE_RUN_MSG"]
+ _file_create_key = _cache_prefix.format(
+ name=_file_create_config,
+ user_id=user_id
+ )
def _itemtype_name(name):
"""Check a list of allowed characters in filenames."""
return re.sub(r'[\/:*"<>|\s]', "_", name)
- def _write_files(item_datas, export_path):
- """Write TSV/CSV data to files.
-
- @param item_datas:
- @param export_path:
- @param list_item_role:
- @return:
- """
- permissions = dict(
- permission_show_hide=lambda a: True,
- check_created_id=lambda a: True,
- hide_meta_data_for_role=lambda a: True,
- current_language=lambda: True,
- )
- headers, records = make_stats_file_with_permission(
- item_datas["item_type_id"],
- item_datas["recids"],
- item_datas["data"],
- permissions,
- export_path
- )
- keys, labels, is_systems, options = headers
- item_datas["recids"].sort()
- item_datas["keys"] = keys
- item_datas["labels"] = labels
- item_datas["is_systems"] = is_systems
- item_datas["options"] = options
- item_datas["data"] = records
- item_type_data = item_datas
-
- file_full_path = "{}/{}.{}".format(export_path, item_type_data.get("name"), _file_format)
- with open(file_full_path, "w", encoding="utf-8-sig") as file:
- file_output = package_export_file(item_type_data)
- file.write(file_output.getvalue())
-
def _get_item_type_list(item_type_id):
"""Get item type list."""
item_types = []
@@ -3360,106 +3335,68 @@ def _get_item_type_list(item_type_id):
def _get_export_data(export_path, item_types, retrys, fromid="", toid="", retry_info={}):
try:
+ write_file_json = {
+ 'start_time': start_time,
+ 'finish_time': '',
+ 'export_path': export_path,
+ 'cancel_flg': False,
+ 'write_file_status': {}
+ }
+ reset_redis_cache(
+ _file_create_key,
+ json.dumps(write_file_json)
+ )
for it in item_types.copy():
item_type_id = it[0]
item_type_name = it[1]
item_datas = {}
- if item_type_id in retry_info:
- counter = retry_info[item_type_id]["counter"]
- file_part = retry_info[item_type_id]["part"]
- from_pid = retry_info[item_type_id]["max"]
- else:
- counter = 0
- file_part = 1
- from_pid = fromid if fromid else "1"
+ pickle_file_name = ''
+ counter, file_part, from_pid = get_retry_info(
+ item_type_id, retry_info, fromid)
current_app.logger.info(
- "Start processing item type {}({}).".format(
+ "Start bulk export of item type {}({}).".format(
item_type_name, item_type_id
)
)
- # get all record id
- if toid:
- recids = db.session.query(
- PersistentIdentifier.pid_value,
- PersistentIdentifier.object_uuid,
- RecordMetadata.json
- ).join(
- ItemMetadata,
- PersistentIdentifier.object_uuid == ItemMetadata.id,
- ).join(
- RecordMetadata,
- PersistentIdentifier.object_uuid == RecordMetadata.id,
- ).filter(
- PersistentIdentifier.pid_type == "recid",
- PersistentIdentifier.status == PIDStatus.REGISTERED,
- PersistentIdentifier.pid_value.notlike("%.%"),
- _func.to_number(
- PersistentIdentifier.pid_value,
- current_app.config["WEKO_SEARCH_UI_TO_NUMBER_FORMAT"]
- ) >= from_pid,
- _func.to_number(
- PersistentIdentifier.pid_value,
- current_app.config["WEKO_SEARCH_UI_TO_NUMBER_FORMAT"]
- ) <= toid,
- ItemMetadata.item_type_id == item_type_id
- ).order_by(_func.to_number(
- PersistentIdentifier.pid_value,
- current_app.config["WEKO_SEARCH_UI_TO_NUMBER_FORMAT"]
- )).all()
- else:
- recids = db.session.query(
- PersistentIdentifier.pid_value,
- PersistentIdentifier.object_uuid,
- RecordMetadata.json
- ).join(
- ItemMetadata,
- PersistentIdentifier.object_uuid == ItemMetadata.id,
- ).join(
- RecordMetadata,
- PersistentIdentifier.object_uuid == RecordMetadata.id,
- ).filter(
- PersistentIdentifier.pid_type == "recid",
- PersistentIdentifier.status == PIDStatus.REGISTERED,
- PersistentIdentifier.pid_value.notlike("%.%"),
- _func.to_number(
- PersistentIdentifier.pid_value,
- current_app.config["WEKO_SEARCH_UI_TO_NUMBER_FORMAT"]
- ) >= from_pid,
- ItemMetadata.item_type_id == item_type_id
- ).order_by(_func.to_number(
- PersistentIdentifier.pid_value,
- current_app.config["WEKO_SEARCH_UI_TO_NUMBER_FORMAT"]
- )).all()
-
- if len(recids) == 0:
+
+ recids = get_all_record_id(toid, from_pid, item_type_id)
+ current_app.logger.info("{}({}) get recids completed:{}".format(item_type_name, item_type_id, recids.count()))
+ if not recids:
item_types.remove(it)
continue
+ record_ids = get_record_ids(recids)
- record_ids = [(recid.pid_value, recid.object_uuid)
- for recid in recids if 'publish_status' in recid.json
- and recid.json['publish_status'] in [PublishStatus.PUBLIC.value, PublishStatus.PRIVATE.value]]
+ # recidsを削除
+ del recids
+ gc.collect()
+
+ file_count = math.ceil(len(record_ids) / current_app.config["WEKO_SEARCH_UI_BULK_EXPORT_LIMIT"])
+ write_file_json = json.loads(get_redis_cache(_file_create_key))
+ for i in range(file_count):
+ write_file_json['write_file_status'][item_type_id + '.' + str(i + 1)] = 'waiting'
+ reset_redis_cache(
+ _file_create_key,
+ json.dumps(write_file_json)
+ )
if len(record_ids) == 0:
item_types.remove(it)
continue
for recid, uuid in record_ids:
- if counter % WEKO_SEARCH_UI_BULK_EXPORT_LIMIT == 0 and item_datas:
+ if counter % current_app.config["WEKO_SEARCH_UI_BULK_EXPORT_LIMIT"] == 0 and item_datas:
# Create export info file
item_datas["name"] = "{}.part{}".format(
item_datas["name"], file_part
)
- _write_files(item_datas, export_path)
- reset_redis_cache(
- _run_msg_key,
- "The latest {} file was created on {}.".format(
- _file_format,
- datetime.now(pytz.timezone(timezone)).strftime("%Y/%m/%d %H:%M:%S"))
- + " Number of retries: {} times.".format(retrys)
- )
- current_app.logger.info(
- "{}.{} has been created.".format(item_datas["name"], _file_format)
+ pickle_file_name = "{}.{}.part{}.pickle".format(
+ user_id, item_type_id, file_part
)
+ with open(pickle_file_name, 'wb') as f:
+ pickle.dump(item_datas, f)
+ del item_datas
+ gc.collect()
+ write_files_task.apply_async(args=(export_path, pickle_file_name, user_id,))
item_datas = {}
file_part += 1
retry_info[item_type_id] = {
@@ -3481,28 +3418,31 @@ def _get_export_data(export_path, item_types, retrys, fromid="", toid="", retry_
"recids": [],
"data": {},
}
+ pickle_file_name = "{}.{}.pickle".format(user_id,item_type_id)
item_datas["recids"].append(recid)
item_datas["data"][recid] = record
counter += 1
+ del record
+ gc.collect()
if file_part != 1:
item_datas["name"] = "{}.part{}".format(
item_datas["name"], file_part
)
+ pickle_file_name = "{}.{}.part{}.pickle".format(
+ user_id, item_type_id,file_part
+ )
+
+ with open(pickle_file_name, 'wb') as f:
+ pickle.dump(item_datas, f)
+
+ del item_datas
+ gc.collect()
+
# Create export info file
- _write_files(item_datas, export_path)
- reset_redis_cache(
- _run_msg_key,
- "The latest {} file was created on {}.".format(
- _file_format,
- datetime.now(pytz.timezone(timezone)).strftime("%Y/%m/%d %H:%M:%S"))
- + " Number of retries: {} times.".format(retrys)
- )
+ write_files_task.apply_async(args=(export_path, pickle_file_name, user_id,))
item_types.remove(it)
- current_app.logger.info(
- "{}.{} has been created.".format(item_datas["name"], _file_format)
- )
current_app.logger.info(
"Processed {} items of item type {}.".format(
counter, item_type_name
@@ -3526,9 +3466,10 @@ def _get_export_data(export_path, item_types, retrys, fromid="", toid="", retry_
reset_redis_cache(_msg_key, "")
reset_redis_cache(_run_msg_key, "")
- temp_path = tempfile.TemporaryDirectory(
- prefix=current_app.config["WEKO_ITEMS_UI_EXPORT_TMP_PREFIX"]
- )
+ reset_redis_cache(_file_create_key, json.dumps({}))
+
+ temp_path = os.getenv('TMPDIR')
+ os.makedirs(temp_path, exist_ok=True)
try:
# Delete old file
_task_config = current_app.config["WEKO_SEARCH_UI_BULK_EXPORT_URI"]
@@ -3540,7 +3481,7 @@ def _get_export_data(export_path, item_types, retrys, fromid="", toid="", retry_
if prev_uri:
delete_exported(prev_uri, _uri_key)
- export_path = temp_path.name + "/" + datetime.utcnow().strftime("%Y%m%d%H%M%S")
+ export_path = temp_path + "/" + datetime.utcnow().strftime("%Y%m%d%H%M%S%f")
os.makedirs(export_path, exist_ok=True)
item_type_id = data.get('item_type_id', "-1")
@@ -3562,25 +3503,211 @@ def _get_export_data(export_path, item_types, retrys, fromid="", toid="", retry_
result = _get_export_data(export_path, item_types, 0, fromid, toid)
if result:
- # Create bag
- bagit.make_bag(export_path)
- shutil.make_archive(export_path, "zip", export_path)
- with open(export_path + ".zip", "rb") as file:
- src = FileInstance.create()
- src.set_contents(file, default_location=Location.get_default().uri)
db.session.commit()
else:
+ json_data = json.loads(get_redis_cache(_file_create_key))
+ json_data['cancel_flg'] = True
+ reset_redis_cache(_file_create_key, json.dumps(json_data))
reset_redis_cache(_msg_key, "Export failed.")
else:
reset_redis_cache(_msg_key, "Export failed. Please check item id range.")
- reset_redis_cache(_run_msg_key, "")
- return src.uri if result and src else ""
except Exception as ex:
db.session.rollback()
current_app.logger.error(ex)
reset_redis_cache(_msg_key, "Export failed.")
reset_redis_cache(_run_msg_key, "")
- return ""
+
+def get_all_record_id(toid, from_pid, item_type_id):
+ """Get all record id.
+
+ Args:
+ toid (str): The ending ID.
+ from_pid (str): The starting ID.
+ item_type_id (str): The item type ID.
+
+ Returns:
+ list: List of record IDs.
+ """
+
+ # get all record id
+ if toid:
+ recids = db.session.query(
+ PersistentIdentifier.pid_value,
+ PersistentIdentifier.object_uuid,
+ RecordMetadata.json
+ ).join(
+ ItemMetadata,
+ PersistentIdentifier.object_uuid == ItemMetadata.id,
+ ).join(
+ RecordMetadata,
+ PersistentIdentifier.object_uuid == RecordMetadata.id,
+ ).filter(
+ PersistentIdentifier.pid_type == "recid",
+ PersistentIdentifier.status == PIDStatus.REGISTERED,
+ PersistentIdentifier.pid_value.notlike("%.%"),
+ _func.to_number(
+ PersistentIdentifier.pid_value,
+ current_app.config["WEKO_SEARCH_UI_TO_NUMBER_FORMAT"]
+ ) >= from_pid,
+ _func.to_number(
+ PersistentIdentifier.pid_value,
+ current_app.config["WEKO_SEARCH_UI_TO_NUMBER_FORMAT"]
+ ) <= toid,
+ ItemMetadata.item_type_id == item_type_id
+ ).order_by(_func.to_number(
+ PersistentIdentifier.pid_value,
+ current_app.config["WEKO_SEARCH_UI_TO_NUMBER_FORMAT"]
+ )).yield_per(500)
+ else:
+ recids = db.session.query(
+ PersistentIdentifier.pid_value,
+ PersistentIdentifier.object_uuid,
+ RecordMetadata.json
+ ).join(
+ ItemMetadata,
+ PersistentIdentifier.object_uuid == ItemMetadata.id,
+ ).join(
+ RecordMetadata,
+ PersistentIdentifier.object_uuid == RecordMetadata.id,
+ ).filter(
+ PersistentIdentifier.pid_type == "recid",
+ PersistentIdentifier.status == PIDStatus.REGISTERED,
+ PersistentIdentifier.pid_value.notlike("%.%"),
+ _func.to_number(
+ PersistentIdentifier.pid_value,
+ current_app.config["WEKO_SEARCH_UI_TO_NUMBER_FORMAT"]
+ ) >= from_pid,
+ ItemMetadata.item_type_id == item_type_id
+ ).order_by(_func.to_number(
+ PersistentIdentifier.pid_value,
+ current_app.config["WEKO_SEARCH_UI_TO_NUMBER_FORMAT"]
+ )).yield_per(500)
+
+ return recids
+
+
+def get_retry_info(item_type_id, retry_info, fromid):
+ """Get retry information for item type.
+
+ Args:
+ item_type_id (str): The item type ID.
+ retry_info (dict): The retry information dictionary.
+ fromid (str): The starting ID.
+
+ Returns:
+ tuple: A tuple containing counter, file_part, and from_pid.
+ """
+ if item_type_id in retry_info:
+ counter = retry_info[item_type_id]["counter"]
+ file_part = retry_info[item_type_id]["part"]
+ from_pid = retry_info[item_type_id]["max"]
+ else:
+ counter = 0
+ file_part = 1
+ from_pid = fromid if fromid else "1"
+ return counter, file_part, from_pid
+
+
+def get_record_ids(recids):
+ """Get record ids.
+
+ Args:
+ recids (list): List of record ids.
+
+ Returns:
+ list: List of record ids.
+ """
+ record_ids = [(recid.pid_value, recid.object_uuid)
+ for recid in recids if recid.json and 'publish_status' in recid.json \
+ and recid.json['publish_status'] in [PublishStatus.PUBLIC.value, PublishStatus.PRIVATE.value]]
+ return record_ids
+
+
+def write_files(item_datas, export_path, user_id, retrys):
+ """Write TSV/CSV data to files.
+ Args:
+ item_datas (json): data for file output
+ export_path (str): file creation destination
+ user_id (int): performing user id
+ retrys (int): retry time
+
+ Returns:
+ bool: task is success or failure.
+ """
+ from weko_items_ui.utils import make_stats_file_with_permission, \
+ package_export_file
+ _cache_prefix = current_app.config["WEKO_ADMIN_CACHE_PREFIX"]
+ _run_msg_config = current_app.config["WEKO_SEARCH_UI_BULK_EXPORT_RUN_MSG"]
+ _run_msg_key = _cache_prefix.format(
+ name=_run_msg_config,
+ user_id=user_id
+ )
+ _timezone = current_app.config.get("WEKO_INDEX_TREE_PUBLIC_DEFAULT_TIMEZONE")
+ _file_format = current_app.config.get('WEKO_ADMIN_OUTPUT_FORMAT', 'tsv').lower()
+
+ try:
+ permissions = dict(
+ permission_show_hide=lambda a: True,
+ check_created_id=lambda a: True,
+ hide_meta_data_for_role=lambda a: True,
+ current_language=lambda: True,
+ )
+ headers, records = make_stats_file_with_permission(
+ item_datas["item_type_id"],
+ item_datas["recids"],
+ item_datas["data"],
+ permissions,
+ export_path
+ )
+ keys, labels, is_systems, options = headers
+ item_datas["recids"].sort()
+ item_datas["keys"] = keys
+ item_datas["labels"] = labels
+ item_datas["is_systems"] = is_systems
+ item_datas["options"] = options
+ item_datas["data"] = records
+ item_type_data = item_datas
+
+ os.makedirs(export_path, exist_ok=True)
+
+ file_full_path = "{}/{}.{}".format(
+ export_path,
+ item_type_data.get("name"),
+ _file_format
+ )
+ with open(file_full_path, "w", encoding="utf-8-sig") as file:
+ file_output = package_export_file(item_type_data)
+ file.write(file_output.getvalue())
+ del file_output,item_type_data
+ gc.collect()
+ reset_redis_cache(
+ _run_msg_key,
+ "The latest {} file was created on {}.".format(
+ _file_format,
+ datetime.now(pytz.timezone(_timezone)).strftime("%Y/%m/%d %H:%M:%S"))
+ + " Number of retries: {} times.".format(retrys)
+ )
+ current_app.logger.info(
+ "{}.{} has been created.".format(item_datas["name"], _file_format)
+ )
+ db.session.commit()
+ del item_datas, headers, records, keys, labels, is_systems, options,permissions
+ gc.collect()
+ return True
+ except SQLAlchemyError as ex:
+ current_app.logger.error(ex)
+ _num_retry = current_app.config["WEKO_SEARCH_UI_BULK_EXPORT_RETRY"]
+ if retrys < _num_retry:
+ retrys += 1
+ current_app.logger.info("retry count: {}".format(retrys))
+ db.session.rollback()
+ sleep(5)
+ result = write_files(
+ item_datas, export_path, user_id, retrys
+ )
+ return result
+ else:
+ return False
def delete_exported(uri, cache_key):
@@ -3606,10 +3733,14 @@ def cancel_export_all():
name=WEKO_SEARCH_UI_BULK_EXPORT_TASK,
user_id=current_user.get_id()
)
+ _file_create_key = current_app.config["WEKO_ADMIN_CACHE_PREFIX"].format(
+ name=WEKO_SEARCH_UI_BULK_EXPORT_FILE_CREATE_RUN_MSG,
+ user_id=current_user.get_id()
+ )
_expired_time=current_app.config["WEKO_SEARCH_UI_BULK_EXPORT_TASKID_EXPIRED_TIME"]
try:
task_id = get_redis_cache(cache_key)
- export_status, _, _, _, _ = get_export_status()
+ export_status, _, _, _, _, _, _ = get_export_status()
if export_status:
revoke(task_id, terminate=True)
@@ -3620,6 +3751,9 @@ def cancel_export_all():
),
countdown=int(_expired_time) * 60
)
+ json_data = json.loads(get_redis_cache(_file_create_key))
+ json_data['cancel_flg'] = True
+ reset_redis_cache(_file_create_key, json.dumps(json_data))
return True
except Exception as ex:
current_app.logger.error(ex)
@@ -3632,6 +3766,8 @@ def get_export_status():
Return: True: Otthers
False: Success / Failed / Revoked
"""
+ from weko_search_ui.tasks import delete_exported_task
+
cache_key = current_app.config["WEKO_ADMIN_CACHE_PREFIX"].format(
name=WEKO_SEARCH_UI_BULK_EXPORT_TASK,
user_id=current_user.get_id()
@@ -3648,27 +3784,91 @@ def get_export_status():
name=WEKO_SEARCH_UI_BULK_EXPORT_RUN_MSG,
user_id=current_user.get_id()
)
+ file_msg = current_app.config["WEKO_ADMIN_CACHE_PREFIX"].format(
+ name=WEKO_SEARCH_UI_BULK_EXPORT_FILE_CREATE_RUN_MSG,
+ user_id=current_user.get_id()
+ )
+ _expired_time = current_app.config["WEKO_SEARCH_UI_BULK_EXPORT_EXPIRED_TIME"]
+
+ def _check_write_file_info(json):
+ status = json.get('write_file_status','before')
+ cancel_flg = json.get('cancel_flg', False)
+ if status == 'before':
+ return 'BEFORE'
+ elif status and ('waiting' not in status.values()) and ('started' not in status.values()):
+ if 'error' in status.values():
+ return ''
+ elif 'canceled' in status.values():
+ return 'REVOKED'
+ else:
+ return 'SUCCESS'
+ elif cancel_flg:
+ return 'REVOKED'
+ elif not status:
+ return 'SUCCESS'
+ else:
+ return 'STARTED'
+
export_status = False
download_uri = None
message = None
run_message = ""
status = ""
+ start_time = ""
+ finish_time = ""
try:
task_id = get_redis_cache(cache_key)
download_uri = get_redis_cache(cache_uri)
message = get_redis_cache(cache_msg)
run_message = get_redis_cache(run_msg)
+ write_file_info = get_redis_cache(file_msg)
if task_id:
- task = AsyncResult(task_id)
- status_cond = task.successful() or task.failed() or task.state == "REVOKED"
- status = task.state
- export_status = True if not status_cond else False
+ write_file_data = json.loads(write_file_info)
+ if write_file_data:
+ write_file_status = _check_write_file_info(write_file_data)
+ task = AsyncResult(task_id)
+ status_cond = (task.successful() or task.failed() or task.state == "REVOKED") \
+ and write_file_status != 'STARTED'
+ if not write_file_status == 'BEFORE':
+ status = write_file_status
+ export_status = True if not status_cond else False
+ start_time = write_file_data.get("start_time")
+ finish_time = write_file_data.get("finish_time")
+ if status_cond and write_file_status == 'SUCCESS':
+ export_path = write_file_data['export_path']
+ is_dir = not os.path.isdir(os.path.join(export_path, 'data'))
+ if is_dir:
+ bagit.make_bag(export_path)
+ shutil.make_archive(export_path, "zip", export_path)
+ with open(export_path + ".zip", "rb") as file:
+ src = FileInstance.create()
+ src.set_contents(file, default_location=Location.get_default().uri)
+ db.session.commit()
+ download_uri = src.uri
+ _timezone = current_app.config.get("WEKO_INDEX_TREE_PUBLIC_DEFAULT_TIMEZONE")
+ finish_time = datetime.now(pytz.timezone(_timezone)).strftime('%Y/%m/%d %H:%M:%S')
+ write_file_data = json.loads(get_redis_cache(file_msg))
+ write_file_data["finish_time"] = finish_time
+ current_app.logger.info("Bulk export all finished at {}.".format(finish_time))
+ reset_redis_cache(file_msg, json.dumps(write_file_data))
+ reset_redis_cache(cache_uri, download_uri)
+ reset_redis_cache(run_msg, "")
+ delete_exported_task.apply_async(
+ args=(
+ download_uri,
+ cache_uri,
+ cache_key,
+ export_path
+ ),
+ countdown=int(_expired_time) * 60,
+ )
+ os.remove(export_path + ".zip")
except Exception as ex:
current_app.logger.error(ex)
export_status = False
- return export_status, download_uri, message, run_message, status
-
+ return export_status, download_uri, message, run_message, \
+ status, start_time, finish_time
def handle_check_item_is_locked(item):
"""Check an item is being edit or deleted.