preprocess.py

Go to the documentation of this file.

 
 
 
 
import re
import ast
import shutil
from Path.getPath import *
from Extract.getCall import getCallFunction, modifyWithName
from Extract.extractCall import WithVisitor
from Tool.tool import getAst,getParameter,getLastAPIParameter,departAPI,departAPI2,ConditionalReturnTransformer, getFileName, getRunFile
from Tool.workspace import getRuntimePaths
 
 
 
 
def countBracket(s):
    minL=0
    minR=0
    midL=0
    midR=0
    huaL=0
    huaR=0
    flag=1
    cnt=0 #计算引号的个数
    for it in s:
        if it=='\'': #引号内的括号不计数
            flag=0
            cnt+=1
            if cnt%2==0:
                flag=1
        elif flag==1:
            if it=='(':
                minL+=1
            elif it==')':
                minR+=1
            elif it=='[':
                midL+=1
            elif it==']':
                midR+=1
            elif it=='{':
                huaL+=1
            elif it=='}':
                huaR+=1
    return minL,minR,midL,midR,huaL,huaR
 
 
 
 
def oneLine(filePath):
    try:
        root=getAst(filePath)
        #重写回文件
        with open(filePath,'w',encoding='UTF-8') as fw:
            newCode=ast.unparse(root)
            # newCode=re.sub(r'""".*?"""','pass',newCode,flags=re.DOTALL) #去掉代码中的注释
            fw.write(f"{newCode}\n")
    except Exception as e:
        print(f"oneLine --> {filePath} parse to ast failed: {e}")
 
 
    
 
def expandConditionalReturn(filePath):
    try:
        root=getAst(filePath)
        transformer = ConditionalReturnTransformer()
        new_root = transformer.visit(root)
        #重写回文件
        with open(filePath,'w',encoding='UTF-8') as fw:
            newCode=ast.unparse(root)
            fw.write(f"{newCode}\n")
    except Exception as e:
        print(f"expandConditionalReturn --> {filePath} parse to ast failed: {e}")
 
 
 
 
def getListVar(root,ansLst):
    for node in ast.iter_child_nodes(root):
        if isinstance(node,ast.ListComp):
            s=ast.unparse(node.generators).lstrip(' ')
            pattern="for (.*?) in"
            lst=re.findall(pattern,s)
            if len(lst)==1:
                temp=lst[0]
                if temp[0]=='(':
                    var=temp[1:-1]
                else:
                    var=temp
                s1=f"{var}=[{temp} {s}][0]"
                ansLst.append(s1)
 
        getListVar(node,ansLst)    
 
 
 
 
def getDictVar(root,ansLst):
    for node in ast.iter_child_nodes(root):
        if isinstance(node,ast.DictComp):
            s=ast.unparse(node.generators).lstrip(' ')
            pattern="for (.*?) in"
            lst=re.findall(pattern,s)
            if len(lst)==1:
                temp=lst[0]
                if temp[0]=='(':
                    var=temp[1:-1]
                else:
                    var=temp
                s1=f"{var}=[{temp} {s}][0]"
                ansLst.append(s1)
 
        getDictVar(node,ansLst)    
 
 
 
 
def convertLocalVar(filePath,libName):
    with open(filePath,'r',encoding='UTF-8') as fr:
        codeLst=fr.readlines()
    
    for i in range(len(codeLst)):
        s=codeLst[i].lstrip(' ').rstrip(' ')
        try:
            root=ast.parse(s,filename='<unknown>',mode='exec')
        except Exception as e:
            continue
        
        #step1:判断代码语句中是否含有列表推导式或字典推导式
        listComp=0
        dictComp=0
        for node in ast.walk(root):
            if isinstance(node,ast.ListComp):
                listComp=1
            if isinstance(node,ast.DictComp):
                dictComp=1
 
        if not listComp and not dictComp:
            continue
 
        #step2:判断列表推导式中是否含有第三方库调用的API
        flag=0
        _,callDict=getCallFunction(filePath,libName)
        callLst=[record['call_text'] for record in callDict.values()]
        flag=0
        for node in ast.walk(root):
            if isinstance(node, ast.Call):
                callState=ast.unparse(node)
                callState=callState.replace(' ','').replace('"','').replace("'",'')
                for it in callLst:
                    if callState==it.replace(' ','').replace('"','').replace("'",''):
                        flag=1
                        break
                if flag==1:
                    break
        if flag==0:
            continue
        
        #step3:提取出列表推导式中的变量
        ansLst=[]
        if listComp:
            getListVar(root,ansLst)
            spaceNum=countSpace(codeLst[i])
            temp=''
            for it in ansLst:
                tryStr="try:\n"
                exceptStr="except:\n"
                passStr="pass\n"
                if spaceNum:
                    tryStr=' '*spaceNum*1+tryStr
                    it=' '*spaceNum*1+' '*4+it+'\n'
                    exceptStr=' '*spaceNum*1+exceptStr
                    passStr=' '*spaceNum*1+' '*4+passStr
                else:
                    spaceNum=4
                    it=' '*spaceNum*1+it+'\n'
                    passStr=' '*spaceNum*1+passStr
                    spaceNum=0 #用完之后就置为0
            
                s=tryStr+it+exceptStr+passStr
                temp=temp+s
            temp+=codeLst[i]
            codeLst[i]=temp
 
        if dictComp:
            getDictVar(root,ansLst)
            spaceNum=countSpace(codeLst[i])
            temp=''
            for it in ansLst:
                tryStr="try:\n"
                exceptStr="except:\n"
                passStr="pass\n"
                if spaceNum:
                    tryStr=' '*spaceNum*1+tryStr
                    it=' '*spaceNum*1+' '*4+it+'\n'
                    exceptStr=' '*spaceNum*1+exceptStr
                    passStr=' '*spaceNum*1+' '*4+passStr
                else:
                    spaceNum=4
                    it=' '*spaceNum*1+it+'\n'
                    passStr=' '*spaceNum*1+passStr
                    spaceNum=0 #用完之后就置为0
            
                s=tryStr+it+exceptStr+passStr
                temp=temp+s
            temp+=codeLst[i]
            codeLst[i]=temp
 
 
    with open(filePath,'w',encoding='UTF-8') as fw:
        for it in codeLst:
            fw.write(it)
 
 
 
 
