guanghulab/scripts/hldp-exec
Guanghu Domestic Migration d1e47f4565
Some checks are pending
自动更新代码和重启 / update-and-restart (push) Waiting to run
CI检查 + 自动部署 / check (push) Waiting to run
CI检查 + 自动部署 / deploy (push) Blocked by required conditions
重启聊天服务 / restart (push) Waiting to run
chore: import sanitized domestic snapshot for REPO-002
Source snapshot: ca48d3ddf926d79aa138306164169baf764bb829
2026-07-17 15:54:41 +08:00

175 lines
5.1 KiB
Python
Executable File
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/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)