def process_task(row):
    from uliweb import settings

    Task = functions.get_model('async_tasks')

    now = date.now()
    #check started status
    if row.status == STARTED:
        #check timeout
        if row.started_time + datetime.timedelta(milliseconds=row.timeout) > now:
            return
        else:
            #检查任务是否还在执行
            keys_queue = settings.get_var('ASYNC_TASKS/tasks_keys_queue')
            redis = functions.get_redis()
            if redis.sismember(keys_queue, row.task_id):
                return
            row.startup_time = now

    handler = row.get_handler(log=log)
    #check depend task
    depend_task = row.check_depend_tasks(log)
    if depend_task:
        #依赖任务或父任务为取消时,当前任务取消
        if depend_task.status == CANCEL or depend_task.current_status == CANCEL:
            msg = "Depend task {} has been cancelled.".format(depend_task.task_id)
            row.cancel(msg, log=log)
            log.debug("Current task {} cancelled because depend task {} status is CANCEL".format(
                row.task_id, depend_task.task_id
            ))
        #如果依赖任务失败,当前任务也置为失败
        elif depend_task.status == FAILED:
            msg = "Current task {} FAILED because depend task {} status is FAILED".format(
                row.task_id, depend_task.task_id
            )
            handler.save(status=FAILED, finished_time=date.now(), message=msg)
            log.info(msg)
        return
    if row.retry_times >= row.max_retry_times:
        msg = ('Async task {} reaches max retry times, '
                      'status changes to FAILED').format(row.task_id)
        handler.save(status=FAILED, finished_time=date.now(), message=msg)
        log.info(msg)
        return
    if row.startup_time and row.startup_time<=now or not row.startup_time:
        #处理父结点,如果current_status为成功,则不执行
        if row.children_count > 0 and row.current_status == SUCCESS:
            return
        handler.save(status=STARTED, started_time=date.now(),
                     retry_times=row.retry_times+1)
        log.info('Async task {0} [{1}({2!r})] started, retry_times={3}'.format(
            row.task_id,
            row.command_name,
            row.command_info.get('parameters', ''),
            row.retry_times))
        run_command(row)

        def post_save(obj, data):
            from uliweb import functions
            from uliweb.utils.common import Serial
            from uliweb.mail import Mail

            Post.filter(Post.c.id == int(parent_id)).update(
                num_replies=Post.c.num_replies + 1, last_post_user=request.user.id, last_reply_on=date.now()
            )
            self._clear_files(obj.slug, data["content"])

            Topic.filter(Topic.c.id == int(topic_id)).update(
                num_replies=Topic.c.num_replies + 1,
                last_post_user=request.user.id,
                last_reply_on=date.now(),
                last_post=obj.id,
            )
            Forum.filter(Forum.c.id == int(forum_id)).update(
                num_posts=Forum.c.num_posts + 1,
                last_post_user=request.user.id,
                last_reply_on=date.now(),
                last_post=obj.id,
            )

            # 増加发送邮件的处理
            emails = []
            for u_id in Post.filter(
                (Post.c.topic == int(topic_id)) & (Post.c.reply_email == True) & (Post.c.id == parent_id)
            ).values(Post.c.posted_by):
                user = User.get(u_id[0])
                if user and user.email and (user.email not in emails) and (user.email != request.user.email):
                    emails.append(user.email)

            if not emails:
                return

            _type = settings.get_var("PARA/FORUM_REPLY_PROCESS", "print")
            url = "%s/forum/%s/%s" % (settings.get_var("PARA/DOMAIN"), forum_id, topic_id)
            d = {"url": str(url)}
            mail = {
                "from_": settings.get_var("PARA/EMAIL_SENDER"),
                "to_": emails,
                "subject": settings.get_var("FORUM_EMAIL/FORUM_EMAIL_TITLE"),
                "message": settings.get_var("FORUM_EMAIL/FORUM_EMAIL_TEXT") % d,
                "html": True,
            }

            if _type == "mail":
                Mail().send_mail(**mail)
            elif _type == "print":
                print mail
            elif _type == "redis":
                redis = functions.get_redis()
                _t = Serial.dump(mail)
                redis.lpush("send_mails", _t)

    def run(self):
        import traceback

        Task = functions.get_model('async_Tasks')

        try:
            self.before()

            if not self.func:
                self.ret = 'empty'
                self.log.debug("Async task {} function is empty".format(self.task_id))
            else:
                f = import_attr(self.func)
                self.log.debug("Run Async task {} function {!r}({!r})".format(
                    self.task_id, f, self.parameters))
                #_args表示位置参数
                args = self.parameters.pop('_args', ())
                self.ret = f(*args, **self.parameters)
                self.log.debug("Execute {!r} with {!r} and the result is {!r}".format(
                    f, self.parameters, self.ret))

            self.on_finish()

            self.after()

            self.save()

        except Exception as e:
            type, value, tb = sys.exc_info()
            txt =  ''.join(traceback.format_exception(type, value, tb))
            if self.task.retry_times >= self.task.max_retry_times:
                status = FAILED
                finished_time = date.now()
                startup_time = self.task.startup_time
                msg =  'Async task {} reaches max retry times, status changes to FAILED'.format(self.task.task_id)
                self.log.info(msg)
            else:
                status = ERROR
                startup_time = (date.now() +
                    datetime.timedelta(milliseconds=self.task.retry_time_interval*(self.task.retry_times-1)))
                finished_time = None
                msg =  'Async task {} runs on error, it will startup at {}'.format(self.task.task_id, startup_time)
                self.log.info(msg)

            self.save(execution_info=txt,
                      finished_time=finished_time,
                      startup_time=startup_time,
                      status=status,
                      message=msg)
            self.log.exception(e)

    def commit(self, session=None):
        Task = functions.get_model('async_tasks')
        if session:
            Task = Task.use(session)
        if self.parent_task and isinstance(self.parent_task, AsyncCommand):
            parent_task_id = self.parent_task.task_id
        else:
            parent_task_id = self.parent_task
        task = Task(task_id=self.task_id,
                    title=self.title,
                    parent_task=parent_task_id,
                    children_count=len(self.tasks),
                    command_name=self.name,
                    command_info=self._make_info(),
                    depend_tasks=self.depend_tasks,
                    startup_time=self.startup_time or date.now(),
                    timeout=self.timeout,
                    max_retry_times=self.max_retry_times,
                    retry_time_interval=self.retry_time_interval,
                    user_id=self.user_id,
                    src_ip=self.src_ip,
                    correlation=self.correlation,
                    correlation_link=self.correlation_link)
        task.update(**self.kwargs)
        task.save()
        self.task = task

        #commit child tasks
        for t in self.tasks.values():
            t.commit()

        log.info('Commit a task %s' % self.task_id)
        return task

 def post_save(obj, data):
     #更新Post表
     post.content = data['content']
     post.updated_on = date.now()
     post.save()
     
     self._clear_files(obj.slug, data['content'])

 def output():
     while self.process.poll() is None:
         line = self.process.stdout.readline().rstrip()
         self.process.timestamp = now()
         self.output('return', self.server.safe_encode(line))
     self.output('cwd', self.server.safe_encode(self.old_cwd))
     self.status = 1 #finished

    def api_login(self):
        lpsust = request.values.get(settings.AUTH_LENOVOID.LENOVOID_WUST_NAME)
        rememberme = request.values.get("rememberme")
        if rememberme:
            rememberme = (rememberme.lower()=="true") or (rememberme=='1')
        if lpsust:
            f,d = authenticate(lpsust=lpsust)
            if f:
                from uliweb.utils.date import now
                user = d
                user.last_login = now()
                user.save()
                request.user = user
                session = functions.get_session()
                session[settings.AUTH_LENOVOID.SESSION_KEY_USER] = user.id
                if session.deleted:
                    session.delete()
                else:
                    if rememberme:
                        timeout = settings.SESSION.remember_me_timeout
                        session.set_expiry(timeout)
                    else:
                        timeout = settings.SESSION.timeout
                    flag = session.save()
                    return json({
                        settings.AUTH_LENOVOID.TOKEN_NAME: session.key,
                        "timeout":timeout,
                        }
                    )
            else:
                return json({"error_message": d.get("error_message")}, status = d.get("error_code"))

        return json({"error_message":"Fail to log in."}, status = 400)

    def clean_dir(self, dir, extensions, exclude_extensions, recursion,
                  days, pattern, verbose=False, test=False):
        from uliweb.utils.common import walk_dirs
        import datetime
        from uliweb.utils import date
        import shutil

        now = date.now()
        i = 0
        for f in walk_dirs(dir, include_ext=extensions, exclude_ext=exclude_extensions,
                           recursion=recursion, file_only=False,
                           use_default_pattern=False, patterns=pattern):
            t = datetime.datetime.fromtimestamp(os.path.getmtime(f))
            if not days or (days and (now-t).days >= days):
                try:
                    if os.path.isfile(f):
                        if not test:
                            os.unlink(f)
                        if test or verbose:
                            print 'Clean filename {}...'.format(f)
                    else:
                        if not test:
                            shutil.rmtree(f)
                        if test or verbose:
                            print 'Clean directory {}...'.format(f)

                    i += 1
                except:
                    import traceback
                    traceback.print_exc()
        print 'Cleaned {} files'.format(i)

 def on_finish(self):
     """
     在处理成功后执行此处理,如果返回True,表示不执行缺省处理,否则按缺省处理进行
     :param ret: 命令执行结果
     :return:
     """
     #增加对返回值的处理,当self.ret返回为AsyncCommandList时,将当前任务置为
     #父结点,增加关联的子结点
     #动态生成的父任务
     if isinstance(self.ret, AsyncCommandList):
         self.ret.commit(self.task)
         self.task.children_count = len(self.ret.tasks)
         self.task.current_status = SUCCESS
         self.task.status = STARTED
         self.task.execution_info=safe_str(self.ret.result)
     #处理父任务
     elif self.task.children_count > 0:
         self.task.current_status = SUCCESS
         self.task.status = STARTED
         self.task.execution_info=safe_str(self.ret)
     #一般任务
     else:
         self.task.status = SUCCESS
         self.task.finished_time = date.now()
         self.task.execution_info = safe_str(self.ret)

 def parse_function(self, f):
     args = inspect.getargspec(f)[0]
     if args:
         args = ['<%s>' % x for x in args]
     if f.__name__ in reserved_keys:
         raise ReservedKeyError, 'The name "%s" is a reversed key, so please change another one' % f.__name__
     appname, path = self._get_path(f)
     if self.rule is None:
         if self.restful:
             rule = '/' + '/'.join([path] + args[:1] + [f.__name__] + args[1:])
         else:
             rule = '/' + '/'.join([path, f.__name__] + args)
     else:
         rule = self.rule
     rule = self._fix_url(appname, rule)
     if inspect.ismethod(f):
         endpoint = '.'.join([f.im_class.__module__, f.im_class.__name__, f.__name__])
     else:
         endpoint = '.'.join([f.__module__, f.__name__])
     f.func_dict['__exposed__'] = True
     f.func_dict['__no_rule__'] = (self.parse_level == 1) or (self.parse_level == 2 and (self.rule is None))
     if not hasattr(f, '__old_rule__'):
         f.func_dict['__old_rule__'] = {}
 
     f.func_dict['__old_rule__'][rule] = self.rule
     #add name parameter process
     if 'name' in self.kwargs:
         url_name = self.kwargs.pop('name')
         __url_names__[url_name] = endpoint
     return f, (appname, endpoint, rule, self.kwargs.copy(), now())

 def delete_chapter(self, id):
     """
     删除章节
     删除时，如果有子结点，则子结点的父结点应变成当前结点的父结点
     """
     obj = self.model_chapters.get_or_notfound(int(id))
     tutorial = obj.tutorial
     count = obj.comments_count
     parent = obj._parent_
     
     #修改所有子结点的父结点
     obj.children_chapters.update(parent=parent)
     
     #删除章节顺序cache
     cache = functions.get_cache()
     cache.delete(self._get_tutorial_chapters_cache_key(obj._tutorial_))
     
     #删除当前章节
     obj.deleted = True
     obj.modified_user = request.user.id
     obj.modified_date = date.now()
     obj.save()
     
     #删除所属教程的评论数目
     tutorial.comments_count = max(0, tutorial.comments_count-count)
     tutorial.save()
     
     #跳转回教程展示页面
     return redirect(url_for(TutorialView.read, id=tutorial.id))

