|
#!/usr/bin/env python3 |
|
# -*- coding: utf-8 -*- |
|
""" |
|
Parse `code --status` output (including "Workspace Stats") and emit JSON. |
|
""" |
|
|
|
import sys, json, re, argparse, subprocess |
|
|
|
def read_input(): |
|
data = sys.stdin.read() |
|
if data.strip(): |
|
return data |
|
try: |
|
out = subprocess.check_output(["code", "--status"]) |
|
for enc in ("utf-8", "cp932", "mbcs"): |
|
try: |
|
return out.decode(enc) |
|
except UnicodeDecodeError: |
|
continue |
|
return out.decode(errors="replace") |
|
except Exception as e: |
|
print(f"Failed to run `code --status`: {e}", file=sys.stderr) |
|
sys.exit(1) |
|
|
|
# ---------- Header parsing (same as前回) ---------- |
|
def parse_headers(lines, i): |
|
headers, gpu, in_gpu = {}, {}, False |
|
while i < len(lines): |
|
line = lines[i].rstrip("\n") |
|
if not line.strip(): |
|
j = i + 1 |
|
while j < len(lines) and not lines[j].strip(): |
|
j += 1 |
|
if j < len(lines) and (lines[j].lstrip().startswith("CPU %") or lines[j].startswith("Workspace Stats:")): |
|
i = j |
|
break |
|
i += 1 |
|
continue |
|
|
|
if in_gpu: |
|
if re.match(r"^\s{2,}\S", line): |
|
k, v = [s.strip() for s in re.split(r":\s+", line.strip(), maxsplit=1)] if ":" in line else (line.strip(), "") |
|
gpu[k] = v |
|
i += 1 |
|
continue |
|
else: |
|
headers["GPU Status"] = gpu |
|
in_gpu = False |
|
|
|
m = re.match(r"^([A-Za-z ]+):\s+(.*)$", line) |
|
if m: |
|
key, val = m.group(1).strip(), m.group(2).strip() |
|
if key == "GPU Status": |
|
in_gpu, gpu = True, {} |
|
else: |
|
headers[key] = val |
|
i += 1 |
|
continue |
|
|
|
if headers: |
|
break |
|
else: |
|
i += 1 |
|
|
|
if in_gpu: |
|
headers["GPU Status"] = gpu |
|
return headers, i |
|
|
|
def parse_process_table(lines, i): |
|
while i < len(lines) and not lines[i].lstrip().startswith("CPU %"): |
|
if lines[i].startswith("Workspace Stats:"): # プロセステーブルがないケースにも対応 |
|
return [], i |
|
i += 1 |
|
if i >= len(lines): return [], i |
|
i += 1 |
|
row_re = re.compile(r"^(?P<indent>\s*)(?P<cpu>-?\d+)\s+(?P<mem>-?\d+)\s+(?P<pid>-?\d+)\s+(?P<proc>.+?)\s*$") |
|
stack, root = [], [] |
|
|
|
def mk(cpu, mem, pid, proc, lvl): |
|
return {"cpu_percent": int(cpu), "mem_mb": int(mem), "pid": int(pid), "process": proc, "_lvl": lvl, "children": []} |
|
|
|
while i < len(lines): |
|
line = lines[i].rstrip("\n") |
|
if not line.strip(): |
|
i += 1; continue |
|
if line.startswith("Workspace Stats:"): # 次セクション |
|
break |
|
m = row_re.match(line) |
|
if not m: break |
|
lvl = len(m.group("indent")) // 2 |
|
node = mk(m.group("cpu"), m.group("mem"), m.group("pid"), m.group("proc"), lvl) |
|
|
|
if not stack: |
|
root.append(node); stack.append(node) |
|
else: |
|
while stack and stack[-1]["_lvl"] >= lvl: |
|
stack.pop() |
|
(stack[-1]["children"] if stack else root).append(node) |
|
stack.append(node) |
|
i += 1 |
|
|
|
def strip(nodes): |
|
for n in nodes: |
|
n.pop("_lvl", None) |
|
if n["children"]: strip(n["children"]) |
|
else: n.pop("children", None) |
|
strip(root) |
|
return root, i |
|
|
|
# ---------- Workspace Stats ここから追加 ---------- |
|
def parse_workspace_stats(lines, i): |
|
""" |
|
Parse block: |
|
Workspace Stats: |
|
| Window (<title>) |
|
| Folder (<name>): N files |
|
| File types: ext1(n1) ext2(n2) ... |
|
| Conf files: ... (optional, may be empty) |
|
Multiple windows/folders can exist; line prefixes use pipes and spaces. |
|
""" |
|
# 探索 |
|
while i < len(lines) and not lines[i].startswith("Workspace Stats:"): |
|
i += 1 |
|
if i >= len(lines): return None, i |
|
i += 1 # skip "Workspace Stats:" |
|
|
|
windows = [] |
|
current_window = None |
|
current_folder = None |
|
|
|
win_re = re.compile(r'^\|\s+Window\s+\((?P<title>.+)\)\s*$') |
|
fol_re = re.compile(r'^\|\s{2,}Folder\s+\((?P<name>.+)\):\s+(?P<count>\d+)\s+files\s*$') |
|
types_re = re.compile(r'^\|\s{4,}File types:\s*(?P<rest>.+)$') |
|
conf_re = re.compile(r'^\|\s{4,}Conf files:\s*(?P<rest>.*)$') |
|
|
|
def parse_kv_counts(s): |
|
# ext(n) ext2(n2) などを dict に |
|
out = {} |
|
for tok in re.split(r"\s+", s.strip()): |
|
m = re.match(r'(?P<key>[^()]+)\((?P<n>\d+)\)$', tok.strip(',')) |
|
if m: |
|
out[m.group("key")] = int(m.group("n")) |
|
return out |
|
|
|
while i < len(lines): |
|
line = lines[i].rstrip("\n") |
|
if not line.strip(): |
|
i += 1; continue |
|
# 次セクション(何か別のヘッダが来たら終了) |
|
if not line.startswith("|"): |
|
break |
|
|
|
m = win_re.match(line) |
|
if m: |
|
current_window = {"title": m.group("title"), "folders": []} |
|
windows.append(current_window) |
|
current_folder = None |
|
i += 1; continue |
|
|
|
m = fol_re.match(line) |
|
if m and current_window is not None: |
|
current_folder = { |
|
"name": m.group("name"), |
|
"file_count": int(m.group("count")), |
|
"file_types": {}, |
|
"conf_files": [] |
|
} |
|
current_window["folders"].append(current_folder) |
|
i += 1; continue |
|
|
|
m = types_re.match(line) |
|
if m and current_folder is not None: |
|
current_folder["file_types"] = parse_kv_counts(m.group("rest")) |
|
i += 1; continue |
|
|
|
m = conf_re.match(line) |
|
if m and current_folder is not None: |
|
rest = m.group("rest").strip() |
|
# Conf files: の行は空のことがある(例の出力がそう) |
|
current_folder["conf_files"] = [s.strip() for s in re.split(r"\s+", rest) if s.strip()] if rest else [] |
|
i += 1; continue |
|
|
|
# それ以外の行はスキップ |
|
i += 1 |
|
|
|
return {"windows": windows}, i |
|
|
|
# ---------- postprocess(簡略) ---------- |
|
def postprocess(headers): |
|
import re |
|
h = dict(headers) |
|
m = re.match(r"^(?P<label>.+?)\s+(?P<ver>[0-9.]+)\s*\((?P<meta>.+)\)$", h.get("Version","")) |
|
if m: |
|
commit, btime = None, None |
|
for p in (s.strip() for s in m.group("meta").split(",")): |
|
if re.fullmatch(r"[0-9a-f]{7,}", p): commit = p |
|
elif re.search(r"\d{4}-\d{2}-\d{2}T", p): btime = p |
|
h["Version"] = {"name": m.group("label"), "version": m.group("ver"), "commit": commit, "build_time": btime, "raw": headers.get("Version","")} |
|
m = re.match(r"^(?P<os>[A-Za-z_]+)\s*(?P<arch>x64|arm64)?\s*(?P<ver>[\w\.\-]+)?$", h.get("OS Version","")) |
|
if m: |
|
h["OS Version"] = {"os": m.group("os"), "arch": m.group("arch"), "version": m.group("ver"), "raw": headers.get("OS Version","")} |
|
m = re.match(r"^(?P<model>.+?)\s*\((?P<count>\d+)\s*x\s*(?P<speed>\d+)\)$", h.get("CPUs","")) |
|
if m: |
|
h["CPUs"] = {"model": m.group("model").strip(), "count": int(m.group("count")), "speed_mhz": int(m.group("speed")), "raw": headers.get("CPUs","")} |
|
m = re.match(r"^(?P<total>[\d\.]+)GB\s*\((?P<free>[\d\.]+)GB free\)$", h.get("Memory (System)","")) |
|
if m: |
|
h["Memory (System)"] = {"total_gb": float(m.group("total")), "free_gb": float(m.group("free")), "raw": headers.get("Memory (System)","")} |
|
if isinstance(h.get("Process Argv"), str) and h["Process Argv"]: |
|
h["Process Argv"] = [a for a in re.split(r"\s+", h["Process Argv"].strip()) if a] |
|
return h |
|
|
|
def main(): |
|
ap = argparse.ArgumentParser(description="Convert `code --status` (incl. Workspace Stats) to JSON.") |
|
ap.add_argument("--pretty", action="store_true") |
|
args = ap.parse_args() |
|
|
|
text = read_input() |
|
lines = text.splitlines() |
|
|
|
headers, idx = parse_headers(lines, 0) |
|
processes, idx = parse_process_table(lines, idx) |
|
workspace, _ = parse_workspace_stats(lines, idx) |
|
|
|
result = { |
|
"headers": postprocess(headers), |
|
"processes": processes, |
|
"workspace_stats": workspace, # None の場合もあり |
|
"raw": text |
|
} |
|
|
|
print(json.dumps(result, ensure_ascii=False, indent=2 if args.pretty else None)) |
|
|
|
if __name__ == "__main__": |
|
main() |