def findAssignCall(root):
    assignLst=[]
    for node in ast.walk(root):
        if isinstance(node,ast.Assign) and isinstance(node.value,ast.Call):
            target=ast.unparse(node.targets)
            assignLst.append(target)
    return assignLst
 
 
 
def getAssignReceiverExpr(root,source,aliasName,lineno):
    expr=None
    bestLine=-1
 
    scopeBodies=[root.body]
    for node in ast.walk(root):
        if isinstance(node,(ast.FunctionDef,ast.AsyncFunctionDef,ast.ClassDef)):
            start=node.lineno
            end=getattr(node,'end_lineno',start)
            if start<lineno<=end:
                scopeBodies.append(node.body)
 
    for body in scopeBodies:
        for stmt in body:
            if not hasattr(stmt,'lineno') or stmt.lineno>=lineno:
                continue
            if isinstance(stmt,(ast.If,ast.For,ast.AsyncFor,ast.While,ast.Try,ast.With,ast.AsyncWith)):
                continue
            value=None
            targets=[]
            if isinstance(stmt,ast.Assign):
                value=stmt.value
                targets=stmt.targets
            elif isinstance(stmt,ast.AnnAssign):
                value=stmt.value
                targets=[stmt.target]
            if value is None:
                continue
            for target in targets:
                if isinstance(target,ast.Name) and target.id==aliasName and stmt.lineno>bestLine:
                    sourceExpr=ast.get_source_segment(source,value)
                    if sourceExpr:
                        expr=sourceExpr.strip()
                        bestLine=stmt.lineno
    return expr
 
 
 
def getSelfReceiverExpr(root,lineno,methodName):
    for node in ast.walk(root):
        if not isinstance(node,ast.ClassDef):
            continue
        start=node.lineno
        end=getattr(node,'end_lineno',start)
        if not (start<lineno<=end) or len(node.bases)==0:
            continue
        defNames=set()
        for item in ast.iter_child_nodes(node):
            if isinstance(item,(ast.FunctionDef,ast.AsyncFunctionDef)):
                defNames.add(item.name)
        if methodName in defNames:
            return None
        return ast.unparse(node.bases[0])
    return None
 
    
 
 
def countSpace(s):
    cntSpace=0
    for it in s:
        if it==' ':
            cntSpace+=1
        else:
            break
    return cntSpace
 
 
 
 
def getImportLine(codeLst):
    #首先判断from import语句中是否含有特殊的__future__
    index=-1
    for i in range(len(codeLst)):
        if 'import' in codeLst[i] and '__future__' in codeLst[i]:
            index=i
 
    #若没有future,再判断开头是否存在'"""'注释 
    count=0
    if index==-1 and '"""' in codeLst[0]:
        for i in range(0,len(codeLst)):
            if '"""' in codeLst[i]:
                index=i
                count+=1
            if count==2:
                break
 
    index+=1
    
    return index
 
 
 
 
def extractDecorator(root):
    decoratorLst = []
    for n in ast.walk(root):
        if isinstance(n, (ast.FunctionDef, ast.ClassDef)) and n.decorator_list:
            for decorator in n.decorator_list:
                if isinstance(decorator, ast.Call): # and isinstance(decorator.func.value, ast.Name):
                    decoratorLst.append(ast.unparse(decorator.func))
 
    return decoratorLst 
 
 
 
def getDecoratorInsertIndex(codeLst,index):
    insertIndex=index
    if not codeLst[index].lstrip().startswith('@'):
        return insertIndex
 
    spaceNum=countSpace(codeLst[index])
    j=index-1
    while j>=0:
        if codeLst[j].lstrip().startswith('@') and countSpace(codeLst[j])==spaceNum:
            insertIndex=j
            j-=1
            continue
        break
 
    return insertIndex
 
 
 
 
def addDictSingle(callAPI,filePath,callKey):
    with open(filePath,'r',encoding='UTF-8') as fr:
        codeLst=fr.readlines()
    source=''.join(codeLst)
    
    lineno=getImportLine(codeLst)
    importDict='from recordValue import paraValueDict\nfrom recordValue import callsiteInfoDict\n'
    codeLst.insert(lineno,importDict)
    
    paraStr=getLastAPIParameter(callAPI) #获取最后一个API的参数
    parameterLst=getParameter(paraStr,space=0) #项目参数不去空格
    root=getAst(filePath)
    targetLst=findAssignCall(root)
 
    #找出树中所有withitem call节点 -- 2025/5/19
    withitem_visitor = WithVisitor()
    withitem_visitor.visit(root)
    withitem_call_names = withitem_visitor.get_withitem_call() #dict
 
    for i in range(len(codeLst)): #每次只会往列表中插入一个元素
        if callAPI.replace(' ','') in codeLst[i].replace(' ','') and 'def ' not in codeLst[i] and 'paraValueDict' not in codeLst[i] and 'callsiteInfoDict' not in codeLst[i]:
            spaceNum=countSpace(codeLst[i])
            dicString1=''
            l=departAPI(callAPI)
            l2=departAPI2(callAPI)
            firstPart=''
            for it in l2:
                if '(' not in it:
                    firstPart+=it+'.'
            firstPart=firstPart.rstrip('.')
            
            callsiteInfo={
                'call_text': callAPI,
                'format_api': callAPI,
            }
            displayCall=callAPI.replace('\n',' ')
            dicStringComment=f'# PCART callsite: {displayCall}\n'
            dicStringKey=f'__pcart_runtime_callsite_key__={repr(callKey)}\n'
            dicString0=f'callsiteInfoDict[__pcart_runtime_callsite_key__]={repr(callsiteInfo)}\n'
            if firstPart and (firstPart.split('.')[0] in targetLst or firstPart.split('.')[0]=='self') and len(l)==1:
                lineNo = i + 1
                receiverExpr=None
                if firstPart.split('.')[0] in targetLst:
                    receiverExpr=getAssignReceiverExpr(root,source,firstPart.split('.')[0],lineNo)
                elif firstPart.split('.')[0]=='self':
                    methodName=callAPI.split('(')[0].split('.')[-1]
                    receiverExpr=getSelfReceiverExpr(root,lineNo,methodName)
                if receiverExpr:
                    receiverExpr=receiverExpr.replace('\\','\\\\').replace('"','\\"')
                    dicString1=f'paraValueDict[\"@\"+__pcart_runtime_callsite_key__]={{}}; paraValueDict[\"@\"+__pcart_runtime_callsite_key__][\"object\"]={firstPart}; paraValueDict[\"@\"+__pcart_runtime_callsite_key__][\"expr\"]=\"{receiverExpr}\"\n'
                else:
                    dicString1=f'paraValueDict[\"@\"+__pcart_runtime_callsite_key__]={firstPart}\n'
            elif len(l)>1: #df.a(x).b(y), np.max(...), torch.nn.Sequential(...)
                dicString1=f'paraValueDict[\"@\"+__pcart_runtime_callsite_key__]={l[-2]}\n'
 
            #判断API是否为withitem中的别名调用 -- 2025/5/19 
            if firstPart and firstPart.split('.')[0] in withitem_call_names:
                lineNo = i + 1
                initialCallName = modifyWithName(firstPart, withitem_call_names, lineNo).rstrip('.')
                # withitem接收者同时保存运行时对象和还原表达式，动态阶段按可用候选依次尝试
                initialCallName=initialCallName.replace('\\','\\\\').replace('"','\\"')
                dicString1=f'paraValueDict[\"@\"+__pcart_runtime_callsite_key__]={{}}; paraValueDict[\"@\"+__pcart_runtime_callsite_key__][\"object\"]={firstPart}; paraValueDict[\"@\"+__pcart_runtime_callsite_key__][\"expr\"]=\"{initialCallName}\"\n'
            
            #再保存API的参数值 
            dicString2='paraValueDict[__pcart_runtime_callsite_key__]=['
            for para in parameterLst: #
                if '=' in para and "'='" not in para and '"="' not in para: #若参数的形式为key=f(x=1),只要确保=的前面不含括号即可
                    pos=para.find('=') #找到第一个=的位置
                    if '(' not in para[0:pos] and "'" not in para[0:pos] and '"' not in para[0:pos] and para[pos+1]!='=': #等号前面也不能出现引号，比如f('x= ',y=1)
                        para=para[pos+1:]
                
                para=para.lstrip('*') #有的参数会带*，2023-12-20
                dicString2=dicString2+para+','
            dicString2=dicString2.rstrip(',')+']\n'
            
            while spaceNum>0:
                dicStringComment=' '+dicStringComment
                dicStringKey=' '+dicStringKey
                dicString0=' '+dicString0
                if dicString1:
                    dicString1=' '+dicString1
                dicString2=' '+dicString2
                spaceNum-=1
 
            #插入的时候要考虑是否含有elif,如果有elif要把它插在elif后面
            if 'elif' not in codeLst[i]:
                insertIndex=getDecoratorInsertIndex(codeLst,i)
                codeLst.insert(insertIndex,dicString2)
                if dicString1:
                    codeLst.insert(insertIndex,dicString1)
                codeLst.insert(insertIndex,dicString0)
                codeLst.insert(insertIndex,dicStringKey)
                codeLst.insert(insertIndex,dicStringComment)
            else:
                if dicString1:
                    dicString1=dicString1.lstrip(' ')#去掉之前添加的空格，重新计算开头的空格数
                dicStringComment=dicStringComment.lstrip(' ')
                dicStringKey=dicStringKey.lstrip(' ')
                dicString0=dicString0.lstrip(' ')
                dicString2=dicString2.lstrip(' ') 
                for j in range(i+1,len(codeLst)):
                    if codeLst[j]!='\n' and '#' not in codeLst[j]:
                        spaceNum=countSpace(codeLst[j])
                        while spaceNum>0:
                            dicStringComment=' '+dicStringComment
                            dicStringKey=' '+dicStringKey
                            dicString0=' '+dicString0
                            if dicString1:
                                dicString1=' '+dicString1
                            dicString2=' '+dicString2
                            spaceNum-=1
                        
                        codeLst.insert(j,dicString2)
                        if dicString1:
                            codeLst.insert(j,dicString1)
                        codeLst.insert(j,dicString0)
                        codeLst.insert(j,dicStringKey)
                        codeLst.insert(j,dicStringComment)
                        break
 
            break
        
 
    if codeLst[0]=='pass\n':
        codeLst=codeLst[1:]
    with open(filePath,'w',encoding='UTF-8') as fw:
        for it in codeLst:
            fw.write(it)
 
 
 
 
def addDictAll(projPath,projName,filePath,copyRoot,runFileLst,libName,runPath,runCommand,pcresolveLookup=None):
    with open(filePath,'r',encoding='UTF-8') as fr:
        code=fr.read()
        fr.seek(0) #将文件指针重新定位到文件的开头
        codeLst=fr.readlines()
    try:
        root=ast.parse(code,filename='<unknown>',mode='exec')
    except Exception as e:
        print(f"addDictAll --> ast.parse failed, {filePath}: {e}")
        return
 
    #处理相关路径
    fileName = os.path.basename(filePath)[0:-3]
    fileRelativePath=os.path.relpath(filePath,os.path.join(copyRoot,projName))
    fileAbsolutePath=os.path.join(projPath,fileRelativePath)
    
    lineno=getImportLine(codeLst)
    importDict='from recordValue import paraValueDict\nfrom recordValue import apiCoveredSet\nfrom recordValue import callsiteInfoDict\n'
    codeLst.insert(lineno,importDict)
    if pcresolveLookup is None:
        _,callDict=getCallFunction(fileAbsolutePath,libName,projPath)
    else:
        _,callDict=getCallFunction(fileAbsolutePath,libName,projPath,pcresolveLookup=pcresolveLookup)
 
    targetLst=findAssignCall(root) #用来区分调用者是否来自赋值语句，比如a.f(), tf.f(), or self.f()
 
    #找出树中所有withitem call节点 -- 2025/5/19
    try:
        withitem_root = getAst(fileAbsolutePath)
    except Exception:
        withitem_root = root
    withitem_visitor = WithVisitor()
    withitem_visitor.visit(withitem_root)
    withitem_call_names = withitem_visitor.get_withitem_call() #dict
 
    insertStartLine=0 #记录每次插桩的行
    preInsertAPI='' #记录上一个插桩的API是哪个
    preInsertAPICount=0 #记录上一个插桩行中出现了几次被插的API
    for artifactId,record in callDict.items(): #key是调用点artifact id，value是结构化调用点记录
        flag=0 #标记API是否找到了插桩的位置
        lineno=int(record['lineno']) #这个lineno是原项目中的行数
        callState=record['call_text']
        paraStr=record['parameters']
        callAPI=callState.replace(' ','')
        if callAPI==preInsertAPI: #判断当前要处理的API与上一个API是否相同
            preInsertAPICount-=1
            if preInsertAPICount<=0:
                insertStartLine+=1
        
        i=insertStartLine #从第i行开始向后找
        while i<len(codeLst):
            #API调用在i行代码中，且i行代码不是函数定义语句、插桩语句和运行覆盖检查语句
            if callAPI in codeLst[i].replace(' ','') and 'def ' not in codeLst[i] and 'paraValueDict' not in codeLst[i] and 'apiCoveredSet' not in codeLst[i] and 'callsiteInfoDict' not in codeLst[i]:
                if callAPI!=preInsertAPI:#只有当前API不等于上一个被插API时，才需要重新计算preAPICount
                    preInsertAPICount=codeLst[i].replace(' ','').count(callAPI)
                    preInsertAPI=callAPI
                flag=1
                spaceNum=countSpace(codeLst[i])
                callSiteKey=artifactId
                callsiteInfo={
                    'artifact_hash': record.get('artifact_hash',''),
                    'rel_path': record.get('rel_path',''),
                    'lineno': record.get('lineno'),
                    'col_offset': record.get('col_offset'),
                    'end_lineno': record.get('end_lineno'),
                    'end_col_offset': record.get('end_col_offset'),
                    'call_text': record.get('call_text',''),
                    'format_api': record.get('format_api',''),
                }
                displayPath=record.get('rel_path','')
                displayLine=record.get('lineno')
                displayColumn=record.get('col_offset')
                displayCall=str(record.get('call_text','')).replace('\n',' ')
                displayPrefix=f'{displayPath}:{displayLine}:{displayColumn} ' if displayPath else ''
                dicStringComment=f'# PCART callsite: {displayPrefix}{displayCall}\n'
                dicStringKey=f'__pcart_runtime_callsite_key__={repr(callSiteKey)}\n'
                dicString0=f'callsiteInfoDict[__pcart_runtime_callsite_key__]={repr(callsiteInfo)}\n'
                l=departAPI(callState)
                l2=departAPI2(callState)
                firstPart=''
                for it in l2:
                    if '(' not in it:
                        firstPart+=it+'.'
                firstPart=firstPart.rstrip('.')
                dicString1=''
                
                #判断API是否具有上文依赖，比如self.f(x), a(x).b(y)中的a(x)，或者a.f(x)中的a
                #df.a(x).b(y)这种情况如何解决
                #a.b.c(x)
                # if '(' not in firstPart and (firstPart in targetLst or firstPart=='self'):
                #self.f(x), a.f(x), a.b.c(x)
                if firstPart and (firstPart.split('.')[0] in targetLst or firstPart.split('.')[0]=='self') and len(l)==1:
                    receiverExpr=None
                    if firstPart.split('.')[0] in targetLst:
                        receiverExpr=getAssignReceiverExpr(root,code,firstPart.split('.')[0],lineno)
                    elif firstPart.split('.')[0]=='self':
                        methodName=callState.split('(')[0].split('.')[-1]
                        receiverExpr=getSelfReceiverExpr(root,lineno,methodName)
                    if receiverExpr:
                        receiverExpr=receiverExpr.replace('\\','\\\\').replace('"','\\"')
                        dicString1=f'paraValueDict[\"@\"+__pcart_runtime_callsite_key__]={{}}; paraValueDict[\"@\"+__pcart_runtime_callsite_key__][\"object\"]={firstPart}; paraValueDict[\"@\"+__pcart_runtime_callsite_key__][\"expr\"]=\"{receiverExpr}\"\n'
                    else:
                        dicString1=f'paraValueDict[\"@\"+__pcart_runtime_callsite_key__]={firstPart}\n'
                elif len(l)>1: #df.a(x).b(y), np.max(...), torch.nn.Sequential(...)
                    dicString1=f'paraValueDict[\"@\"+__pcart_runtime_callsite_key__]={l[-2]}\n'
               
                #判断API是否为withitem中的别名调用 -- 2025/5/19 
                if firstPart and firstPart.split('.')[0] in withitem_call_names:
                    initialCallName = modifyWithName(firstPart, withitem_call_names, lineno).rstrip('.')
                    initialCallName = initialCallName.rstrip('.')
                    # withitem接收者同时保存运行时对象和还原表达式，非withitem调用仍保持原有单值保存
                    initialCallName=initialCallName.replace('\\','\\\\').replace('"','\\"')
                    dicString1=f'paraValueDict[\"@\"+__pcart_runtime_callsite_key__]={{}}; paraValueDict[\"@\"+__pcart_runtime_callsite_key__][\"object\"]={firstPart}; paraValueDict[\"@\"+__pcart_runtime_callsite_key__][\"expr\"]=\"{initialCallName}\"\n'
                #再保存API的参数值
                dicString2='paraValueDict[__pcart_runtime_callsite_key__]=['
                paraLst=getParameter(paraStr,space=0) #项目参数不去空格2023-12-14
                for para in paraLst: 
                    if '=' in para and "'='" not in para and '"="' not in para: #若参数的形式为key=f(x=1),只要确保=的前面不含括号即可
                        pos=para.find('=') #找到第一个=的位置,存在x=(a==b)和a==b形式
                        if '(' not in para[0:pos] and "'" not in para[0:pos] and '"' not in para[0:pos] and para[pos+1]!='=': #等号前面也不能出现引号，比如f('x= ',y=1)
                            para=para[pos+1:] #把参数的值保存下来
                    
                    para=para.lstrip('*') #有的参数会带*号，2023-12-20
                    dicString2=dicString2+para+','
                dicString2=dicString2.rstrip(',')+']\n'
                
                dicString3='apiCoveredSet.add(__pcart_runtime_callsite_key__)\n'
                while spaceNum>0:
                    dicStringComment=' '+dicStringComment
                    dicStringKey=' '+dicStringKey
                    dicString0=' '+dicString0
                    if dicString1:
                        dicString1=' '+dicString1
                    dicString2=' '+dicString2
                    dicString3=' '+dicString3
                    spaceNum-=1
 
                #插入的时候要考虑是否含有elif,如果有elif要把它插在elif后面
                #不能将字典插入到if和elif之间，所以需要找到elif后的下一个非空行再插入
                if 'elif' not in codeLst[i]:
                    insertIndex=getDecoratorInsertIndex(codeLst,i)
                    codeLst.insert(insertIndex,dicString3)
                    codeLst.insert(insertIndex,dicString2)
                    if dicString1:
                        codeLst.insert(insertIndex,dicString1)
                    codeLst.insert(insertIndex,dicString0)
                    codeLst.insert(insertIndex,dicStringKey)
                    codeLst.insert(insertIndex,dicStringComment)
                    #记录当前插在了哪一行
                    if dicString1:
                        insertStartLine=insertIndex+6
                    else:
                        insertStartLine=insertIndex+5
                else:
                    if dicString1:
                        dicString1=dicString1.lstrip(' ') #去掉之前添加的空格，重新计算开头的空格数
                    dicStringComment=dicStringComment.lstrip(' ')
                    dicStringKey=dicStringKey.lstrip(' ')
                    dicString0=dicString0.lstrip(' ')
                    dicString2=dicString2.lstrip(' ') 
                    dicString3=dicString3.lstrip(' ') 
                    for j in range(i+1,len(codeLst)):
                        if codeLst[j]!='\n' and '#' not in codeLst[j]:
                            spaceNum=countSpace(codeLst[j])
                            while spaceNum>0:
                                dicStringComment=' '+dicStringComment
                                dicStringKey=' '+dicStringKey
                                dicString0=' '+dicString0
                                if dicString1:
                                    dicString1=' '+dicString1
                                dicString2=' '+dicString2
                                dicString3=' '+dicString3
                                spaceNum-=1
                            codeLst.insert(j,dicString3)
                            codeLst.insert(j,dicString2)
                            if dicString1:
                                codeLst.insert(j,dicString1)
                            codeLst.insert(j,dicString0)
                            codeLst.insert(j,dicStringKey)
                            codeLst.insert(j,dicStringComment)
                            #记录当前插在了哪一行
                            if dicString1:
                                insertStartLine=j+6
                            else:
                                insertStartLine=j+5
                            break
  
                break
            
            i+=1
        
        if flag==0:
            # Debug aid: log uninstrumentable callsites for diagnosis
            # 调试辅助：记录无法插桩的调用点以便诊断
            print(f"{fileName}#{lineno}-->{callState}\n")
 
    if codeLst[0]=='pass\n':
        codeLst=codeLst[1:]
    with open(filePath,'w',encoding='UTF-8') as fw:
        for it in codeLst:
            fw.write(it)
 
 
 
 
def handleRunFile(file,runPath,runCommand):
    with open(file,'r',encoding='UTF-8') as fr:
        codeLst=fr.readlines()
    lineno=getImportLine(codeLst)
    codeLst.insert(lineno,f"from recordValue import paraValueDict\n")
 
    if codeLst[0]=='pass\n':
        codeLst=codeLst[1:]
    with open(file,'w',encoding='UTF-8') as fw:
        for it in codeLst:
            fw.write(it)
 
 
def obtainDef(sourcePath):
    with open(sourcePath,'r',encoding='UTF-8') as fr:
        code=fr.read()
    try:
        root=ast.parse(code,filename='<unknown>',mode='exec') #将源码解析成AST语法树
    except:
        print(sourcePath)
        return
    fw=open('Copy/defFile.py','a',encoding='UTF-8')
    for node in ast.iter_child_nodes(root):
        if isinstance(node,ast.ClassDef) or isinstance(node,ast.FunctionDef):
            s=ast.unparse(node)
            fw.write(f"{s}\n")
    fw.close()
 
 
 
def modifyFromImport(filePath,importStatement):
    # with open(filePath,'r') as fr:
    with open(filePath, 'r', encoding='UTF-8') as fr:
        codeLst=fr.readlines()
 
    s='\n'.join(importStatement)+'\n'
    codeLst.insert(0,s)
    # with open(filePath,'w') as fw:
    with open(filePath, 'w', encoding='UTF-8') as fw:
        for it in codeLst:
            fw.write(it)
 
 
 
 
