matthias/mayan-edms
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Remove SSL timeout option, added POP3_EMAIL_PROCESSING_INTERVAL configuration option

Browse Source
This commit is contained in:
Roberto Rosario
2012-03-10 03:30:21 -04:00
parent 29c3746163
commit 0e060aec46
4 changed files with 29 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
apps/sources/__init__.py
View File

@@ -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)

11
apps/sources/conf/settings.py
View File

@@ -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,
)

15
apps/sources/models.py
View File

@@ -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()

77
apps/sources/tasks.py
View File

@@ -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')
"""