<commit_before># Copyright (c) 2008-2009 gocept gmbh & co. kg
# See also LICENSE.txt
"""Publish and retract actions."""

from __future__ import with_statement
from zeit.cms.i18n import MessageFactory as _
import ZODB.POSException
import datetime
import logging
import lovely.remotetask.interfaces
import os.path
import pytz
import random
import subprocess
import threading
import time
import transaction
import zeit.cms.interfaces
import zeit.cms.repository.interfaces
import zeit.cms.workflow.interfaces
import zeit.cms.workingcopy.workingcopy
import zeit.connector.interfaces
import zeit.objectlog.interfaces
import zope.app.appsetup.product
import zope.app.security.interfaces
import zope.component
import zope.event
import zope.interface
import zope.publisher.interfaces
import zope.security.management


logger = logging.getLogger(__name__)
timer_logger = logging.getLogger('zeit.workflow.timer')


PUBLISHED_FUTURE_SHIFT = 60


class Timer(threading.local):

    def start(self, message):
        self.times = []
        self.mark(message)

    def mark(self, message):
        self.times.append((time.time(), message))

    def get_timings(self):
        result = []
        last = None
        total = 0
        for when, message in self.times:
            if last is None:
                diff = 0
            else:
                diff = when - last
            total += diff
            last = when
            result.append((diff, total, message))
        return result

    def __unicode__(self):
        result = []
        for diff, total, message in self.get_timings():
            result.append(u'%2.4f %2.4f %s' % (diff, total, message))
        return u'\n'.join(result)


timer = Timer()


class TaskDescription(object):
    """Data to be passed to publish/retract tasks."""

    def __init__(self, obj):
        self.uniqueId = obj.uniqueId
        self.principal = self.get_principal().id

    @staticmethod
    def get_principal():
        interaction = zope.security.management.getInteraction()
        for p in interaction.participations:
            return p.principal


class Publish(object):

    zope.interface.implements(zeit.cms.workflow.interfaces.IPublish)
    zope.component.adapts(zeit.cms.interfaces.ICMSContent)

    def __init__(self, context):
        self.context = context

    def publish(self):
        """Publish object."""
        info = zeit.cms.workflow.interfaces.IPublishInfo(self.context)
        if not info.can_publish():
            raise zeit.cms.workflow.interfaces.PublishingError(
                "Publish pre-conditions not satisifed.")

        self.log(self.context, _('Publication scheduled'))
        return self.tasks.add(u'zeit.workflow.publish',
                       TaskDescription(self.context))

    def retract(self):
        """Retract object."""
        self.log(self.context, _('Retracting scheduled'))
        return self.tasks.add(u'zeit.workflow.retract',
                       TaskDescription(self.context))

    @property
    def tasks(self):
        return zope.component.getUtility(
            lovely.remotetask.interfaces.ITaskService, 'general')

    def log(self, obj, msg):
        log = zope.component.getUtility(zeit.objectlog.interfaces.IObjectLog)
        log.log(obj, msg)



active_objects = set()
active_objects_lock = threading.Lock()