def saveConstantAssign(astNode, constantVar, nonConstantVar, decoratorLst):
    flag = 0
    astBody = []
    for n in ast.walk(astNode):
        if isinstance(n,ast.Assign):
            value = n.value
            valueAstContent = ast.dump(n.value)
            targetAstContent = ast.dump(n.targets[0])
            if isinstance(n.targets[0], ast.Name):
                varName = n.targets[0].id
                #如果赋值语句的变量名在装饰器列表中出现过，则保留该赋值语句 2025/5/13 
                if any(varName in decorator.split('.') for decorator in decoratorLst):
                    continue
                #如果赋值语句的变量名是常量变量名，则保留该赋值语句 2025/5/31 
                if isinstance(value, ast.Constant):
                    if varName in nonConstantVar:
                        flag=1
                        break
                    else:   
                        targets = n.targets
                        if targets:
                            target = targets[0]
                            if isinstance(target, ast.Name):
                                constantVar.append(target.id)
                else:
                    # 使用正则表达式匹配单引号包围的内容
                    pattern = r"id='([^']*)'"
                    valueMatches = re.findall(pattern, valueAstContent)
                    targetMatches = re.findall(pattern, targetAstContent)
                    if not len(valueMatches):
                        if targetMatches[0] in nonConstantVar:
                            flag=1
                            break
                    else:
                        for match in valueMatches:
                            if match not in constantVar:
                                nonConstantVar.append(targetMatches[0])
                                flag=1
                                break
            else:
                flag=1
                break
    if flag==1:
        return flag
    else:
        astBody.append(astNode)
        return astBody
 
 
 
 
