Post-processing after all phases are complete.
(message:dict, attributes:dict[str,object])
| 71 | return message |
| 72 | |
| 73 | def post_processing(message:dict, attributes:dict[str,object]) -> dict: |
| 74 | """ |
| 75 | Post-processing after all phases are complete. |
| 76 | """ |
| 77 | write_meta(directory=attributes['directory'], attributes=attributes) |
| 78 | |
| 79 | # If git management is enabled, automatically commit the final code |
| 80 | if attributes["git_management"]: |
| 81 | log_git_info = "**[Git Information]**\n\n" |
| 82 | os.system("cd {}; git add .".format(attributes["directory"])) |
| 83 | log_git_info += "cd {}; git add .\n".format(attributes["directory"]) |
| 84 | os.system("cd {}; git commit -m \"v{} Final Version\"".format(attributes["directory"], attributes["code_manager"].version)) |
| 85 | log_git_info += "cd {}; git commit -m \"v{} Final Version\"\n".format(attributes["directory"], attributes["code_manager"].version) |
| 86 | log_visualize(log_git_info) |
| 87 | |
| 88 | # Record git log output for auditing |
| 89 | git_info = "**[Git Log]**\n\n" |
| 90 | import subprocess |
| 91 | command = "cd {}; git log".format(attributes["directory"]) |
| 92 | completed_process = subprocess.run(command, shell=True, text=True, stdout=subprocess.PIPE) |
| 93 | if completed_process.returncode == 0: |
| 94 | log_output = completed_process.stdout |
| 95 | else: |
| 96 | log_output = "Error when executing " + command |
| 97 | git_info += log_output |
| 98 | log_visualize(git_info) |
| 99 | |
| 100 | # Collect project summary and elapsed time metrics |
| 101 | post_info = "**[Post Info]**\n\n" |
| 102 | now_time = now() |
| 103 | time_format = "%Y%m%d%H%M%S" |
| 104 | datetime1 = datetime.strptime(attributes["start_time"], time_format) |
| 105 | datetime2 = datetime.strptime(now_time, time_format) |
| 106 | duration = (datetime2 - datetime1).total_seconds() |
| 107 | post_info += "Software Info: {}".format( |
| 108 | get_info(attributes["directory"], attributes["log_filepath"]) + "\n\n🕑**duration**={:.2f}s\n\n".format( |
| 109 | duration)) |
| 110 | post_info += "ChatDev Starts ({})".format(attributes["start_time"]) + "\n\n" |
| 111 | post_info += "ChatDev Ends ({})".format(now_time) + "\n\n" |
| 112 | |
| 113 | # Remove any __pycache__ directories if configured to clear structure |
| 114 | directory = attributes['directory'] |
| 115 | if attributes["config"]["clear_structure"]: |
| 116 | for filename in os.listdir(directory): |
| 117 | file_path = os.path.join(directory, filename) |
| 118 | if os.path.isdir(file_path) and file_path.endswith("__pycache__"): |
| 119 | shutil.rmtree(file_path, ignore_errors=True) |
| 120 | post_info += "{} Removed.".format(file_path) + "\n\n" |
| 121 | |
| 122 | log_visualize(post_info) |
| 123 | logging.shutdown() |
| 124 | time.sleep(1) |
| 125 | |
| 126 | root = os.path.dirname(os.path.dirname(__file__)) |
| 127 | # Move the aggregated log file into the final project directory |
| 128 | shutil.move(attributes["log_filepath"], |
| 129 | os.path.join(root + "/assets/output/WareHouse", "_".join([attributes["project_name"], attributes["org_name"], attributes["start_time"]]), |
| 130 | os.path.basename(attributes["log_filepath"]))) |
nothing calls this directly
no test coverage detected