# coding=utf-8 """ 构建合同建单 Main.seq —— 按业务段换行，画布上能直接读出步骤。 艺赛旗「插入代码块」的标题固定为「插入代码块#1」，「业务流程描述」节点 打开后再保存标题会被清空。画布上真正能看见的是代码块的备注（NOTE）。 因此本脚本： 1. 不再插入空的业务流程描述节点 2. 把「S06 填写签约方」写进 NOTE，作为画布副标题 3. 按 准备 / 进入起草 / 填写表单 / 提交回写 四段换行 关闭 Studio 后运行本脚本，再重新打开项目编译。 """ from __future__ import annotations import base64 import json from pathlib import Path HERE = Path(__file__).resolve().parent OUT = HERE / "studio" / "Main.seq" COL_W = 280 ROW_H = 200 ORIGIN_X = 360 ORIGIN_Y = 220 BEGIN_TAG = "202608141000000001" def b64_apa(val): # 艺赛旗 EP_CONTROL / EP_Gen 字段值末尾带换行，缺失会导致字段解析失败 return json.dumps({"val": base64.b64encode(str(val).encode()).decode()}, indent=4) + "\n" def b64_raw(val): return base64.b64encode(str(val).encode()).decode() STEP_INIT = '''import sys, os, json from pathlib import Path _REPO_ROOT = next( p for p in Path(__file__).resolve().parents if (p / "rpa_yisaiqi" / "__init__.py").exists() ) if str(_REPO_ROOT) not in sys.path: sys.path.insert(0, str(_REPO_ROOT)) import rpa_yisaiqi._flow_state as _st from rpa_yisaiqi.contract_create_node import ContractCreateError, load_input_data, build_spec_from_data from rpa_yisaiqi.callback_client import StateCenterClient from rpa_yisaiqi.apa_browser_adapter import ApaBrowserAdapter def _first(*values): for value in values: if value is not None and str(value).strip(): return str(value).strip() return "" raw = (self.input_arg or "").strip() if not raw: raw = os.getenv("CONTRACT_INPUT_JSON", "").strip() if not raw: raw = str(_REPO_ROOT / "rpa_yisaiqi" / "examples" / "contract_create_input_dry_run.json") print("未传入 -i/--input，使用默认联调入参：", raw) if not raw: raise ContractCreateError("缺少业务入参：请通过 -i/--input 传入 JSON 字符串或文件路径") data = load_input_data(raw) state_center_url = _first( data.get("stateCenterUrl"), data.get("stateCenterURL"), os.getenv("STATE_CENTER_URL"), "http://192.168.105.69:8000", ) agent_token = _first(data.get("agentToken"), os.getenv("AGENT_TOKEN"), "dev-agent-token") _st.spec = build_spec_from_data(data) _st.client = StateCenterClient(state_center_url, agent_token) _st.ctx = _st.spec.flow_context _st.browser = ApaBrowserAdapter(url=_st.ctx.contract_url, flow="CONTRACT_CREATE") _st.is_dry_run = (_st.spec.submit_mode == "dry_run") if not _st.is_dry_run: print("回调状态中心 start", _st.spec.job_id) _st.client.start_node(_st.spec.job_id, "CONTRACT_CREATE")''' STEP_OPEN = '''import rpa_yisaiqi._flow_state as _st from rpa_yisaiqi.contract_flow_steps import open_contract_system open_contract_system(_st.browser, _st.ctx)''' STEP_LOGIN = '''import rpa_yisaiqi._flow_state as _st from rpa_yisaiqi.contract_flow_steps import login_contract_system login_contract_system(_st.browser, _st.ctx)''' STEP_ENTER = '''import rpa_yisaiqi._flow_state as _st from rpa_yisaiqi.contract_flow_steps import enter_contract_create enter_contract_create(_st.browser)''' STEP_TEMPLATE = '''import rpa_yisaiqi._flow_state as _st from rpa_yisaiqi.contract_flow_steps import select_contract_template select_contract_template(_st.browser, _st.ctx)''' STEP_SIGNATORY = '''import rpa_yisaiqi._flow_state as _st from rpa_yisaiqi.contract_flow_steps import fill_contract_signatory_info fill_contract_signatory_info(_st.browser, _st.ctx)''' STEP_BASIC = '''import rpa_yisaiqi._flow_state as _st from rpa_yisaiqi.contract_flow_steps import fill_contract_basic_info fill_contract_basic_info(_st.browser, _st.ctx)''' STEP_AMOUNT = '''import rpa_yisaiqi._flow_state as _st from rpa_yisaiqi.contract_flow_steps import fill_contract_amount_and_dates fill_contract_amount_and_dates(_st.browser, _st.ctx)''' STEP_GOODS = '''import rpa_yisaiqi._flow_state as _st from rpa_yisaiqi.contract_flow_steps import fill_contract_goods_lines fill_contract_goods_lines(_st.browser, _st.ctx)''' STEP_PAYMENT = '''import rpa_yisaiqi._flow_state as _st from rpa_yisaiqi.contract_flow_steps import fill_contract_payment_info fill_contract_payment_info(_st.browser, _st.ctx)''' STEP_UPLOAD = '''import rpa_yisaiqi._flow_state as _st from rpa_yisaiqi.contract_flow_steps import upload_contract_document_if_needed upload_contract_document_if_needed(_st.browser, _st.ctx)''' STEP_SUBMIT = '''import rpa_yisaiqi._flow_state as _st from rpa_yisaiqi.contract_flow_steps import submit_contract_for_approval _st.submitted = submit_contract_for_approval(_st.browser, _st.ctx)''' STEP_READ_NO = '''import rpa_yisaiqi._flow_state as _st from rpa_yisaiqi.contract_flow_steps import read_contract_number if _st.submitted: _st.contract_no = read_contract_number(_st.browser, _st.ctx)''' STEP_COMPLETE = '''import json import rpa_yisaiqi._flow_state as _st if not _st.is_dry_run and _st.contract_no: print("回调状态中心 complete", _st.spec.job_id, _st.contract_no) _st.client.complete_node(_st.spec.job_id, "CONTRACT_CREATE", { "businessRefs": {"contractSystemNo": _st.contract_no}, }) print(json.dumps({"status": "done", "contractSystemNo": _st.contract_no}, ensure_ascii=False))''' # 每段一行。NOTE 会显示在画布「插入代码块#1」下方，务必写短步骤名。 PHASES = [ ( "准备", [ ("S01 初始化并回调 start", STEP_INIT), ("S02 打开合同系统", STEP_OPEN), ("S03 登录合同系统", STEP_LOGIN), ], ), ( "进入起草", [ ("S04 进入合同起草", STEP_ENTER), ("S05 选择模板", STEP_TEMPLATE), ], ), ( "填写表单", [ ("S06 填写签约方", STEP_SIGNATORY), ("S07 填写基础信息", STEP_BASIC), ("S08 填写金额日期", STEP_AMOUNT), ("S09 填写货品明细", STEP_GOODS), ("S10 填写付款信息", STEP_PAYMENT), ], ), ( "提交回写", [ ("S11 上传合同PDF", STEP_UPLOAD), ("S12 提交送审", STEP_SUBMIT), ("S13 读取合同编号", STEP_READ_NO), ("S14 回调 complete", STEP_COMPLETE), ], ), ] def code_tag(i: int) -> str: return f"202608143{i:02d}0000000" def line_tag(i: int) -> str: return f"202608149{i:02d}0000000" def _iter_steps(): index = 1 for phase_name, items in PHASES: for title, code in items: yield index, phase_name, title, code index += 1 def build_begin_node(px: int, py: int) -> dict: return { "NodePanelData": { "ExpandedGroups": [], "InputProperty": {"mouse_x": 0, "mouse_y": 0}, "PubProperty": "", "RetPins": [], }, "NodeProperty": "", "NodeUI": { "NodeElementID": "", "NodeModel": "2", "NodePropErr": False, "NodePx": str(px), "NodePy": str(py), "NodeSubType": "EnumWidget_Begin", "NodeTag": BEGIN_TAG, "NodeTitle": "开始", }, } def build_code_node(tag: str, title: str, code: str, px: int, py: int) -> dict: body = f'print("[{title}]")\n' + code return { "NodePanelData": { "ExpandedGroups": [], "InputProperty": {"mouse_x": 0, "mouse_y": 0}, "PubProperty": "", "RetPins": [], }, "NodeProperty": { "EP_CONTROL": { "EP_CONTROL_IMPORT": json.dumps( {"PImport": "", "PBCode": b64_raw(body)}, indent=4 ) + "\n" }, "EP_Gen": { "EP_Gen_AFTERDELAY": b64_apa("0"), "EP_Gen_BEFOREDELAY": b64_apa("0"), "EP_Gen_NOTE": b64_apa(title), }, }, "NodeUI": { "NodeElementID": "", "NodeModel": "2", "NodePropErr": False, "NodePx": str(px), "NodePy": str(py), "NodeSubType": "EnumWidget_Code_Custom", "NodeTag": tag, "NodeTitle": title, }, } def build_nodes(): nodes = [] placed = [] # (tag, x, y) begin_x, begin_y = ORIGIN_X, ORIGIN_Y nodes.append(build_begin_node(begin_x, begin_y)) placed.append((BEGIN_TAG, begin_x, begin_y)) # 第一段从「开始」右侧接着排；后续每段单独一行 col = 1 row = 0 current_phase = None for index, phase_name, title, code in _iter_steps(): if current_phase is None: current_phase = phase_name elif phase_name != current_phase: current_phase = phase_name row += 1 col = 0 x = ORIGIN_X + col * COL_W y = ORIGIN_Y + row * ROW_H tag = code_tag(index) nodes.append(build_code_node(tag, title, code, x, y)) placed.append((tag, x, y)) col += 1 return nodes, placed def build_lines(placed: list) -> list: lines = [] for k in range(len(placed) - 1): prev_tag, prev_x, prev_y = placed[k] next_tag, next_x, next_y = placed[k + 1] same_row = prev_y == next_y if same_row: start_x, start_y = prev_x + 150, prev_y + 40 end_x, end_y = next_x - 4, next_y + 40 else: start_x, start_y = prev_x + 70, prev_y + 90 end_x, end_y = next_x + 70, next_y - 8 lines.append({ "LineColor": "255, 255, 255", "LineCommented": 0, "LineEnd": { "Pidx": f"{next_tag}/EPin_Flow_In", "Px": str(end_x), "Py": str(end_y), }, "LineNote": "", "LineSplit": {"Px": "", "Py": ""}, "LineStart": { "Pidx": f"{prev_tag}/EPin_Flow_Out", "Px": str(start_x), "Py": str(start_y), }, "LineTag": line_tag(k + 1), "LineType": 0, }) return lines def main() -> None: nodes, placed = build_nodes() xs = [x for _, x, _ in placed] ys = [y for _, _, y in placed] main_seq = { "LineList": build_lines(placed), "NodeList": nodes, "SeqTag": "Main", "ViewCenterPx": str((min(xs) + max(xs)) // 2), "ViewCenterPy": str((min(ys) + max(ys)) // 2), "ViewScaling": "0.7", } OUT.write_text(json.dumps(main_seq, ensure_ascii=False, indent=4), encoding="utf-8") print(f"多节点 Main.seq 已生成：{OUT}") print(f"节点数：{len(nodes)}（1 开始 + {len(placed) - 1} 代码块，已去掉空业务标识）") print(f"连线数：{len(main_seq['LineList'])}") current_phase = None for index, phase_name, title, _ in _iter_steps(): if phase_name != current_phase: current_phase = phase_name print(f" [{phase_name}]") print(f" {title}") if __name__ == "__main__": main()