<commit_before># -*- coding: utf-8 -*-

from datetime import datetime
import logging
import os
import sys
from signal import signal, SIGINT, SIGTERM

from apscheduler.schedulers import SchedulerNotRunningError  # type: ignore
from apscheduler.triggers.date import DateTrigger  # type: ignore
from fasteners import InterProcessLock  # type: ignore

from deck_chores import __version__
from deck_chores.caches import get_image_labels_for_container
from deck_chores.config import cfg, generate_config
from deck_chores.exceptions import ConfigurationError
from deck_chores.indexes import locking_container_to_services_map
from deck_chores import jobs
import deck_chores.parsers as parse
from deck_chores.utils import from_json, generate_id, trueish


####


lock = InterProcessLock('/tmp/deck-chores.lock')


def there_is_another_deck_chores_container() -> bool:
    matched_containers = 0
    for container in cfg.client.containers():
        if container['State'] in 'created,exited':
            continue
        labels = get_image_labels_for_container(container['Id'])
        if labels.get('org.label-schema.name', '') == 'deck-chores':
            matched_containers += 1
        if matched_containers > 1:
            return True
    return False


####


def sigint_handler(signum, frame):
    log.info('Keyboard interrupt.')
    raise SystemExit(0)


def sigterm_handler(signum, frame):
    log.info('Received SIGTERM.')
    raise SystemExit(0)


signal(SIGINT, sigint_handler)
signal(SIGTERM, sigterm_handler)


####


log = logging.getLogger('deck_chores')
log_handler = logging.StreamHandler(sys.stdout)
log.addHandler(log_handler)
log.setLevel(logging.DEBUG if trueish(os.getenv('DEBUG', 'no')) else logging.INFO)


####


def process_running_container_labels(container_id: str) -> None:
    service_id, options, definitions = parse.labels(container_id)
    if not definitions:
        return
    if service_id and 'service' in options:
        if locking_container_to_services_map.values():
            log.debug('Service id has a registered job: %s' % service_id)
            return
        log.info('Locking service id: %s' % service_id)
        locking_container_to_services_map[container_id] = service_id
    jobs.add(container_id, definitions)


def inspect_running_containers() -> datetime:
    log.debug('Fetching running containers')
    containers = cfg.client.containers(filters={'status': 'running'})
    inspection_time = datetime.utcnow()
    jobs.scheduler.add_job(exec_inspection, trigger=DateTrigger(), args=(containers,),
                           id='container_inspection')
    return inspection_time


def exec_inspection(containers: dict) -> None:
    log.info('Inspecting running containers.')
    for container in containers:
        process_running_container_labels(container['Id'])


def listen(since: datetime = datetime.utcnow()) -> None:
    log.info('Listening to events.')
    for event_json in cfg.client.events(since=since):
        if b'container' not in event_json:
            continue
        if not any((x in event_json) for x in (b'start', b'die', b'pause', b'unpause')):<commit_msg>fix check on locked services<commit_after># -*- coding: utf-8 -*-

from datetime import datetime
import logging
import os
import sys
from signal import signal, SIGINT, SIGTERM

from apscheduler.schedulers import SchedulerNotRunningError  # type: ignore
from apscheduler.triggers.date import DateTrigger  # type: ignore
from fasteners import InterProcessLock  # type: ignore

from deck_chores import __version__
from deck_chores.caches import get_image_labels_for_container
from deck_chores.config import cfg, generate_config
from deck_chores.exceptions import ConfigurationError
from deck_chores.indexes import locking_container_to_services_map
from deck_chores import jobs
import deck_chores.parsers as parse
from deck_chores.utils import from_json, generate_id, trueish


####


lock = InterProcessLock('/tmp/deck-chores.lock')


def there_is_another_deck_chores_container() -> bool:
    matched_containers = 0
    for container in cfg.client.containers():
        if container['State'] in 'created,exited':
            continue
        labels = get_image_labels_for_container(container['Id'])
        if labels.get('org.label-schema.name', '') == 'deck-chores':
            matched_containers += 1
        if matched_containers > 1:
            return True
    return False


####


def sigint_handler(signum, frame):
    log.info('Keyboard interrupt.')
    raise SystemExit(0)


def sigterm_handler(signum, frame):
    log.info('Received SIGTERM.')
    raise SystemExit(0)


signal(SIGINT, sigint_handler)
signal(SIGTERM, sigterm_handler)


####


log = logging.getLogger('deck_chores')
log_handler = logging.StreamHandler(sys.stdout)
log.addHandler(log_handler)
log.setLevel(logging.DEBUG if trueish(os.getenv('DEBUG', 'no')) else logging.INFO)


####


def process_running_container_labels(container_id: str) -> None:
    service_id, options, definitions = parse.labels(container_id)
    if not definitions:
        return
    if service_id and 'service' in options:
        if service_id in locking_container_to_services_map.values():
            log.debug('Service id has a registered job: %s' % service_id)
            return
        log.info('Locking service id: %s' % service_id)
        locking_container_to_services_map[container_id] = service_id
    jobs.add(container_id, definitions)


def inspect_running_containers() -> datetime:
    log.debug('Fetching running containers')
    containers = cfg.client.containers(filters={'status': 'running'})
    inspection_time = datetime.utcnow()
    jobs.scheduler.add_job(exec_inspection, trigger=DateTrigger(), args=(containers,),
                           id='container_inspection')
    return inspection_time


def exec_inspection(containers: dict) -> None:
    log.info('Inspecting running containers.')
    for container in containers:
        process_running_container_labels(container['Id'])


def listen(since: datetime = datetime.utcnow()) -> None:
    log.info('Listening to events.')
    for event_json in cfg.client.events(since=since):
        if b'container' not in event_json:
            continue
        if not any((x in event_json) for x in (b'start', b'die', b'pause', b'unpause')):
{'fraction_non_alphanumeric': 0.09918931807343824, 'fraction_numerical': 0.0019074868860276585, 'mean_word_length': 4.061946902654867, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 15, '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': '2598093', 'n_tokens_mistral': 1871, 'n_tokens_neox': 1842, 'n_words': 461}