import http.server import socketserver import subprocess from urllib.parse import urlparse, parse_qs import base64 import json import re def compile_and_deploy_contract( user_path: str, post_path: str, comment_path: str, password: str, node_wallet: str, run_start_funcs: bool=False ) -> dict[str, str]: user_compile_command = f"sshpass -p {password} neo-go contract compile -i {user_path}/user.go -c {user_path}/user.yml -m {user_path}/user.json" subprocess.run(user_compile_command, shell=True, check=False) user_deploy_command = f"sshpass -p {password} neo-go contract deploy -i {user_path}/user.nef -w {node_wallet} -r http://localhost:30333 -m {user_path}/user.json --force" proc = subprocess.Popen(user_deploy_command, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE) user_out_list = [] while True: line = proc.stdout.readline() if not line: break user_out_list.append(line.rstrip()) user_hash = str(user_out_list[-1], encoding='utf-8').replace("Contract: ", "") post_compile_command = f"sshpass -p {password} neo-go contract compile -i {post_path}/post_contract.go -c {post_path}/post_contract.yml -m {post_path}/post_contract.json" subprocess.run(post_compile_command, shell=True, check=False) post_deploy_command = f"sshpass -p {password} neo-go contract deploy -i {post_path}/post_contract.nef -w {node_wallet} -r http://localhost:30333 -m {post_path}/post_contract.json --force" proc = subprocess.Popen(post_deploy_command, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE) post_out_list = [] while True: line = proc.stdout.readline() if not line: break post_out_list.append(line.rstrip()) post_hash = str(post_out_list[-1], encoding='utf-8').replace("Contract: ", "") comment_compile_command = f"sshpass -p {password} neo-go contract compile -i {comment_path}/comment.go -c {comment_path}/comment.yml -m {comment_path}/comment.json" subprocess.run(comment_compile_command, shell=True, check=False) comment_deploy_command = f"sshpass -p {password} neo-go contract deploy -i {comment_path}/comment.nef -w {node_wallet} -r http://localhost:30333 -m {comment_path}/comment.json --force" proc = subprocess.Popen(comment_deploy_command, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE) comment_out_list = [] while True: line = proc.stdout.readline() if not line: break comment_out_list.append(line.rstrip()) comment_hash = str(comment_out_list[-1], encoding='utf-8').replace("Contract: ", "") print(user_hash, post_hash, comment_hash) if run_start_funcs: comment_deploy_command = f"sshpass -p {password} neo-go contract invokefunction -w {node_wallet} -r http://localhost:30333 {user_hash} newUser user_test1 user_test1 user_test1 1 Nhfg3TbpwogLvDGVvAvqyThbsHgoSUKwtn --force" proc = subprocess.Popen(comment_deploy_command, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE) comment_deploy_command = f"sshpass -p {password} neo-go contract invokefunction -w {node_wallet} -r http://localhost:30333 {user_hash} newUser user_test2 user_test2 user_test2 1 Nhfg3TbpwogLvDGVvAvqyThbsHgoSUKwtn --force" proc = subprocess.Popen(comment_deploy_command, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE) comment_deploy_command = f"sshpass -p {password} neo-go contract invokefunction -w {node_wallet} -r http://localhost:30333 {post_hash} newPost user_test1 test test --force" proc = subprocess.Popen(comment_deploy_command, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE) comment_deploy_command = f"sshpass -p {password} neo-go contract invokefunction -w {node_wallet} -r http://localhost:30333 {post_hash} newPost user_test2 test test --force" proc = subprocess.Popen(comment_deploy_command, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE) comment_deploy_command = f"sshpass -p {password} neo-go contract invokefunction -w {node_wallet} -r http://localhost:30333 {comment_hash} createNewComment user_test1 post_1 some_text --force" proc = subprocess.Popen(comment_deploy_command, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE) comment_deploy_command = f"sshpass -p {password} neo-go contract invokefunction -w {node_wallet} -r http://localhost:30333 {comment_hash} createNewComment user_test1 post_1 some_text --force" proc = subprocess.Popen(comment_deploy_command, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE) comment_deploy_command = f"sshpass -p {password} neo-go contract invokefunction -w {node_wallet} -r http://localhost:30333 {comment_hash} createNewComment user_test2 post_1 some_text --force" proc = subprocess.Popen(comment_deploy_command, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE) return user_hash, post_hash, comment_hash user_hash, post_hash, comment_hash = compile_and_deploy_contract( "../User", "../Post", "../Comment", "one", "/home/ftmi-krasotka/Desktop/frostfs_test/frostfs-aio/morph/node-wallet.json", False ) class MyHttpRequestHandler(http.server.SimpleHTTPRequestHandler): def do_GET(self): with open("./route.json", "r") as f: route_dict = json.load(f) with open("logs.log", "w") as f: f.write("starting...") with open("logs.log", "a") as f: f.write(f"user_hash: {user_hash}\npost_hash: {post_hash}\ncomment_hash: {comment_hash}") parsed_path = urlparse(self.path) query = parse_qs(parsed_path.query) if "command_type" in query and "command_id" in query: self.send_response(200) self.send_header("Content-type", "text/html") self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() params = query.get("args", None) parsed_params = " " + ' '.join(params) if params else "" command_type = query["command_type"][0] contract_name = query["contract"][0] method = query["method"][0] with open("logs.log", "a") as f: f.write("reading params...\n") if command_type == "get": hash_for_req = None if contract_name == "post": hash_for_req = post_hash elif contract_name == "user": hash_for_req = user_hash elif contract_name == "comment": hash_for_req = comment_hash with open("logs.log", "a") as f: f.write(f"params: {[command_type, contract_name, method, parsed_params]}\n") command = f"neo-go contract testinvokefunction -r http://localhost:30333 {hash_for_req} {method}{parsed_params}" print(command) proc = subprocess.Popen(command, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE) out_list = [] while True: if proc.stdout is None: break line = proc.stdout.readline() if not line: break out_list.append(line.rstrip()) answer = " ".join(map(lambda x: x.decode("utf-8"), out_list)) answer = re.sub(r"[\n\t\s]*", "", answer) print(json.loads(answer)["stack"][0]) if json.loads(answer)["stack"][0]["type"] == "Struct": answer_dict = json.loads(answer)["stack"][0]["value"] obj_fiedls = [] for field in answer_dict: if field["type"] in ["ByteString", "Buffer"]: encoded_string = field["value"] result_string = base64.b64decode(encoded_string) try: result_string = result_string.decode('utf-8') except: pass obj_fiedls.append(result_string) elif field["type"] == "Integer": encoded_string = field["value"] obj_fiedls.append(encoded_string) self.wfile.write(bytes(f"{obj_fiedls}", "utf-8")) else: answer_dict = json.loads(answer)["stack"][0]["value"] answer_list = [] for obj_i in answer_dict: obj = obj_i["value"] obj_fiedls = [] for field in obj: if field["type"] in ["ByteString", "Buffer"]: encoded_string = field["value"] result_string = base64.b64decode(encoded_string) result_string = result_string.decode('utf-8') obj_fiedls.append(result_string) elif field["type"] == "Integer": encoded_string = field["value"] obj_fiedls.append(encoded_string) answer_list.append(obj_fiedls) self.wfile.write(bytes(f"{answer_list}", "utf-8")) else: self.send_response(400) self.send_header("Content-type", "text/html") self.end_headers() self.wfile.write(bytes("Bad request", "utf-8")) if __name__ == "__main__": PORT = 2281 Handler = MyHttpRequestHandler with socketserver.TCPServer(("", PORT), Handler) as httpd: print("Сервер запущен на порту", PORT) httpd.serve_forever() httpd.server_close()