<commit_before># Copyright (c) 2013, Web Notes Technologies Pvt. Ltd. and Contributors
# MIT License. See license.txt

from __future__ import unicode_literals
import frappe
import subprocess
from frappe.utils.scheduler import enqueue_events
from frappe.celery_app import get_celery, celery_task, task_logger, LONGJOBS_PREFIX
from frappe.cli import get_sites
from frappe.utils.file_lock import create_lock, delete_lock

@celery_task()
def sync_queues():
	"""notifies workers to monitor newly added sites"""
	app = get_celery()
	shortjob_workers, longjob_workers = get_workers(app)

	if shortjob_workers:
		for worker in shortjob_workers:
			sync_worker(app, worker)

	if longjob_workers:
		for worker in longjob_workers:
			sync_worker(app, worker, prefix=LONGJOBS_PREFIX)

def get_workers(app):
	longjob_workers = []
	shortjob_workers = []

	active_queues = app.control.inspect().active_queues()
	for worker in active_queues:
		if worker.startswith(LONGJOBS_PREFIX):
			longjob_workers.append(worker)
		else:
			shortjob_workers.append(worker)

	return shortjob_workers, longjob_workers

def sync_worker(app, worker, prefix=''):
	active_queues = set(get_active_queues(app, worker))
	required_queues = set(get_required_queues(app, prefix=prefix))
	to_add = required_queues - active_queues
	to_remove = active_queues - required_queues
	for queue in to_add:
		app.control.broadcast('add_consumer', arguments={
				'queue': queue
		}, reply=True, destination=[worker])
	for queue in to_remove:
		app.control.broadcast('cancel_consumer', arguments={
				'queue': queue
		}, reply=True, destination=[worker])


def get_active_queues(app, worker):
	active_queues = app.control.inspect().active_queues()
	if not (active_queues and active_queues.get(worker)):
		return []
	return [queue['name'] for queue in active_queues[worker]]

def get_required_queues(app, prefix=''):
	ret = []
	for site in get_sites():
		ret.append('{}{}'.format(prefix, site))
	ret.append(app.conf['CELERY_DEFAULT_QUEUE'])
	return ret

@celery_task()
def scheduler_task(site, event, handler, now=False):
	from frappe.utils.scheduler import log
	traceback = ""
	task_logger.info('running {handler} for {site} for event: {event}'.format(handler=handler, site=site, event=event))
	try:
		frappe.init(site=site)
		if not create_lock(handler):
			return
		if not now:
			frappe.connect(site=site)
		frappe.get_attr(handler)()

	except Exception:
		frappe.db.rollback()
		traceback = log(handler, "Method: {event}, Handler: {handler}".format(event=event, handler=handler))
		task_logger.warn(traceback)
		raise

	else:
		frappe.db.commit()

	finally:
		delete_lock(handler)

		if not now:
			frappe.destroy()

	task_logger.info('ran {handler} for {site} for event: {event}'.format(handler=handler, site=site, event=event))

@celery_task()
def enqueue_scheduler_events():
	for site in get_sites():
		if site not in ['tailorpad.com', 'testfirst']:
			exec_cmd("../env/bin/frappe --use %s"%(site), cwd = "/home/erpnext/admin_site/frappe-bench/sites")	
			enqueue_events_for_site.delay(site=site)

@celery_task()
def enqueue_events_for_site(site):
	try:
		frappe.init(site=site)
		if frappe.local.conf.maintenance_mode:
			return
		frappe.connect(site=site)
		if site not in ['tailorpad.com', 'testfirst']:
			exec_cmd("../env/bin/frappe --use %s"%(site), cwd = "/home/erpnext/admin_site/frappe-bench/sites")	
			enqueue_events(site)
	finally:
		frappe.destroy()

def exec_cmd(cmd, cwd='.'):
	try:
		subprocess.check_call(cmd, cwd=cwd, shell=True)
	except subprocess.CalledProcessError, e:
		print "Error:", getattr(e, "output", None) or getattr(e, "error", None)
		raise
<commit_msg>new ch<commit_after># Copyright (c) 2013, Web Notes Technologies Pvt. Ltd. and Contributors
# MIT License. See license.txt