def start_task(task_id, now=None):
    from uliweb.utils import date

    now =  now or date.now()

    try:
        task = functions.get_object('cron_task', task_id)
        log.info(u'Starting task [{title}] at {now}'.format(title=task.label, now=now))

        d = parse_command(task)

        c = functions.AsyncCommand(
                            title=task.label,
                            category='cron',
                            message_source='l',
                            correlation=task.id,
                            queue=(task.queue or 'default').split(','),
                            timeout=task.timeout or None,
                            # correlation_link='/async_task/view/{0}?next=/cron/{1}'.format(job.id, obj.id)
                            **d
                            )


        functions.call_async(c)
        return c
    except Exception as e:
        log.exception(e)
        raise

def run_command(row):
    import json
    from uliweb import settings

    keys_queue = settings.get_var('ASYNC_TASKS/tasks_keys_queue')

    kw = row.command_info.copy()
    kw['task_id'] = row.task_id
    queue =kw['queue']
    try:
        redis = functions.get_redis()
        if not isinstance(queue, (tuple, list)):
            queue = [queue]
        msg = json.dumps(kw)
        #在添加消息到公共队列的同时,将任务ID添加到keys_queue中,用来记录当前将要,或正在执行的任务
        for q in queue:
            pipe = redis.pipeline()
            name = settings.ASYNC_TASKS_QUEUES.get(q or 'default') or q
            pipe.lpush(name, msg).sadd(keys_queue, row.task_id)
            pipe.execute()
            log.info('Put task_id {} to queue=[{}] and keys_queue=[{}]'.format(row.task_id, name, keys_queue))
    except Exception as e:
        _type, value, tb = sys.exc_info()
        txt =  ''.join(traceback.format_exception(_type, value, tb))
        handler = row.get_handler(log=log)
        handler.save(status=ERROR, execution_info=txt, finished_time=date.now())
        log.error('Run async task {} failed when push with redis'.format(row.task_id))
        log.exception(e)

    def enroll(self):
        """
        报名处理
        """
        
        issue = int(request.POST.get('issue'))
        class_id = int(request.POST.get('class_id'))
        obj = self.model_issue.get(
            and_(self.model_issue.c.class_obj==class_id, 
                self.model_issue.c.issue==issue), for_update=True)

        row = self._get_student(obj)
        if row and not row.deleted:
            return json({'success':False, 'message':'你已经报过名了！'})
        
        if obj.students_num < obj.need_num:
            obj.students_num += 1
            obj.save()
            if not row:
                row = self.model_study(
                    issue=issue,
                    class_obj=class_id,
                    student=request.user.id,
                )
            else:
                row.deleted = False
                row.create_date = now()
            row.save()
            d = obj.to_dict()
            d['enrolled'] = True
            return json({'success':True, 'message':'报名成功！', 'data':d})
        else:
            return json({'success':False, 'message':'对不起，报名人数已经满了，请等下期再报名！'})

