175 lines
5.1 KiB
Plaintext
175 lines
5.1 KiB
Plaintext
|
|
#!/usr/bin/env python3
|
|||
|
|
"""HLDP执行器
|
|||
|
|
收到HLDP文件 → 解析@exec/@check → 查.code-map → 执行 → 返回
|
|||
|
|
|
|||
|
|
用法:
|
|||
|
|
hldp-exec <file.hldp> 执行HLDP文件中的@exec指令
|
|||
|
|
hldp-exec <file.hldp> --dry-run 只翻译不执行
|
|||
|
|
hldp-exec --code-map 显示当前.code-map内容
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
import sys, os, subprocess, re, json
|
|||
|
|
|
|||
|
|
REPO = os.path.expanduser("~/guanghulab")
|
|||
|
|
CODE_MAP_PATH = os.path.join(REPO, ".code-map")
|
|||
|
|
|
|||
|
|
def load_code_map():
|
|||
|
|
"""加载 .code-map → {编号: 路径}"""
|
|||
|
|
cm = {}
|
|||
|
|
if not os.path.exists(CODE_MAP_PATH):
|
|||
|
|
return cm
|
|||
|
|
with open(CODE_MAP_PATH) as f:
|
|||
|
|
for line in f:
|
|||
|
|
line = line.strip()
|
|||
|
|
if not line or line.startswith("#"):
|
|||
|
|
continue
|
|||
|
|
if "=" in line:
|
|||
|
|
k, v = line.split("=", 1)
|
|||
|
|
cm[k.strip()] = v.strip()
|
|||
|
|
return cm
|
|||
|
|
|
|||
|
|
def resolve(code, code_map):
|
|||
|
|
"""编号 → 实际命令"""
|
|||
|
|
if code in code_map:
|
|||
|
|
return code_map[code]
|
|||
|
|
# 尝试模糊匹配
|
|||
|
|
for k, v in code_map.items():
|
|||
|
|
if code in k:
|
|||
|
|
return v
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
def parse_hldp(path):
|
|||
|
|
"""解析HLDP文件,提取@exec和@check字段"""
|
|||
|
|
execs = []
|
|||
|
|
checks = []
|
|||
|
|
in_exec = False
|
|||
|
|
in_check = False
|
|||
|
|
|
|||
|
|
with open(path) as f:
|
|||
|
|
for line in f:
|
|||
|
|
line = line.strip()
|
|||
|
|
if line.startswith("@exec:"):
|
|||
|
|
in_exec = True
|
|||
|
|
in_check = False
|
|||
|
|
cmd = line.split(":", 1)[1].strip()
|
|||
|
|
if cmd:
|
|||
|
|
execs.append(cmd)
|
|||
|
|
elif line.startswith("@check:"):
|
|||
|
|
in_exec = False
|
|||
|
|
in_check = True
|
|||
|
|
cmd = line.split(":", 1)[1].strip()
|
|||
|
|
if cmd:
|
|||
|
|
checks.append(cmd)
|
|||
|
|
elif line.startswith("@"):
|
|||
|
|
in_exec = False
|
|||
|
|
in_check = False
|
|||
|
|
elif in_exec and line.strip():
|
|||
|
|
execs.append(line.strip())
|
|||
|
|
elif in_check and line.strip():
|
|||
|
|
checks.append(line.strip())
|
|||
|
|
|
|||
|
|
return execs, checks
|
|||
|
|
|
|||
|
|
def translate(commands, code_map):
|
|||
|
|
"""翻译编号→真实命令"""
|
|||
|
|
result = []
|
|||
|
|
for cmd in commands:
|
|||
|
|
parts = cmd.split()
|
|||
|
|
if not parts:
|
|||
|
|
continue
|
|||
|
|
code = parts[0]
|
|||
|
|
resolved = resolve(code, code_map)
|
|||
|
|
if not resolved:
|
|||
|
|
result.append(("UNKNOWN", f"未知编号: {code}", cmd))
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
# 密钥类: 返回加载指令
|
|||
|
|
if code.startswith("SC-"):
|
|||
|
|
result.append(("SECRET", resolved, cmd))
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
# 脚本/文件类: 组成完整命令
|
|||
|
|
full_path = os.path.join(REPO, resolved)
|
|||
|
|
args = parts[1:]
|
|||
|
|
|
|||
|
|
if resolved.endswith(".py"):
|
|||
|
|
full_cmd = f"python3 {full_path} " + " ".join(args)
|
|||
|
|
elif resolved.endswith(".js"):
|
|||
|
|
full_cmd = f"node {full_path} " + " ".join(args)
|
|||
|
|
else:
|
|||
|
|
full_cmd = f"{full_path} " + " ".join(args)
|
|||
|
|
|
|||
|
|
result.append(("EXEC", full_cmd, cmd))
|
|||
|
|
|
|||
|
|
return result
|
|||
|
|
|
|||
|
|
def run(translated, dry_run=False):
|
|||
|
|
"""执行翻译后的命令"""
|
|||
|
|
for i, (typ, cmd, original) in enumerate(translated):
|
|||
|
|
print(f"[{i+1}] {original}")
|
|||
|
|
print(f" → {cmd}")
|
|||
|
|
|
|||
|
|
if typ == "UNKNOWN":
|
|||
|
|
print(f" ✗ 无法执行: 编号未注册")
|
|||
|
|
continue
|
|||
|
|
if typ == "SECRET":
|
|||
|
|
print(f" ✓ 密钥指令已就绪(不在此处执行)")
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
if dry_run:
|
|||
|
|
print(f" (dry-run, 未实际执行)")
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
proc = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=300)
|
|||
|
|
if proc.returncode == 0:
|
|||
|
|
print(f" ✓ 执行成功")
|
|||
|
|
if proc.stdout.strip():
|
|||
|
|
# 截取前500字符
|
|||
|
|
out = proc.stdout.strip()[:500]
|
|||
|
|
print(f" {out}")
|
|||
|
|
else:
|
|||
|
|
print(f" ✗ 执行失败 (code={proc.returncode})")
|
|||
|
|
if proc.stderr.strip():
|
|||
|
|
print(f" {proc.stderr.strip()[:300]}")
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f" ✗ 异常: {e}")
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
if len(sys.argv) < 2:
|
|||
|
|
print(__doc__)
|
|||
|
|
sys.exit(1)
|
|||
|
|
|
|||
|
|
arg = sys.argv[1]
|
|||
|
|
|
|||
|
|
if arg == "--code-map":
|
|||
|
|
cm = load_code_map()
|
|||
|
|
for k, v in sorted(cm.items()):
|
|||
|
|
print(f" {k:<12} = {v}")
|
|||
|
|
sys.exit(0)
|
|||
|
|
|
|||
|
|
filepath = arg
|
|||
|
|
dry_run = "--dry-run" in sys.argv
|
|||
|
|
|
|||
|
|
if not os.path.exists(filepath):
|
|||
|
|
print(f"✗ 文件不存在: {filepath}")
|
|||
|
|
sys.exit(1)
|
|||
|
|
|
|||
|
|
cm = load_code_map()
|
|||
|
|
execs, checks = parse_hldp(filepath)
|
|||
|
|
|
|||
|
|
if not execs and not checks:
|
|||
|
|
print("⚠ 该HLDP文件中没有 @exec 或 @check 字段")
|
|||
|
|
sys.exit(0)
|
|||
|
|
|
|||
|
|
print(f"=== 执行 {filepath} ===")
|
|||
|
|
|
|||
|
|
if execs:
|
|||
|
|
translated = translate(execs, cm)
|
|||
|
|
run(translated, dry_run)
|
|||
|
|
|
|||
|
|
if checks:
|
|||
|
|
print(f"\n--- @check 校验 ---")
|
|||
|
|
translated = translate(checks, cm)
|
|||
|
|
run(translated, dry_run)
|