Fix harvest validation (advice on proper PR creation needed)

.gitignore

@@ -13,4 +13,5 @@ development.ini
 node_modules
 *.project
 .eggs
-.vscode/
+.idea/
+.vscode/

ckanext/harvest/harvesters/ckanharvester.py

@@ -11,6 +11,7 @@
 from ckan.logic import ValidationError, NotFound, get_action
 from ckan.lib.helpers import json
 from ckan.plugins import toolkit
+import ckan.lib.plugins as lib_plugins
 
 from ckanext.harvest.model import HarvestObject
 from .base import HarvesterBase
@@ -547,6 +548,43 @@ def get_extra(key, package_dict):
 
             package_dict = self.modify_package_dict(package_dict, harvest_object)
 
+            # validate packages if needed
+            validate_packages = True
+            if validate_packages:
+                if 'type' not in package_dict:
+                    package_plugin = lib_plugins.lookup_package_plugin()
+                    try:
+                        # use first type as default if user didn't provide type
+                        package_type = package_plugin.package_types()[0]
+                    except (AttributeError, IndexError):
+                        package_type = 'dataset'
+                        # in case a 'dataset' plugin was registered w/o fallback
+                        package_plugin = lib_plugins.lookup_package_plugin(package_type)
+                    package_dict['type'] = package_type
+                else:
+                    package_plugin = lib_plugins.lookup_package_plugin(package_dict['type'])
+
+
+                errors = {}
+                # if package has been previously imported
+                try:
+                    existing_package_dict = self._find_existing_package(package_dict)
+
+                    if not 'metadata_modified' in package_dict or \
+                            package_dict['metadata_modified'] > existing_package_dict.get('metadata_modified'):
+                        schema = package_plugin.update_package_schema()
+                        data, errors = lib_plugins.plugin_validate(
+                            package_plugin, base_context, package_dict, schema, 'package_update')
+
+                except NotFound:
+
+                    schema = package_plugin.create_package_schema()
+                    data, errors = lib_plugins.plugin_validate(
+                        package_plugin, base_context, package_dict, schema, 'package_create')
+
+                if errors:
+                    raise ValidationError(errors)
+
             result = self._create_or_update_package(
                 package_dict, harvest_object, package_dict_form='package_show')
 

ckanext/harvest/logic/action/notify.py

@@ -0,0 +1,117 @@
+
+import logging
+
+from pylons import config, app_globals
+
+from ckan import logic
+from ckan.logic import get_action
+
+from ckan.plugins import toolkit
+
+
+from ckanext.harvest.logic.dictization import harvest_job_dictize
+
+import ckan.lib.mailer as mailer
+
+log = logging.getLogger(__name__)
+
+
+
+def send_error_mail_ncar(context, job_obj):
+
+    sql = 'select name from package where id = :source_id;'
+
+    model = context['model']
+
+    q = model.Session.execute(sql, {'source_id': job_obj.source.id})
+
+    for row in q:
+        harvest_name = str(row['name'])
+
+    ckan_site_url = config.get('ckan.site_url')
+    job_url = ckan_site_url + '/harvest/' + harvest_name + '/job/' + job_obj.id
+
+    msg = 'This is a failure-notification of the latest harvest job on ' + ckan_site_url + '.\n\n'
+    msg += 'Harvest Job URL: ' + job_url + '\n\n'
+
+    sql = '''select g.title as org, s.title as job_title from member m
+            join public.group g on m.group_id = g.id
+            join harvest_source s on s.id = m.table_id
+            where table_id = :source_id;'''
+
+    q = model.Session.execute(sql, {'source_id': job_obj.source.id})
+
+    for row in q:
+        orgName = str(row['org'])
+        msg += 'Organization: ' + str(row['org']) + '\n\n'
+        msg += 'Harvest Source: ' + str(row['job_title']) + '\n\n'
+
+    msg += 'Date of Harvest: ' + str(job_obj.created) + ' GMT\n\n'
+
+    out = {
+        'last_job': None,
+    }
+
+    out['last_job'] = harvest_job_dictize(job_obj, context)
+
+    job_dict = get_action('harvest_job_report')(context, {'id': job_obj.id})
+    error_dicts = job_dict['object_errors']
+    errored_object_keys = error_dicts.keys()
+    numRecordsInError = len(errored_object_keys)
+    msg += 'Records in Error: ' + str(numRecordsInError) + '\n\n'
+
+    msg += 'For help, please contact the NCAR Data Stewardship Coordinator (mailto:datahelp@ucar.edu).\n\n\n'
+
+    if numRecordsInError <= 20:
+       errored_object_keys = errored_object_keys[:20]
+       for key in errored_object_keys:
+           error_dict = error_dicts[key]
+           msg += error_dict['original_url'] + ' :\n\n'
+           for error in error_dict['errors']:
+               msg += error['message']
+               if error['line']:
+                  msg += ' (line ' + str(error['line']) + ')\n\n'
+               else:
+                  msg += '\n'
+           msg += '\n\n'
+    else:
+       for key in errored_object_keys:
+           msg += error_dicts[key]['original_url'] + '\n\n'
+       msg += '\n\nError Messages are suppressed if there are more than 20 records with errors.\n'
+
+    log.debug("msg == " + msg)
+
+    if numRecordsInError > 0:
+        msg += '\n--\nYou are receiving this email because you are currently set-up as a member of the Organization "' + orgName + '" for ' + config.get('ckan.site_title') + '. Please do not reply to this email as it was sent from a non-monitored address.'
+
+        # get org info
+        log.debug('orgName == ' + orgName)
+        org_dict = toolkit.get_action('organization_show')(context, {'id' : orgName.lower(), 'include_users': True})
+
+        # get usernames in org
+        usernames = [x['name'] for x in org_dict['users']]
+        log.debug("usernames == " + ','.join(usernames))
+
+        # get emails for users 
+        email_recipients = []
+        for username in usernames:
+            user_dict = toolkit.get_action('user_show')(context, {'id' : username})
+            email_recipients.append(user_dict['email'])
+
+        log.debug("email_recipients == " + ','.join(email_recipients))
+        emails = {}
+
+        for recipient in email_recipients:
+            email = {'recipient_name': recipient,
+                      'recipient_email': recipient,
+                      'subject': config.get('ckan.site_title') + ' - Harvesting Job - Error Notification',
+                      'body': msg}
+
+            try:
+                app_globals._push_object(config['pylons.app_globals'])
+                mailer.mail_recipient(**email)
+            except Exception as e:
+                log.exception(e)
+                log.error('Sending Harvest-Notification-Mail failed. Message: ' + msg)
+
+

