def run(self):
     try:
         self.close()
     except Exception as e:
         import traceback
         sendLog('closor','failed on %s due to %s and %s'%( self.wfo.name, str(e), traceback.format_exc()), level='critical')
         self.failed = True

def cachor(spec=None):
    if duplicateLock(silent=True): 
        print "currently running"
        return
    try:
        all_checks = json.loads(open('cached_transfer_statuses.json').read())
    except:
        all_checks = {}

    all_transfers = [transfer for transfer in session.query(Transfer).filter(Transfer.phedexid>0).all()]
    random.shuffle( all_transfers )

    existing = all_checks.keys() ## strng keys
    new = (set([str(transfer.phedexid) for transfer in all_transfers]) - set(existing))

    print len(new),"transfers not look out at all, will do those first"
    if spec: new = [spec]

    for transfer in all_transfers:
        if new and str(transfer.phedexid)!=sorted(new)[0]: continue
        print "running the check on",transfer.phedexid
        new_check = checkTransferStatus(url, transfer.phedexid, nocollapse=True)
        if new_check : 
            all_checks[str(transfer.phedexid)] = new_check
        else:
            print "withouth an update, we are in deep shit"
            sendLog('cachor','Failed transfer status check on %s'% transfer.phedexid, level='critical')
        #do only one
        break

    for pid in sorted(all_checks.keys()):
        if not all_checks[pid]:
            all_checks.pop(pid)
    open('cached_transfer_statuses.json','w').write(json.dumps( all_checks , indent=2))

def cachor(spec=None):
    if duplicateLock(silent=True): 
        print "currently running"
        return
    try:
        all_checks = json.loads(open('cached_transfer_statuses.json').read())
    except:
        all_checks = {}

    ## pop all that are now in inactive
    for phedexid in all_checks.keys():
        transfers = session.query(TransferImp).filter(TransferImp.phedexid==int(phedexid)).filter(TransferImp.active==True).all()
        if not transfers:
            print phedexid,"does not look relevant to be in cache anymore. poping"
            print all_checks.pop( phedexid )

    #all_transfers = [transfer for transfer in session.query(Transfer).filter(Transfer.phedexid>0).all()]
    all_transfers = list(set([imp.phedexid for imp in session.query(TransferImp).filter(TransferImp.active==True).all()]))
    random.shuffle( all_transfers )

    existing = map(int,all_checks.keys()) ## strng keys
    new = (set(all_transfers) - set(existing))

    print len(new),"transfers not look out at all, will do those first",new
    if spec: new = [int(spec)]

    #for transfer in all_transfers:
    for phedexid in all_transfers:    
        #print phedexid
        if new and phedexid!=sorted(new)[0]: continue
        print "running the check on",phedexid
        new_check = checkTransferStatus(url, phedexid, nocollapse=True)
        if new_check : 
            print json.dumps( new_check ,indent=2)
            all_checks[str(phedexid)] = new_check
        else:
            print "withouth an update, we are in some trouble."
            sendLog('cachor','Failed transfer status check on %s'% phedexid, level='critical')
        #do only one
        break

    for pid in sorted(all_checks.keys()):
        if not all_checks[pid]:
            print "Removing empty report for",pid
            all_checks.pop(pid)
    open('cached_transfer_statuses.json','w').write(json.dumps( all_checks , indent=2))

    def close(self):
        if os.path.isfile('.closor_stop'):
            print "The closing of workflows is shortened"
            return 

        url = self.url
        batch_go = self.batch_go
        CI = self.CI
        UC = self.UC
        wfo = self.wfo

        jump_the_line = self.jump_the_line
        batch_goodness = self.batch_goodness
        check_fullcopy_to_announce = UC.get('check_fullcopy_to_announce')

        ## what is the expected #lumis 
        self.wfi = workflowInfo(url, wfo.name )
        wfi = self.wfi
        wfo.wm_status = wfi.request['RequestStatus']

        if wfi.isRelval():
            has_batch_go = False
            batch_name = wfi.getCampaign()
            if not batch_name in batch_go:
                ## do the esimatation whethere this can be announced : only once per batch
                in_batches = getWorkflowByCampaign(url , batch_name, details=True)
                batch_go[ batch_name ]  = all(map(lambda s : not s in ['completed','running-open','running-closed','acquired','assigned','assignment-approved'], [r['RequestStatus'] for r in in_batches]))
            ## already verified
            has_batch_go = batch_go[batch_name]
            if not has_batch_go:
                wfi.sendLog('closor', 'Cannot close for now because the batch <a href=https://dmytro.web.cern.ch/dmytro/cmsprodmon/workflows.php?campaign=%s>%s</a> is not all close'%( batch_name, batch_name))
                return


        if wfi.request['RequestStatus'] in  ['announced','normal-archived'] and not options.force:
            ## manually announced ??
            self.to_status = 'done'
            self.to_wm_status = wfi.request['RequestStatus']
            wfi.sendLog('closor','%s is announced already : %s'%( wfo.name,self.to_wm_status))
            return 

        if jump_the_line:
            wfi.sendLog('closor','Announcing while completing')

        expected_lumis = 1
        if not 'TotalInputLumis' in wfi.request:
            print wfo.name,"has not been assigned yet, or the database is corrupted"
        elif wfi.request['TotalInputLumis']==0:
            print wfo.name,"is corrupted with 0 expected lumis"
        else:
            expected_lumis = wfi.request['TotalInputLumis']

        ## what are the outputs
        outputs = wfi.request['OutputDatasets']
        ## check whether the number of lumis is as expected for each
        all_OK = defaultdict(lambda : False)
        stats = defaultdict(int)
        #print outputs
        if len(outputs): 
            print wfo.name,wfi.request['RequestStatus']
        for out in outputs:
            event_count,lumi_count = getDatasetEventsAndLumis(dataset=out)
            self.outs.append( Output( datasetname = out ))
            odb = self.outs[-1]
            odb.workflow = wfo
            odb.nlumis = lumi_count
            odb.nevents = event_count
            odb.workfow_id = wfo.id
            if odb.expectedlumis < expected_lumis:
                odb.expectedlumis = expected_lumis
            else:
                expected_lumis = odb.expectedlumis
            odb.date = time.mktime(time.gmtime())

            fraction = lumi_count/float(expected_lumis)*100.

            completion_line = "%60s %d/%d = %3.2f%%"%(out,lumi_count,expected_lumis,fraction)
            wfi.sendLog('closor',"\t%s"% completion_line)
            if wfi.isRelval() and fraction < batch_goodness:
                self.batch_warnings[ wfi.getCampaign()].add( completion_line )
            stats[out] = lumi_count
            all_OK[out] = True 


        ## check for at least one full copy prior to moving on
        in_full = {}
        for out in outputs:
            in_full[out] = []
            presence = getDatasetPresence( url, out )
            where = [site for site,info in presence.items() if info[0]]
            if where:
                all_OK[out] = True
                print out,"is in full at",",".join(where)
                in_full[out] = copy.deepcopy(where)
            else:

                going_to = wfi.request['NonCustodialSites']+wfi.request['CustodialSites']
                wfi.sendLog('closor',"%s is not in full anywhere. send to %s"%(out, ",".join(sorted(going_to))))
                at_destination = dict([(k,v) for (k,v) in presence.items() if k in going_to])
                else_where = dict([(k,v) for (k,v) in presence.items() if not k in going_to])
