Phân tích một cơ sở mã khổng lồ với Python

Oct 15 2020

Tôi đã viết một chương trình Python để phân tích một cơ sở mã C ++ khổng lồ vượt quá hàng triệu dòng mã. Công việc của chương trình chỉ đơn giản là tìm kiếm tất cả các bình luận kiểu C và C ++ và trích xuất tất cả các bình luận có chứa các từ khóa và cụm từ cụ thể và ghi các bình luận đó vào một tệp liên kết. Cho đến nay chương trình hoạt động đủ tốt. Nó khá nhanh chóng và dễ dàng để cấu hình và nó thực hiện công việc, nhưng nó không tìm kiếm ở bất kỳ đâu nhanh như tôi muốn và tôi muốn một số lời khuyên về cách làm cho mọi thứ chạy nhanh hơn.

Cấu hình

Chương trình rất nhanh chóng và dễ dàng cấu hình. Nếu bạn muốn tìm kiếm cơ sở mã cho một biểu thức duy nhất, tất cả những gì bạn phải làm là chuyển nó biểu thức đó làm đối số và thư mục để tìm kiếm và nó sẽ thực hiện phần còn lại cho bạn. Để tìm kiếm các bình luận có chứa từ "hack" trong thư mục src /, bạn chỉ cần viết:

./main.py -e hack -s ./src/

Tuy nhiên, nếu bạn muốn truyền nhiều biểu thức, bạn cần sử dụng tệp JSON được tạo đặc biệt và chuyển tên của tệp đó làm đối số. Một tệp JSON mẫu có thể trông giống như sau:

[
  {
    "hack-stuff": [
      "hack",
      "hacked",
      "hacky"
    ]
  },
  "WARNING",
  "DANGER",
  [
    "bad",
    "badly"
  ]
]

Sau đó, tệp được chuyển tới chương trình như sau:

./main.py -f test-words.json -s ./src/

Điều này sẽ tạo ra bốn tệp: "hack-things", "WARNING", "DANGER" và "bad". Tệp "hack-things" sẽ là nơi đặt tất cả các bình luận có chứa từ hack, hack và hacky, "bad" sẽ là nơi đặt tất cả các bình luận có chứa "bad" và "bad", và "WARNING" và " NGUY HIỂM "sẽ là nơi đặt các bình luận có chứa" CẢNH BÁO "và" NGUY HIỂM ". Ví dụ này trình bày ba cách (chuỗi, danh sách hoặc từ điển) bạn có thể chỉ định nhận xét nào phù hợp mà bạn muốn đặt trong tệp nào.

Cấu trúc chương trình

Cốt lõi của chương trình là lớp SourceWalker chứa tất cả logic nội bộ cần thiết để phân tích cơ sở mã và viết nhận xét vào các tệp tương ứng của chúng. Cũng có một hàm chính trong một tệp riêng biệt, hàm này đọc các đối số thành các biến có liên quan của chúng và thực hiện xử lý JSON (nếu cần) trước khi khởi tạo và chạy một phiên bản của lớp thông qua phương thức walk ().

Hiệu suất

Tôi đã thử một số cách để làm cho chương trình hoạt động hiệu quả nhất có thể, bao gồm cả việc kết hợp đa xử lý mang lại những cải tiến lớn, nhưng tôi không chắc mình có thể làm gì để làm cho nó nhanh hơn vào thời điểm này. Sự chậm lại chính là do vòng lặp for trong _process_files () trên dòng 117. Đây là phần của chương trình chạy trong từng quy trình con và tìm kiếm qua từng tệp để tìm các nhận xét hợp lệ trước khi kiểm tra chúng với một loạt các quy trình các biểu thức chính quy được biên dịch để xem liệu chúng có khớp với một trong các biểu thức mà chúng tôi đang tìm kiếm hay không. Tôi chắc rằng có nhiều cách tốt hơn để trích xuất các nhận xét từ mỗi tệp và sau đó tìm kiếm chúng nhưng tôi không chắc chúng sẽ như thế nào. Bất kỳ đề xuất nào ở đây sẽ được đánh giá rất cao.

Ý kiến ​​khác

Mặc dù hiệu suất là mối quan tâm chính của tôi ở đây, tôi cũng đánh giá cao bất kỳ phản hồi nào về tính đúng đắn và phong cách của chương trình của tôi. Nó có vẻ hoạt động như dự định nhưng tôi không thể đảm bảo không có một số trường hợp cạnh mà tôi đã bỏ qua trong quá trình thử nghiệm của mình.

