<commit_before># Copyright (c) 2013 AnsibleWorks, Inc.
# All Rights Reserved.

# Python
import ConfigParser
import cStringIO
import datetime
import distutils.version
import json
import logging
import os
import pipes
import re
import subprocess
import stat
import tempfile
import time
import traceback
import urlparse
import uuid

# Pexpect
import pexpect

# Celery
from celery import Task

# Django
from django.conf import settings
from django.db import transaction
from django.utils.datastructures import SortedDict
from django.utils.timezone import now

# AWX
from awx.main.models import Job, ProjectUpdate, InventoryUpdate
from awx.main.utils import get_ansible_version, decrypt_field, update_scm_url

__all__ = ['RunJob', 'RunProjectUpdate', 'RunInventoryImport']

logger = logging.getLogger('awx.main.tasks')

# FIXME: Cleanly cancel task when celery worker is stopped.

class BaseTask(Task):
    
    name = None
    model = None

    def update_model(self, pk, **updates):
        '''
        Reload model from database and update the given fields.
        '''
        output_replacements = updates.pop('output_replacements', None) or []
        # Commit outstanding transaction so that we fetch the latest object
        # from the database.
        transaction.commit()
        instance = self.model.objects.get(pk=pk)
        if updates:
            update_fields = []
            for field, value in updates.items():
                if field in ('result_stdout', 'result_traceback'):
                    for srch, repl in output_replacements:
                        value = value.replace(srch, repl)
                setattr(instance, field, value)
                update_fields.append(field)
                if field == 'status':
                    update_fields.append('failed')
            instance.save(update_fields=update_fields)
            transaction.commit()
        return instance

    def get_model(self, pk):
        return self.model.objects.get(pk=pk)

    def get_path_to(self, *args):
        '''
        Return absolute path relative to this file.
        '''
        return os.path.abspath(os.path.join(os.path.dirname(__file__), *args))

    def build_private_data(self, instance, **kwargs):
        '''
        Return any private data that needs to be written to a temporary file
        for this task.
        '''

    def build_private_data_file(self, instance, **kwargs):
        '''
        Create a temporary file containing the private data.
        '''
        private_data = self.build_private_data(instance, **kwargs)
        if private_data is not None:
            handle, path = tempfile.mkstemp()
            f = os.fdopen(handle, 'w')
            f.write(private_data)
            f.close()
            os.chmod(path, stat.S_IRUSR|stat.S_IWUSR)
            return path
        else:
            return ''

    def build_passwords(self, instance, **kwargs):
        '''
        Build a dictionary of passwords for responding to prompts.
        '''
        return {
            'yes': 'yes',
            'no': 'no',
            '': '',
        }

    def build_env(self, instance, **kwargs):
        '''
        Build environment dictionary for ansible-playbook.
        '''
        env = dict(os.environ.items())
        # Add ANSIBLE_* settings to the subprocess environment.
        for attr in dir(settings):
            if attr == attr.upper() and attr.startswith('ANSIBLE_'):
                env[attr] = str(getattr(settings, attr))
        # Also set environment variables configured in AWX_TASK_ENV setting.
        for key, value in settings.AWX_TASK_ENV.items():
            env[key] = str(value)
        # Set environment variables needed for inventory and job event
        # callbacks to work.
        env['ANSIBLE_NOCOLOR'] = '1' # Prevent output of escape sequences.
        return env

    def build_safe_env(self, instance, **kwargs):
        hidden_re = re.compile('API|TOKEN|KEY|SECRET|PASS')
        env = self.build_env(instance, **kwargs)
        for k,v in env.items():
            if hidden_re.search(k):
                env[k] = '*'*len(str(v))
        return env

    def args2cmdline(self, *args):
        return ' '.join([pipes.quote(a) for a in args])

    def build_args(self, instance, **kwargs):
        raise NotImplementedError

    def build_safe_args(self, instance, **kwargs):
        return self.build_args(instance, **kwargs)

    def build_cwd(self, instance, **kwargs):
        raise NotImplementedError

    def build_output_replacements(self, instance, **kwargs):
        return []

    def get_idle_timeout(self):
        return None

    def get_password_prompts(self):
        '''
        Return a dictionary of prompt regular expressions and password lookup
        keys.
        '''
        return SortedDict()

    def run_pexpect(self, instance, args, cwd, env, passwords, task_stdout_handle,
                    output_replacements=None):
        '''
        Run the given command using pexpect to capture output and provide
        passwords when requested.
        '''
        status, stdout = 'error', ''
        logfile = task_stdout_handle#cStringIO.StringIO()
        logfile_pos = logfile.tell()
        child = pexpect.spawn(args[0], args[1:], cwd=cwd, env=env)
        child.logfile_read = logfile
        canceled = False
        last_stdout_update = time.time()
        idle_timeout = self.get_idle_timeout()
        expect_list = []
        expect_passwords = {}
        pexpect_timeout = getattr(settings, 'PEXPECT_TIMEOUT', 5)
        for n, item in enumerate(self.get_password_prompts().items()):
            expect_list.append(item[0])
            expect_passwords[n] = passwords.get(item[1], '') or ''
        expect_list.extend([pexpect.TIMEOUT, pexpect.EOF])
        self.update_model(instance.pk, status='running', output_replacements=output_replacements)
        while child.isalive():
            result_id = child.expect(expect_list, timeout=pexpect_timeout)
            #print 'pexpect result_id', result_id, expect_list[result_id], expect_passwords.get(result_id, None)
            if result_id in expect_passwords:
                child.sendline(expect_passwords[result_id])
            if logfile_pos != logfile.tell():
                #old_logfile_pos = logfile_pos
                logfile_pos = logfile.tell()
                #updates['result_stdout'] = logfile.getvalue()
                #task_stdout_handle.write(logfile.getvalue()[old_logfile_pos:logfile_pos])
                #task_stdout_handle.flush()
                last_stdout_update = time.time()
            instance = self.get_model(instance.pk)
            # Commit transaction needed when running unit tests. FIXME: Is it
            # needed or breaks anything for normal operation?
            transaction.commit()
            if instance.cancel_flag:
                child.close(True)<commit_msg>AC-620 Continue to update status while a task is running, so that long running tasks don't get inadvertently canceled.<commit_after># Copyright (c) 2013 AnsibleWorks, Inc.
