diff --git a/apps/sources/__init__.py b/apps/sources/__init__.py index 3cd5b85b62..1ac08e782a 100644 --- a/apps/sources/__init__.py +++ b/apps/sources/__init__.py @@ -18,6 +18,7 @@ from .permissions import (PERMISSION_SOURCES_SETUP_VIEW, PERMISSION_SOURCES_SETUP_EDIT, PERMISSION_SOURCES_SETUP_DELETE, PERMISSION_SOURCES_SETUP_CREATE) from .tasks import task_fetch_pop3_emails +from .conf.settings import POP3_EMAIL_PROCESSING_INTERVAL staging_file_preview = {'text': _(u'preview'), 'class': 'fancybox-noscaling', 'view': 'staging_file_preview', 'args': ['source.source_type', 'source.pk', 'object.id'], 'famfam': 'zoom', 'permissions': [PERMISSION_DOCUMENT_NEW_VERSION, PERMISSION_DOCUMENT_CREATE]} staging_file_delete = {'text': _(u'delete'), 'view': 'staging_file_delete', 'args': ['source.source_type', 'source.pk', 'object.id'], 'famfam': 'delete', 'keep_query': True, 'permissions': [PERMISSION_DOCUMENT_NEW_VERSION, PERMISSION_DOCUMENT_CREATE]} @@ -79,4 +80,4 @@ register_model_list_columns(StagingFile, [ register_setup(setup_sources) -#register_interval_job('task_fetch_pop3_emails', _(u'Connects to the POP3 email sources and fetches the attached documents.'), task_fetch_pop3_emails, seconds=5)#QUEUE_PROCESSING_INTERVAL) +register_interval_job('task_fetch_pop3_emails', _(u'Connects to the POP3 email sources and fetches the attached documents.'), task_fetch_pop3_emails, seconds=POP3_EMAIL_PROCESSING_INTERVAL) diff --git a/apps/sources/conf/settings.py b/apps/sources/conf/settings.py index 9babb1d457..c4bf07fb46 100644 --- a/apps/sources/conf/settings.py +++ b/apps/sources/conf/settings.py @@ -8,9 +8,20 @@ from smart_settings.api import Setting, SettingNamespace namespace = SettingNamespace('sources', _(u'Sources'), module='sources.conf.settings') +POP3_DEFAULT_TIMEOUT = 5 +POP3_DEFAULT_EMAIL_PROCESSING_INTERVAL = 15 * 60 + Setting( namespace=namespace, name='POP3_TIMEOUT', global_name='SOURCES_POP3_TIMEOUT', default=5, ) + +Setting( + namespace=namespace, + name='POP3_EMAIL_PROCESSING_INTERVAL', + global_name='SOURCES_POP3_EMAIL_PROCESSING_INTERVAL', + default=POP3_DEFAULT_EMAIL_PROCESSING_INTERVAL, +) + diff --git a/apps/sources/models.py b/apps/sources/models.py index e227661a47..f4254c5ebc 100644 --- a/apps/sources/models.py +++ b/apps/sources/models.py @@ -188,16 +188,16 @@ class POP3Email(BaseModel): username = models.CharField(max_length=64, verbose_name=_(u'username')) password = models.CharField(max_length=64, verbose_name=_(u'password')) uncompress = models.CharField(max_length=1, choices=SOURCE_UNCOMPRESS_CHOICES, verbose_name=_(u'uncompress'), help_text=_(u'Whether to expand or not compressed archives.')) - delete_messages = models.BooleanField(verbose_name=_(u'delete messages'), help_text=_(u'Delete messages after downloading their respective attached documents.')) + delete_messages = models.BooleanField(default=True, verbose_name=_(u'delete messages'), help_text=_(u'Delete messages after downloading their respective attached documents.')) # From: http://bookmarks.honewatson.com/2009/08/11/python-gmail-imaplib-search-subject-get-attachments/ @staticmethod - def process_message(source, message): + def process_message(source, message, expand=False): email = message_from_string(message) counter = 1 for part in email.walk(): - disposition = part.get('Content-Disposition') + disposition = part.get('Content-Disposition', 'attachment') logger.debug('Disposition: %s' % disposition) if disposition.startswith('attachment'): @@ -212,7 +212,7 @@ class POP3Email(BaseModel): logger.debug('filename: %s' % filename) document_file = Attachment(part, name=filename) - source.upload_file(document_file, filename=filename) + source.upload_file(document_file, filename=filename, expand=expand) def fetch_mail(self): @@ -222,7 +222,7 @@ class POP3Email(BaseModel): if self.ssl: port = self.port or POP3_SSL_PORT logger.debug('port: %d' % port) - mailbox = poplib.POP3_SSL(self.host, int(port))#, POP3_TIMEOUT) + mailbox = poplib.POP3_SSL(self.host, int(port)) else: port = self.port or POP3_PORT logger.debug('port: %d' % port) @@ -244,8 +244,9 @@ class POP3Email(BaseModel): complete_message = '\n'.join(mailbox.retr(message_number)[1]) - POP3Email.process_message(self, complete_message) - mailbox.dele(message_number) + POP3Email.process_message(source=self, message=complete_message, expand=self.uncompress) + if self.delete_messages: + mailbox.dele(message_number) mailbox.quit() diff --git a/apps/sources/tasks.py b/apps/sources/tasks.py index be2a11ac3c..c7da93e28a 100644 --- a/apps/sources/tasks.py +++ b/apps/sources/tasks.py @@ -9,56 +9,24 @@ from django.db.models import Q from job_processor.api import process_job from lock_manager import Lock, LockError -#from .api import do_document_ocr -#from .literals import (QUEUEDOCUMENT_STATE_PENDING, -# QUEUEDOCUMENT_STATE_PROCESSING, DOCUMENTQUEUE_STATE_ACTIVE, -# QUEUEDOCUMENT_STATE_ERROR) from .models import POP3Email -#from .conf.settings import (NODE_CONCURRENT_EXECUTION, REPLICATION_DELAY, -# QUEUE_PROCESSING_INTERVAL) - -LOCK_EXPIRE = 30 * 1 # Lock expires in 10 minutes -# TODO: Tie LOCK_EXPIRATION with hard task timeout +from .conf.settings import POP3_TIMEOUT logger = logging.getLogger(__name__) -""" -def task_process_queue_document(queue_document_id): - lock_id = u'task_proc_queue_doc-%d' % queue_document_id - try: - logger.debug('trying to acquire lock: %s' % lock_id) - lock = Lock.acquire_lock(lock_id, LOCK_EXPIRE) - logger.debug('acquired lock: %s' % lock_id) - queue_document = QueueDocument.objects.get(pk=queue_document_id) - queue_document.state = QUEUEDOCUMENT_STATE_PROCESSING - queue_document.node_name = platform.node() - queue_document.save() - try: - do_document_ocr(queue_document) - queue_document.delete() - except Exception, e: - queue_document.state = QUEUEDOCUMENT_STATE_ERROR - queue_document.result = e - queue_document.save() - - lock.release() - except LockError: - logger.debug('unable to obtain lock') - pass -""" def task_fetch_single_pop3_email(pop3_email): try: lock_id = u'task_fetch_pop3_email-%d' % pop3_email.pk logger.debug('trying to acquire lock: %s' % lock_id) - lock = Lock.acquire_lock(lock_id, LOCK_EXPIRE) + lock = Lock.acquire_lock(lock_id, POP3_TIMEOUT + 60) # Lock expiration = POP3 timeout + 60 seconds logger.debug('acquired lock: %s' % lock_id) - #try: - pop3_email.fetch_mail() - #except Exception, exc: - #raise - #finally: - lock.release() + try: + pop3_email.fetch_mail() + except Exception, exc: + raise + finally: + lock.release() except LockError: logger.error('unable to obtain lock') pass @@ -71,32 +39,3 @@ def task_fetch_pop3_emails(): task_fetch_single_pop3_email(pop3_email) except Exception, exc: logger.error('Unhandled exception: %s' % exc) - - """ - # TODO: reset_orphans() - q_pending = Q(state=QUEUEDOCUMENT_STATE_PENDING) - q_delayed = Q(delay=True) - q_delay_interval = Q(datetime_submitted__lt=datetime.now() - timedelta(seconds=REPLICATION_DELAY)) - for document_queue in DocumentQueue.objects.filter(state=DOCUMENTQUEUE_STATE_ACTIVE): - current_local_processing_count = QueueDocument.objects.filter( - state=QUEUEDOCUMENT_STATE_PROCESSING).filter( - node_name=platform.node()).count() - if current_local_processing_count < NODE_CONCURRENT_EXECUTION: - try: - oldest_queued_document_qs = document_queue.queuedocument_set.filter( - (q_pending & ~q_delayed) | (q_pending & q_delayed & q_delay_interval)) - - if oldest_queued_document_qs: - oldest_queued_document = oldest_queued_document_qs.order_by('datetime_submitted')[0] - process_job(task_process_queue_document, oldest_queued_document.pk) - except Exception, e: - pass - #print 'DocumentQueueWatcher exception: %s' % e - finally: - # Don't process anymore from this queryset, might be stale - break - else: - logger.debug('already processing maximun') - else: - logger.debug('nothing to process') - """