# -*- coding: utf-8 -*- """ STEP105-14 Patch V2 Publish Worker - Final Publish 연결 실행 위치: D:\honghee\blog_api 실행: python tools\patch_v2_worker_final_publish.py """ from pathlib import Path import shutil import sys ROOT = Path(__file__).resolve().parents[1] TARGET = ROOT / "workers" / "v2_publish_worker.py" BACKUP = ROOT / "workers" / "v2_publish_worker.py.bak_step105_14" def fail(msg): print("[PATCH FAILED]", msg) sys.exit(1) def replace_once(text, old, new, label): if old not in text: fail(f"pattern not found: {label}") if text.count(old) != 1: fail(f"pattern count != 1: {label}, count={text.count(old)}") return text.replace(old, new, 1) def insert_after_once(text, marker, addition, label): if addition.strip() in text: print(f"[SKIP] already exists: {label}") return text if marker not in text: fail(f"marker not found: {label}") if text.count(marker) != 1: fail(f"marker count != 1: {label}, count={text.count(marker)}") return text.replace(marker, marker + addition, 1) def main(): if not TARGET.exists(): fail(f"target not found: {TARGET}") text = TARGET.read_text(encoding="utf-8") if "from services.v2.publish.final_publish import final_publish" not in text: marker = "from services.v2.publish.publish_dialog import inspect_publish_dialog\n" text = insert_after_once( text, marker, "from services.v2.publish.final_publish import final_publish\n", "import final_publish", ) text = replace_once( text, " check_publish_dialog=False,\n headless=False,\n):", " check_publish_dialog=False,\n check_final_publish=False,\n headless=False,\n):", "run_v2_flow_check signature", ) text = replace_once( text, " check_publish_button, check_publish_click, check_publish_dialog\n ]):", " check_publish_button, check_publish_click, check_publish_dialog, check_final_publish\n ]):", "check_browser any list", ) text = replace_once( text, " check_publish_button, check_publish_click, check_publish_dialog\n ]):", " check_publish_button, check_publish_click, check_publish_dialog, check_final_publish\n ]):", "check_write any list", ) text = replace_once( text, " publish_dialog_result = None\n\n if any([check_title, check_body, check_publish_button, check_publish_click, check_publish_dialog]):", " publish_dialog_result = None\n final_publish_result = None\n\n if any([check_title, check_body, check_publish_button, check_publish_click, check_publish_dialog, check_final_publish]):", "final_publish_result init and title condition", ) text = replace_once( text, " if any([check_body, check_publish_button, check_publish_click, check_publish_dialog]):", " if any([check_body, check_publish_button, check_publish_click, check_publish_dialog, check_final_publish]):", "body condition", ) text = replace_once( text, " if any([check_publish_button, check_publish_click, check_publish_dialog]):", " if any([check_publish_button, check_publish_click, check_publish_dialog, check_final_publish]):", "publish button condition", ) text = replace_once( text, " if check_publish_click or check_publish_dialog:", " if check_publish_click or check_publish_dialog or check_final_publish:", "publish click condition", ) text = replace_once( text, " if check_publish_dialog:", " if check_publish_dialog or check_final_publish:", "publish dialog condition", ) marker = """ publish_dialog_result = inspect_publish_dialog( page=page, write_frame=write_frame, logger=logger, session=session, verbose=True, ) return { """ addition = """ publish_dialog_result = inspect_publish_dialog( page=page, write_frame=write_frame, logger=logger, session=session, verbose=True, ) if check_final_publish: final_publish_result = final_publish( page=page, write_frame=write_frame, dialog_result=publish_dialog_result, execute=False, logger=logger, session=session, verbose=True, ) return { """ text = replace_once(text, marker, addition, "insert final publish call") text = replace_once( text, ' "V2 publish dialog inspect completed"\n if check_publish_dialog\n else "V2 publish click probe completed"', ' "V2 final publish ready check completed"\n if check_final_publish\n else "V2 publish dialog inspect completed"\n if check_publish_dialog\n else "V2 publish click probe completed"', "message condition", ) text = replace_once( text, ' "publish_dialog": publish_dialog_result,\n "url": page.url,\n }', ' "publish_dialog": publish_dialog_result,\n "final_publish": final_publish_result,\n "url": page.url,\n }', "return final_publish result", ) text = replace_once( text, " check_publish_dialog=False,\n headless=False,\n):", " check_publish_dialog=False,\n check_final_publish=False,\n headless=False,\n):", "run_v2_publish_once signature", ) text = insert_after_once( text, ' print("check_publish_dialog:", check_publish_dialog)\n', ' print("check_final_publish:", check_final_publish)\n', "print check_final_publish", ) text = replace_once( text, " check_publish_dialog,\n ]):", " check_publish_dialog,\n check_final_publish,\n ]):", "handler any list", ) text = insert_after_once( text, " check_publish_dialog=check_publish_dialog,\n", " check_final_publish=check_final_publish,\n", "pass check_final_publish to flow", ) old_exec = ' raise NotImplementedError("V2 actual publish is not implemented yet")' new_exec = """ return run_v2_flow_check( queue=queue, session=hpe_session, logger=logger, check_final_publish=True, headless=headless, )""" text = replace_once(text, old_exec, new_exec, "execute final publish path") text = insert_after_once( text, ' parser.add_argument("--check-publish-dialog", action="store_true")\n', ' parser.add_argument("--check-final-publish", action="store_true")\n', "argparse check-final-publish", ) text = insert_after_once( text, " check_publish_dialog=args.check_publish_dialog,\n", " check_final_publish=args.check_final_publish,\n", "main call check_final_publish", ) if not BACKUP.exists(): shutil.copy2(TARGET, BACKUP) print("[BACKUP CREATED]", BACKUP) else: print("[BACKUP EXISTS]", BACKUP) TARGET.write_text(text, encoding="utf-8") print("[PATCH SUCCESS]", TARGET) print("[NEXT TEST]") print("python -m workers.v2_publish_worker --realtor-id 1 --check-final-publish") if __name__ == "__main__": main()