admin
89f820a0f8
用户口径：「项目管理管的是本机所有工作区的项目 不是单个工作区的，可以 通过项目管理工具 创建新目标（新工作区），管理目标」。 数据面 - ⛔ 不再拉「单一上游看板」（并非每个区都起着 board.py ⇒ 会静默漏区） - ✅ 直读各区文件：goal.json / tasks.json / collabd-state.json / collabd.config.json - 「装了机制」判据＝有 collabd.config.json 或 goal.json；跳过的也带出来 （skipped 列表，同族 P0「作用域不许静默排除」） 增删 - 增：POST /api/create → init_workspace.py（骨架＋目标登记） - 删（轻）：POST /api/delete{mode:"goal"} ⇒ goal.json 改名归档（可逆） - 删（重）：POST /api/delete{mode:"ws"} ⇒ 送系统回收站（可逆） 回收站：本机 gio/trash-put/trash 全没有，PowerShell Add-Type 被安全策略 拦 ⇒ 改走纯 ctypes 调 Windows Shell API SHFileOperationW。两个坑： ① pFrom 必须手搓双 NUL 结尾缓冲（LPCWSTR 会在第一个 NUL 截断） ② 判据⛔ 不是 rc==0 —— 实测成功时 rc 恒为 2（Shell 脏状态）⇒ 假报错。 真判据＝原路径不存在 且 fAnyOperationsAborted 为假 其它修复 - init_workspace.py 子进程输出按 GBK 解码 ⇒ 中文乱码 ⇒ 显式 encoding=utf-8 - 起服务只许触发计划任务（裸 & 活不过工具调用边界，被 SIGTERM） 实测：列表 3 个项目 ＋ 12 个未装机制目录；增删两种粒度 HTTP 端到端全通； _pmtest-* 五个测试目录已进回收站且可恢复（$I 元数据逐条核对）。
552 lines
24 KiB
Python
552 lines
24 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""项目管理界面 —— 管**本机所有工作区**的项目（照 dsh-task-board 的「项目 × 卡片」模式）。
|
||
|
||
## 这是什么（用户原话）
|
||
|
||
> 「不动现有任务会话看板，不是让你整体搬，是照的他的模式做一个 项目管理的界面
|
||
> 来管理 执行任务，**目标等于项目**」
|
||
|
||
> 「项目管理管的是**本机所有工作区的项目** 不是单个工作区的，
|
||
> 可以通过项目管理工具 **创建新目标（新工作区）**，管理目标」
|
||
|
||
⇒ 硬约束：
|
||
① ⛔ **不动**现有看板（`board.py` / `board.html` / 20099 一律零改动）；
|
||
② ⛔ **不整体搬** dsh-task-board（DSH 插件、20k 行 TS，宿主不同）；
|
||
③ ✅ 照它的模式：项目列表 ＋ 卡片列 ＋ 点卡片看详情；
|
||
④ ✅ 范围 ＝ **本机所有装了机制的工作区**（⛔ 不是单区、⛔ 不是靠同伴清单）；
|
||
⑤ ✅ 能**建新目标（新工作区）**、能**删**、能**查**；⛔ **不改状态**（状态交给会话机制自己处理）。
|
||
|
||
## 数据从哪来（🔴 只读 ＋ 一处写）
|
||
|
||
**读**：直接扫本机工作区根目录下每个区自己的文件 ——
|
||
· `<区>/tmp/supervise-inbox/goal.json` → **项目**（title／short／lifecycle／acceptance）
|
||
· `<区>/tmp/supervise-inbox/tasks.json` → **卡片**（台账，`state` → 列）
|
||
· `<区>/tmp/supervise-inbox/collabd-state.json` → 队列统计
|
||
· `<区>/.workbuddy/collab/collabd.config.json` → 该区判「装没装机制」的依据
|
||
⚠️ **为什么直读文件而不是拉各区看板**：并非每个区都起着 `board.py`（实测本机 3 个装了
|
||
机制的区里，只有 1 个在跑看板）⇒ 拉看板会**静默漏掉没起看板的区**（＝"只看到一部分"）。
|
||
直读文件是**各区自己的真源**，不依赖谁在跑。
|
||
|
||
**写**：只有两处，且都是**动作型**（⛔ 不在轮询里写任何东西）——
|
||
· 建新目标／新工作区：调技能自带的 `init_workspace.py`（幂等，五步齐全）
|
||
· 删：把该工作区**整体送回收站**（⛔ 不 `rm`；可逆），或只删该区的目标登记（轻量，见下）
|
||
|
||
## 映射
|
||
|
||
· 一个**工作区** → 一个**项目**（`goal.json` 的 `short`／`title` 是项目名）
|
||
· 该区**台账条目** → **卡片**（`state` → 四列）
|
||
· 该区**会话** → 卡片可跳的会话（`id8` 是跳转键）
|
||
|
||
## 列（⛔ 用机制真实四态，不硬套它那五列）
|
||
|
||
本机制台账只有 **待执行 / 执行中 / 有阻碍 / 已完成** 四态；
|
||
硬套 dsh-task-board 的五列（含「待规划」）会凭空多出机制里**不存在**的一列。
|
||
|
||
## 起法
|
||
|
||
python project-board.py --serve 20100 # ⛔ 与 20099 各起各的，互不影响
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
import urllib.request
|
||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||
from pathlib import Path
|
||
|
||
HERE = Path(__file__).resolve().parent
|
||
PAGE = HERE / "project-board.html"
|
||
|
||
# 🔴 扫哪些根目录找工作区（`;` 分隔；可用环境变量覆盖）
|
||
WS_ROOTS = [Path(p) for p in (
|
||
os.environ.get("DSH_PM_BOARD_ROOTS") or "E:/ProgramData/AIProject"
|
||
).split(";") if p.strip()]
|
||
|
||
# 技能自带的新建入口（⛔ 不自己另写一套建区逻辑 —— 同族"一条规则两份实现"必漂）
|
||
SKILL_SCRIPTS = Path(os.environ.get("DSH_SKILL_SCRIPTS")
|
||
or r"E:/ProgramData/.workbuddy/skills/session-mechanism/scripts")
|
||
|
||
# 可选：某个区的旧看板（**只作补充**，主力是直读文件）
|
||
UPSTREAM = os.environ.get("DSH_PROJECT_BOARD_UPSTREAM") or ""
|
||
|
||
_OPENER = urllib.request.build_opener(urllib.request.ProxyHandler({})) # ⛔ 绕过 HTTP_PROXY
|
||
|
||
|
||
def _say(msg: str) -> None:
|
||
"""🔴🔴 **稳健的输出**（2026-10-06 实测栽过）。
|
||
|
||
病根：计划任务用 **`pythonw.exe`** 跑 ⇒ **没有控制台** ⇒ `sys.stdout is None`
|
||
⇒ `print()` / `sys.stdout.flush()` 直接 `AttributeError: 'NoneType' object has no
|
||
attribute 'flush'` ⇒ **进程当场崩**（而 `Result=1` 只说"失败"，看不出为什么）。
|
||
✅ 有 stdout 才写；没有就**静默丢掉**（这是常驻进程，⛔ 不该因为没地方打印就崩）。
|
||
"""
|
||
f = getattr(sys, "stdout", None)
|
||
if f is None:
|
||
return
|
||
try:
|
||
f.write(msg + "\n")
|
||
f.flush()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
# 台账四态 → 列（🔴 与 board.py::_queue 的 `by` 键**逐字一致**）
|
||
COLS = [
|
||
("pending", "待执行"),
|
||
("running", "执行中"),
|
||
("blocked", "有阻碍"),
|
||
("done", "已完成"),
|
||
]
|
||
|
||
|
||
def _load_json(p: Path, default=None):
|
||
try:
|
||
return json.loads(p.read_text(encoding="utf-8"))
|
||
except Exception:
|
||
return default
|
||
|
||
|
||
def _acc_is_pass(v) -> bool:
|
||
"""一条验收判据算不算过 —— **复用技能真身**（⛔ 不自己写词表；见 pitfalls P0-87）。"""
|
||
try:
|
||
if str(SKILL_SCRIPTS) not in sys.path:
|
||
sys.path.insert(0, str(SKILL_SCRIPTS))
|
||
import board as _b
|
||
return bool(_b.acc_is_pass(v))
|
||
except Exception:
|
||
pass
|
||
s = str(v or "").strip()
|
||
if not s:
|
||
return False
|
||
head = re.split(r"[（(]", s, maxsplit=1)[0]
|
||
for _seg in re.split(r"[：:]", head):
|
||
_t = _seg.strip().lstrip("*").strip().lower()
|
||
for _w in ("pass", "过", "通过", "达", "达标", "合格", "完成"):
|
||
if _t.startswith(_w):
|
||
return True
|
||
_t2 = _t
|
||
for _ in range(3):
|
||
_cut = ""
|
||
for _a in ("已", "已经", "均", "都", "经复核", "复核后"):
|
||
if _t2.startswith(_a) and len(_a) > len(_cut):
|
||
_cut = _a
|
||
if not _cut:
|
||
break
|
||
_t2 = _t2[len(_cut):].strip().lstrip("*").strip()
|
||
for _w in ("pass", "过", "通过", "达", "达标", "合格", "完成"):
|
||
if _t2.startswith(_w):
|
||
return True
|
||
return False
|
||
|
||
|
||
def scan_workspaces() -> list:
|
||
"""扫本机所有工作区根目录，返回**装了机制**的那些区（⛔ 没装的跳过但**单独登记**）。
|
||
|
||
🔴 「装了机制」的判据（两条任一）：
|
||
① 有 `<区>/.workbuddy/collab/collabd.config.json`（＝部署配置，机制跑起来的必要条件）
|
||
② 有 `<区>/tmp/supervise-inbox/goal.json`（＝登记过目标）
|
||
⚠️ 返回 `(installed, skipped)` 两个列表 —— **跳过的也带出来**，
|
||
⛔ 不静默丢（同族 P0：「作用域不许静默排除」）。用户口径是"只列装了机制的"，
|
||
但界面上要能看见"扫到了但没装"，否则"少了一个区"永远查不出来。
|
||
"""
|
||
installed, skipped = [], []
|
||
seen = set()
|
||
for root in WS_ROOTS:
|
||
if not root.is_dir():
|
||
continue
|
||
for d in sorted(root.iterdir()):
|
||
if not d.is_dir() or d.name.startswith("_") or d.name.startswith("."):
|
||
continue
|
||
try:
|
||
key = str(d.resolve()).lower()
|
||
except Exception:
|
||
continue
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
cfg = d / ".workbuddy" / "collab" / "collabd.config.json"
|
||
gj = d / "tmp" / "supervise-inbox" / "goal.json"
|
||
if cfg.is_file() or gj.is_file():
|
||
installed.append(d)
|
||
else:
|
||
skipped.append(d.name)
|
||
return installed, skipped
|
||
|
||
|
||
def _region_dir(ws: Path, goal: dict) -> Path:
|
||
"""目标文件夹（`执行会话/目标-<区名>-<短哈希>`）—— ⛔ 与技能同口径，不从目录名猜。"""
|
||
ex = str((goal or {}).get("execution_doc") or "").strip()
|
||
if ex:
|
||
p = ws / ex
|
||
if p.parent.is_dir():
|
||
return p.parent
|
||
return ws / "执行会话"
|
||
|
||
|
||
def build_project(ws: Path) -> dict:
|
||
"""把**一个工作区**读成「项目」（纯读，⛔ 不写任何东西）。"""
|
||
inbox = ws / "tmp" / "supervise-inbox"
|
||
goal = _load_json(inbox / "goal.json", {}) or {}
|
||
tasks = _load_json(inbox / "tasks.json", {}) or {}
|
||
st = _load_json(inbox / "collabd-state.json", {}) or {}
|
||
cfg = _load_json(ws / ".workbuddy" / "collab" / "collabd.config.json", {}) or {}
|
||
|
||
short = str(goal.get("short") or "").strip()
|
||
title = str(goal.get("title") or "").strip()
|
||
name = short or title or ("(未命名 · %s)" % ws.name)
|
||
|
||
# ── 卡片（台账条目；只取四态里的）
|
||
keys = {c[0] for c in COLS}
|
||
cards = []
|
||
for tid, t in (tasks or {}).items():
|
||
t = t or {}
|
||
s = str(t.get("state") or "").strip().lower()
|
||
if s not in keys:
|
||
continue
|
||
cards.append({
|
||
"id": tid, "state": s,
|
||
"title": str(t.get("title") or tid),
|
||
"by": str(t.get("by") or ""),
|
||
"line": str(t.get("line") or ""),
|
||
"artifact": str(t.get("artifact") or ""),
|
||
"block_reason": str(t.get("block_reason") or ""),
|
||
"t_start": str(t.get("t_start") or ""),
|
||
"t_end": str(t.get("t_end") or ""),
|
||
})
|
||
by = {k: 0 for k in keys}
|
||
for c in cards:
|
||
by[c["state"]] += 1
|
||
|
||
# ── 队列统计（以 collabd-state.json 为准，⛔ 不重算）
|
||
qi = st.get("queue_info") or {}
|
||
qby = qi.get("by") or {}
|
||
|
||
# ── 验收判据（🔴 走真身归一化，⛔ 不比字面值）
|
||
acc = goal.get("acceptance_state") or {}
|
||
eff = [k for k in acc if not str(k).startswith("_")]
|
||
acc_pass = sum(1 for k in eff if _acc_is_pass(acc[k]))
|
||
acc_items = [{"k": k, "v": str(acc[k]), "pass": _acc_is_pass(acc[k])} for k in eff]
|
||
|
||
# ── 工作流分组（按台账的 `line` 聚合）
|
||
flows = {}
|
||
for c in cards:
|
||
ln = c["line"] or "(未分组)"
|
||
f = flows.setdefault(ln, {"line": ln, "total": 0, "done": 0, "open": 0})
|
||
f["total"] += 1
|
||
if c["state"] == "done":
|
||
f["done"] += 1
|
||
else:
|
||
f["open"] += 1
|
||
|
||
topics = [str(x) for x in (goal.get("topics") or []) if str(x).strip()]
|
||
return {
|
||
"ws": str(ws).replace("\\", "/"),
|
||
"dir": ws.name,
|
||
"name": name,
|
||
"title": title,
|
||
"topics": topics,
|
||
"life": str(goal.get("lifecycle") or "").strip(),
|
||
"life_at": str(goal.get("lifecycle_at") or "").strip(),
|
||
"declared_at": str(goal.get("declared_at") or "").strip(),
|
||
"has_goal": bool(title or short),
|
||
"cards": cards,
|
||
"flows": [flows[k] for k in sorted(flows)],
|
||
"queue": {
|
||
"total": qi.get("n") or len(cards),
|
||
"pending": qby.get("pending") or by.get("pending", 0),
|
||
"running": qby.get("running") or by.get("running", 0),
|
||
"blocked": qby.get("blocked") or by.get("blocked", 0),
|
||
"done": qby.get("done") or by.get("done", 0),
|
||
"open": (qby.get("pending") or 0) + (qby.get("running") or 0)
|
||
+ (qby.get("blocked") or 0),
|
||
},
|
||
"acc": {"n": len(eff), "pass": acc_pass, "items": acc_items},
|
||
"region": str(_region_dir(ws, goal)).replace("\\", "/"),
|
||
"port": cfg.get("singleton_port"),
|
||
"goal_dir": str(inbox).replace("\\", "/"),
|
||
}
|
||
|
||
|
||
def to_project_view() -> dict:
|
||
"""本机所有装了机制的工作区 → 项目管理视图。"""
|
||
installed, skipped = scan_workspaces()
|
||
projects = [build_project(w) for w in installed]
|
||
# 🔴 排序：先按「有目标」→ 再按**建立时间**升序（与看板 tab 同口径：次序不漂）
|
||
projects.sort(key=lambda p: (0 if p["has_goal"] else 1, p["declared_at"] or "~", p["name"]))
|
||
for i, p in enumerate(projects):
|
||
p["i"] = i
|
||
return {
|
||
"ok": True,
|
||
"ts": time.strftime("%Y-%m-%d %H:%M:%S"),
|
||
"roots": [str(r).replace("\\", "/") for r in WS_ROOTS],
|
||
"cols": [{"key": k, "label": lb} for k, lb in COLS],
|
||
"projects": projects,
|
||
"skipped": skipped,
|
||
}
|
||
|
||
|
||
# ── 动作层（⛔ 只在点按钮时跑，⛔ 不在轮询里写任何东西）──────────────────────
|
||
|
||
def _valid_ws_name(n: str) -> str:
|
||
"""工作区目录名校验（⛔ 不许路径穿越、⛔ 不许空）。"""
|
||
n = str(n or "").strip()
|
||
if not n or n in (".", "..") or "/" in n or "\\" in n:
|
||
return ""
|
||
if re.search(r'[<>:"|?*]', n):
|
||
return ""
|
||
return n
|
||
|
||
|
||
def act_create(root: str, name: str, title: str, short: str = "") -> dict:
|
||
"""**建新目标（新工作区）** —— 调技能自带的 `init_workspace.py`（幂等）。
|
||
|
||
🔴 ⛔ 不自己另写一套建区逻辑（那会与技能漂成两份）。只做：
|
||
① 校验名字 ② 拼路径 ③ 传给技能脚本 ④ 如实回报它的输出。
|
||
⚠️ 非 ASCII 路径：走**列表参数**调子进程（⛔ 不经 shell），避免编码问题。
|
||
"""
|
||
rootp = Path(root)
|
||
nm = _valid_ws_name(name)
|
||
if not nm:
|
||
return {"ok": False, "err": "工作区名不合法（不许空、不许带路径分隔符或 <>:\"|?*）"}
|
||
if not rootp.is_dir():
|
||
return {"ok": False, "err": "根目录不存在：%s" % rootp}
|
||
ws = rootp / nm
|
||
if ws.exists():
|
||
return {"ok": False, "err": "已存在：%s（⛔ 不覆盖；要改就进去改）" % ws}
|
||
script = SKILL_SCRIPTS / "init_workspace.py"
|
||
if not script.is_file():
|
||
return {"ok": False, "err": "找不到技能脚本：%s" % script}
|
||
cmd = [sys.executable, str(script), str(ws),
|
||
"--title", str(title or nm), "--short", str(short or nm)]
|
||
# 🔴🔴 **必须显式指定编码**（2026-10-06 实测踩到）：
|
||
# 子进程输出的是 **UTF-8**（`init_workspace.py` 开头就 `reconfigure("utf-8")`），
|
||
# 而不给 `encoding` 时 Python 按**本机默认编码**（中文 Windows ＝ GBK）解码
|
||
# ⇒ 中文全变乱码（`锛屽埆璺戞妧鑳界洰褰�`）⇒ 界面上给用户看的日志**完全没法读**。
|
||
# ⚠️ 同族：本项目已登记「出口一律声明编码」（P0-10）—— 这里是**入口**侧的同一条。
|
||
# ⚠️ 顺带设 `PYTHONIOENCODING`，兜住子进程自己内部再起孙进程的情况。
|
||
_env = dict(os.environ)
|
||
_env["PYTHONIOENCODING"] = "utf-8"
|
||
try:
|
||
r = subprocess.run(cmd, capture_output=True, text=True, errors="replace",
|
||
encoding="utf-8", timeout=300, cwd=str(rootp), env=_env)
|
||
except Exception as e:
|
||
return {"ok": False, "err": "跑 init_workspace 失败：%r" % (e,)}
|
||
out = ((r.stdout or "") + (r.stderr or "")).strip()
|
||
return {"ok": r.returncode == 0, "rc": r.returncode, "ws": str(ws).replace("\\", "/"),
|
||
"out": out[-4000:], "cmd": " ".join(cmd)}
|
||
|
||
|
||
def _trash_dir(p: Path):
|
||
"""把目录送进**系统回收站**（🔴 可逆；⛔ 绝不 `rm -rf`）。返回 `(ok, info)`。
|
||
|
||
🔴 2026-10-06 实测：本机 **PortableGit 里 `gio` / `trash-put` / `trash` 一个都没有**
|
||
⇒ 原来的 `gio trash` 那条路**恒失败**。而 WorkBuddy 的 `Add-Type` 被安全策略拦
|
||
⇒ 也用不了 `Microsoft.VisualBasic.FileIO`。
|
||
|
||
⇒ 现走 **纯 ctypes 调 Windows Shell API `SHFileOperationW`**（不编译任何代码，
|
||
不碰 .NET，不弹 UAC）—— 这是本机唯一实测可用的回收站途径。
|
||
"""
|
||
# ① Windows：SHFileOperationW ＋ FO_DELETE ＋ FOF_ALLOWUNDO（=真正进回收站）
|
||
if os.name == "nt":
|
||
try:
|
||
import ctypes
|
||
from ctypes import wintypes
|
||
|
||
class SHFILEOPSTRUCTW(ctypes.Structure):
|
||
_fields_ = [
|
||
("hwnd", wintypes.HWND),
|
||
("wFunc", wintypes.UINT),
|
||
("pFrom", ctypes.c_void_p), # 🔴 手搓双 \0 结尾缓冲，见下
|
||
("pTo", ctypes.c_void_p),
|
||
("fFlags", ctypes.c_uint16),
|
||
("fAnyOperationsAborted", wintypes.BOOL),
|
||
("hNameMappings", ctypes.c_void_p),
|
||
("lpszProgressTitle", ctypes.c_void_p),
|
||
]
|
||
|
||
FO_DELETE = 0x0003
|
||
FOF_SILENT = 0x0004
|
||
FOF_NOCONFIRMATION = 0x0010
|
||
FOF_ALLOWUNDO = 0x0040 # 🔴 有它才进回收站；没它就=永久删除
|
||
FOF_NOERRORUI = 0x0400
|
||
FOF_NOCONFIRMMKDIR = 0x0200
|
||
|
||
# 🔴🔴 `pFrom` 必须**双 \0 结尾**。ctypes 的 `LPCWSTR` **会在第一个 \0 截断**
|
||
# ⇒ 必须手搓 `c_wchar_p` 缓冲并把地址塞进 `c_void_p` 字段。
|
||
abspath = str(p.resolve()).replace("/", "\\")
|
||
_buf = ctypes.create_unicode_buffer(abspath, len(abspath) + 2)
|
||
op = SHFILEOPSTRUCTW()
|
||
op.hwnd = None
|
||
op.wFunc = FO_DELETE
|
||
op.pFrom = ctypes.cast(_buf, ctypes.c_void_p)
|
||
op.pTo = None
|
||
op.fFlags = (FOF_ALLOWUNDO | FOF_NOCONFIRMATION | FOF_SILENT
|
||
| FOF_NOERRORUI | FOF_NOCONFIRMMKDIR)
|
||
rc = ctypes.windll.shell32.SHFileOperationW(ctypes.byref(op))
|
||
# 🔴🔴 **判据＝「东西没了 且 没被中止」，⛔ 不是 `rc == 0`**。
|
||
# 2026-10-06 实测：送回收站**成功**（目录消失、回收站里 18 条 `$I` 元数据
|
||
# 都能解出原路径），但返回值恒为 **rc=2（ERROR_FILE_NOT_FOUND）** ——
|
||
# 这是 Shell 内部探测残留的 `GetLastError`，**不是真失败**。
|
||
# ⇒ 拿 `rc` 当判据会**假报错**（用户看到"失败"但东西已经进回收站了）。
|
||
# ✅ 真判据只有两条：① 原路径不存在了 ② `fAnyOperationsAborted` 为假。
|
||
if (not p.exists()) and (not op.fAnyOperationsAborted):
|
||
return True, abspath
|
||
return False, "SHFileOperationW rc=%s aborted=%s exists=%s" % (
|
||
rc, bool(op.fAnyOperationsAborted), p.exists())
|
||
except Exception as e: # noqa: BLE001
|
||
_sh_err = "SHFileOperationW 异常：%r" % (e,)
|
||
else:
|
||
_sh_err = "非 Windows 平台"
|
||
|
||
# ② 兜底：类 Unix 的 `gio trash`（本机实测没有，但留着以备换机）
|
||
for cmd in (["gio", "trash", str(p)], ["trash-put", str(p)]):
|
||
try:
|
||
r = subprocess.run(cmd, capture_output=True, text=True, errors="replace",
|
||
encoding="utf-8", timeout=120)
|
||
if r.returncode == 0:
|
||
return True, str(p).replace("\\", "/")
|
||
except FileNotFoundError:
|
||
continue
|
||
except Exception: # noqa: BLE001
|
||
continue
|
||
|
||
return False, "%s；且 gio/trash-put 均不可用" % _sh_err
|
||
|
||
|
||
def act_delete(ws_path: str, mode: str = "goal") -> dict:
|
||
"""**删**（⛔ 不改状态；两种粒度）。
|
||
|
||
· `mode="goal"`（默认，轻量）：只删该区的**目标登记**（`goal.json` ⇒ 改名归档），
|
||
工作区与台账**原样保留** ⇒ 该项目从列表消失，但东西都还在（可逆）。
|
||
· `mode="ws"`（重）：把**整个工作区**送回收站（⛔ 不 `rm`）⇒ 需要二次确认。
|
||
|
||
🔴 两条都**可逆**；⛔ 绝不 `rm -rf`；⛔ 不碰别的区。
|
||
"""
|
||
ws = Path(ws_path)
|
||
if not ws.is_dir():
|
||
return {"ok": False, "err": "工作区不存在：%s" % ws}
|
||
if mode == "goal":
|
||
gj = ws / "tmp" / "supervise-inbox" / "goal.json"
|
||
if not gj.is_file():
|
||
return {"ok": False, "err": "该区没有目标登记（%s）" % gj}
|
||
bak = gj.with_name("goal.json.archived-%s" % time.strftime("%Y%m%d-%H%M%S"))
|
||
try:
|
||
gj.rename(bak)
|
||
except Exception as e:
|
||
return {"ok": False, "err": "归档目标失败：%r" % (e,)}
|
||
return {"ok": True, "mode": "goal", "moved_to": str(bak).replace("\\", "/")}
|
||
if mode == "ws":
|
||
ok, info = _trash_dir(ws)
|
||
if ok:
|
||
return {"ok": True, "mode": "ws", "trashed": info}
|
||
# 🔴 **fail-closed**：本机没有任何可用的回收站途径 ⇒ **如实报错，⛔ 不 rm 兜底**
|
||
return {"ok": False, "err": "送回收站失败（⛔ 不用 rm 兜底）：%s" % info}
|
||
return {"ok": False, "err": "未知 mode：%s（只支持 goal / ws）" % mode}
|
||
|
||
|
||
# ── HTTP ─────────────────────────────────────────────────────────────────
|
||
|
||
class Handler(BaseHTTPRequestHandler):
|
||
server_version = "ProjectBoard/2.0"
|
||
|
||
def _send(self, code: int, body: bytes, ctype: str) -> None:
|
||
try:
|
||
self.send_response(code)
|
||
self.send_header("Content-Type", ctype)
|
||
self.send_header("Content-Length", str(len(body)))
|
||
self.send_header("Cache-Control", "no-store")
|
||
self.end_headers()
|
||
self.wfile.write(body)
|
||
except Exception:
|
||
pass
|
||
|
||
def _json(self, code: int, obj) -> None:
|
||
self._send(code, json.dumps(obj, ensure_ascii=False).encode("utf-8"),
|
||
"application/json; charset=utf-8")
|
||
|
||
def do_GET(self): # noqa: N802
|
||
path = self.path.split("?")[0]
|
||
if path in ("/", "/index.html", "/project-board.html"):
|
||
try:
|
||
self._send(200, PAGE.read_bytes(), "text/html; charset=utf-8")
|
||
except Exception as e:
|
||
self._send(500, ("页面读不到：%s" % e).encode("utf-8"),
|
||
"text/plain; charset=utf-8")
|
||
elif path == "/api/projects":
|
||
try:
|
||
self._json(200, to_project_view())
|
||
except Exception as e:
|
||
self._json(500, {"ok": False, "err": "扫工作区失败：%r" % (e,)})
|
||
elif path == "/healthz":
|
||
self._json(200, {"ok": True, "roots": [str(r) for r in WS_ROOTS]})
|
||
else:
|
||
self._send(404, b"not found", "text/plain; charset=utf-8")
|
||
|
||
def do_POST(self): # noqa: N802
|
||
path = self.path.split("?")[0]
|
||
try:
|
||
ln = int(self.headers.get("Content-Length") or 0)
|
||
raw = self.rfile.read(ln) if ln else b"{}"
|
||
body = json.loads(raw.decode("utf-8") or "{}")
|
||
except Exception as e:
|
||
self._json(400, {"ok": False, "err": "body 解析失败：%r" % (e,)})
|
||
return
|
||
if path == "/api/create":
|
||
self._json(200, act_create(str(body.get("root") or ""),
|
||
str(body.get("name") or ""),
|
||
str(body.get("title") or ""),
|
||
str(body.get("short") or "")))
|
||
elif path == "/api/delete":
|
||
self._json(200, act_delete(str(body.get("ws") or ""),
|
||
str(body.get("mode") or "goal")))
|
||
else:
|
||
self._send(404, b"not found", "text/plain; charset=utf-8")
|
||
|
||
def log_message(self, *a):
|
||
pass
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser(description="项目管理界面（本机所有工作区 · 增删查）")
|
||
ap.add_argument("--serve", type=int, default=0, metavar="PORT",
|
||
help="起 HTTP 服务（默认不启，只做一次扫描并打印摘要）")
|
||
ap.add_argument("--once", action="store_true", help="扫一次并打印摘要")
|
||
a = ap.parse_args()
|
||
|
||
if a.serve:
|
||
srv = ThreadingHTTPServer(("127.0.0.1", a.serve), Handler)
|
||
_say("项目管理界面已起：http://127.0.0.1:%d/ （根目录 %s）"
|
||
% (a.serve, ", ".join(str(r) for r in WS_ROOTS)))
|
||
try:
|
||
srv.serve_forever()
|
||
except KeyboardInterrupt:
|
||
pass
|
||
return 0
|
||
|
||
if a.once:
|
||
v = to_project_view()
|
||
_say("根目录＝%s" % ", ".join(v["roots"]))
|
||
_say("项目 %d 个：" % len(v["projects"]))
|
||
for p in v["projects"]:
|
||
_say(" %-22r 建立=%-16s 生命=%-5s 卡片=%d 台账:总%s/待%s/中%s/阻%s/完成%s 验收 %d/%d"
|
||
% (p["name"][:22], p["declared_at"] or "-", p["life"] or "-",
|
||
len(p["cards"]), p["queue"]["total"], p["queue"]["pending"],
|
||
p["queue"]["running"], p["queue"]["blocked"], p["queue"]["done"],
|
||
p["acc"]["pass"], p["acc"]["n"]))
|
||
if v["skipped"]:
|
||
_say("（扫到但**没装机制**、未列出：%s）" % "、".join(v["skipped"]))
|
||
return 0
|
||
ap.print_help()
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|