批量修复golang ci

golangci-lint-fix.py

import re
import subprocess

# 不自动处理的规则
ex = ("SA4006", "structcheck")  # 不自动处理的规则

# 忽略error返回白名单，白名单才处理
errwl = {
    "ymetrics": "_ = ",
    "yalert": "_ = ",
    "proto.Unmarshal": "_ = ",
    "json.Unmarshal": "_ = ",
    "jsoniter.Unmarshal": "_ = ",
    "c.Send": " _ = ",
}

# strings忽略大小写的比较
reef = re.compile("^strings\\.\\w+\\(([\\.\\w]+)\\) == [\\.\\w]+\\(([\\.\\w]+)\\)")

files = {}


# 获取文件名、行号、错误类型
def getfilelinetype(l):
    ss = l.split(":")
    print(ss)
    return ss[0], int(ss[1]) - 1, int(ss[2])


# startswith，跳过\t
def sw(l, p):
    return l.strip().startswith(p)


# endswith
def ew(l, s):
    return l.strip().endswith(s)


# 处理一个文件
class File:
    def __init__(self, name):
        self.name = name
        fh = open(name, "r")
        self.ls = fh.readlines()
        fh.close()
        self.comments = set({})  # 被注释的行
        self.dels = set({})  # 被删除的行

    def Unused(self, l, offset):
        if not ew(self.ls[l], "{"):
            self.comments.add(l)
        else:
            for i in range(l, len(self.ls), 1):
                self.comments.add(i)
                if self.ls[i].startswith("}"):  # 不能用sw，不用中间的block也会有一个独立的}
                    break
                elif len(self.ls[i]) > offset and self.ls[i][offset:].startswith("}"):
                    break

    def RemoveBlankReceiver(self, l):
        self.ls[l] = self.ls[l].replace(", _", "")

    def MapSimple(self, l):
        self.ls[l] = self.ls[l].replace(", 0)", ")")

    def EqualFold(self, l, offset):
        self.ls[l] = self.ls[l][:offset] + re.sub(
            reef, "strings.EqualFold(\\1, \\2)", self.ls[l][offset:]
        )

    def ErrorCheck(self, l):  # 白名单的函数才会处理，其他的还是得看看需不需要判断
        if sw(self.ls[l], "defer "):  # defer的这种，需要加个函数体才行，这里的修改不一定对，返回值的数量都默认按1个处理了
            ts = False
            nts = False
            for i in range(l, len(self.ls)):
                if self.ls[i].find("time.Now().Unix()") >= 0:
                    ts = True
                    self.ls[i] = self.ls[i].replace("time.Now().Unix()", "ts", -1)
                if self.ls[i].find("time.Now().UnixNano()") >= 0:
                    nts = True
                    self.ls[i] = self.ls[i].replace("time.Now().UnixNano()", "nts", -1)
                if ew(self.ls[i], ")"):
                    p1 = ""
                    p2 = ""
                    if ts:
                        if nts:
                            p1 = "ts, nts int64"
                            p2 = "time.Now().Unix(), time.Now().UnixNano()"
                        else:
                            p1 = "ts int64"
                            ps = "time.Now().Unix()"
                    elif nts:
                        p1 = "nts int64"
                        p2 = "time.Now().UnixNano()"
                    self.ls[l] = self.ls[l].replace(
                        "defer ", "defer func({}) {{ _ = ".format(p1)
                    )
                    self.ls[i] = self.ls[i].replace("\n", "}} ({})\n".format(p2))
                    break
        elif sw(self.ls[l], "go "):
            self.ls[l] = self.ls[l].replace("go ", "go func() { _ = ")
            for i in range(l, len(self.ls)):
                if ew(self.ls[i], ")"):
                    self.ls[i] = self.ls[i].replace("\n", "} ()\n")
                    break
        else:
            for w in errwl:
                if sw(self.ls[l], w):
                    self.ls[l] = errwl[w] + self.ls[l]
                    break

    def ContextNil(self, l, offset):
        self.ls[l] = self.ls[l][:offset] + self.ls[l][offset:].replace(
            "nil", "context.Background()", 1
        )

    def SprintConst(self, l, offset):
        self.ls[l] = self.ls[l][:offset] + self.ls[l][offset:].replace(
            "fmt.Sprintf(", "", 1
        ).replace('")', '"', 1)

    def SprintNoArg(self, l, offset):
        self.ls[l] = self.ls[l][:offset] + self.ls[l][offset:].replace(
            "fmt.Sprintf(", "", 1
        ).replace(")", "", 1)

    def SimpleReturn(
        self, l, line
    ):  # should use 'return status.IsVideoOpen' instead of 'if status.IsVideoOpen { return true }; return false' (gosimple)
        idx = line.find(" should use '")
        line = line[idx + len(" should use '") :]
        idx = line.find("'")
        simple = line[:idx]
        idx = line.rfind("return ")
        p = line[idx:]
        idx = p.find("'")
        end = p[:idx]
        # print('simple is:',simple, ' end is:', end)
        if self.ls[l].find(end) > 0:
            self.ls[l] = simple + "\n"
        else:
            self.ls[l] = simple + "\n"
            for i in range(l + 1, len(self.ls)):
                self.dels.add(i)
                if self.ls[i].find(end) >= 0:
                    break

    def Write(self):
        fh = open(self.name, "w")
        for i, l in enumerate(self.ls):
            if i in self.dels:
                continue
            elif i in self.comments:
                fh.write("//")
            fh.write(l)
        fh.close()


def getfile(name):
    if name in files:
        return files[name]
    f = File(name)
    files[name] = f
    return f


def process(k):
    for i in range(0, len(k), 3):
        line = k[i].decode("utf-8").strip()
        notcheck = False
        for t in ex:
            if line.find(t) >= 0:
                notcheck = True
        if notcheck:
            continue
        name, l, t = getfilelinetype(line)
        f = getfile(name)
        if (
            line.find("is unused (deadcode)") >= 0
            or line.find("is unused (unused)") >= 0
        ):  # unused
            f.Unused(l, len(k[i + 2]) - 1)
        elif line.find("redundant `return` statement") >= 0:  # 多余的return
            f.Unused(l, 0)
        elif (
            line.find("unnecessary assignment to the blank identifier") >= 0
        ):  # 取map的变量的时候，用了a,_ := m[xxx]
            f.RemoveBlankReceiver(l)
        elif line.find("should use make(map") >= 0:
            f.MapSimple(l)
        elif line.find("should use strings.EqualFold instead") >= 0:
            f.EqualFold(l, len(k[i + 2]) - 1)
        elif line.find("is not checked (errcheck)") >= 0:
            f.ErrorCheck(l)
        elif (
            line.find("pass context.TODO if you are unsure about which Context to use")
            >= 0
        ):
            f.ContextNil(l, len(k[i + 2]) - 1)
        elif line.find("unnecessary use of fmt.Sprintf (gosimple)") >= 0:
            f.SprintConst(l, len(k[i + 2]) - 1)
        elif (
            line.find(
                "printf-style function with dynamic format string and no further arguments should use print-style function instead"
            )
            >= 0
        ):
            f.SprintNoArg(l, len(k[i + 2]) - 1)
        elif line.find("should use 'return ") >= 0 and line.endswith("(gosimple)"):
            f.SimpleReturn(l, line)


cnt = 0
lk = 0
while True:
    cnt += 1
    files = {}
    print("running golangci-lint(第{}次) ...".format(cnt), end="", flush=True)
    ret = subprocess.run(
        [
            "golangci-lint",
            "run",
            "--tests=false",
            "--max-issues-per-linter=0",
            "--max-same-issues=0",
        ],
        stdout=subprocess.PIPE,
    )
    k = ret.stdout.splitlines()
    print("done. has {} issues".format(len(k) / 3))
    if lk == len(k):
        print("issue 数量稳定。lint结束")
        break
    else:
        lk = len(k)

    print("fixing issues ...", end="")
    process(k)
    for f in files:
        files[f].Write()
    print("done")

    print("running goimports ...", end="", flush=True)
    subprocess.run(["goimports", "-w", "./"], stdout=subprocess.PIPE)
    print("done")

print("running gofmt ...", end="", flush=True)
subprocess.run(["gofmt", "-w", "./"], stdout=subprocess.PIPE)
print("done")