Mật mã

main.py

#!/usr/bin/python3

import sys
import json
import os
import argparse
import SourceWalker

def initialiseParser():
    parser = argparse.ArgumentParser(description = "Search the contents of comments within source code files")

    parser.add_argument("--file_name", "--file", "-f", help = "Name of the file containing the JSON list of expressions to search for", type = str)
    parser.add_argument("--source-dir", "--src", "-s", help = "The root directory of the source files to search over", type = str)
    parser.add_argument("--output-dir", "--out", "-o", help = "The directory the output files will be placed in", type = str)
    parser.add_argument("--expression", "--expr", "-e", help = "The expression to search for within the source comments", type = str)
    parser.add_argument("--language", "--lang", "-l", help = "The style of comments to look for within the file", type = str)

    return parser

def main():
    parser = initialiseParser()
    args = parser.parse_args()

    if args.source_dir:
        source_dir = args.source_dir
    else:
        sys.exit("Source directory must be specified!")

    if args.file_name:
        file_name = args.file_name
        input_file = open(file_name, "r")
        expressions = json.loads(input_file.read())
    elif args.expression:
        expressions = []
        expressions.append(str(args.expression))
    else:
        sys.exit("Error: Expression or file containing expressions must be specified!")

    output_dir = "./comments/"

    if args.output_dir:
        output_dir = args.output_dir

    if not os.path.exists(output_dir):
        os.makedirs(output_dir)

    walker = SourceWalker.SourceWalker(source_dir, output_dir, expressions, extensions=[ ".c", ".cpp", ".h", ".cxx" ])
    walker.walk()

    if "input_file" in locals():
        input_file.close()

    return

if __name__=="__main__":
    main()

SourceWalker.py #! / Usr / bin / python3

import sys
import json
import re
import os
import multiprocessing
import codecs
from pathlib import Path

class SourceWalkerException(Exception):
    """
    Raised when there is an error processing the given expressions
    TODO: Make error handling more informative and clean up. Should display a message to the user explaing what went wrong and close all open files.
    """
    pass