def get_exe_time(job):
    import croniter
    from uliweb import settings

    now = date.now(settings.CRON.TIMEZONE)

    cron = croniter.croniter(job.time, now)
    return cron.get_next()

 def post_save(obj, data):
     obj.tutorial.modified_date = date.now()
     obj.tutorial.modified_user = request.user.id
     obj.tutorial.save();
     
     #删除章节顺序cache
     cache = functions.get_cache()
     cache.delete(self._get_tutorial_chapters_cache_key(int(t_id)))

    def _upload_file(self, image=False):
        import os
        import Image
        from uliweb.contrib.upload import get_url, save_file, get_filename

        if image:
            from forms import ImageUploadForm as Form
        else:
            from forms import FileUploadForm as Form
        from uliweb.utils.image import thumbnail_image, fix_filename

        File = get_model("forumattachment")

        forum_id = request.GET.get("forum_id")
        slug = request.GET.get("slug")
        form = Form()
        suffix = date.now().strftime("_%Y_%m_%d")
        if request.method == "GET":
            form.bind({"is_thumbnail": True})
            return {"form": form}
        else:
            flag = form.validate(request.values, request.files)
            if flag:
                f = form.data["filedata"]
                _f = os.path.basename(f["filename"])
                # 文件格式为forum/<forum_id>/<filename_yyyy_mm_dd>
                filename = fix_filename("forum/%s/%s" % (forum_id, _f), suffix)
                if image:
                    filename = save_file(filename, f["file"])
                    if form.data["is_thumbnail"]:
                        # process thumbnail
                        rfilename, thumbnail = thumbnail_image(
                            get_filename(filename, filesystem=True),
                            filename,
                            settings.get_var("PARA/FORUM_THUMBNAIL_SIZE"),
                        )
                        _file = get_url(thumbnail)
                    else:
                        _file = get_url(filename)
                    name = get_url(filename)
                else:
                    filename = save_file(filename, f["file"])
                    _file = get_url(filename)
                    name = form.data["title"]
                    if not name:
                        name = _f
                ff = File(slug=slug, file_name=filename, name=name)
                ff.save()
                return """<script type="text/javascript">
var url='%s||%s';
setTimeout(function(){callback(url);},100);
</script>
""" % (
                    _file,
                    name,
                )
            else:
                return {"form": form}

 def post_save(obj, data):
     from uliweb import functions
     from uliweb.utils.common import Serial
     from uliweb.mail import Mail
     
     Post.filter(Post.c.id==int(parent_id)).update(num_replies=Post.c.num_replies+1, last_post_user=request.user.id, last_reply_on=date.now())            
     self._clear_files(obj.slug, data['content'])
     
     Topic.filter(Topic.c.id==int(topic_id)).update(
         num_replies=Topic.c.num_replies+1, 
         last_post_user=request.user.id, 
         last_reply_on=date.now(),
         last_post=obj.id)
     Forum.filter(Forum.c.id==int(forum_id)).update(
         num_posts=Forum.c.num_posts+1, 
         last_post_user=request.user.id, 
         last_reply_on=date.now(),
         last_post=obj.id)
     
     #増加发送邮件的处理
     emails = []
     for u_id in Post.filter((Post.c.topic==int(topic_id)) & (Post.c.reply_email==True) & (Post.c.id==parent_id)).values(Post.c.posted_by):
         user = User.get(u_id[0])
         if user and user.email and (user.email not in emails) and (user.email!=request.user.email):
             emails.append(user.email)
     
     if not emails:
         return
     
     _type = settings.get_var('PARA/FORUM_REPLY_PROCESS', 'print')
     url = '%s/%s/%s' % (settings.get_var('PARA/DOMAIN'), forum_id, topic_id)
     d = {'url':str(url)}
     mail = {'from_':settings.get_var('PARA/EMAIL_SENDER'), 'to_':emails,
         'subject':settings.get_var('FORUM_EMAIL/FORUM_EMAIL_TITLE'),
         'message':settings.get_var('FORUM_EMAIL/FORUM_EMAIL_TEXT') % d,
         'html':True}
     
     if _type == 'mail':
         Mail().send_mail(**mail)
     elif _type == 'print':
         print mail
     elif _type == 'redis':
         redis = functions.get_redis()
         _t = Serial.dump(mail)
         redis.lpush('send_mails', _t)

    def _upload_file(self, image=False, show_filename=True):
        import os
        import Image
        if image:
            from forms import ImageUploadForm as Form
        else:
            from forms import FileUploadForm as Form
        from uliweb.utils.image import thumbnail_image, fix_filename
        from uliweb import json_dumps
        
        File = get_model('forumattachment')
        
        forum_id = request.GET.get('forum_id')
        slug = request.GET.get('slug')
        form = Form()
        suffix = date.now().strftime('_%Y_%m_%d')
        if request.method == 'GET':
            form.bind({'is_thumbnail':True})
            return {'form':form}
        else:
            flag = form.validate(request.values, request.files)
            if flag:
                f = form.data['filedata']
                _f = os.path.basename(f['filename'])
                #文件格式为<forum_id>/<filename_yyyy_mm_dd>
                filename = fix_filename('%s/%s' % (forum_id, _f), suffix)
                if image:
                    filename = functions.save_file(filename, f['file'])
                    if form.data['is_thumbnail']:
                        #process thumbnail
                        rfilename, thumbnail = thumbnail_image(functions.get_filename(filename, filesystem=True), filename, settings.get_var('PARA/FORUM_THUMBNAIL_SIZE'))
                        _file = functions.get_href(thumbnail)
                    else:
                        _file = functions.get_href(filename)
                    name = functions.get_href(filename)
                else:
                    filename = functions.save_file(filename, f['file'])
                    name = form.data['title']
                    if not name:
                        name = _f
                    _file = functions.get_href(filename, alt=name)
                ff = File(slug=slug, file_name=filename, name=name)
                ff.save()
                name = json_dumps(name, unicode=True)
                if show_filename:
                    fargs = '||%s' % name[1:-1]
                else:
                    fargs = ''
                return '''<script type="text/javascript">
var url='%s%s';
setTimeout(function(){callback(url);},100);
</script>
''' % (_file, fargs)
            else:
                return {'form':form}