ckanext/harvest/logic/action/update.py

@@ -36,6 +36,8 @@
 from ckanext.harvest.logic.action.get import (
     harvest_source_show, harvest_job_list, _get_sources_for_user)
 
+from ckanext.harvest.logic.action.notify import send_error_mail_ncar
+
 import ckan.lib.mailer as mailer
 from itertools import islice
 
@@ -668,6 +670,8 @@ def harvest_jobs_run(context, data_dict):
                     log.debug('Notifications: All:{} On error:{} Errors:{}'.format(notify_all, notify_errors, last_job_errors))
 
                     if last_job_errors > 0 and (notify_all or notify_errors):
+                        #send_error_mail_ncar(context, job_obj)
+                        #get_mail_extra_vars(context, job_obj.source.id, status)
                         send_error_email(context, job_obj.source.id, status)
                     elif notify_all:
                         send_summary_email(context, job_obj.source.id, status)
@@ -698,8 +702,20 @@ def get_mail_extra_vars(context, source_id, status):
         harvest_object_error = report.get(
             'object_errors')[harvest_object_error_key]['errors']
 
-        for error in harvest_object_error:
-            obj_errors.append(error['message'])
+    ckan_site_url = config.get('ckan.site_url')
+    job_url = toolkit.url_for('harvest_job_show', source=source['id'], id=last_job['id'])
+
+    msg = 'This is a failure-notification of the latest harvest job on ' + ckan_site_url + '.\n\n'
+    msg += 'Harvest Job URL: ' + ckan_site_url + job_url + '\n\n'
+
+    msg += toolkit._('Harvest Source: {0}').format(source['title']) + '\n'
+    if source.get('config'):
+        msg += toolkit._('Harvester-Configuration: {0}').format(source['config']) + '\n'
+    msg += '\n\n'
+
+    if source['organization']:
+        msg += toolkit._('Organization: {0}').format(source['organization']['name'])
+        msg += '\n\n'
 
     for harvest_gather_error in islice(report.get('gather_errors'), 0, 20):
         job_errors.append(harvest_gather_error['message'])
@@ -709,6 +725,8 @@ def get_mail_extra_vars(context, source_id, status):
     else:
         organization = 'Not specified'
 
+    msg += 'For help, please contact the NCAR Data Stewardship Coordinator (mailto:datahelp@ucar.edu).\n\n\n'
+
     harvest_configuration = source.get('config')
 
     if harvest_configuration in [None, '', '{}']:
@@ -774,6 +792,18 @@ def send_error_email(context, source_id, status):
     send_mail(recipients, subject, body)
 
 
+#   for harvest_object_error_key in islice(report.get('object_errors'), 0, 20):
+#       harvest_object_error = report.get('object_errors')[harvest_object_error_key]['errors']
+#       harvest_object_url = report.get('object_errors')[harvest_object_error_key]['original_url']
+#       for error in harvest_object_error:
+#           obj_error += harvest_object_url + ' :\n\n'
+#           obj_error += error['message']
+#           if error['line']:
+#               obj_error += ' (line ' + str(error['line']) + ')\n\n'
+#           else:
+#               obj_error += '\n'
+
+
 def send_mail(recipients, subject, body):
 
     for recipient in recipients: