def rejector(url, specific, options=None):

    up = componentInfo()

    if specific.startswith('/'):
        pass
    else:
        wfo = session.query(Workflow).filter(Workflow.name == specific).first()
        if not wfo:
            print "cannot reject",spec
            return
        results=[]
        wfi = workflowInfo(url, wfo.name)
        reqMgrClient.invalidateWorkflow(url, wfo.name, current_status=wfi.request['RequestStatus'])
        #if wfi.request['RequestStatus'] in ['assignment-approved','new','completed']:
        #    #results.append( reqMgrClient.rejectWorkflow(url, wfo.name))
        #    reqMgrClient.rejectWorkflow(url, wfo.name)
        #else:
        #    #results.append( reqMgrClient.abortWorkflow(url, wfo.name))
        #    reqMgrClient.abortWorkflow(url, wfo.name)
        
        datasets = wfi.request['OutputDatasets']
        for dataset in datasets:
            if options.keep:
                print "keeping",dataset,"in its current status"
            else:
                results.append( setDatasetStatus(dataset, 'INVALID') )

        if all(map(lambda result : result in ['None',None,True],results)):
            wfo.status = 'forget'
            session.commit()
            print wfo.name,"and",datasets,"are rejected"
            if options and options.clone:
                schema = wfi.getSchema()
                schema['Requestor'] = os.getenv('USER')
                schema['Group'] = 'DATAOPS'
                schema['OriginalRequestName'] = wfo.name
                if 'ProcessingVersion' in schema:
                    schema['ProcessingVersion']+=1
                else:
                    schema['ProcessingVersion']=2
                ##schema.pop('RequestDate') ## ok then, let's not reset the time stamp
                if options.Memory:
                    schema['Memory'] = options.Memory
                response = reqMgrClient.submitWorkflow(url, schema)
                m = re.search("details\/(.*)\'",response)
                if not m:
                    print "error in cloning",wfo.name
                    print response
                    return 
                newWorkflow = m.group(1)
                data = reqMgrClient.setWorkflowApproved(url, newWorkflow)
                print data
                wfo.status = 'trouble'
                session.commit()
        else:
            print "error in rejecting",wfo.name,results

def htmlor( caller = ""):
    up = componentInfo(mcm=False, soft=['mcm'])
    if not up.check(): return 
        
    try:
        boost = json.loads(open('%s/equalizor.json'%monitor_dir).read())['modifications']
    except:
        boost = {}
    cache = getWorkflows(reqmgr_url,'assignment-approved', details=True)
    cache.extend( getWorkflows(reqmgr_url,'acquired', details=True) )
    cache.extend( getWorkflows(reqmgr_url,'running-open', details=True) )
    cache.extend( getWorkflows(reqmgr_url,'running-closed', details=True) )
    def getWL( wfn ):
        cached = filter(lambda d : d['RequestName']==wfn, cache)
        if cached:
            wl = cached[0]
        else:
            wl = getWorkLoad(reqmgr_url,wfn)
        return wl

    def wfl(wf,view=False,p=False,ms=False,within=False,ongoing=False,status=False,update=False):
        wfn = wf.name
        wfs = wf.wm_status
        wl = None
        pid = None
        wl_pid = None
        pids=filter(lambda seg: seg.count('-')==2, wf.name.split('_'))
        if len(pids):
            pids = pids[:1]
            pid=pids[0]
            
        if not pids:
            wl = getWL( wf.name )
            pids = getPrepIDs( wl )
            pid = pids[0]

        wl_pid = pid
        if 'task' in wf.name:
            wl_pid = 'task_'+pid

        
        text=', '.join([
                #wfn,
                #'<a href="https://cmsweb.cern.ch/reqmgr/view/details/%s" target="_blank">%s</a> '%(wfn,wfn),
                #'<table><tr><td>%s</td></tr></table>'%(wfn),
                #'<span>%s</span>'%(wfn),
                "%s "%wfn,
                '(%s) <br>'%wfs])
        text+=', '.join([
                '<a href="https://%s/reqmgr2/fetch?rid=%s" target="_blank">dts</a>'%(reqmgr_url,wfn),
                '<a href="https://cmsweb.cern.ch/reqmgr/view/details/%s" target="_blank">dts-req1</a>'%wfn,
                #TOFIX '<a href=https://cmsweb.cern.ch/reqmgr/view/showWorkload?requestName=%s target="_blank">wkl</a>'%wfn,
                '<a href="https://%s/couchdb/reqmgr_workload_cache/%s" target="_blank">wfc</a>'%(reqmgr_url,wfn),
                '<a href="https://%s/reqmgr2/data/request?name=%s" target="_blank">req</a>'%(reqmgr_url,wfn),
                #'<a href="https://cmsweb.cern.ch/reqmgr/reqMgr/request?requestName=%s" target="_blank">dwkc</a>'%wfn,
                #TOFIX '<a href="https://cmsweb.cern.ch/reqmgr/view/splitting/%s" target="_blank">spl</a>'%wfn,
                '<a href="https://cms-pdmv.cern.ch/stats/?RN=%s" target="_blank">vw</a>'%wfn,
                '<a href="https://cms-pdmv.cern.ch/stats/restapi/get_one/%s" target="_blank">vwo</a>'%wfn,
                '<a href="https://cms-logbook.cern.ch/elog/Workflow+processing/?mode=full&reverse=0&reverse=1&npp=20&subtext=%s&sall=q" target="_blank">elog</a>'%pid,
                '<a href="http://cms-gwmsmon.cern.ch/prodview/%s" target="_blank">pv</a>'%wfn,
                #deprecated '<a href="https://cmsweb.cern.ch/reqmgr/reqMgr/outputDatasetsByRequestName/%s" target="_blank">out</a>'%wfn,
                '<a href="closeout.html#%s" target="_blank">clo</a>'%wfn,
                '<a href="statuses.html#%s" target="_blank">st</a>'%wfn,
                '<a href="https://%s/couchdb/workloadsummary/_design/WorkloadSummary/_show/histogramByWorkflow/%s" target="_blank">perf</a>'%(reqmgr_url,wfn)
                ])
        if within and (not view or wfs=='completed'):
            wl = getWL( wfn )
            dataset =None
            if 'InputDataset' in wl:
                dataset = wl['InputDataset']                
            if 'Task1' in wl and 'InputDataset' in wl['Task1']:
                dataset = wl['Task1']['InputDataset']

            if dataset:
                text+=', '.join(['',
                                 '<a href=https://cmsweb.cern.ch/das/request?input=%s target=_blank>input</a>'%dataset,
                                 '<a href=https://cmsweb.cern.ch/phedex/prod/Data::Subscriptions#state=create_since=0;filter=%s target=_blank>sub</a>'%dataset,
                                 '<a href=https://cmsweb.cern.ch/phedex/datasvc/xml/prod/subscriptions?dataset=%s&collapse=n target=_blank>ds</a>'%dataset,
                                 '<a href=https://cmsweb.cern.ch/phedex/datasvc/xml/prod/blockreplicas?dataset=%s target=_blank>rep</a>'%dataset,
                                 ])

        if p:
            cached = filter(lambda d : d['RequestName']==wfn, cache)
            if cached:
                wl = cached[0]
            else:
                wl = getWorkLoad('cmsweb.cern.ch',wfn)
            text+=', (%s)'%(wl['RequestPriority'])
            pass

        if pid:
            if ms:
                mcm_s = json.loads(os.popen('curl https://cms-pdmv.cern.ch/mcm/public/restapi/requests/get_status/%s --insecure'%pid).read())[pid]
                text+=', <a href="https://cms-pdmv.cern.ch/mcm/requests?prepid=%s" target="_blank">mcm (%s)</a>'%(pid,mcm_s)
            else:
                text+=', <a href="https://cms-pdmv.cern.ch/mcm/requests?prepid=%s" target="_blank">mcm</a>'%(pid)
                text+=', <a href="https://dmytro.web.cern.ch/dmytro/cmsprodmon/workflows.php?prep_id=%s" target="_blank">ac</a>'%(wl_pid)
                
        if status:
            if wf.status.startswith('assistance'):
#.........这里部分代码省略.........

def recoveror(url,specific,options=None):
    if userLock('recoveror'): return

    up = componentInfo()
    CI = campaignInfo()

    UC = unifiedConfiguration()

    def make_int_keys( d ):
        for code in d:
            d[int(code)] = d.pop(code)

    error_codes_to_recover = UC.get('error_codes_to_recover')
    error_codes_to_block = UC.get('error_codes_to_block')
    error_codes_to_notify = UC.get('error_codes_to_notify')
    make_int_keys( error_codes_to_recover )
    make_int_keys( error_codes_to_block )
    make_int_keys( error_codes_to_notify )

    wfs = session.query(Workflow).filter(Workflow.status == 'assistance-recovery').all()
    if specific:
        wfs.extend( session.query(Workflow).filter(Workflow.status == 'assistance-manual').all() )

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

        if not specific and 'manual' in wfo.status: continue
        
        wfi = workflowInfo(url, wfo.name, deprecated=True) ## need deprecated info for mergedlfnbase

        ## need a way to verify that this is the first round of ACDC, since the second round will have to be on the ACDC themselves

        all_errors = None
        try:
            wfi.getSummary()
            all_errors = wfi.summary['errors']
        except:
            pass

        print '-'*100        
        print "Looking at",wfo.name,"for recovery options"
        
        if not len(all_errors): 
            print "\tno error for",wfo.name

        task_to_recover = defaultdict(list)
        message_to_ops = ""
        message_to_user = ""

        recover=True
        if 'LheInputFilese' in wfi.request and wfi.request['LheInputFiles']:
            ## we do not try to recover pLHE
            recover = False

        if 'Campaign' in wfi.request:
            c = wfi.request['Campaign']
            if c in CI.campaigns and 'recover' in CI.campaigns[c]:
                recover=CI.campaigns[c]['recover']

        for task,errors in all_errors.items():
            print "\tTask",task
            ## collect all error codes and #jobs regardless of step at which it occured
            all_codes = []
            for name, codes in errors.items():
                if type(codes)==int: continue
                all_codes.extend( [(int(code),info['jobs'],name,list(set([e['type'] for e in info['errors']])),list(set([e['details'] for e in info['errors']])) ) for code,info in codes.items()] )

            all_codes.sort(key=lambda i:i[1], reverse=True)
            sum_failed = sum([l[1] for l in all_codes])

            for errorCode,njobs,name,types,details in all_codes:
                rate = 100*njobs/float(sum_failed)
                #print ("\t\t %10d (%6s%%) failures with error code %10d (%"+str(max_legend)+"s) at stage %s")%(njobs, "%4.2f"%rate, errorCode, legend, name)
                print ("\t\t %10d (%6s%%) failures with error code %10d (%30s) at stage %s")%(njobs, "%4.2f"%rate, errorCode, ','.join(types), name)
                    
                added_in_recover=False

                #if options.go:
                # force the recovery of any task with error ?

                if errorCode in error_codes_to_recover:
                    ## the error code is registered
                    for case in error_codes_to_recover[errorCode]:
                        match = case['details']
                        matched= (match==None)
                        if not matched:
                            matched=False
                            for detail in details:
                                if match in detail:
                                    print "[recover] Could find keyword",match,"in"
                                    print 50*"#"
                                    print detail
                                    print 50*"#"
                                    matched = True
                                    break
                        if matched and rate > case['rate']:
                            print "\t\t => we should be able to recover that", case['legend']
                            task_to_recover[task].append( (code,case) )
                            added_in_recover=True
                            message_to_user = ""
#.........这里部分代码省略.........

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

    UC = unifiedConfiguration()
    CI = campaignInfo()
    #SI = siteInfo()
    SI = global_SI()
    #NLI = newLockInfo()
    #if not NLI.free() and not options.go: return
    LI = lockInfo()
    if not LI.free() and not options.go: return

    n_assigned = 0
    n_stalled = 0

    wfos=[]
    fetch_from = []
    if specific or options.early:
        fetch_from.extend(['considered','staging'])
    if specific:
        fetch_from.extend(['considered-tried'])
    
    fetch_from.extend(['staged'])

    if options.from_status:
        fetch_from = options.from_status.split(',')
        print "Overriding to read from",fetch_from

    for status in fetch_from:
        wfos.extend(session.query(Workflow).filter(Workflow.status==status).all())

    ## in case of partial, go for fetching a list from json ?
    #if options.partial and not specific:
    #    pass

    dataset_endpoints = json.loads(open('%s/dataset_endpoints.json'%monitor_dir).read())
    aaa_mapping = json.loads(open('%s/equalizor.json'%monitor_pub_dir).read())['mapping']

    all_stuck = set()
    all_stuck.update( json.loads( open('%s/stuck_transfers.json'%monitor_dir).read() ))
    all_stuck.update( getAllStuckDataset()) 

    max_per_round = UC.get('max_per_round').get('assignor',None)
    max_cpuh_block = UC.get('max_cpuh_block')
    random.shuffle( wfos )
    for wfo in wfos:
        
        if options.limit and (n_stalled+n_assigned)>options.limit:
            break

        if max_per_round and (n_stalled+n_assigned)>max_per_round:
            break

        if specific:
            if not any(map(lambda sp: sp in wfo.name,specific.split(','))): continue
            #if not specific in wfo.name: continue
        print "\n\n"
        wfh = workflowInfo( url, wfo.name)

        if options.priority and int(wfh.request['RequestPriority']) < options.priority:
            continue

        options_text=""
        if options.early: options_text+=", early option is ON"
        if options.partial: 
            options_text+=", partial option is ON"
            options_text+=", good fraction is %.2f"%options.good_enough
        


        wfh.sendLog('assignor',"%s to be assigned%s"%(wfo.name, options_text))

        ## the site whitelist takes into account siteInfo, campaignInfo, memory and cores
        (lheinput,primary,parent,secondary, sites_allowed) = wfh.getSiteWhiteList()
        output_tiers = list(set([o.split('/')[-1] for o in wfh.request['OutputDatasets']]))
        
        
        is_stuck = (all_stuck & primary)
        if is_stuck:
            wfh.sendLog('assignor',"%s are stuck input"%(','.join( is_stuck)))

        ## check if by configuration we gave it a GO
        no_go = False
        if not wfh.go(log=True) and not options.go:
            no_go = True

        allowed_secondary = {}
        assign_parameters = {}
        check_secondary = False
        for campaign in wfh.getCampaigns():
            if campaign in CI.campaigns:
                assign_parameters.update( CI.campaigns[campaign] )

            if campaign in CI.campaigns and 'secondaries' in CI.campaigns[campaign]:
                if CI.campaigns[campaign]['secondaries']:
                    allowed_secondary.update( CI.campaigns[campaign]['secondaries'] )
                    check_secondary = True
            if campaign in CI.campaigns and 'banned_tier' in CI.campaigns[campaign]:
#.........这里部分代码省略.........

def collector(url, specific, options):
    up = componentInfo(mcm=False, soft=['mcm'])
    if not up.check(): return 

    SI = siteInfo()
    dss = DSS()
    #NL = newLockInfo()
    mcm = McMClient(dev=False)
    fetch_in_campaigns = ['RunIISummer15GS']
    mcm_statuses=['new']#,'validation','defined','approved']

    will_be_used = defaultdict(list)
    secondary_used = defaultdict(list)
    for campaign,status in itertools.product( fetch_in_campaigns, mcm_statuses):
        queries=[]
        if campaign:
            print "getting for",campaign
            queries.append('member_of_campaign=%s'%campaign)
        if status:
            print "getting for",status
            queries.append('status=%s'%status)
        rs = mcm.getA('requests', query='&'.join(queries))
        for r in rs:
            #if r['type'] != 'Prod': continue
            dataset = r['input_dataset']
            if dataset:
                #print r['prepid'],dataset
                will_be_used[dataset].append( r )
            pileup = r['pileup_dataset_name']
            if pileup:
                secondary_used['pileup'].append( r )

    all_transfers = defaultdict(list)
    print len(will_be_used),"datasets that can be pre-fetched"
    ## for secondary we really need to have the campaign right
    print len(secondary_used),"pileup will be used"

    datasets = will_be_used.keys()
    if options.limit:
        print "Restricting to randomly picked",options.limit
        random.shuffle( datasets )
        datasets = datasets[:options.limit]
    
    for dataset in datasets:
        print "\tlooking at",dataset
        #presence = getDatasetPresence(url, dataset)#, within_sites=['T2_CH_CERN'])
        ## lock all those, and pre-fecth them
        #NL.lock( dataset )
        ## we could get the reqmgr dictionnary from McM if it was implemented and use standard workflowInfo !!!
        for request in will_be_used[dataset]:
            print "will be used by",request['prepid']
            campaign = request['member_of_campaign']
            ## based on the campaign, pre-fetch a site list
            sites_allowed = SI.sites_T1s + SI.sites_with_goodIO
            if options.spread:
                ## pick up the number of copies from campaign
                copies_needed = 1 ## hard coded for now
            else:
                copies_needed = 1 ## hard coded for now        

            print "Will look for",copies_needed,"of",dataset
            ## figure out where it is and going
            destinations, all_block_names = getDatasetDestinations(url, dataset, within_sites = [SI.CE_to_SE(site) for site in sites_allowed])
            print json.dumps( destinations, indent=2)
            prim_location = [site for (site,info) in destinations.items() if info['completion']==100 and info['data_fraction']==1]
            prim_destination = [site for site in destinations.keys() if not site in prim_location]
            prim_destination = [site for site in prim_destination if not any([osite.startswith(site) for osite in SI.sites_veto_transfer])]
            copies_needed = max(0,copies_needed - len(prim_location))
            copies_being_made = [ sum([info['blocks'].keys().count(block) for site,info in destinations.items() if site in prim_destination]) for block in all_block_names]
            
            prim_to_distribute = [site for site in sites_allowed if not SI.CE_to_SE(site) in prim_location]
            prim_to_distribute = [site for site in prim_to_distribute if not SI.CE_to_SE(site) in prim_destination]
                ## take out the ones that cannot receive transfers
            prim_to_distribute = [site for site in prim_to_distribute if not any([osite.startswith(site) for osite in SI.sites_veto_transfer])]
            copies_needed = max(0,copies_needed - min(copies_being_made))
            spreading = {}
            if copies_needed:
                print "needing",copies_needed 
                chops,sizes = getDatasetChops(dataset, chop_threshold = options.chopsize)
                spreading = distributeToSites( chops, prim_to_distribute, n_copies = copies_needed, weights=SI.cpu_pledges, sizes=sizes)
            else:
                print "no copy needed for",dataset
            for (site,items) in spreading.items():
                all_transfers[site].extend( items )
    
    print "accumulated transfers"
    print json.dumps(all_transfers, indent=2)
    if not options.test:
        sendEmail('dataset to be fetched',
                  'the following datasets and location were figured from mcm up-coming requests\n%s'%( json.dumps(all_transfers, indent=2) ),
                  destination=['[email protected]'])
    
    ## now collect and make transfer request
    for (site,items_to_transfer) in all_transfers.iteritems():
        print "Directing at",site
        items_to_transfer = list(set(items_to_transfer))

        site_se = SI.CE_to_SE(site)
        blocks = [it for it in items_to_transfer if '#' in it]
        datasets = [it for it in items_to_transfer if not '#' in it]
#.........这里部分代码省略.........

def stagor(url,specific =None, options=None):
    
    if not componentInfo().check(): return
    SI = siteInfo()
    CI = campaignInfo()
    UC = unifiedConfiguration()

    done_by_wf_id = {}
    done_by_input = {}
    completion_by_input = {}
    good_enough = 100.0
    
    lost = json.loads(open('lost_blocks_datasets.json').read())
    still_lost = []
    for dataset in lost:
        l = findLostBlocks(url ,dataset)
        if not l:
            print dataset,"is not really lost"
        else:
            still_lost.append( dataset )
    open('lost_blocks_datasets.json','w').write( json.dumps( still_lost, indent=2) )

    if options.fast:
        print "doing the fast check of staged with threshold:",options.goodavailability
        for wfo in session.query(Workflow).filter(Workflow.status == 'staging').all():
            if specific and not specific in wfo.name: continue
            wfi = workflowInfo(url, wfo.name)
            sites_allowed = getSiteWhiteList( wfi.getIO() )
            if 'SiteWhitelist' in CI.parameters(wfi.request['Campaign']):
                sites_allowed = CI.parameters(wfi.request['Campaign'])['SiteWhitelist']
            if 'SiteBlacklist' in CI.parameters(wfi.request['Campaign']):
                sites_allowed = list(set(sites_allowed) - set(CI.parameters(wfi.request['Campaign'])['SiteBlacklist']))
            _,primaries,_,secondaries = wfi.getIO()
            se_allowed = [SI.CE_to_SE(site) for site in sites_allowed] 
            all_check = True
            for dataset in list(primaries):#+list(secondaries) ?
                #print se_allowed
                available = getDatasetBlocksFraction( url , dataset , sites=se_allowed )
                all_check &= (available >= options.goodavailability)
                if not all_check: break

            if all_check:
                print "\t\t",wfo.name,"can go staged"
                wfo.status = 'staged'
                session.commit()
            else:
                print "\t",wfo.name,"can wait a bit more"
        return 

    for wfo in session.query(Workflow).filter(Workflow.status == 'staging').all():
        wfi = workflowInfo(url, wfo.name)
        _,primaries,_,secondaries = wfi.getIO()
        for dataset in list(primaries)+list(secondaries):
            done_by_input[dataset] = {}
            completion_by_input[dataset] = {}
            print wfo.name,"needs",dataset

    for transfer in session.query(Transfer).all():
        if specific  and str(transfer.phedexid)!=str(specific): continue

        skip=True
        for wfid in transfer.workflows_id:
            tr_wf = session.query(Workflow).get(wfid)
            if tr_wf: 
                if tr_wf.status == 'staging':
                    print "\t",transfer.phedexid,"is staging for",tr_wf.name
                    skip=False

        if skip: continue
        if transfer.phedexid<0: continue

        ## check the status of transfers
        checks = checkTransferApproval(url,  transfer.phedexid)
        approved = all(checks.values())
        if not approved:
            print transfer.phedexid,"is not yet approved"
            approveSubscription(url, transfer.phedexid)
            continue

        ## check on transfer completion
        checks = checkTransferStatus(url, transfer.phedexid, nocollapse=True)

        if not specific:
            for dsname in checks:
                if not dsname in done_by_input: done_by_input[dsname]={}
                if not dsname in completion_by_input: completion_by_input[dsname] = {}
                done_by_input[dsname][transfer.phedexid]=all(map(lambda i:i>=good_enough, checks[dsname].values()))
                completion_by_input[dsname][transfer.phedexid]=checks[dsname].values()
        if checks:
            print "Checks for",transfer.phedexid,[node.values() for node in checks.values()]
            done = all(map(lambda i:i>=good_enough,list(itertools.chain.from_iterable([node.values() for node in checks.values()]))))
        else:
            ## it is empty, is that a sign that all is done and away ?
            print "ERROR with the scubscriptions API of ",transfer.phedexid
            print "Most likely something else is overiding the transfer request. Need to work on finding the replacement automatically, if the replacement exists"
            done = False

        ## the thing above is NOT giving the right number
        #done = False

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

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 checkor(url, spec=None, options=None):
    if userLock():   return
    if duplicateLock() and not options.go:  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']

    def time_point(label="",sub_lap=False):
        now = time.mktime(time.gmtime())
        nows = time.asctime(time.gmtime())

        print "Time check (%s) point at : %s"%(label, nows)
        print "Since start: %s [s]"% ( now - time_point.start)
        if sub_lap:
            print "Sub Lap : %s [s]"% ( now - time_point.sub_lap ) 
            time_point.sub_lap = now
        else:
            print "Lap : %s [s]"% ( now - time_point.lap ) 
            time_point.lap = now            
            time_point.sub_lap = now

    time_point.sub_lap = time_point.lap = time_point.start = time.mktime(time.gmtime())
    
    runnings = session.query(Workflow).filter(Workflow.status == 'away').all()
    standings = session.query(Workflow).filter(Workflow.status.startswith('assistance')).all()

    ## intersect with what is actually in completed status in request manager now
    all_completed = set(getWorkflows(url, 'completed' ))

    wfs=[]

    if options.strict:
        ## the one which were running and now have completed
        print "strict option is on: checking workflows that freshly completed"
        wfs.extend( filter(lambda wfo: wfo.name in all_completed , runnings))
    if options.update:
        print "update option is on: checking workflows that have not completed yet"
        wfs.extend( filter(lambda wfo: not wfo.name in all_completed , runnings))

    if options.clear:
        print "clear option is on: checking workflows that are ready to toggle closed-out"
        wfs.extend( filter(lambda wfo: 'custodial' in wfo.status, standings))
    if options.review:
        print "review option is on: checking the workflows that needed intervention"
        wfs.extend( filter(lambda wfo: not 'custodial' in wfo.status, standings))

    ## what is left out are the wf which were running and ended up aborted/failed/...

    

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

    def get_campaign(output, wfi):
        ## this should be a perfect matching of output->task->campaign
        campaign = None
        era = None
        wf_campaign = None
        if 'Campaign' in wfi.request:   wf_campaign = wfi.request['Campaign']
        try:
            era = output.split('/')[2].split('-')[0]
        except:
            era = None
            
        if wfi.isRelval(): 
            campaign = wf_campaign
        else:
            campaign = era if era else wf_campaign
        return campaign

    ## retrieve bypass and onhold configuration
    bypasses = []
    forcings = []
    overrides = getForceCompletes()
    holdings = []

    
    actors = UC.get('allowed_bypass')

    for bypassor,email in actors:
        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)
#.........这里部分代码省略.........

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

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

    n_assigned = 0
    n_stalled = 0

    wfos = []
    if specific or options.early:
        wfos.extend(session.query(Workflow).filter(Workflow.status == "considered").all())
        wfos.extend(session.query(Workflow).filter(Workflow.status == "staging").all())
    if specific:
        wfos.extend(session.query(Workflow).filter(Workflow.status == "considered-tried").all())
    wfos.extend(session.query(Workflow).filter(Workflow.status == "staged").all())

    dataset_endpoints = json.loads(open("%s/dataset_endpoints.json" % monitor_dir).read())

    max_per_round = UC.get("max_per_round").get("assignor", None)
    max_cpuh_block = UC.get("max_cpuh_block")
    random.shuffle(wfos)
    for wfo in wfos:
        if options.limit and (n_stalled + n_assigned) > options.limit:
            break

        if max_per_round and (n_stalled + n_assigned) > max_per_round:
            break

        if specific:
            if not any(map(lambda sp: sp in wfo.name, specific.split(","))):
                continue
            # if not specific in wfo.name: continue
        print "\n\n"
        wfh = workflowInfo(url, wfo.name)
        wfh.sendLog("assignor", "%s to be assigned" % wfo.name)

        ## the site whitelist takes into account siteInfo, campaignInfo, memory and cores
        (lheinput, primary, parent, secondary, sites_allowed) = wfh.getSiteWhiteList()

        ## check if by configuration we gave it a GO
        no_go = False
        if not wfh.go(log=True) and not options.go:
            n_stalled += 1
            no_go = True

        allowed_secondary = set()
        for campaign in wfh.getCampaigns():
            if campaign in CI.campaigns and "secondaries" in CI.campaigns[campaign]:
                allowed_secondary.update(CI.campaigns[campaign]["secondaries"])
        if (secondary and allowed_secondary) and (set(secondary) & allowed_secondary != set(secondary)):
            wfh.sendLog("assignor", "%s is not an allowed secondary" % (", ".join(set(secondary) - allowed_secondary)))
            # sendEmail('secondary not allowed','%s is not an allowed secondary'%( ', '.join(set(secondary)-allowed_secondary)))
            sendLog(
                "assignor",
                "%s is not an allowed secondary" % (", ".join(set(secondary) - allowed_secondary)),
                level="critical",
            )
            if not options.go:
                n_stalled += 1
                no_go = True

        if no_go:
            continue

        ## check on current status for by-passed assignment
        if wfh.request["RequestStatus"] != "assignment-approved":
            if not options.test:
                wfh.sendLog("assignor", "setting %s away and skipping" % wfo.name)
                ## the module picking up from away will do what is necessary of it
                wfo.wm_status = wfh.request["RequestStatus"]
                wfo.status = "away"
                session.commit()
                continue
            else:
                print wfo.name, wfh.request["RequestStatus"]

        ## retrieve from the schema, dbs and reqMgr what should be the next version
        version = wfh.getNextVersion()
        if not version:
            if options and options.ProcessingVersion:
                version = options.ProcessingVersion
            else:
                wfh.sendLog("assignor", "cannot decide on version number")
                n_stalled += 1
                wfo.status = "trouble"
                session.commit()
                continue

        original_sites_allowed = copy.deepcopy(sites_allowed)
        wfh.sendLog("assignor", "Site white list %s" % sorted(sites_allowed))
        override_sec_location = CI.get(wfh.request["Campaign"], "SecondaryLocation", [])

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

def recoveror(url,specific,options=None):
    if userLock('recoveror'): return

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

    CI = campaignInfo()
    SI = siteInfo()
    UC = unifiedConfiguration()
    use_recoveror = UC.get('use_recoveror')

    if not use_recoveror and not options.go:
        print "We are told not to run recoveror"
        return 

    def make_int_keys( d ):
        for code in d:
            d[int(code)] = d.pop(code)

    error_codes_to_recover = UC.get('error_codes_to_recover')
    error_codes_to_block = UC.get('error_codes_to_block')
    error_codes_to_notify = UC.get('error_codes_to_notify')
    make_int_keys( error_codes_to_recover )
    make_int_keys( error_codes_to_block )
    make_int_keys( error_codes_to_notify )

    #wfs = session.query(Workflow).filter(Workflow.status == 'assistance-recovery').all()
    wfs = session.query(Workflow).filter(Workflow.status.contains('recovery')).all()
    if specific:
        wfs.extend( session.query(Workflow).filter(Workflow.status == 'assistance-manual').all() )

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

        if not specific and 'manual' in wfo.status: continue
        
        wfi = workflowInfo(url, wfo.name)

        ## need a way to verify that this is the first round of ACDC, since the second round will have to be on the ACDC themselves

        all_errors = {}
        try:
            ## this is clearly very truncated and should be changed completely
            wfi.getSummary()
            all_errors = wfi.summary['errors']
        except:
            pass

        print '-'*100        
        print "Looking at",wfo.name,"for recovery options"

        recover = True       

        if not 'MergedLFNBase' in wfi.request:
            print "fucked up"
            sendEmail('missing lfn','%s wl cache is screwed up'%wfo.name)
            recover = False
 
        if not len(all_errors): 
            print "\tno error for",wfo.name
            recover = False

        task_to_recover = defaultdict(list)
        message_to_ops = ""
        message_to_user = ""

        if 'LheInputFilese' in wfi.request and wfi.request['LheInputFiles']:
            ## we do not try to recover pLHE
            recover = False

        if wfi.request['RequestType'] in  ['MonteCarlo','ReReco']:
            recover = False

        if 'Campaign' in wfi.request:
            c = wfi.request['Campaign']
            if c in CI.campaigns and 'recover' in CI.campaigns[c]:
                recover=CI.campaigns[c]['recover']

        for task,errors in all_errors.items():
            print "\tTask",task
            ## collect all error codes and #jobs regardless of step at which it occured
            all_codes = []
            for name, codes in errors.items():
                if type(codes)==int: continue
                all_codes.extend( [(int(code),info['jobs'],name,list(set([e['type'] for e in info['errors']])),list(set([e['details'] for e in info['errors']])) ) for code,info in codes.items()] )

            all_codes.sort(key=lambda i:i[1], reverse=True)
            sum_failed = sum([l[1] for l in all_codes])

            for errorCode,njobs,name,types,details in all_codes:
                rate = 100*njobs/float(sum_failed)
                #print ("\t\t %10d (%6s%%) failures with error code %10d (%"+str(max_legend)+"s) at stage %s")%(njobs, "%4.2f"%rate, errorCode, legend, name)
                print ("\t\t %10d (%6s%%) failures with error code %10d (%30s) at stage %s")%(njobs, "%4.2f"%rate, errorCode, ','.join(types), name)
                    
                added_in_recover=False

                #if options.go:
                # force the recovery of any task with error ?

                if errorCode in error_codes_to_recover:
#.........这里部分代码省略.........

def new_recoveror(url, specific, options=None):
    if userLock('recoveror'): return

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

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

    wfs = session.query(Workflow).filter(Workflow.status.contains('recovery')).all()
    if specific:
        wfs.extend( session.query(Workflow).filter(Workflow.status == 'assistance-manual').all() )    

    try:
        from_operator = json.loads(os.popen('curl -s http://vocms0113.cern.ch/actions/test.json').read())
        ## now we have a list of things that we can take action on
    except:
        pass




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

        if not specific and 'manual' in wfo.status: continue
        
        wfi = workflowInfo(url, wfo.name)
    
        send_recovery = False ## will make all acdc
        send_clone = False ## will make a clone
        send_back = False ## should just reject. manual ?
        send_manual = False ## will set in manual

        where_to_run, missing_to_run = wfi.getRecoveryInfo()

        task_to_recover = where_to_run.keys()

        ## if the site at which the recovery could run in drain or out ?
        for task in task_to_recover:
            not_ready = set(where_to_run[task]) - set(SI.sites_ready)
            if not_ready:
                print "the following sites are not ready for the ACDC",",".join( sorted(not_ready) )
                ## do we have a way of telling if a site is going to be out for a long time ?
                # check on priority: high prio, restart
                if wfi.request['RequestPriority'] >= 85000:
                    send_clone = True
                # check on age of the request
                injection_time = time.mktime(time.strptime('.'.join(map(str,wfi.request['RequestDate'])),"%Y.%m.%d.%H.%M.%S")) / (60.*60.)
                now = time.mktime(time.gmtime()) / (60.*60.)
                if float(now - injection_time) <14.:
                    ## less than 14 days, start over
                    send_clone = True
                else:
                    send_manual = True

        
        if not send_recovery:
            ## check on whether the stats is very low
            pass

        if send_recovery:
            ## make acdc for all tasks
            for task in task_to_recover:
                actions = list(set([case['solution'] for code,case in task_to_recover[task]  ]))
                acdc = singleRecovery(url, task, wfi.request , actions, do = True)
        elif send_clone:
            ## this will get it cloned
            wfo.status = 'assistance-clone'
            session.commit()
        elif send_manual:
            wfo.status = 'assistance-manual'

def completor(url, specific):
    mlock = moduleLock(silent=True)
    if mlock(): return 


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

    safe_mode = False

    CI = campaignInfo()
    SI = siteInfo()