class SourceWalker:
    _output_file_names = []
    _regexes = {}
    _expr_file_names = {}

    def __init__(self, source_dir, output_dir, expr_list, encoding = "ISO-8859-1", process_count = 12, extensions = [ ".c", ".h" ]):
        try:
            if not os.path.exists(source_dir) or not os.path.exists(output_dir):
                raise NotADirectoryError

            if process_count < 1:
                raise SourceWalkerException("Process count cannot be less than one!")

            codecs.lookup(encoding)

            if not isinstance(extensions, list):
                raise SourceWalkerException("Extensions must be passed as a list!")

            for extension in extensions:
                if extension[0] != '.':
                    raise SourceWalkerException("Extensions must start with a \'.\'!")
                elif len(extension) <= 1:
                    raise SourceWalkerException("Extensions must be more than one character long!")
        except NotADirectoryError as exception:
            raise SourceWalkerException("Directory does not exist! " + str(exception))
        else:
            self._source_dir = source_dir
            self._output_dir = output_dir
            self._encoding = encoding
            self._expr_list = expr_list
            self._process_count = process_count
            self._extensions = extensions
            self._process_expr_list()

    def _process_expr_list(self):
        for expr in self._expr_list:
            try:
                if isinstance(expr, list):
                    if len(expr) == 0:
                        raise SourceWalkerException("Expression list cannot be empty!")
                    
                    output_file_name = expr[0]

                    if not isinstance(output_file_name, str):
                        raise SourceWalkerException("Expression sub-lists can only contain strings!")
                    
                    for sub_expr in expr:
                        if not isinstance(sub_expr, str):
                            raise SourceWalkerException("Expression sub-lists can only contain strings!")
                        elif sub_expr in self._regexes.keys():
                            raise SourceWalkerException("Expressions can only appear once in the expression list!")

                        self._regexes[sub_expr] = re.compile("\s+%s(\s|,|:|;|\n)+" % (sub_expr)) # Naieve regex to catch expressions
                        self._expr_file_names[sub_expr] = self._output_dir + output_file_name

                    self._output_file_names.append(self._output_dir + output_file_name)
                elif isinstance(expr, dict):
                    if len(expr.keys()) == 0:
                        raise SourceWalkerException("Expression dictionary cannot be empty!")

                    output_file_name = list(expr)[0]

                    if not isinstance(expr[output_file_name], list):
                        raise SourceWalkerException("Expression dictionary cannot be empty!")
                    
                    for sub_expr in expr[output_file_name]:
                        if not isinstance(sub_expr, str):
                            raise SourceWalkerException("Expression sub-lists can only contain strings!")
                        elif sub_expr in self._regexes.keys():
                            raise SourceWalkerException("Expressions can only appear once in the expression list!")

                        self._regexes[sub_expr] = re.compile("\s+%s(\s|,|:|;|\n)+" % (sub_expr))
                        self._expr_file_names[sub_expr] = self._output_dir + output_file_name

                    self._output_file_names.append(self._output_dir + output_file_name)
                elif isinstance(expr, str):
                    if expr in self._regexes.keys():
                        raise SourceWalkerException("Expressions can only appear once in the expression list!")

                    self._output_file_names.append(self._output_dir + expr)
                    self._regexes[expr] = re.compile("\s+%s(\s|,|:|;|\n)+" % (expr))
                    self._expr_file_names[expr] = self._output_dir + expr
                else:
                    raise SourceWalkerException("Expression list can only contain dictionaries, lists, and strings!")
            except SourceWalkerException as exception:
                self.cleanup()
                raise

    def _process_files(self, input_files, output_files, mutexes): # Find way to process different types of source file, I'd rather not be limited to C only...
        for file_name in iter(input_files.get, None):
            with open(file_name, "r", encoding = self._encoding) as file_object:
                in_multi_comment = False
                in_single_comment = False
                in_string = False
                prev_char = ''
                comment = ''

                for line_num, line in enumerate(file_object, 1):
                    for char in line:
                        if char == '/':
                            if in_string or in_single_comment:
                                prev_char = char
                                continue
                            if prev_char == '*':
                                in_multi_comment = False
                                comment += char

                                for expr in self._regexes.keys():
                                    if self._regexes[expr].search(comment):
                                        mutexes[expr].acquire()
                                        os.write(output_files[expr], ("%s: %s %s\n" % (file_name, str(line_num), comment)).encode())
                                        mutexes[expr].release()

                                comment = ''
                            elif prev_char == '/':
                                in_single_comment = True
                                comment += prev_char
                        elif char == '*':
                            if in_string or in_single_comment or in_multi_comment:
                                if in_single_comment or in_multi_comment:
                                    comment += char

                                prev_char = char
                                continue
                            if prev_char == '/':
                                in_multi_comment = True
                                comment += prev_char
                        elif char == '"':
                            if prev_char == '\\' or in_single_comment or in_multi_comment:
                                prev_char = char
                                continue
                            
                            in_string = not in_string

                        prev_char = char

                        if in_single_comment or in_multi_comment:
                            comment += char

                    if in_single_comment:
                        in_single_comment = False
                        
                        for expr in self._regexes.keys():
                            if self._regexes[expr].search(comment):
                                mutexes[expr].acquire()
                                os.write(output_files[expr], ("%s: %s %s" % (file_name, str(line_num), comment)).encode())
                                mutexes[expr].release()

                        comment = ''

    def walk(self):
        input_files = multiprocessing.Queue(0)
        processes = []
        mutexes = {}
        output_files = {}

        for fname in self._output_file_names:
            try:
                file_handle = os.open(fname, os.O_WRONLY | os.O_CREAT)
                mutex = multiprocessing.Lock()
            except IOError:
                for file in output_files.keys():
                    output_files[file].close()

                raise SourceWalkerException("Error: Could not open output file %s, skipping!" % fname)

            for expr in self._expr_file_names.keys():
                if self._expr_file_names[expr] == fname:
                    output_files[expr] = file_handle
                    mutexes[expr] = mutex

        for root, dirs, file_names in os.walk(self._source_dir):
            for file_name in file_names:
                if any(ext in Path(file_name).suffix for ext in self._extensions):
                    input_files.put(os.path.join(root, file_name))

        for i in range(self._process_count):
            input_files.put(None)
        
        for cur_process in range(self._process_count):
            process = multiprocessing.Process(target = self._process_files, args = (input_files, output_files, mutexes))
            processes.append(process)
            process.start()

        for i in range(1, self._process_count):
            processes[i].join()

        for file in output_files.keys(): # Close the file associated with each expression
            try:
                os.close(output_files[file]) # Since multiple expressions can be associated with the same file we need to avoid invalid file closures
            except:
                pass

