bad_clone_prediction/FunctionIdUpdater.py

from typing import List

import GlobalConstants
from models.MethodInfo import MethodInfo
from models.RepoInfo import RepoInfo
from MySQLOperator import MySQLOperator


def update_function_id(repoInfo: RepoInfo, mysqlOp: MySQLOperator):
    blob_methods_tablename = (
        "{ownername}{separator}{reponame}{separator}blob_methods".format(
            ownername=repoInfo.ownername,
            reponame=repoInfo.reponame,
            separator=GlobalConstants.SEPARATOR,
        )
    )

    method_function_relation_tablename = (
        "{ownername}{separator}{reponame}{separator}method_function_relations".format(
            ownername=repoInfo.ownername,
            reponame=repoInfo.reponame,
            separator=GlobalConstants.SEPARATOR,
        )
    )

    def get_related_methods(method_id: int) -> List[int]:
        sql = """
            with recursive temp as (
                select `method_id_2` from `{method_function_relation_tablename}` where `method_id_1`={method_id_1}
                union distinct
                select bm.`method_id_2` from `{method_function_relation_tablename}` bm inner join temp on temp.`method_id_2`=bm.`method_id_1`
            )
            select `method_id_2` from temp;
        """.format(
            method_function_relation_tablename=method_function_relation_tablename,
            method_id_1=method_id,
        )
        mysqlOp.cursor.execute(sql)
        method_ids_2 = mysqlOp.cursor.fetchall()
        method_ids_2 = [method_id["method_id_2"] for method_id in method_ids_2]

        sql = """
            with recursive temp as (
                select `method_id_1` from `{method_function_relation_tablename}` where `method_id_2`={method_id_2}
                union distinct
                select bm.`method_id_1` from `{method_function_relation_tablename}` bm inner join temp on temp.`method_id_1`=bm.`method_id_2`
            )
            select `method_id_1` from temp;
        """.format(
            method_function_relation_tablename=method_function_relation_tablename,
            method_id_2=method_id,
        )
        mysqlOp.cursor.execute(sql)
        method_ids_1 = mysqlOp.cursor.fetchall()
        method_ids_1 = [method_id["method_id_1"] for method_id in method_ids_1]

        method_ids = [method_id]
        method_ids = list(
            (set(method_ids).union(set(method_ids_1))).union(set(method_ids_2))
        )

        return method_ids

    mysqlOp.cursor.execute(
        "select id from `{blob_methods_tablename}` where function_id is null".format(
            blob_methods_tablename=blob_methods_tablename
        )
    )
    methods = mysqlOp.cursor.fetchall()
    methods = [method["id"] for method in methods]
    while len(methods) > 0:
        method_id = methods[0]
        method_ids = get_related_methods(method_id=method_id)
        mysqlOp.cursor.execute(
            "select max(function_id) as max_function_id from `{blob_methods_tablename}`".format(
                blob_methods_tablename=blob_methods_tablename
            )
        )
        max_function_id = mysqlOp.cursor.fetchone()["max_function_id"]
        if max_function_id is None:
            max_function_id = 0
        method_ids_str = [str(method_id) for method_id in method_ids]
        method_id_str = "({method_id_str})".format(
            method_id_str=",".join(method_ids_str)
        )
        mysqlOp.cursor.execute(
            "update `{blob_methods_tablename}` set function_id=%s where id in {method_id_str}".format(
                blob_methods_tablename=blob_methods_tablename,
                method_id_str=method_id_str,
            ),
            ((max_function_id + 1),),
        )

        for method_id in method_ids:
            if method_id in methods:
                methods.remove(method_id)
    mysqlOp.connection.commit()