#.........这里部分代码省略.........

def completor(url, specific):
    use_mcm = True
    up = componentInfo(mcm=use_mcm, soft=['mcm'])
    if not up.check(): return
    use_mcm = up.status['mcm']
    if use_mcm:
        mcm = McMClient(dev=False)

    CI = campaignInfo()
    SI = siteInfo()
    UC = unifiedConfiguration()

    wfs = []
    wfs.extend( session.query(Workflow).filter(Workflow.status == 'away').all() )
    wfs.extend( session.query(Workflow).filter(Workflow.status.startswith('assistance')).all() )

    ## just take it in random order so that not always the same is seen
    random.shuffle( wfs )

    max_per_round = UC.get('max_per_round').get('completor',None)
    if max_per_round and not specific: wfs = wfs[:max_per_round]

    ## by workflow a list of fraction / timestamps
    completions = json.loads( open('%s/completions.json'%monitor_dir).read())
    
    good_fractions = {}
    timeout = {}
    for c in CI.campaigns:
        if 'force-complete' in CI.campaigns[c]:
            good_fractions[c] = CI.campaigns[c]['force-complete']
        if 'force-timeout' in CI.campaigns[c]:
            timeout[c] = CI.campaigns[c]['force-timeout']

    long_lasting = {}

    overrides = getForceCompletes()
    if use_mcm:    
        ## add all workflow that mcm wants to get force completed
        mcm_force = mcm.get('/restapi/requests/forcecomplete')
        ## assuming this will be a list of actual prepids
        overrides['mcm'] = mcm_force

    print "can force complete on"
    print json.dumps( good_fractions ,indent=2)
    print json.dumps( overrides, indent=2)
    max_force = UC.get("max_force_complete")
    
    #wfs_no_location_in_GQ = set()
    #block_locations = defaultdict(lambda : defaultdict(list))
    #wfs_no_location_in_GQ = defaultdict(list)

    set_force_complete = set()


    for wfo in wfs:
        if specific and not specific in wfo.name: continue

        print "looking at",wfo.name
        ## get all of the same
        wfi = workflowInfo(url, wfo.name)
        pids = wfi.getPrepIDs()
        skip=False
        if not any([c in wfo.name for c in good_fractions]): skip=True
        for user,spec in overrides.items():

            if wfi.request['RequestStatus']!='force-complete':
                if any(s in wfo.name for s in spec) or (wfo.name in spec) or any(pid in spec for pid in pids) or any(s in pids for s in spec):
                    sendEmail('force-complete requested','%s is asking for %s to be force complete'%(user,wfo.name))
                    wfi = workflowInfo(url, wfo.name)
                    forceComplete(url , wfi )
                    skip=True
                    wfi.notifyRequestor("The workflow %s was force completed by request of %s"%(wfo.name,user), do_batch=False)
                    wfi.sendLog('completor','%s is asking for %s to be force complete'%(user,wfo.name))
                    break
    
        if wfo.status.startswith('assistance'): skip = True

        if skip: 
            continue

        priority = wfi.request['RequestPriority']

        if not 'Campaign' in wfi.request: continue

        if not wfi.request['RequestStatus'] in ['acquired','running-open','running-closed']: continue

        c = wfi.request['Campaign']
        if not c in good_fractions: continue
        good_fraction = good_fractions[c]
        ignore_fraction = 2.
        
        lumi_expected = None
        event_expected = None
        if not 'TotalInputEvents' in wfi.request: 
            if 'RequestNumEvents' in wfi.request:
                event_expected = wfi.request['RequestNumEvents']
            else:
                print "truncated, cannot do anything"
                continue
        else:
#.........这里部分代码省略.........

def injector(url, options, specific):
    mlock = moduleLock()
    if mlock(): return

    use_mcm = True
    up = componentInfo(soft=['mcm','wtc','jira'] )
    if not up.check(): return
    use_mcm = up.status['mcm']

    UC = unifiedConfiguration()

    transform_keywords = UC.get('convert_to_stepchain')

    workflows = getWorkflows(url, status=options.wmstatus, user=options.user)
    for user in UC.get("user_rereco"):
        workflows.extend( getWorkflows(url, status=options.wmstatus, user=user, rtype="ReReco")) 
    for user in (options.user_relval.split(',') if options.user_relval else UC.get("user_relval")) :
        workflows.extend( getWorkflows(url, status=options.wmstatus, user=user, rtype="TaskChain")) 
    for user in (options.user_storeresults.split(',') if options.user_storeresults else UC.get("user_storeresults")) :
        workflows.extend( getWorkflows(url, status=options.wmstatus, user=user, rtype="StoreResults"))

    print len(workflows),"in line"
    cannot_inject = set()
    to_convert = set()
    status_cache = defaultdict(str)

    ## browse for assignment-approved requests, browsed for ours, insert the diff
    for wf in workflows:
        if specific and not specific in wf: continue

        exists = session.query(Workflow).filter(Workflow.name == wf ).first()
        if not exists:
            wfi = workflowInfo(url, wf)
            ## check first that there isn't related here with something valid
            can_add = True
            ## first try at finding a match
            familly = session.query(Workflow).filter(Workflow.name.contains(wfi.request['PrepID'])).all()
            if not familly:
                pids = wfi.getPrepIDs()
                req_familly = []
                for pid in pids:
                    req_familly.extend( getWorkflowById( url, pid, details=True) )
                    
                familly = []
                print len(req_familly),"members"
                for req_member in req_familly:
                    #print "member",req_member['RequestName']
                    owfi = workflowInfo(url, req_member['RequestName'], request=req_member)
                    other_pids = owfi.getPrepIDs()
                    if set(pids) == set(other_pids):
                        ## this is a real match
                        familly.extend( session.query(Workflow).filter(Workflow.name == req_member['RequestName']).all() )

            for lwfo in familly:
                if lwfo:
                    ## we have it already
                    if not lwfo.status in ['forget','trouble','forget-unlock','forget-out-unlock']:
                        wfi.sendLog('injector',"Should not put %s because of %s %s"%( wf, lwfo.name,lwfo.status ))
                        sendLog('injector',"Should not put %s because of %s %s"%( wf, lwfo.name,lwfo.status ), level='critical')
                        print "Should not put",wf,"because of",lwfo.name,lwfo.status
                        cannot_inject.add( wf )
                        can_add = False
            ## add a check on validity of input datasets
            _,prim,par,sec = wfi.getIO()
            for d in list(prim)+list(par)+list(sec):
                if not d in status_cache:
                    status_cache[d] = getDatasetStatus(d)
                if status_cache[d] != 'VALID':
                    wfi.sendLog('injector',"One of the input is not VALID. %s : %s"%( d, status_cache[d]))
                    sendLog('injector',"One of the input of %s is not VALID. %s : %s"%( wf, d, status_cache[d]), level='critical')
                    can_add = False
                #else:
                #    ##make sure that all blocks get closed
                #    closeAllBlocks(url, d)

                ## check for any file in phedex, to verify existence
                _,ph_files,_,_ = getDatasetFiles(url, d)
                if not ph_files and not ( 'StoreResults' == wfi.request.setdefault('RequestType',None) ):
                    wfi.sendLog('injector',"One of the input has no file in phedex: %s" % d )
                    sendLog('injector',"One of the input has no file in phedex: %s"% d, level='critical')
                    can_add = False

            ### ban some workflow that you don't like anymore
            #outputs = wfi.request['OutputDatasets']



            if not can_add: continue

            ## temporary hack to transform specific taskchain into stepchains
            good_for_stepchain = wfi.isGoodToConvertToStepChain( keywords = transform_keywords)
            #good_for_stepchain = wfi.isGoodToConvertToStepChain( keywords = None) 


            ## match keywords and technical constraints
            if (not options.no_convert) and good_for_stepchain and not wfi.isRelval():
                to_convert.add( wf )
                wfi.sendLog('injector','Transforming %s TaskChain into StepChain'%wf)
                sendEmail('convertion to stepchain','Transforming %s TaskChain into StepChain'%wf)

#.........这里部分代码省略.........

def singleRecovery(url, task, initial, actions, do=False):
    print "Inside single recovery!"
    payload = {
        "Requestor" : os.getenv('USER'),
        "Group" : 'DATAOPS',
        "RequestType" : "Resubmission",
        "ACDCServer" : initial['CouchURL'],
        "ACDCDatabase" : "acdcserver",
        "OriginalRequestName" : initial['RequestName'],
        "OpenRunningTimeout" : 0
    }
    copy_over = ['PrepID','Campaign','RequestPriority', 'TimePerEvent', 'SizePerEvent', 'Group', 'Memory', 'RequestString' ,'CMSSWVersion']
    for c in copy_over:
        if c in initial:
            payload[c] = copy.deepcopy(initial[c])
        else:
            print c,"not in the initial payload"

    #a massage ? boost the recovery over the initial wf
#    payload['RequestPriority'] *= 10
    #Max priority is 1M
    payload['RequestPriority'] = min(500000,  payload['RequestPriority']*2 ) ## never above 500k

    #change parameters based on actions here
    if actions:
        for action in actions:
            if action.startswith('mem') and actions[action] != "" and actions[action] != 'Same':
                payload['Memory'] = actions[action]
                print "Memory set to " + actions[action]
                ## Taskchains needs to be treated special to set the memory to all tasks
                if 'TaskChain' in initial:
                    it = 1
                    while True:
                        t = 'Task%d'%it
                        it += 1
                        if t in initial:
                            payload[t] = copy.deepcopy(initial[t])
                            payload[t]['Memory'] = actions[action]
                        else:
                            break

            if action.startswith('split'):
                split_alert = (initial['RequestType'] in ['MonteCarlo'] )
                for key in initial:
                    if key == 'SplittingAlgo' and (initial[key] in ['EventBased']):
                        split_alert = True
                    elif key.startswith('Task') and key != 'TaskChain':
                        for key2 in initial[key]:
                            if key2 == 'TaskName':
                                print "task",task.split('/')[-1]
                                print "TaskName",initial[key][key2]
                                if (initial[key][key2] == task) and (initial[key][key2] in ['EventBased']):
                                    split_alert = True
                if split_alert:
                    sendLog('actor','Cannot change splitting for %s'%initial['RequestName'],level='warning')
                    print "I should not be doing splitting for this type of request",initial['RequestName']
                    return None

    acdc_round = 0
    initial_string = payload['RequestString']
    if initial_string.startswith('ACDC'):
        if initial_string[4].isdigit():
            acdc_round = int(initial_string[4])
        acdc_round += 1

    initial_string = initial_string.replace('ACDC_','').replace('ACDC%d_'%(acdc_round-1),'')
    payload['RequestString'] = 'ACDC%d_%s'%(acdc_round,initial_string)
    payload['InitialTaskPath'] = task

    if not do:
        print json.dumps( payload, indent=2)
        return None

    print "ACDC payload"
#    print json.dumps( payload , indent=2)
    print actions

    ## submit here
    acdc = reqMgrClient.submitWorkflow(url, payload)
    if not acdc:
        print "Error in making ACDC for",initial["RequestName"]
        acdc = reqMgrClient.submitWorkflow(url, payload)
        if not acdc:
            print "Error twice in making ACDC for",initial["RequestName"]
            sendLog('actor','Failed twice in making ACDCs for %s!'%initial['RequestName'],level='critical')                
            return None

    ## change splitting if requested
    if actions:
        for action in actions:
            if action.startswith('split'):
                acdcInfo = workflowInfo(url, acdc)
                splittings = acdcInfo.getSplittings()
                if actions[action] != 'Same' and actions[action] != 'max':
                    factor = int(actions[action][0:-1]) if 'x' in actions[action] else 2
                    for split in splittings:
                        for act in ['avg_events_per_job','events_per_job','lumis_per_job']:
                            if act in split:
                                print "Changing %s (%d) by a factor %d"%( act, split[act], factor),
                                split[act] /= factor
#.........这里部分代码省略.........

def singleClone(url, wfname, actions, comment, do=False):
    
    wfi = workflowInfo(url, wfname)
    payload = wfi.getSchema()
    initial = wfi.request

    payload['Requestor']           = os.getenv('USER')
    payload['Group']               = 'DATAOPS'
    payload['OriginalRequestName'] = initial['RequestName']
    payload['RequestPriority'] = initial['RequestPriority']

    if 'ProcessingVersion' in initial:
        payload['ProcessingVersion'] = int(initial['ProcessingVersion']) +1
    else:
        payload['ProcessingVersion'] = 2


## drop parameters on the way to reqmgr2
    paramBlacklist = ['BlockCloseMaxEvents', 'BlockCloseMaxFiles', 'BlockCloseMaxSize', 'BlockCloseMaxWaitTime',
                  'CouchWorkloadDBName', 'CustodialGroup', 'CustodialSubType', 'Dashboard',
                  'GracePeriod', 'HardTimeout', 'InitialPriority', 'inputMode', 'MaxMergeEvents', 'MaxMergeSize',
                  'MaxRSS', 'MaxVSize', 'MinMergeSize', 'NonCustodialGroup', 'NonCustodialSubType',
                  'OutputDatasets', 'ReqMgr2Only', 'RequestDate' 'RequestorDN', 'RequestName', 'RequestStatus',
                  'RequestTransition', 'RequestWorkflow', 'SiteWhitelist', 'SoftTimeout', 'SoftwareVersions',
                  'SubscriptionPriority', 'Team', 'timeStamp', 'TrustSitelists', 'TrustPUSitelists',
                  'TotalEstimatedJobs', 'TotalInputEvents', 'TotalInputLumis', 'TotalInputFiles','checkbox',
                  'DN', 'AutoApproveSubscriptionSites', 'NonCustodialSites', 'CustodialSites', 'OriginalRequestName', 'Teams', 'OutputModulesLFNBases', 
                  'SiteBlacklist', 'AllowOpportunistic', '_id']
    for p in paramBlacklist:
        if p in payload:
            payload.pop( p )
            pass

    if actions:
        for action in actions:
            if action.startswith('mem') and actions[action] != "" and actions[action] != 'Same':
                if 'TaskChain' in payload:
                    print "Setting memory for clone of task chain"
                    it=1
                    while True:
                        t = 'Task%d'%it
                        it+=1
                        if t in payload:
                            payload[t]['Memory'] = actions[action]
                            print "Memory set for Task%d"%it
                        else:
                            break
                else:
                    print "Setting memory for non-taskchain workflow"
                    payload['Memory'] = actions[action]
                print "Memory set to " + actions[action]
                #This line is doesn't work for some reason
#                wfi.sendLog('actor','Memory of clone set to %d'%actions[action])

    print "Clone payload"
#    print json.dumps( payload , indent=2)
    print actions

    #Create clone
    clone = reqMgrClient.submitWorkflow(url, payload)
    if not clone:
        print "Error in making clone for",initial["RequestName"]
        clone = reqMgrClient.submitWorkflow(url, payload)
        if not clone:
            print "Error twice in making clone for",initial["RequestName"]
            sendLog('actor','Failed to make a clone twice for %s!'%initial["RequestName"],level='critical')
            wfi.sendLog('actor','Failed to make a clone twice for %s!'%initial["RequestName"])
            return None

    if actions:
        for action in actions:
            if action.startswith('split'):
                cloneinfo = workflowInfo(url, clone)
                splittings = cloneinfo.getSplittings()
                if actions[action] != 'Same' and actions[action] != 'max' and actions[action] != '':
                    factor = int(actions[action][0:-1]) if 'x' in actions[action] else 2
                    for split in splittings:
                        for act in ['avg_events_per_job','events_per_job','lumis_per_job']:
                            if act in split:
                                wfi.sendLog('actor','Changing %s (%d) by a factor %d'%( act, split[act], factor))
                                print "Changing %s (%d) by a factor %d"%( act, split[act], factor),
                                split[act] /= factor
                                print "to",split[act]
                                break
                        split['requestName'] = clone
                        print "changing the splitting of",clone
                        print json.dumps( split, indent=2 )
                        print reqMgrClient.setWorkflowSplitting(url, clone, split )
                elif 'max' in actions[action]:
                    for split in splittings:
                        for act in ['avg_events_per_job','events_per_job','lumis_per_job']:
                            if act in split:
                                wfi.sendLog('actor','Max splitting set for %s (%d'%( act, split[act]))
                                print "Changing %s (%d) "%( act, split[act]),
                                split[act] = 1
                                print "to max splitting ",split[act]
                                break
                        split['requestName'] = clone
                        print "changing the splitting of",clone
                        print json.dumps( split, indent=2 )
#.........这里部分代码省略.........

url = reqmgr_url



wfs = getWorkflows(url, 'assigned', details=True)

now = time.mktime( time.gmtime())
for wf in wfs:
    assigned_log = filter(lambda change : change["Status"] in ["assigned"],wf['RequestTransition'])
    if assigned_log:
        then = assigned_log[-1]['UpdateTime']
        since = (now-then)/float(1*24*60*60.)
        if since>1.:
            print "workflow",wf['RequestName'],"is assigned since",then," that is",since,"days"
            sendLog('GQ','The workflow %s has been assigned for %.2f days'%(wf['RequestName'], since), level='critical')

may_have_one=set()
may_have_one.update([wfo.name for wfo in session.query(Workflow).filter(Workflow.status.startswith('away')).all()])
may_have_one.update([wfo.name for wfo in session.query(Workflow).filter(Workflow.status.startswith('assistance')).all()])

wfs = []
wfs.extend( getWorkflows(url, 'running-open', details=True))
wfs.extend( getWorkflows(url, 'running-closed', details=True))
wfs.extend( getWorkflows(url, 'completed', details=True))

may_have_one_too = set()
for wf in wfs:
    if wf['RequestName'] in may_have_one:
        #print wf['RequestName'],"and familly"
        may_have_one_too.update( getWorkflowById(url, wf['PrepID']) )

def checkor(url, spec=None, options=None):
    if userLock():   return
    if duplicateLock():  return


    fDB = closeoutInfo()

    UC = unifiedConfiguration()
    use_mcm = True
    up = componentInfo(mcm=use_mcm, soft=['mcm'])
    if not up.check(): return
    use_mcm = up.status['mcm']

    wfs=[]
    if options.new:
        ## get all in running and check

        ## you want to intersect with what is completed !
        if options.strict:
            completed_wfi = getWorkflows(url, status='completed')
            for wfo in session.query(Workflow).filter(Workflow.status == 'away').all():
                if wfo.name in completed_wfi:
                    wfs.append( wfo )
                else:
                    print wfo.name,"is not completed"
                    sendLog('checkor','%s is not completed'%( wfo.name))
        else:
            wfs.extend( session.query(Workflow).filter(Workflow.status == 'away').all() )

    if options.current:
        ## recheck those already there, probably to just pass them along
        wfs.extend( session.query(Workflow).filter(Workflow.status== 'assistance').all() )

    if options.old:
        ## than get all in need for assistance
        wfs.extend( session.query(Workflow).filter(Workflow.status.startswith('assistance-')).all() )


    custodials = defaultdict(list) #sites : dataset list
    transfers = defaultdict(list) #sites : dataset list
    invalidations = [] #a list of files
    SI = global_SI
    CI = campaignInfo()
    mcm = McMClient(dev=False)

    def get_campaign(output, wfi):
        campaign = None
        try:
            campaign = output.split('/')[2].split('-')[0]
        except:
            if 'Campaign' in wfi.request:
                campaign = wfi.request['Campaign']
        return campaign

    ## retrieve bypass and onhold configuration
    bypasses = []
    holdings = []
    #try:
    #    already_notified = json.loads(open('already_notifified.json').read())
    #except:
    #    print "no record of already notified workflow. starting fresh"
    #    already_notified = []

    for bypassor,email in [('vlimant','[email protected]'),('jen_a','[email protected]')]:
        bypass_file = '/afs/cern.ch/user/%s/%s/public/ops/bypass.json'%(bypassor[0],bypassor)
        if not os.path.isfile(bypass_file):
            #sendLog('checkor','no file %s',bypass_file)
            continue
        try:
            bypasses.extend( json.loads(open(bypass_file).read()))
        except:
            sendLog('checkor',"cannot get by-passes from %s for %s"%(bypass_file ,bypassor))
            sendEmail("malformated by-pass information","%s is not json readable"%(bypass_file), destination=[email])
        
        holding_file = '/afs/cern.ch/user/%s/%s/public/ops/onhold.json'%(bypassor[0],bypassor)
        if not os.path.isfile(holding_file):
            #sendLog('checkor',"no file %s"%holding_file)
            continue
        try:
            holdings.extend( json.loads(open(holding_file).read()))
        except:
            sendLog('checkor',"cannot get holdings from %s for %s"%(holding_file, bypassor))
            sendEmail("malformated by-pass information","%s is not json readable"%(holding_file), destination=[email])

    ## once this was force-completed, you want to bypass
    for rider,email in [('vlimant','[email protected]'),('jen_a','[email protected]'),('srimanob','[email protected]')]:
        rider_file = '/afs/cern.ch/user/%s/%s/public/ops/forcecomplete.json'%(rider[0],rider)
        if not os.path.isfile(rider_file):
            print "no file",rider_file
            #sendLog('checkor',"no file %s"%rider_file)
            continue
        try:
            bypasses.extend( json.loads(open( rider_file ).read() ) )
        except:
            sendLog('checkor',"cannot get force complete list from %s"%rider)
            sendEmail("malformated force complet file","%s is not json readable"%rider_file, destination=[email])

    if use_mcm:
        mcm_force = mcm.get('/restapi/requests/forcecomplete')
        bypasses.extend( mcm_force )
#.........这里部分代码省略.........

                unlock = True
            else:
                print "keep a lock on secondary within",delay_days,"days"
                unlock = False
                newly_locking.add(dataset)
                continue

        time_point("Checked as useful secondary", sub_lap=True)

        tier = dataset.split('/')[-1]
        creators = getWorkflowByOutput( url, dataset , details=True)
        if not creators and not tier == 'RAW' and not '-PromptReco-' in dataset:
            ds_status = getDatasetStatus( dataset )
            if not '-v0/' in dataset and ds_status!=None:
                #sendEmail('failing get by output','%s has not been produced by anything?'%dataset)
                sendLog('lockor','failing get by output, %s has not been produced by anything?'%dataset, level='critical')
                newly_locking.add(dataset)
                continue
            else:
                # does not matter, cannot be an OK dataset
                unlock = True
                bad_ds = True
        creators_status = [r['RequestStatus'] for r in creators]
        print "Statuses of workflow that made the dataset",dataset,"are",creators_status
        if len(creators_status) and all([status in ['failed','aborted','rejected','aborted-archived','rejected-archived'] for status in creators_status]):
            ## crap 
            print "\tunlocking",dataset,"for bad workflow statuses"
            unlock = True
            bad_ds = True

        time_point("Check as necessary output", sub_lap=True)

def batchor( url ):
    UC = unifiedConfiguration()
    ## get all workflows in assignment-approved with SubRequestType = relval
    all_wfs = []
    for user in UC.get("user_relval"):
        all_wfs = getWorkflows(url, 'assignment-approved', details=True, user=user, rtype='TaskChain')

    wfs = filter( lambda r :r['SubRequestType'] == 'RelVal' if 'SubRequestType' in r else False, all_wfs)
    ## need a special treatment for those
    hi_wfs = filter( lambda r :r['SubRequestType'] == 'HIRelVal' if 'SubRequestType' in r else False, all_wfs)

    by_campaign = defaultdict(set)
    by_hi_campaign = defaultdict(set)
    for wf in wfs:
        print "Relval:",wf['RequestName'], wf['Campaign']
        by_campaign[wf['Campaign']].add( wf['RequestName'] )
    for wf in hi_wfs:
        print "HI Relval:",wf['RequestName'], wf['Campaign']
        by_hi_campaign[wf['Campaign']].add( wf['RequestName'] )
        
    default_setup = {
        "go" :True,
        "parameters" : {
            "SiteWhitelist": [ "T1_US_FNAL" ],
            "MergedLFNBase": "/store/relval",
            "Team" : "relval",
            "NonCustodialGroup" : "RelVal"
            },
        "custodial" : "T1_US_FNAL_MSS",
        "phedex_group" : "RelVal",
        "lumisize" : -1,
        "fractionpass" : 0.0,
        "maxcopies" : 1
        }
    default_hi_setup = copy.deepcopy( default_setup )

    add_on = {}
    batches = json.loads( open('batches.json').read() )
    for campaign in by_campaign:
        ## get a bunch of information
        setup  = copy.deepcopy( default_setup )
        add_on[campaign] = setup
        sendLog('batchor','Adding the relval campaigns %s with parameters \n%s'%( campaign, json.dumps( setup, indent=2)),level='critical')
        if not campaign in batches: batches[campaign] = []
        batches[campaign] = list(set(list(copy.deepcopy( by_campaign[campaign] )) + batches[campaign] ))
    for campaign in by_hi_campaign:
        ## get a bunch of information
        setup  = copy.deepcopy( default_hi_setup )
        hi_site = random.choice(["T1_DE_KIT","T1_FR_CCIN2P3"])
        setup["parameters"]["SiteWhitelist"]=[ hi_site ]

        add_on[campaign] = setup
        sendLog('batchor','Adding the HI relval campaigns %s with parameters \n%s'%( campaign, json.dumps( setup, indent=2)),level='critical')
        if not campaign in batches: batches[campaign] = []
        batches[campaign] = list(set(list(copy.deepcopy( by_hi_campaign[campaign] )) + batches[campaign] ))
        
    
    open('batches.json','w').write( json.dumps( batches , indent=2 ) )

    ## open the campaign configuration 
    campaigns = json.loads( open('campaigns.relval.json').read() )


    ## protect for overwriting ??
    for new_campaign in list(set(add_on.keys())-set(campaigns.keys())):
        ## this is new, and can be announced as such
        print new_campaign,"is new stuff"
        workflows = by_campaign[new_campaign]
        requester = list(set([wf.split('_')[0] for wf in workflows]))
        subject = "Request of RelVal samples batch %s"% new_campaign
        text="""Dear all, 
A new batch of relval workflows was requested.

Batch ID:

%s

Requestor:

%s

Details of the workflows:

https://dmytro.web.cern.ch/dmytro/cmsprodmon/requests.php?campaign=%s

This is an automated message"""%( new_campaign, 
                                  ', '.join(requester),
                                  new_campaign,
                                  #'\n'.join( sorted(workflows) ) 
                                  )


        print subject
        print text
        to = ['[email protected]']
        sendEmail(subject, text, destination=to)
        sendLog('batchor',text, level='critical')

    ## merge all anyways
    campaigns.update( add_on )