Trả lời

3 user985366 Oct 15 2020 at 05:32

Nếu tôi hiểu mô tả của bạn, bạn chỉ đang tìm kiếm nhận xét nhưng bạn đang tìm kiếm toàn bộ cơ sở mã mọi lúc. Vì bình luận thường là một phần nhỏ của mã (ít hơn 10%?) Tôi khuyên bạn nên thực hiện bước tiền xử lý trước, nơi bạn chỉ cần trích xuất tất cả các bình luận và sau đó thực hiện tìm kiếm thực tế trên những bình luận đó.

Bằng cách "giải nén", tôi có nghĩa là lưu nhận xét trong các tệp riêng biệt để bạn chỉ có thể tìm kiếm trong các tệp đó.

Đối với các tìm kiếm lặp lại trong cùng một cơ sở mã, điều này sẽ hữu ích vì quá trình xử lý trước có thể được thực hiện một lần và sau đó tìm kiếm thực tế có ít văn bản hơn để xem qua.

3 l0b0 Oct 15 2020 at 06:15

Bạn có thể chỉ định đối số cần thiết trong argparsechứ không phải xử lý chúng mình. Bạn cũng có thể chỉ định các giá trị mặc định , chẳng hạn như output_dir.

os.makedirsmất exist_ok=Trueđể chỉ ra rằng nó chỉ nên tạo thư mục nếu nó không tồn tại được nêu ra.

Đối extensionssố cho hàm SourceWalkertạo có lẽ không nên được mã hóa cứng.

Sử dụng with open(file_name, "r") as file_handle:để đảm bảo tệp luôn được đóng khi rời khỏi ngữ cảnh đó. Một lưu ý liên quan, locals()IMO chỉ nên được sử dụng như một phương sách cuối cùng, vì thật khó để theo dõi mã nơi các chuỗi được sử dụng để tham chiếu đến các biến và IDE không thể làm bất cứ điều gì hữu ích với mã đó. Ví dụ: nếu bạn định đổi tên input_filethành input_file_path(tôi khuyên bạn nên làm vậy trong mọi trường hợp), việc quên thay đổi tham chiếu chuỗi sẽ rất nhỏ.

Tôi khuyên bạn nên sử dụng blackvà isortđịnh dạng mã. Theo cách đó, nó sẽ gần với phong cách thành ngữ hơn, không cần làm thủ công.

Sau khi định dạng, tôi khuyên bạn nên chạy flake8để tìm mã không có thành ngữ khác. Ví dụ, có vẻ như SourceWalkermã bị hỏng - các phương thức ở cùng cấp với định nghĩa lớp.

Tạo một ngoại lệ với một chuỗi tĩnh, chẳng hạn như SourceWalkerException("Process count cannot be less than one!"), là một mùi mã. Nó không nhất thiết là xấu, nhưng giống như các tham số boolean, nó thường có nghĩa là một cái gì đó nên được tách ra. Trong trường hợp này, tốt hơn là nên có một lớp ngoại lệ riêng cho số quá trình quá thấp (và các điều kiện lỗi khác), có thể là một cái gì đó đơn giản như vậy ProcessCountTooLowError. Bằng cách đó, điều duy nhất được truyền cho một hàm tạo ngoại lệ (nếu có) là bất kỳ nội dung động nào có thể được sử dụng để gỡ lỗi ngoại lệ đó. Ngoài ra, điều này có nghĩa là nếu bạn muốn xử lý ngoại lệ, bạn có thể except ProcessCountTooLowErrorthay vì phải phân tích cú pháp thông báo lỗi bên trong một thông báo chung except SourceWalkerException. (Ồ, và các lớp ngoại lệ tùy chỉnh phải kết thúc bằng Error, không phải Exception.)

Thay vì kiểm tra những thứ như liệu thư mục đầu ra có tồn tại hay không, tôi sẽ để ứng dụng bị lỗi khi nó đạt đến mã cố gắng ghi vào thư mục đó. Đây được gọi là "thời gian kiểm tra đến thời điểm sử dụng" - về cơ bản, liệu thư mục có tồn tại khi bạn thực hiện kiểm tra đó hay không không liên quan đến việc liệu nó có tồn tại khi mã thực sự cố gắng sử dụng nó hay không. Nó cũng là một kiểm tra quá hẹp, vì ví dụ, thư mục cũng không thể ghi được bởi người dùng hiện tại, hoặc hệ thống tệp có thể đầy.

