# Path: celeryapp/crawl_tasks.py
# def request(self, dict_item, **kwargs):
# def request_priority(self, dict_item, **kwargs):
# def parse(self, dict_item, **kwargs):
# def parse_priority(self, dict_item, **kwargs):
# def pipeline(self, dict_item, **kwargs):
# def schedule(self, tasks=None, check_task=False, save_filter=False):
# def crawl_synchronous(self, dict_item, **kwargs):
# 
# Path: celeryapp/celery.py
# def get_message_queue_size(queue_name):
#     client = redis.Redis(IP.findall(config.broker)[0])
#     length = client.llen(queue_name)
#     return length


import os
import sys
import config
import celeryconfig
import logging
import gevent
import json
import time
from celeryapp.crawl_tasks import app
from celeryapp.celery import get_message_queue_size
from celery import group
from gevent import monkey
from gevent.pool import Pool
from gevent.lock import Semaphore
    filepath = os.path.join(root_path, 'data/%s' % crawler.get('scheduler').filter_file_name)
    filepath = os.path.abspath(filepath)
    if not os.path.exists(filepath):
        #tasks[crawler['name'] + '.schedule'].delay()
        for task in tasks[crawler['name']+'.schedule'].scheduler.init_generator():
            if task.get('priority', None):
                #app.tasks[self.crawler_name+'.request_priority'].delay(task)
                tasks[crawler['name']+'.request_priority'].apply_async((task, ), compression='zlib')
            else:
                #tasks[crawler['name']+'.request'].delay(task)
                tasks[crawler['name']+'.request'].apply_async((task, ), compression='zlib')


    #for dict_item in crawler['scheduler'].generator():
    #    group(tasks[crawler['name'] + '.request'].s(dict_item) 
    #            | tasks[crawler['name'] + '.parse'].s() 
    #           | tasks[crawler['name'] + '.pipeline'].s()).delay()

monkey.patch_all()

scheduel_lock = Semaphore()

filtered_newtasks = []

def schedule():
    global filtered_newtasks
    while 1:
        try:
            if (get_message_queue_size('parse') < config.max_task_queue_size * 2 and 
                get_message_queue_size('pipeline') < config.max_task_queue_size * 2 and

# Path: celeryapp/crawl_tasks.py
# def request(self, dict_item, **kwargs):
# def request_priority(self, dict_item, **kwargs):
# def parse(self, dict_item, **kwargs):
# def parse_priority(self, dict_item, **kwargs):
# def pipeline(self, dict_item, **kwargs):
# def schedule(self, tasks=None, check_task=False, save_filter=False):
# def crawl_synchronous(self, dict_item, **kwargs):
# 
# Path: celeryapp/celery.py
# def get_message_queue_size(queue_name):
#     client = redis.Redis(IP.findall(config.broker)[0])
#     length = client.llen(queue_name)
#     return length


# Path: bin/run.py
import os
import sys
import config
import celeryconfig
import logging
import gevent
import json
import time
from celeryapp.crawl_tasks import app
from celeryapp.celery import get_message_queue_size
from celery import group
from gevent import monkey
from gevent.pool import Pool
from gevent.lock import Semaphore

    filepath = os.path.join(root_path, 'data/%s' % crawler.get('scheduler').filter_file_name)
    filepath = os.path.abspath(filepath)
    if not os.path.exists(filepath):
        #tasks[crawler['name'] + '.schedule'].delay()
        for task in tasks[crawler['name']+'.schedule'].scheduler.init_generator():
            if task.get('priority', None):
                #app.tasks[self.crawler_name+'.request_priority'].delay(task)
                tasks[crawler['name']+'.request_priority'].apply_async((task, ), compression='zlib')
            else:
                #tasks[crawler['name']+'.request'].delay(task)
                tasks[crawler['name']+'.request'].apply_async((task, ), compression='zlib')


    #for dict_item in crawler['scheduler'].generator():
    #    group(tasks[crawler['name'] + '.request'].s(dict_item) 
    #            | tasks[crawler['name'] + '.parse'].s() 
    #           | tasks[crawler['name'] + '.pipeline'].s()).delay()

monkey.patch_all()

scheduel_lock = Semaphore()

filtered_newtasks = []

def schedule():
    global filtered_newtasks
    while 1:
        try:
            if (get_message_queue_size('parse') < config.max_task_queue_size * 2 and 
                get_message_queue_size('pipeline') < config.max_task_queue_size * 2 and

                get_message_queue_size('schedule') < config.max_task_queue_size * 2 and