65 lines
2.1 KiB
Python
65 lines
2.1 KiB
Python
# coding=utf-8
|
||
"""APA 启动 Main.py 前会加载本文件。Studio 重新编译不会改这里。
|
||
|
||
Studio 点运行只带 -p，不带 -i。这里自动补上默认联调 JSON，
|
||
避免 S01 因 input_arg 为空直接报「缺少业务入参」。
|
||
账号过期把 Main.py 覆盖成空注释时，改走 run_contract.py。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import sys
|
||
|
||
|
||
def _codes_dir() -> str:
|
||
return os.path.dirname(os.path.abspath(__file__))
|
||
|
||
|
||
def _default_input_path() -> str:
|
||
repo = os.path.abspath(os.path.join(_codes_dir(), "..", "..", ".."))
|
||
return os.path.join(repo, "rpa_yisaiqi", "examples", "contract_create_input_dry_run.json")
|
||
|
||
|
||
def _is_apa_main() -> bool:
|
||
if not sys.argv:
|
||
return False
|
||
return os.path.basename(str(sys.argv[0])).lower() == "main.py"
|
||
|
||
|
||
def _argv_has_input(argv: list) -> bool:
|
||
for i, arg in enumerate(argv):
|
||
if arg in ("-i", "--input") and i + 1 < len(argv) and str(argv[i + 1]).strip():
|
||
return True
|
||
if str(arg).startswith("--input=") and str(arg).split("=", 1)[1].strip():
|
||
return True
|
||
return False
|
||
|
||
|
||
def _main_is_expired_stub(path: str) -> bool:
|
||
try:
|
||
text = open(path, encoding="utf-8").read()
|
||
except Exception:
|
||
return False
|
||
return "账号过期" in text
|
||
|
||
|
||
# Windows 虚拟机只回调 Mac 状态中心，不在本机起库。
|
||
os.environ.setdefault("STATE_CENTER_URL", "http://192.168.105.69:8000")
|
||
os.environ.setdefault("AGENT_TOKEN", "dev-agent-token")
|
||
os.environ.setdefault("RPA_CALLBACK_STATE_CENTER", "1")
|
||
|
||
if _is_apa_main():
|
||
if _main_is_expired_stub(sys.argv[0]):
|
||
print("检测到 Main.py 被艺赛旗覆盖（账号过期），改走独立入口", flush=True)
|
||
if _codes_dir() not in sys.path:
|
||
sys.path.insert(0, _codes_dir())
|
||
from run_contract import main
|
||
|
||
raise SystemExit(main())
|
||
|
||
if not _argv_has_input(sys.argv):
|
||
default_input = _default_input_path()
|
||
if os.path.isfile(default_input):
|
||
sys.argv.extend(["-i", default_input])
|
||
print("未传入 -i，自动使用默认联调入参：", default_input, flush=True)
|