def saveStructure(projPath,libName):
    pathObj=Path('DF')
    pathObj.getPath(projPath)
    filePath=[it for it in pathObj.path if it.endswith('py')]
    for file in filePath:
        _,callDict=getCallFunction(file,libName)
        callLst=[record['call_text'] for record in callDict.values()]
        # with open(file,'r') as fr:
        with open(file, 'r', encoding='UTF-8') as fr:
            code=fr.read()
        try:
            root=ast.parse(code,filename='<unknown>',mode='exec')
        except Exception as e:
            print(f"saveStructure --> ast parse failed in {file}: {e}")
            continue
        newBody=[]
        constantVar = []
        nonConstantVar = []
        decoratorLst = extractDecorator(root)
        #保留Import语句、函数和类定义
        for node in root.body:
            #增加AsyncFunctionDef节点信息保存 -- 2025/5/19 
            if isinstance(node,ast.Import) or isinstance(node,ast.ImportFrom) or isinstance(node,ast.ClassDef) or isinstance(node,ast.FunctionDef) or isinstance(node,ast.AsyncFunctionDef):
                newBody.append(node)
 
            #保留包含常量的全局赋值语句和装饰器调用相关的赋值语句，例如:
            # Case 1:
            #1.  a = 1
            #2.  b = 1
            #3.  c = a + b
            #4.  c = func(a,b)
            #仅保留1，2，3行代码
            # Case 2:
            #1. app = Flask(__name__)
            #2. @app.route("/")
            #保留 app = Flask(__name__)以解决NameError 
            if isinstance(node,ast.Assign):
                result = saveConstantAssign(node,constantVar,nonConstantVar,decoratorLst)
                if result==1:
                    continue
                else:
                   newBody+=result
 
        #保留包含常量的局部赋值语句
        for node in ast.walk(root):
            if isinstance(node,ast.Assign):
                result = saveConstantAssign(node,constantVar,nonConstantVar,decoratorLst)
                if result==1:
                    continue
                else:
                   for item in result:
                       if ast.dump(item) not in [ast.dump(i) for i in newBody]: 
                           newBody.append(item)
    
        root.body=newBody
        newFile=ast.unparse(root)
        with open(file,'w',encoding='utf-8') as fw:
            fw.write(f"{newFile}\n")
 
 
 
 
def ignore_sym_links(directory, files):
    return [f for f in files if os.path.islink(os.path.join(directory, f))]
 
 
 
 
def getLibImportLst(projPath,libName):
    lst=[]
    pathObj=Path('DF')
    pathObj.getPath(projPath)
    filePath=[file for file in pathObj.path if file.endswith('.py')]
    pattern=rf"(from {libName}|import {libName})" #确保库的前面不会出现其它字符
    for file in filePath:#下面的所有操作都是对项目副本进行的
        # with open(file,'r') as fr:
        with open(file, 'r', encoding='UTF-8') as fr:
            code=fr.read()
        try:
            root=ast.parse(code,filename='<unknown>',mode='exec')
            for node in ast.walk(root):
                if isinstance(node,ast.Import) or isinstance(node,ast.ImportFrom):
                    s=ast.unparse(node)
                    if bool(re.search(pattern,s)):
                        lst.append(s)
        except Exception as e:
            print(f"getLibImportLst: ast parse failed, {file}, {e}")
 
 
    ansLst=list(set(lst))
    ansLst.sort(key=lst.index) 
    return ansLst 
 
 
 
 
def convertTabsToSpaces(directory):
    for root, dirs, files in os.walk(directory):
        for file in files:
            if file.endswith('.py'):
                file_path = os.path.join(root, file)
                try:
                    with open(file_path, 'r', encoding='utf-8') as f:
                        content = f.read()
                    # 将制表符转换为4个空格
                    content = content.expandtabs(4)
                    with open(file_path, 'w', encoding='utf-8') as f:
                        f.write(content)
                except Exception as e:
                    print(f"Error converting file {file_path}: {e}")
 
 
 
 
def writeRecordValue(filePath,pklRelPath,useCallsiteName):
    scriptPath=os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
                            'Script','recordValue.py')
    with open(scriptPath,'r',encoding='UTF-8') as fr:
        content=fr.read()
    content=content.replace("'__PCART_PKL_REL_PATH__'",repr(pklRelPath))
    content=content.replace('__PCART_USE_CALLSITE_NAME__',str(bool(useCallsiteName)))
    with open(filePath,'w',encoding='UTF-8') as fw:
        fw.write(content)
 
 
 
def scriptPath(scriptName):
    return os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
                        'Script',scriptName)
 
 
 