class PublishRetractTask(object):

    zope.interface.implements(lovely.remotetask.interfaces.ITask)
    #inputSchema = zope.schema.Object()  # XXX
    #outputSchema = None or an error message

    def __call__(self, service, jobid, input):
        timer.start(u'Job %s started: %s (%s)' % (
            type(self).__name__, input.uniqueId, jobid))
        logger.info("Running job %s" % jobid)
        uniqueId = input.uniqueId
        principal = input.principal

        self.login(principal)
        timer.mark('Logged in')

        obj = self.repository.getContent(input.uniqueId)
        info = zeit.cms.workflow.interfaces.IPublishInfo(obj)
        timer.mark('Looked up object')

        retries = 0
        while True:
            try:
                acquired = self.acquire_active_lock(uniqueId)
                if acquired:
                    self.run(obj, info)
                else:
                    self.log(
                        obj, _('A publish/retract job is already active.'
                               ' Aborting'))
                    logger.info("Aborting parallel publish/retract of %r" % (
                        uniqueId))
                transaction.commit()
                timer.mark('Commited')
            except ZODB.POSException.ConflictError, e:
                timer.mark('Conflict')
                retries += 1
                if retries >= 3:
                    raise
                # Spiels noch einmal, Sam.
                logger.exception(e)
                transaction.abort()
                # Stagger retry:
                time.sleep(random.uniform(0, 2**(retries)))
            except Exception, e:
                logger.error("Error during publish/retract", exc_info=True)
                message = _("Error during publish/retract: ${exc}: ${message}",
                            mapping=dict(
                                exc=e.__class__.__name__,
                                message=str(e)))
                self.log(obj, message)
                break
            else:
                # Everything okay.
                message = None
                break
            finally:
                if acquired:
                    self.release_active_lock(uniqueId)
        timer.mark('Done %s' % input.uniqueId)
        timer_logger.debug('Timings:\n%s' % (unicode(timer).encode('utf8'),))
        dummy, total, message = timer.get_timings()[-1]
        logger.info('%s (%2.4fs)' % (message, total))
        return message

    def acquire_active_lock(self, uniqueId):
        with active_objects_lock:
            if uniqueId in active_objects:
                return False
            active_objects.add(uniqueId)
            return True

    def release_active_lock(self, uniqueId):
        with active_objects_lock:
            active_objects.remove(uniqueId)

    def cycle(self, obj):
        """checkout/checkin obj to sync data as necessary.

        The basic idea is that there are some event handlers which sync
        properties to xml on checkout/checkin.<commit_msg>fix workflow return value<commit_after># Copyright (c) 2008-2009 gocept gmbh & co. kg
# See also LICENSE.txt
"""Publish and retract actions."""

from __future__ import with_statement
from zeit.cms.i18n import MessageFactory as _
import ZODB.POSException
import datetime
import logging
import lovely.remotetask.interfaces
import os.path
import pytz
import random
import subprocess
import threading
import time
import transaction
import zeit.cms.interfaces
import zeit.cms.repository.interfaces
import zeit.cms.workflow.interfaces
import zeit.cms.workingcopy.workingcopy
import zeit.connector.interfaces
import zeit.objectlog.interfaces
import zope.app.appsetup.product
import zope.app.security.interfaces
import zope.component
import zope.event
import zope.interface
import zope.publisher.interfaces
import zope.security.management


logger = logging.getLogger(__name__)
timer_logger = logging.getLogger('zeit.workflow.timer')


PUBLISHED_FUTURE_SHIFT = 60


class Timer(threading.local):

    def start(self, message):
        self.times = []
        self.mark(message)

    def mark(self, message):
        self.times.append((time.time(), message))

    def get_timings(self):
        result = []
        last = None
        total = 0
        for when, message in self.times:
            if last is None:
                diff = 0
            else:
                diff = when - last
            total += diff
            last = when
            result.append((diff, total, message))
        return result

    def __unicode__(self):
        result = []
        for diff, total, message in self.get_timings():
            result.append(u'%2.4f %2.4f %s' % (diff, total, message))
        return u'\n'.join(result)


timer = Timer()


class TaskDescription(object):
    """Data to be passed to publish/retract tasks."""

    def __init__(self, obj):
        self.uniqueId = obj.uniqueId
        self.principal = self.get_principal().id

    @staticmethod
    def get_principal():
        interaction = zope.security.management.getInteraction()
        for p in interaction.participations:
            return p.principal


class Publish(object):

    zope.interface.implements(zeit.cms.workflow.interfaces.IPublish)
    zope.component.adapts(zeit.cms.interfaces.ICMSContent)

    def __init__(self, context):
        self.context = context

    def publish(self):
        """Publish object."""
        info = zeit.cms.workflow.interfaces.IPublishInfo(self.context)
        if not info.can_publish():
            raise zeit.cms.workflow.interfaces.PublishingError(
                "Publish pre-conditions not satisifed.")

        self.log(self.context, _('Publication scheduled'))
        return self.tasks.add(u'zeit.workflow.publish',
                       TaskDescription(self.context))

    def retract(self):
        """Retract object."""
        self.log(self.context, _('Retracting scheduled'))
        return self.tasks.add(u'zeit.workflow.retract',
                       TaskDescription(self.context))

    @property
    def tasks(self):
        return zope.component.getUtility(
            lovely.remotetask.interfaces.ITaskService, 'general')

    def log(self, obj, msg):
        log = zope.component.getUtility(zeit.objectlog.interfaces.IObjectLog)
        log.log(obj, msg)



active_objects = set()
active_objects_lock = threading.Lock()


class PublishRetractTask(object):

    zope.interface.implements(lovely.remotetask.interfaces.ITask)
    #inputSchema = zope.schema.Object()  # XXX
    #outputSchema = None or an error message

    def __call__(self, service, jobid, input):
        timer.start(u'Job %s started: %s (%s)' % (
            type(self).__name__, input.uniqueId, jobid))
        logger.info("Running job %s" % jobid)
        uniqueId = input.uniqueId
        principal = input.principal

        self.login(principal)
        timer.mark('Logged in')

        obj = self.repository.getContent(input.uniqueId)
        info = zeit.cms.workflow.interfaces.IPublishInfo(obj)
        timer.mark('Looked up object')

        retries = 0
        while True:
            try:
                acquired = self.acquire_active_lock(uniqueId)
                if acquired:
                    self.run(obj, info)
                else:
                    self.log(
                        obj, _('A publish/retract job is already active.'
                               ' Aborting'))
                    logger.info("Aborting parallel publish/retract of %r" % (
                        uniqueId))
                transaction.commit()
                timer.mark('Commited')
            except ZODB.POSException.ConflictError, e:
                timer.mark('Conflict')
                retries += 1
                if retries >= 3:
                    raise
                # Spiels noch einmal, Sam.
                logger.exception(e)
                transaction.abort()
                # Stagger retry:
                time.sleep(random.uniform(0, 2**(retries)))
            except Exception, e:
                logger.error("Error during publish/retract", exc_info=True)
                message = _("Error during publish/retract: ${exc}: ${message}",
                            mapping=dict(
                                exc=e.__class__.__name__,
                                message=str(e)))
                self.log(obj, message)
                break
            else:
                # Everything okay.
                message = None
                break
            finally:
                if acquired:
                    self.release_active_lock(uniqueId)
        timer.mark('Done %s' % input.uniqueId)
        timer_logger.debug('Timings:\n%s' % (unicode(timer).encode('utf8'),))
        dummy, total, timer_message = timer.get_timings()[-1]
        logger.info('%s (%2.4fs)' % (timer_message, total))
        return message

    def acquire_active_lock(self, uniqueId):
        with active_objects_lock:
            if uniqueId in active_objects:
                return False
            active_objects.add(uniqueId)
            return True

    def release_active_lock(self, uniqueId):
        with active_objects_lock:
            active_objects.remove(uniqueId)

    def cycle(self, obj):
        """checkout/checkin obj to sync data as necessary.

        The basic idea is that there are some event handlers which sync
        properties to xml on checkout/checkin.
{'fraction_non_alphanumeric': 0.09787367104440275, 'fraction_numerical': 0.004221388367729831, 'mean_word_length': 2.0300805305542395, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '216211', 'n_tokens_mistral': 3674, 'n_tokens_neox': 3470, 'n_words': 820}