# All Rights Reserved.

# Python
import ConfigParser
import cStringIO
import datetime
import distutils.version
import json
import logging
import os
import pipes
import re
import subprocess
import stat
import tempfile
import time
import traceback
import urlparse
import uuid

# Pexpect
import pexpect

# Celery
from celery import Task

# Django
from django.conf import settings
from django.db import transaction
from django.utils.datastructures import SortedDict
from django.utils.timezone import now

# AWX
from awx.main.models import Job, ProjectUpdate, InventoryUpdate
from awx.main.utils import get_ansible_version, decrypt_field, update_scm_url

__all__ = ['RunJob', 'RunProjectUpdate', 'RunInventoryImport']

logger = logging.getLogger('awx.main.tasks')

# FIXME: Cleanly cancel task when celery worker is stopped.

class BaseTask(Task):
    
    name = None
    model = None

    def update_model(self, pk, **updates):
        '''
        Reload model from database and update the given fields.
        '''
        output_replacements = updates.pop('output_replacements', None) or []
        # Commit outstanding transaction so that we fetch the latest object
        # from the database.
        transaction.commit()
        instance = self.model.objects.get(pk=pk)
        if updates:
            update_fields = []
            for field, value in updates.items():
                if field in ('result_stdout', 'result_traceback'):
                    for srch, repl in output_replacements:
                        value = value.replace(srch, repl)
                setattr(instance, field, value)
                update_fields.append(field)
                if field == 'status':
                    update_fields.append('failed')
            instance.save(update_fields=update_fields)
            transaction.commit()
        return instance

    def get_model(self, pk):
        return self.model.objects.get(pk=pk)

    def get_path_to(self, *args):
        '''
        Return absolute path relative to this file.
        '''
        return os.path.abspath(os.path.join(os.path.dirname(__file__), *args))

    def build_private_data(self, instance, **kwargs):
        '''
        Return any private data that needs to be written to a temporary file
        for this task.
        '''

    def build_private_data_file(self, instance, **kwargs):
        '''
        Create a temporary file containing the private data.
        '''
        private_data = self.build_private_data(instance, **kwargs)
        if private_data is not None:
            handle, path = tempfile.mkstemp()
            f = os.fdopen(handle, 'w')
            f.write(private_data)
            f.close()
            os.chmod(path, stat.S_IRUSR|stat.S_IWUSR)
            return path
        else:
            return ''

    def build_passwords(self, instance, **kwargs):
        '''
        Build a dictionary of passwords for responding to prompts.
        '''
        return {
            'yes': 'yes',
            'no': 'no',
            '': '',
        }

    def build_env(self, instance, **kwargs):
        '''
        Build environment dictionary for ansible-playbook.
        '''
        env = dict(os.environ.items())
        # Add ANSIBLE_* settings to the subprocess environment.
        for attr in dir(settings):
            if attr == attr.upper() and attr.startswith('ANSIBLE_'):
                env[attr] = str(getattr(settings, attr))
        # Also set environment variables configured in AWX_TASK_ENV setting.
        for key, value in settings.AWX_TASK_ENV.items():
            env[key] = str(value)
        # Set environment variables needed for inventory and job event
        # callbacks to work.
        env['ANSIBLE_NOCOLOR'] = '1' # Prevent output of escape sequences.
        return env

    def build_safe_env(self, instance, **kwargs):
        hidden_re = re.compile('API|TOKEN|KEY|SECRET|PASS')
        env = self.build_env(instance, **kwargs)
        for k,v in env.items():
            if hidden_re.search(k):
                env[k] = '*'*len(str(v))
        return env

    def args2cmdline(self, *args):
        return ' '.join([pipes.quote(a) for a in args])

    def build_args(self, instance, **kwargs):
        raise NotImplementedError

    def build_safe_args(self, instance, **kwargs):
        return self.build_args(instance, **kwargs)

    def build_cwd(self, instance, **kwargs):
        raise NotImplementedError

    def build_output_replacements(self, instance, **kwargs):
        return []

    def get_idle_timeout(self):
        return None

    def get_password_prompts(self):
        '''
        Return a dictionary of prompt regular expressions and password lookup
        keys.
        '''
        return SortedDict()

    def run_pexpect(self, instance, args, cwd, env, passwords, task_stdout_handle,
                    output_replacements=None):
        '''
        Run the given command using pexpect to capture output and provide
        passwords when requested.
        '''
        status, stdout = 'error', ''
        logfile = task_stdout_handle#cStringIO.StringIO()
        logfile_pos = logfile.tell()
        child = pexpect.spawn(args[0], args[1:], cwd=cwd, env=env)
        child.logfile_read = logfile
        canceled = False
        last_stdout_update = time.time()
        idle_timeout = self.get_idle_timeout()
        expect_list = []
        expect_passwords = {}
        pexpect_timeout = getattr(settings, 'PEXPECT_TIMEOUT', 5)
        for n, item in enumerate(self.get_password_prompts().items()):
            expect_list.append(item[0])
            expect_passwords[n] = passwords.get(item[1], '') or ''
        expect_list.extend([pexpect.TIMEOUT, pexpect.EOF])
        self.update_model(instance.pk, status='running', output_replacements=output_replacements)
        while child.isalive():
            result_id = child.expect(expect_list, timeout=pexpect_timeout)
            #print 'pexpect result_id', result_id, expect_list[result_id], expect_passwords.get(result_id, None)
            if result_id in expect_passwords:
                child.sendline(expect_passwords[result_id])
            if logfile_pos != logfile.tell():
                #old_logfile_pos = logfile_pos
                logfile_pos = logfile.tell()
                #updates['result_stdout'] = logfile.getvalue()
                #task_stdout_handle.write(logfile.getvalue()[old_logfile_pos:logfile_pos])
                #task_stdout_handle.flush()
                last_stdout_update = time.time()
            # Update instance status here (also updates modified timestamp, so
            # we have a way to know the task is still running, otherwise the
            # post_run_hook below would cancel long-running tasks that are
            # really still active).
            instance = self.update_model(instance.pk, status='running')
            if instance.cancel_flag:
                child.close(True)
{'fraction_non_alphanumeric': 0.095, 'fraction_numerical': 0.0017857142857142857, 'mean_word_length': 2.3696750902527075, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2176610', 'n_tokens_mistral': 3962, 'n_tokens_neox': 3847, 'n_words': 1053}