def codeProcess(projPath,runCommand,runPath,libName,workspace,pcresolveLookup=None):
    runtimePaths=getRuntimePaths(workspace)
    copyRoot=runtimePaths['copy_root']
    dynamicRoot=runtimePaths['dynamic_root']
    dataDir=runtimePaths['data_dir']
    #提取运行的文件
    runFileLst=[]
    runFile=getRunFile(runCommand)
    prefix='' #运行文件所在的目录，默认是在项目的一级子目录下
    # if '/' in runFile:
    if runFile and ('/' in runFile or '\\' in runFile):
        # prefix=runFile.rsplit('/',1)[0]
        prefix = os.path.dirname(runFile)
        # runFile=runFile.rsplit('/',1)[1] #去掉路径前缀，只保留文件名即run.py
        runFile = os.path.basename(runFile)
    if runFile:
        runFileLst.append(runFile)
    #这种情况针对于python run.py, run.py在其它目录中比如src，则prefix就是src 
    #若是python src/run.py, 则prexfix和runPath是一致的
    if runPath!='' and prefix!=runPath:
        prefix=runPath
    # projName=projPath.split('/')[-1]
    projName = os.path.basename(projPath)
    copyProjPath=os.path.join(copyRoot,projName)
    importStatement=''
    for it in runFileLst:
        # it=it.rstrip('.py') #把文件名中的后缀去掉，但遇到display.py会变成displa
        it=it[0:-3]
        importStatement+=f"from {it} import *\n"
 
 
    #找出项目中所有和第三方库相关的import语句
    libImportLst=getLibImportLst(projPath,libName)
    libImportLst.append(importStatement) 
 
    #清除Copy和Dynamic中遗留的项目信息，然后把新项目的信息拷贝进去
    if os.path.isdir(copyRoot):
        shutil.rmtree(copyRoot)
    os.makedirs(os.path.dirname(copyRoot) or '.',exist_ok=True)
    shutil.copytree(projPath,os.path.join(copyRoot,projName),ignore=ignore_sym_links)
    os.mkdir(os.path.join(copyRoot,'pkl'))
    if os.path.isdir(dynamicRoot):
        shutil.rmtree(dynamicRoot)
    os.makedirs(os.path.dirname(dynamicRoot) or '.',exist_ok=True)
    shutil.copytree(projPath,os.path.join(dynamicRoot,projName),ignore=ignore_sym_links)
    
    #去掉项目代码中的冗余信息，仅保存项目代码的结构信息（import,functionDef, classDef）
    saveStructure(os.path.join(dynamicRoot,projName),libName)
    
    dynamicScriptDir=os.path.join(dynamicRoot,projName,prefix)
    shutil.copy2(scriptPath('addValueForAPI.py'),dynamicScriptDir)
    shutil.copy2(scriptPath('codeUtils.py'),dynamicScriptDir)
    shutil.copy2(scriptPath('dynamicMatch.py'),dynamicScriptDir)
    shutil.copy2(scriptPath('verifySingle.py'),dynamicScriptDir)
    
    #更新脚本中的from ... import ...语句,因为加载pkl的时候需要依赖于项目的结构信息
    modifyFromImport(os.path.join(dynamicScriptDir,'addValueForAPI.py'),libImportLst)
    modifyFromImport(os.path.join(dynamicScriptDir,'dynamicMatch.py'),libImportLst)
    modifyFromImport(os.path.join(dynamicScriptDir,'verifySingle.py'),libImportLst)
    
    #清除data中的数据
    if os.path.isdir(dataDir):
        shutil.rmtree(dataDir)
    os.makedirs(dataDir)
    
    
    #然后再把Copy中的项目制表符统一转化为空格,目的是为了插入字典的时候计算空格缩进
    convertTabsToSpaces(copyRoot)
 
    #把代码换行写的合成一行，并添加字典
    pathObj=Path('DF')
    pathObj.getPath(copyProjPath)
    filePath=[file for file in pathObj.path if file.endswith('.py')]
    for file in filePath:#下面的所有操作都是对项目副本进行的
        oneLine(file)
    
    #处理单行条件返回语句
    for file in filePath:
        expandConditionalReturn(file) 
 
    #处理局部变量
    for file in filePath:
        convertLocalVar(file,libName) 
 
 
    shutil.copytree(os.path.join(copyRoot,projName),os.path.join(copyRoot,f'bak_{projName}'))
 
    # 计算recordValue.py到Copy/pkl的相对路径
    pklDepth=1
    if prefix:
        pklDepth+=len([s for s in prefix.replace('\\','/').strip('/').split('/') if s])
    pklRelPath='/'.join(['..']*pklDepth+['pkl'])
    writeRecordValue(os.path.join(copyRoot,f'bak_{projName}',prefix,'recordValue.py'),pklRelPath,0)
    
    
    for file in filePath:
        if pcresolveLookup is None:
            addDictAll(projPath,projName,file,copyRoot,runFileLst,libName,runPath,runCommand)
        else:
            addDictAll(projPath,projName,file,copyRoot,runFileLst,libName,runPath,runCommand,pcresolveLookup=pcresolveLookup)
    
    #再对bak_proj中的运行文件进行插桩
    for file in runFileLst:
        file=os.path.join(copyRoot,f'bak_{projName}',prefix,file)
        handleRunFile(file,runPath,runCommand) 
    # bak项目会在current pkl生成后换回Copy/{projName}，其运行文件也依赖codeUtils
    shutil.copy2(scriptPath('codeUtils.py'),os.path.join(copyRoot,f'bak_{projName}',prefix))
    
    #处理完项目所有文件后，再给项目添加一个新的文件
    writeRecordValue(os.path.join(copyRoot,projName,prefix,'recordValue.py'),pklRelPath,1)
    
    shutil.copy2(scriptPath('codeUtils.py'),os.path.join(copyRoot,projName,runPath))
    if prefix!=runPath:
        shutil.copy2(scriptPath('codeUtils.py'),os.path.join(copyRoot,projName,prefix))