#.........这里部分代码省略.........

#.........这里部分代码省略.........
            if any([agent in agents.get(wqs,{}).keys() for wqs,agent in itertools.product( wqss, bad_agents)]):
                print "overriding the need for bad agent"
                needs_overide = True
            return needs_overide

        ## now parse this for action
        for i_task,(task,campaign) in enumerate(tasks_and_campaigns):
            if options.augment:
                print task.pathName
                print campaign
    
            
            tune = CI.get(campaign,'tune',options.tune)
            if tune and not campaign in tune_performance:
                tune_performance.append( campaign )

            overflow = CI.get(campaign,'overflow',{})
            if overflow:
                if "PU" in overflow and not campaign in PU_overflow:
                    PU_overflow[campaign] = copy.deepcopy(overflow['PU'])
                    print "adding",campaign,"to PU overflow rules"
                if "LHE" in overflow and not campaign in LHE_overflow:
                    print "adding",campaign,"to light input overflow rules"
                    site_list = overflow['LHE']['site_list']
                    LHE_overflow[campaign] = copy.deepcopy( getattr(SI,site_list) )
                    

            ### get the task performance, for further massaging.
            if campaign in tune_performance or options.tune:
                print "performance",task.taskType,task.pathName
                if task.taskType in ['Processing','Production']:
                    set_memory,set_time = getPerf( task.pathName )
                    #print "Performance %s GB %s min"%( set_memory,set_time)
                    wfi.sendLog('equalizor','Performance tuning to %s GB %s min'%( set_memory,set_time))
                    ## get values from gmwsmon
                    # massage the values : 95% percentile
                    performance[task.pathName] = {}
                    if set_memory:
                        performance[task.pathName]['memory']=set_memory
                    if set_time and False:
                        performance[task.pathName]['time'] = set_time
            
            ### rule to avoid the issue of taskchain secondary jobs being stuck at sites processing the initial step
            if campaign in LHE_overflow:
                if task.taskType in ['Processing']:
                    needs, task_name, running, idled = needs_action(wfi, task)
                    needs_overide = overide_from_agent( wfi, needs_overide)
                    extend_to = list(set(copy.deepcopy( LHE_overflow[campaign] )))
                    if stay_within_site_whitelist:
                        extend_to = list(set(extend_to) & set(wfi.request['SiteWhitelist'])) ## restrict to stupid-site-whitelist
                    extend_to = list(set(extend_to) & set(SI.sites_ready + force_sites))

                    if extend_to and needs or needs_overide:

                        modifications[wfo.name][task.pathName] = { "ReplaceSiteWhitelist" : extend_to ,"Running" : running, "Pending" : idled, "Priority" : wfi.request['RequestPriority']}
                        wfi.sendLog('equalizor','%s of %s is running %d and pending %d, taking action : ReplaceSiteWhitelist \n %s'%( task_name,
                                                                                                                                      wfo.name,
                                                                                                                                      running,
                                                                                                                                      idled ,
                                                                                                                                      json.dumps( sorted(modifications[wfo.name][task.pathName]['ReplaceSiteWhitelist']))))

                        altered_tasks.add( task.pathName )
                    else:
                        wfi.sendLog('equalizor','%s of %s is running %d and pending %d'%( task_name, wfo.name, running, idled))
                        

def closor(url, specific=None, options=None):
    if userLock(): return
    if duplicateLock(): return
    if not componentInfo().check(): return


    UC = unifiedConfiguration()
    CI = campaignInfo()

    all_late_files = []
    check_fullcopy_to_announce = UC.get('check_fullcopy_to_announce')

    jump_the_line = options.announce if options else False
    if jump_the_line:
        wfs = session.query(Workflow).filter(Workflow.status.contains('announce')).filter(sqlalchemy.not_(Workflow.status.contains('announced'))).all()
    else:
        wfs = session.query(Workflow).filter(Workflow.status=='close').all()

    held = set()

    print len(wfs),"closing"
    max_per_round = UC.get('max_per_round').get('closor',None)
    if options.limit: max_per_round = options.limit
    random.shuffle( wfs )    
    if max_per_round: wfs = wfs[:max_per_round]

    batch_go = {}
    batch_warnings = defaultdict(set)
    batch_goodness = UC.get("batch_goodness")

    for wfo in wfs:

        if specific and not specific in wfo.name: continue

        ## what is the expected #lumis 
        wfi = workflowInfo(url, wfo.name )
        wfo.wm_status = wfi.request['RequestStatus']

        if wfi.isRelval():
            has_batch_go = False
            batch_name = wfi.getCampaign()
            if not batch_name in batch_go:
                ## do the esimatation whethere this can be announced : only once per batch
                in_batches = getWorkflowByCampaign(url , batch_name, details=True)
                batch_go[ batch_name ]  = all(map(lambda s : not s in ['completed','running-open','running-closed','acquired','assigned','assignment-approved'], [r['RequestStatus'] for r in in_batches]))
            ## already verified
            has_