def start_job(job, now=None):
    from uliweb.utils import date

    now =  now or date.now()

    log.info(u'Starting job [{title} -- {time}] at {now}'.format(title=job.title, now=now,
                                                           time=job.time))
    Task = functions.get_model('cron_task')

    Begin()
    try:
        parent = functions.AsyncCommand('', title=job.title, startup_time=now,
                                       category='cron', message_source='l',
                                       timeout=job.timeout,
                                       #todo add correlation to job
                                       )
        commands = []
        ids = {}

        #process task
        for row in Task.filter(Task.c.cron_job==job.id):
            #process async task
            d = parse_command(row)

            c = functions.AsyncCommand(
                                title=row.label,
                                category='cron',
                                message_source='l',
                                correlation=job.id,
                                queue=(row.queue or 'default').split(','),
                                timeout=row.timeout or None,
                                # correlation_link='/async_task/view/{0}?next=/cron/{1}'.format(job.id, obj.id)
                                **d
                                )

            commands.append((c, row.depend_tasks))
            ids[row.id] = c.task_id


            parent.add_child(c)

        #fix depends
        for c, depends in commands:
            _d = [ids[x] for x in depends]
            c.depend_tasks = _d


        functions.call_async(parent)
        job.instances.add(parent.task_id)
        Commit()
    except Exception as e:
        Rollback()
        log.exception(e)
        raise