from __future__ import unicode_literals
import frappe
import subprocess
from frappe.utils.scheduler import enqueue_events
from frappe.celery_app import get_celery, celery_task, task_logger, LONGJOBS_PREFIX
from frappe.cli import get_sites
from frappe.utils.file_lock import create_lock, delete_lock

@celery_task()
def sync_queues():
	"""notifies workers to monitor newly added sites"""
	app = get_celery()
	shortjob_workers, longjob_workers = get_workers(app)

	if shortjob_workers:
		for worker in shortjob_workers:
			sync_worker(app, worker)

	if longjob_workers:
		for worker in longjob_workers:
			sync_worker(app, worker, prefix=LONGJOBS_PREFIX)

def get_workers(app):
	longjob_workers = []
	shortjob_workers = []

	active_queues = app.control.inspect().active_queues()
	for worker in active_queues:
		if worker.startswith(LONGJOBS_PREFIX):
			longjob_workers.append(worker)
		else:
			shortjob_workers.append(worker)

	return shortjob_workers, longjob_workers

def sync_worker(app, worker, prefix=''):
	active_queues = set(get_active_queues(app, worker))
	required_queues = set(get_required_queues(app, prefix=prefix))
	to_add = required_queues - active_queues
	to_remove = active_queues - required_queues
	for queue in to_add:
		app.control.broadcast('add_consumer', arguments={
				'queue': queue
		}, reply=True, destination=[worker])
	for queue in to_remove:
		app.control.broadcast('cancel_consumer', arguments={
				'queue': queue
		}, reply=True, destination=[worker])


def get_active_queues(app, worker):
	active_queues = app.control.inspect().active_queues()
	if not (active_queues and active_queues.get(worker)):
		return []
	return [queue['name'] for queue in active_queues[worker]]

def get_required_queues(app, prefix=''):
	ret = []
	for site in get_sites():
		ret.append('{}{}'.format(prefix, site))
	ret.append(app.conf['CELERY_DEFAULT_QUEUE'])
	return ret

@celery_task()
def scheduler_task(site, event, handler, now=False):
	from frappe.utils.scheduler import log
	traceback = ""
	task_logger.info('running {handler} for {site} for event: {event}'.format(handler=handler, site=site, event=event))
	try:
		if site not in ['tailorpad.com', 'testfirst']:
			exec_cmd("../env/bin/frappe --use %s"%(site), cwd = "/home/erpnext/admin_site/frappe-bench/sites")
			frappe.init(site=site)
			if not create_lock(handler):
				return
			if not now:
				frappe.connect(site=site)
			frappe.get_attr(handler)()

	except Exception:
		frappe.db.rollback()
		traceback = log(handler, "Method: {event}, Handler: {handler}".format(event=event, handler=handler))
		task_logger.warn(traceback)
		raise

	else:
		frappe.db.commit()

	finally:
		delete_lock(handler)

		if not now:
			frappe.destroy()

	task_logger.info('ran {handler} for {site} for event: {event}'.format(handler=handler, site=site, event=event))

@celery_task()
def enqueue_scheduler_events():
	for site in get_sites():
		if site not in ['tailorpad.com', 'testfirst']:
			exec_cmd("../env/bin/frappe --use %s"%(site), cwd = "/home/erpnext/admin_site/frappe-bench/sites")	
			enqueue_events_for_site.delay(site=site)

@celery_task()
def enqueue_events_for_site(site):
	try:
		frappe.init(site=site)
		if frappe.local.conf.maintenance_mode:
			return
		frappe.connect(site=site)
		if site not in ['tailorpad.com', 'testfirst']:
			exec_cmd("../env/bin/frappe --use %s"%(site), cwd = "/home/erpnext/admin_site/frappe-bench/sites")	
			enqueue_events(site)
	finally:
		frappe.destroy()

def exec_cmd(cmd, cwd='.'):
	try:
		subprocess.check_call(cmd, cwd=cwd, shell=True)
	except subprocess.CalledProcessError, e:
		print "Error:", getattr(e, "output", None) or getattr(e, "error", None)
		raise

{'fraction_non_alphanumeric': 0.12924071082390953, 'fraction_numerical': 0.0010770059235325794, 'mean_word_length': 5.499562554680665, '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': '297157', 'n_tokens_mistral': 2780, 'n_tokens_neox': 2630, 'n_words': 574}