Trên một lưu ý tương tự, chạy codecs.lookup(encoding)để kiểm tra mã hóa tồn tại trước khi thực sự sử dụng nó trong một cuộc gọi hoàn toàn khác có vẻ như nó có thể dễ dàng là một kiểm tra chưa hoàn thành. Có thể tốt hơn nếu giới hạn tham số mã hóa optionschỉ với các mã hóa có sẵn. Bằng cách đó, nó được kiểm tra càng sớm càng tốt, người dùng nhận được một lỗi có thể xử lý tốt và văn bản trợ giúp hiển thị các giá trị có thể.

Số nhiều của "regex" là "regex e s".

Không sử dụng các đối số mặc định có thể thay đổi ( thảo luận ). Trên một lưu ý liên quan, khi nào bạn nên sử dụng các đối số mặc định? Hai quy tắc chung:

  1. Nếu mặc định không thực sự được sử dụng vì tất cả các lệnh gọi chỉ định một giá trị, thì mặc định là vô nghĩa.
  2. Nếu không có lệnh gọi nào ghi đè mặc định thì nó cũng vô nghĩa - nó cũng có thể là một biến hoặc hằng số.

if not isinstance(extensions, list):không phải là thành ngữ; nó phải hoàn toàn hợp lệ để chuyển vào bất kỳ có thể lặp lại nào chẳng hạn như a tuple.

Nói chung, các trình xây dựng không nên làm gì phức tạp hơn việc đặt các giá trị trường. Sau khi hoàn tất run, một processhoặc phương thức khác sẽ được chạy riêng để xử lý thực tế. Tôi không nhớ lần đầu tiên tôi thấy điều này được giải thích rõ ràng ở đâu, nhưng hãy xem ví dụ Tại sao sử dụng tác dụng phụ lại là phương pháp không tốt trong các trình tạo JavaScript? và Người xây dựng phải không có mã

1 Reinderien Oct 17 2020 at 01:32

Liệt kê các chữ

    expressions = []
    expressions.append(str(args.expression))

chỉ nên là

    expressions = [str(args.expression)]

Pathlib

Điều này:

if not os.path.exists(output_dir):
    os.makedirs(output_dir)

nên sử dụng pathlibtương đương mới sáng bóng :

Path(output_dir).mkdir(exist_ok=True)

Cùng một mô-đun có thể được sử dụng cho

if not os.path.exists(source_dir) or not os.path.exists(output_dir):

Sự tồn tại biến đổi

if "input_file" in locals():
    input_file.close()

là sơ sài. Thông thường, cách để chỉ ra trong Python rằng một biến có giá trị hay không là có khả năng nhận None, không có khả năng không được khai báo. Bạn thậm chí có thể đánh dấu nó là có thể-không có bằng cách sử dụng Optionalgợi ý loại.

Trở về

Đơn returnở cuối main()là thừa.

Độ dài từ điển

if len(expr.keys()) == 0:

có thể

if len(expr) == 0:

hoặc thậm chí

if not expr:

Regex biên dịch

                self._regexs[sub_expr] = re.compile("\s+%s(\s|,|:|;|\n)+" % (sub_expr))

ít nhất cũng cần một dẫn rđể làm cho chuỗi đó trở thành nguyên bản.

Nói chung, lấy đầu vào từ một tệp và coi nó như một biểu thức phụ không được xác thực là một ý tưởng tồi. Nó không evaltệ ở cấp độ, nhưng nó không tốt. Đây là các biểu thức chính quy thực tế hay chỉ là các chuỗi con? Nếu chúng chỉ là các chuỗi con, hãy gọi escapechúng trước khi chèn chúng vào biểu thức chính quy bên ngoài của bạn.

Nếu chúng thực sự là biểu thức chính quy của riêng chúng, ít nhất bạn sẽ muốn đặt từng biểu thức đó vào nhóm không bắt của riêng nó trong biểu thức bên ngoài, để tránh những bất ngờ khó chịu.

Các ngoại lệ bị xích

except NotADirectoryError as exception:
    raise SourceWalkerException("Directory does not exist! " + str(exception))

nên là

except NotADirectoryError as exception:
    raise SourceWalkerException("Directory does not exist! " + str(exception)) from exception