#!/usr/bin/env python3 # -*- coding: utf-8 -*- """项目管理界面 —— 照 dsh-task-board 的「项目 × 卡片」模式做的**只读**看板。 ## 这是什么（用户 2026-10-06 原话） > 「不动现有任务会话看板，不是让你整体搬，是照的他的模式做一个 项目管理的界面 > 来管理 执行任务，**目标等于项目**」 ⇒ 三条硬约束： ① ⛔ **不动**现有看板（`board.py` / `board.html` / 20099 一律零改动）； ② ⛔ **不整体搬** dsh-task-board（它是 DSH 插件、20k 行 TS，宿主不同，搬不动）； ③ ✅ **照它的模式**：项目下拉 ＋ 卡片列 ＋ 点卡片看详情。 ## 数据从哪来（🔴 只读，零写） · 复用现有看板已经算好的 `GET /board.json`（20099）—— ⛔ **不重算、不直读台账** （重算 ＝ 把 `board.py` 那套逻辑抄第二份 ⇒ 同族 P0「一条规则两份实现」必漂）。 · 本项目**自己不再产数据**，只做**渲染**。 ## 映射（目标 → 项目） · `goals[]` → **项目**（`goal.short`＝项目名，`tab_created`＝建立时间） · `labor[]` → 该项目的**工作流/分组**（`line`／`items`） · `tasks{}`（台账） → **卡片**（`state` → 列） · `sessions[]` → 卡片可跳的**会话**（`id8` 是跳转键） ## 列（⛔ 用机制真实四态，不硬套它那五列） dsh-task-board 那套是 待规划/待办/进行中/已完成/已失败； 本机制台账只有 **待执行 / 执行中 / 已完成 / 有阻碍** 四态 ⇒ 硬套五列会凭空多出机制里**不存在**的「待规划」⇒ 看着像有、其实永远空。 ✅ 用户问「什么工作流状态 干啥用的」⇒ 就是这一列的含义；**按真实态列**。 ## 起法 python project-board.py --serve 20100 # ⛔ 与 20099 各起各的，互不影响 """ from __future__ import annotations import argparse import json import os 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" # 🔴 上游＝**现有看板**（⛔ 不直读台账）。允许用环境变量指到别的实例。 UPSTREAM = os.environ.get("DSH_PROJECT_BOARD_UPSTREAM") or "http://127.0.0.1:20099/board.json" _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` 只说"失败"，看不出为什么）。 ⚠️ 这正是"终端里好好的、一进计划任务就死"的典型 —— ⛔ 别用裸 `print` 起常驻。 ✅ 有 stdout 才写；没有就**静默丢掉**（这是常驻进程，⛔ 不该因为没地方打印就崩）。 """ f = getattr(sys, "stdout", None) if f is None: return try: f.write(msg + "\n") f.flush() except Exception: pass def fetch_upstream(timeout: float = 5.0) -> dict: """拉现有看板的快照（⛔ 只读；失败**如实报错**，⛔ 不返回空壳假装没数据）。""" req = urllib.request.Request(UPSTREAM, headers={"Accept": "application/json"}) with _OPENER.open(req, timeout=timeout) as r: return json.loads(r.read().decode("utf-8")) # 台账四态 → 列（🔴 与 board.py::_queue 的 `by` 键**逐字一致**） COLS = [ ("pending", "待执行"), ("running", "执行中"), ("blocked", "有阻碍"), ("done", "已完成"), ] def to_project_view(snap: dict) -> dict: """把现有看板的 `/board.json` 转成**项目管理视图**（纯映射，⛔ 不重算业务）。""" goals = snap.get("goals") or [] projects = [] for gi, g in enumerate(goals): gg = g.get("goal") or {} q = g.get("queue") or {} qby = q.get("by") or {} tasks = g.get("tasks") or {} # 台账条目 → 卡片（只取当前目标的那几条；无归属的按 legacy 单列，⛔ 不混进项目） cards = [] for tid, t in (tasks or {}).items(): st = str((t or {}).get("state") or "").strip().lower() if st not in {c[0] for c in COLS}: continue cards.append({ "id": tid, "state": st, "by": str((t or {}).get("by") or ""), "line": str((t or {}).get("line") or ""), "artifact": str((t or {}).get("artifact") or ""), "block_reason": str((t or {}).get("block_reason") or ""), "t_start": str((t or {}).get("t_start") or ""), "t_end": str((t or {}).get("t_end") or ""), }) # 会话（卡片"点开到对应会话"用；`id8` ＝ 跳转键） sess = [{"id8": s.get("id8"), "title": s.get("title") or "", "status": s.get("status") or "", "role": s.get("role") or "", "age_min": s.get("age_min")} for s in (g.get("sessions") or [])] # 工作流分组（labor） flows = [{"line": x.get("line") or x.get("name") or "", "name": x.get("name") or x.get("line") or "", "total": x.get("total") or 0, "done": x.get("done") or 0, "open": x.get("open") or 0, "state": x.get("state") or ""} for x in (g.get("labor") or [])] orphan = g.get("orphan") or {} projects.append({ "i": gi, "name": str(gg.get("short") or gg.get("title") or "(未命名)"), "title": str(gg.get("title") or ""), "peer": str(g.get("peer") or ""), "created": str(g.get("tab_created") or ""), "life": str(g.get("life") or ""), "life_at": str(g.get("life_at") or ""), "cards": cards, "sessions": sess, "flows": flows, "orphan": {"n": orphan.get("n") or 0, "lines": orphan.get("lines") or []}, "queue": {"total": q.get("total") or 0, "open": q.get("open") or 0, "done": qby.get("done") or 0, "pending": qby.get("pending") or 0, "running": qby.get("running") or 0, "blocked": qby.get("blocked") or 0, "legacy": q.get("legacy") or 0}, "acc": g.get("acc_summary") or "", }) return { "ok": True, "ts": snap.get("ts") or "", "ws_name": snap.get("ws_name") or "", "ws_root": snap.get("ws_root") or "", "upstream": UPSTREAM, "cols": [{"key": k, "label": lb} for k, lb in COLS], "projects": projects, } class Handler(BaseHTTPRequestHandler): server_version = "ProjectBoard/1.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 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: snap = fetch_upstream() body = json.dumps(to_project_view(snap), ensure_ascii=False).encode("utf-8") self._send(200, body, "application/json; charset=utf-8") except Exception as e: # 🔴 上游读不到 ⇒ **如实报错**（⛔ 不返回空列表假装"没有项目"） self._send(502, json.dumps( {"ok": False, "err": "读不到上游看板：%s" % e, "upstream": UPSTREAM}, ensure_ascii=False).encode("utf-8"), "application/json; charset=utf-8") elif path == "/healthz": self._send(200, json.dumps({"ok": True, "upstream": UPSTREAM}, ensure_ascii=False).encode("utf-8"), "application/json; charset=utf-8") 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, UPSTREAM)) try: srv.serve_forever() except KeyboardInterrupt: pass return 0 if a.once: try: v = to_project_view(fetch_upstream()) except Exception as e: _say("⛔ 读不到上游：%s" % e) return 2 _say("上游＝%s｜工作区＝%s" % (v["upstream"], v["ws_name"])) for p in v["projects"]: _say(" 项目 %-18r 建立=%s 卡片=%d 会话=%d 工作流=%d 台账:总%s/待%s/中%s/阻%s/完成%s%s" % (p["name"][:18], p["created"] or "-", len(p["cards"]), len(p["sessions"]), len(p["flows"]), p["queue"]["total"], p["queue"]["pending"], p["queue"]["running"], p["queue"]["blocked"], p["queue"]["done"], ("（另有历史 %d 件）" % p["queue"]["legacy"]) if p["queue"]["legacy"] else "")) return 0 ap.print_help() return 0 if __name__ == "__main__": sys.exit(main())