การวิเคราะห์ Codebase ขนาดใหญ่ด้วย Python
ฉันเขียนโปรแกรม Python เพื่อวิเคราะห์ฐานรหัส C ++ ขนาดใหญ่ที่มีโค้ดเกินกว่าล้านบรรทัด งานของโปรแกรมคือค้นหาความคิดเห็นสไตล์ C และ C ++ ทั้งหมดและดึงความคิดเห็นทั้งหมดที่มีคำหลักและวลีที่เฉพาะเจาะจงและเขียนความคิดเห็นเหล่านั้นไปยังไฟล์ที่เกี่ยวข้อง จนถึงตอนนี้โปรแกรมทำงานได้ดีพอ มันค่อนข้างรวดเร็วและง่ายในการกำหนดค่าและทำงานได้ดี แต่มันไม่ได้ค้นหาที่ใดก็ได้ใกล้เร็วที่สุดเท่าที่ฉันต้องการและฉันต้องการคำแนะนำในการทำให้สิ่งต่างๆทำงานเร็วขึ้น
การกำหนดค่า
โปรแกรมนี้ง่ายและรวดเร็วในการกำหนดค่า หากคุณต้องการค้นหา codebase สำหรับนิพจน์เดียวสิ่งที่คุณต้องทำคือส่งนิพจน์นั้นเป็นอาร์กิวเมนต์และไดเร็กทอรีที่จะค้นหาและมันจะดำเนินการส่วนที่เหลือให้คุณ หากต้องการค้นหาความคิดเห็นที่มีคำว่า "แฮ็ก" ในไดเรกทอรี src / คุณสามารถเขียน:
./main.py -e hack -s ./src/
อย่างไรก็ตามหากคุณต้องการส่งผ่านหลาย ๆ นิพจน์คุณต้องใช้ไฟล์ JSON ที่สร้างขึ้นมาเป็นพิเศษและส่งชื่อไฟล์นั้นเป็นอาร์กิวเมนต์ ไฟล์ JSON ตัวอย่างอาจมีลักษณะดังนี้:
[
{
"hack-stuff": [
"hack",
"hacked",
"hacky"
]
},
"WARNING",
"DANGER",
[
"bad",
"badly"
]
]
จากนั้นไฟล์จะถูกส่งไปยังโปรแกรมดังนี้:
./main.py -f test-words.json -s ./src/
การดำเนินการนี้จะสร้างไฟล์สี่ไฟล์: "hack-stuff", "WARNING", "DANGER" และ "bad" ไฟล์ "hack-stuff" จะเป็นที่ซึ่งความคิดเห็นทั้งหมดที่มีคำว่า hack, hacked และ hacky จะถูกวางไว้ "bad" จะอยู่ที่ความคิดเห็นทั้งหมดที่มี "bad" และ "ไม่ดี" และ "คำเตือน" และ " DANGER "จะเป็นจุดที่แสดงความคิดเห็นที่มี" คำเตือน "และ" อันตราย "ตามลำดับ ตัวอย่างนี้แสดงให้เห็นถึงสามวิธี (สตริงรายการหรือพจนานุกรม) ที่คุณสามารถระบุความคิดเห็นที่ตรงกับที่คุณต้องการวางในไฟล์
โครงสร้างโปรแกรม
แกนหลักของโปรแกรมคือคลาส SourceWalker ซึ่งมีตรรกะภายในทั้งหมดที่จำเป็นในการวิเคราะห์โค้ดเบสและเขียนความคิดเห็นไปยังไฟล์ที่เกี่ยวข้อง นอกจากนี้ยังมีฟังก์ชันหลักในไฟล์แยกต่างหากซึ่งจะอ่านอาร์กิวเมนต์ในตัวแปรที่เกี่ยวข้องและดำเนินการประมวลผล JSON (ถ้าจำเป็น) ก่อนที่จะเริ่มต้นและเรียกใช้อินสแตนซ์ของคลาสผ่านวิธีการ walk ()
ประสิทธิภาพ
ฉันได้ลองทำหลายอย่างเพื่อให้โปรแกรมมีประสิทธิภาพมากที่สุดรวมถึงการรวมการประมวลผลหลายขั้นตอนซึ่งได้รับการปรับปรุงอย่างมาก แต่ฉันไม่แน่ใจว่าจะทำอย่างไรเพื่อให้เร็วขึ้น ณ จุดนี้ การชะลอตัวหลักเกิดจาก for loop ภายใน _process_files () ในบรรทัด 117 นี่คือส่วนของโปรแกรมที่ทำงานในแต่ละกระบวนการย่อยและค้นหาในแต่ละไฟล์เพื่อค้นหาความคิดเห็นที่ถูกต้องก่อนที่จะตรวจสอบกับชุดของ pre- รวบรวมนิพจน์ทั่วไปเพื่อดูว่าตรงกับนิพจน์ที่เราต้องการหรือไม่ ฉันแน่ใจว่ามีวิธีที่ดีกว่าในการแยกความคิดเห็นจากแต่ละไฟล์จากนั้นจึงค้นหาข้อมูลเหล่านั้น แต่ฉันไม่แน่ใจว่าจะเป็นอย่างไร ข้อเสนอแนะใด ๆ ที่นี่จะได้รับการชื่นชมอย่างมาก
ความคิดเห็นเพิ่มเติม
แม้ว่าประสิทธิภาพจะเป็นข้อกังวลหลักของฉันที่นี่ แต่ฉันก็ขอขอบคุณข้อเสนอแนะเกี่ยวกับความถูกต้องและรูปแบบของโปรแกรมของฉัน ดูเหมือนว่าจะทำงานได้ตามที่ตั้งใจไว้ แต่ฉันไม่สามารถรับประกันได้ว่าจะไม่มีข้อผิดพลาดบางอย่างที่ฉันพลาดระหว่างการทดสอบ
รหัส
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
คำตอบ
หากฉันเข้าใจคำอธิบายของคุณคุณกำลังมองหาเพียงความคิดเห็น แต่คุณกำลังค้นหาด้วยฐานรหัสแบบเต็มทุกครั้ง เนื่องจากโดยปกติความคิดเห็นเป็นส่วนเล็ก ๆ ของโค้ด (น้อยกว่า 10%?) ฉันขอแนะนำให้ทำขั้นตอนก่อนดำเนินการก่อนโดยคุณเพียงแค่ดึงความคิดเห็นทั้งหมดออกจากนั้นทำการค้นหาจริงในโค้ดเหล่านั้น
โดย "แตกไฟล์" ฉันหมายถึงบันทึกความคิดเห็นในไฟล์แยกต่างหากเพื่อให้คุณสามารถค้นหาได้เฉพาะในไฟล์เหล่านั้น
สำหรับการค้นหาซ้ำใน codebase เดียวกันสิ่งนี้จะช่วยได้เนื่องจากการประมวลผลล่วงหน้าสามารถทำได้เพียงครั้งเดียวจากนั้นการค้นหาจริงจะมีข้อความให้ค้นหาน้อยลง
คุณสามารถระบุข้อโต้แย้งที่จำเป็นในargparseมากกว่าการจัดการพวกเขาด้วยตัวคุณเอง คุณยังสามารถระบุค่าเริ่มต้นตัวอย่างเช่นสำหรับoutput_dir.
os.makedirsใช้exist_ok=Trueเพื่อระบุว่าควรสร้างไดเร็กทอรีหากยังไม่มีอยู่
extensionsอาร์กิวเมนต์SourceWalkerคอนสตรัคอาจจะไม่ควรจะ hardcoded
ใช้with open(file_name, "r") as file_handle:เพื่อให้แน่ใจว่าไฟล์ปิดอยู่เสมอเมื่อออกจากบริบทนั้น ในบันทึกที่เกี่ยวข้องlocals()ควรใช้ IMO เป็นทางเลือกสุดท้ายเท่านั้นเนื่องจากเป็นเรื่องยากที่จะปฏิบัติตามรหัสที่ใช้สตริงเพื่ออ้างถึงตัวแปรและ IDE ไม่สามารถทำอะไรที่เป็นประโยชน์กับรหัสนั้นได้ ตัวอย่างเช่นหากคุณจะเปลี่ยนชื่อinput_fileเป็นinput_file_path(ฉันขอแนะนำว่าไม่ว่าในกรณีใด ๆ ) การลืมเปลี่ยนการอ้างอิงสตริงจะเป็นเรื่องเล็กน้อย
ฉันอยากจะแนะนำให้ใช้blackและisortเพื่อจัดรูปแบบรหัส มันจะใกล้เคียงกับสำนวนสไตล์นั้นมากขึ้นโดยไม่ต้องทำงานด้วยตนเอง
หลังจากจัดรูปแบบแล้วฉันขอแนะนำให้เรียกใช้flake8เพื่อค้นหารหัสที่ไม่ใช่สำนวนอื่น ๆ ตัวอย่างเช่นดูเหมือนว่าSourceWalkerรหัสจะเสีย - วิธีการอยู่ในระดับเดียวกับการกำหนดคลาส
การสร้างข้อยกเว้นด้วยสตริงแบบคงที่เช่นเป็นSourceWalkerException("Process count cannot be less than one!")กลิ่นรหัส ไม่จำเป็นต้องแย่เสมอไป แต่เช่นเดียวกับพารามิเตอร์บูลีนมักหมายความว่าควรดึงบางสิ่งออกจากกัน ในกรณีนี้จะเป็นการดีกว่าหากมีคลาสข้อยกเว้นแยกต่างหากสำหรับจำนวนกระบวนการที่ต่ำเกินไป (และเงื่อนไขข้อผิดพลาดอื่น ๆ ) ซึ่งอาจเป็นอะไรที่ง่ายพอProcessCountTooLowErrorๆ ด้วยวิธีนี้สิ่งเดียวที่ส่งผ่านไปยังตัวสร้างข้อยกเว้น (ถ้ามี) คือเนื้อหาแบบไดนามิกใด ๆ ที่สามารถใช้เพื่อดีบักข้อยกเว้นนั้นได้ นอกจากนี้หมายถึงว่าถ้าคุณเคยจบลงด้วยความต้องการที่จะจัดการข้อยกเว้นคุณสามารถมากกว่าที่มีการแยกข้อความข้อผิดพลาดภายในทั่วไปexcept ProcessCountTooLowError except SourceWalkerException(โอ้และคลาสข้อยกเว้นที่กำหนดเองควรลงท้ายErrorด้วยไม่ใช่Exception)
แทนที่จะตรวจสอบสิ่งต่างๆเช่นมีไดเร็กทอรีเอาต์พุตหรือไม่ฉันจะปล่อยให้แอปพลิเคชันล้มเหลวเมื่อถึงโค้ดที่พยายามเขียนไปยังไดเร็กทอรีนั้น สิ่งนี้เรียกว่า "time of check to time of use" โดยพื้นฐานแล้วว่ามีไดเร็กทอรีอยู่หรือไม่เมื่อคุณทำการตรวจสอบนั้นไม่มีผลว่ามีอยู่จริงหรือไม่เมื่อโค้ดพยายามใช้จริง นอกจากนี้ยังเป็นการตรวจสอบที่แคบเกินไปเนื่องจากเช่นไดเร็กทอรีไม่สามารถเขียนได้โดยผู้ใช้ปัจจุบันหรือระบบไฟล์อาจเต็ม
ในบันทึกที่คล้ายกันการเรียกใช้codecs.lookup(encoding)เพื่อตรวจสอบการเข้ารหัสนั้นมีอยู่ก่อนที่จะใช้งานจริงในการเรียกที่แตกต่างกันโดยสิ้นเชิงดูเหมือนว่าจะเป็นการตรวจสอบที่ไม่สมบูรณ์ อาจเป็นการดีกว่าที่จะ จำกัดพารามิเตอร์การ เข้ารหัสoptionsเฉพาะการเข้ารหัสที่มีอยู่ ด้วยวิธีนี้จะตรวจสอบโดยเร็วที่สุดผู้ใช้จะได้รับข้อผิดพลาดที่ดำเนินการได้ดีและข้อความช่วยเหลือจะแสดงค่าที่เป็นไปได้
พหูพจน์ของ "regex" คือ "regex e s"
อย่าใช้อาร์กิวเมนต์เริ่มต้นที่ไม่แน่นอน ( การอภิปราย ) ในบันทึกที่เกี่ยวข้องคุณควรใช้อาร์กิวเมนต์เริ่มต้นเมื่อใด สองกฎง่ายๆ:
- หากไม่เคยใช้ค่าเริ่มต้นจริงเนื่องจากการเรียกทั้งหมดระบุค่าค่าเริ่มต้นจะไม่มีจุดหมาย
- หากไม่มีการเรียกใดแทนที่ค่าดีฟอลต์มันก็ไม่มีจุดหมายเช่นกัน - อาจเป็นตัวแปรหรือค่าคงที่เช่นกัน
if not isinstance(extensions, list):ไม่ใช่สำนวน; มันควรจะใช้ได้อย่างสมบูรณ์แบบในการทำซ้ำเช่นไฟล์tuple.
โดยทั่วไปตัวสร้างไม่ควรทำอะไรที่ซับซ้อนไปกว่าการตั้งค่าฟิลด์ เมื่อทำเสร็จrunแล้วprocessหรือวิธีอื่นควรเรียกใช้แยกกันเพื่อทำการประมวลผลจริง ฉันจำไม่ได้ว่าฉันเห็นสิ่งนี้ครั้งแรกที่อธิบายไว้อย่างชัดเจน แต่ดูตัวอย่างทำไมการใช้ผลข้างเคียงที่ไม่ดีในตัวสร้าง JavaScript และตัวสร้างต้องไม่มีรหัส
แสดงรายการตัวอักษร
expressions = []
expressions.append(str(args.expression))
ควรจะเป็น
expressions = [str(args.expression)]
Pathlib
นี้:
if not os.path.exists(output_dir):
os.makedirs(output_dir)
ควรใช้pathlibเทียบเท่าใหม่ที่เป็นมันเงา:
Path(output_dir).mkdir(exist_ok=True)
สามารถใช้โมดูลเดียวกันสำหรับ
if not os.path.exists(source_dir) or not os.path.exists(output_dir):
การดำรงอยู่ของตัวแปร
if "input_file" in locals():
input_file.close()
เป็นร่าง โดยปกติวิธีการระบุใน Python ว่าตัวแปรมีค่าหรือไม่นั้นเป็นไปNoneได้ที่จะไม่ได้รับการประกาศ คุณสามารถทำเครื่องหมายว่าอาจไม่มีเลยโดยใช้Optionalคำใบ้ประเภท
กลับ
ซิงเกิ้ลreturnต่อท้ายmain()ซ้ำซ้อน
ความยาวพจนานุกรม
if len(expr.keys()) == 0:
เป็นไปได้
if len(expr) == 0:
หรือแม้กระทั่ง
if not expr:
การรวบรวม Regex
self._regexs[sub_expr] = re.compile("\s+%s(\s|,|:|;|\n)+" % (sub_expr))
อย่างน้อยที่สุดความต้องการที่นำrไปสู่การทำให้สตริงที่แท้จริงนั้นดิบ
โดยทั่วไปแล้วการรับอินพุตจากไฟล์และถือว่าเป็นนิพจน์ย่อยปกติที่ไม่ผ่านการตรวจสอบความถูกต้องเป็นความคิดที่ไม่ดี ไม่ใช่evalระดับไม่ดี แต่ก็ไม่ดี เป็นนิพจน์ทั่วไปจริงหรือเป็นเพียงสตริงย่อย? หากเป็นเพียงสตริงย่อยให้เรียกescapeใช้ก่อนที่จะแทรกลงในนิพจน์ทั่วไปภายนอกของคุณ
หากพวกเขาเป็นจริงการแสดงออกปกติของตัวเองคุณจะต้องการอย่างน้อยใส่แต่ละคนในกลุ่มที่ไม่ได้จับภาพของตัวเองในการแสดงออกด้านนอกเพื่อหลีกเลี่ยงความผิดที่น่ารังเกียจ
ข้อยกเว้นที่ถูกล่ามโซ่
except NotADirectoryError as exception:
raise SourceWalkerException("Directory does not exist! " + str(exception))
ควรจะเป็น
except NotADirectoryError as exception:
raise SourceWalkerException("Directory does not exist! " + str(exception)) from exception