import os import http.server import socketserver import urllib.parse import html import json import sqlite3 import markdown from markdown.extensions.toc import TocExtension PORT = 9120 BOARDS_DIR = "/root/.hermes/kanban/boards" FILES_ROOT = "/" FILES_BLOCKED = [ "auth.json", ".env", "google_token.json", ".git", "__pycache__", "node_modules", ".key", ".pem", "credentials.json", "token.pickle", "state.db", ] CTYPE = { ".html": "text/html; charset=utf-8", ".js": "application/javascript; charset=utf-8", ".css": "text/css; charset=utf-8", ".json": "application/json; charset=utf-8", ".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".gif": "image/gif", ".svg": "image/svg+xml", ".pdf": "application/pdf" } RAW_EXT = { ".md", ".txt", ".yaml", ".yml", ".json", ".csv", ".toml", ".ini", ".cfg", ".conf", ".log", ".xml", ".html", ".sh", ".py", ".js", ".css", ".env.example" } def _is_blocked(fpath): fpath = fpath.replace("\\", "/") parts = fpath.split("/") for b in FILES_BLOCKED: if b in parts: return True return False def _get_available_boards(): try: return sorted([d for d in os.listdir(BOARDS_DIR) if os.path.isdir(os.path.join(BOARDS_DIR, d))]) except Exception: return ["nv-office"] def _get_kanban_db_path(board_name): board_name = "".join([c for c in board_name if c.isalnum() or c in ("-", "_")]) return os.path.join(BOARDS_DIR, board_name, "kanban.db") def _get_kanban_data(board_name, sort_by="priority"): db_path = _get_kanban_db_path(board_name) if not os.path.exists(db_path): return [] try: con = sqlite3.connect("file:%s?mode=ro" % db_path, uri=True, timeout=2) con.row_factory = sqlite3.Row if sort_by == "created": order = "created_at DESC" elif sort_by == "created_asc": order = "created_at ASC" elif sort_by == "title": order = "title ASC" elif sort_by == "title_desc": order = "title DESC" elif sort_by == "priority_asc": order = "priority ASC" else: # priority desc order = "priority DESC, created_at DESC" tasks = con.execute( f"SELECT id, title, assignee, status, priority FROM tasks WHERE status != 'archived' ORDER BY {order}" ).fetchall() con.close() return tasks except Exception as e: print(f"Error reading kanban for board {board_name}: {e}") return [] CSS = """ * { box-sizing: border-box; } body { font-family: 'Inter', -apple-system, BlinkMacSystemFont, sans-serif; background: #0f1117; color: #c9d1d9; padding: 1.5rem; margin: 0; } .container { max-width: 1400px; margin: 0 auto; } h1 { color: #f0c040; font-size: 1.5rem; margin: 0 0 1rem 0; display: flex; align-items: center; gap: 0.5rem; } h1 small { font-size: 0.8rem; color: #8b949e; font-weight: normal; } /* TOP BAR */ .topbar { background: #161b22; border: 1px solid #30363d; border-radius: 8px; padding: 0.75rem 1rem; margin-bottom: 1rem; display: flex; align-items: center; flex-wrap: wrap; gap: 0.75rem; } .topbar-label { color: #8b949e; font-size: 0.8rem; font-weight: 600; text-transform: uppercase; letter-spacing: 0.5px; } .board-selector-group { display: flex; gap: 0.5rem; } .board-btn { background: #21262d; color: #c9d1d9; padding: 0.4rem 0.8rem; border-radius: 6px; text-decoration: none; border: 1px solid #30363d; font-size: 0.85rem; font-weight: 500; transition: all 0.15s; } .board-btn:hover { background: #30363d; border-color: #58a6ff; } .board-btn.active-board { background: #1f6feb; color: #fff; border-color: #1f6feb; } .sort-select { background: #21262d; color: #c9d1d9; border: 1px solid #30363d; border-radius: 6px; padding: 0.4rem 0.6rem; font-size: 0.85rem; cursor: pointer; } .controls-row { display: flex; align-items: center; gap: 0.75rem; flex-wrap: wrap; } .controls-row a { color: #58a6ff; text-decoration: none; font-size: 0.85rem; } .controls-row a:hover { text-decoration: underline; } /* BOARD */ .board { display: flex; gap: 1rem; overflow-x: auto; align-items: flex-start; } .column { background: #161b22; border: 1px solid #30363d; border-radius: 8px; min-width: 260px; width: 260px; flex-shrink: 0; display: flex; flex-direction: column; } .column-header { padding: 0.75rem 1rem; border-bottom: 1px solid #30363d; font-weight: 600; font-size: 0.85rem; color: #8b949e; text-transform: uppercase; letter-spacing: 0.5px; display: flex; justify-content: space-between; align-items: center; } .column-count { background: #21262d; color: #c9d1d9; padding: 0.1rem 0.5rem; border-radius: 10px; font-size: 0.75rem; } .task-list { padding: 0.5rem; flex: 1; } .task { background: #0d1117; border: 1px solid #30363d; border-left: 3px solid #58a6ff; border-radius: 6px; padding: 0.75rem; margin-bottom: 0.5rem; transition: border-color 0.15s; } .task:hover { border-color: #8b949e; } .task-priority-high { border-left-color: #f85149; } .task-priority-med { border-left-color: #d29922; } .task-title { color: #e6edf3; font-size: 0.82rem; line-height: 1.4; margin-bottom: 0.5rem; } .task-meta { font-size: 0.72rem; color: #8b949e; display: flex; justify-content: space-between; align-items: center; } .task-assignee { display: flex; align-items: center; gap: 0.3rem; } .task-tag { background: #21262d; padding: 0.1rem 0.4rem; border-radius: 4px; font-family: monospace; font-size: 0.7rem; } /* Collect/Collapse */ .toggle-btn { display: block; width: 100%; text-align: center; background: #21262d; border: 1px dashed #30363d; border-radius: 6px; color: #8b949e; padding: 0.5rem; font-size: 0.78rem; cursor: pointer; margin-top: 0.25rem; transition: all 0.15s; } .toggle-btn:hover { background: #30363d; color: #c9d1d9; border-style: solid; } .hidden-tasks { display: none; } .hidden-tasks.visible { display: block; } """ def _render_kanban_html(board_name, tasks, sort_by): available = _get_available_boards() board_btns = [] for b in available: cls = "active-board" if b == board_name else "" board_btns.append( f"{html.escape(b)}" ) sort_options = [ ("priority", "Priority Cao→Thấp"), ("priority_asc", "Priority Thấp→Cao"), ("created", "Mới nhất"), ("created_asc", "Cũ nhất"), ("title", "A→Z"), ("title_desc", "Z→A"), ] sort_opts = "".join( f"" for v, l in sort_options ) cols = [ ("triage", "Cần làm"), ("running", "Đang chạy"), ("blocked", "Tắc"), ("done", "Xong"), ] cols_html = [] for status_key, status_name in cols: col_tasks = [ t for t in tasks if t["status"] == status_key or (status_key == "triage" and t["status"] in ("todo", "ready")) ] count = len(col_tasks) show_initially = min(count, 5) task_items = [] for i, t in enumerate(col_tasks): pri_class = ( "task-priority-high" if int(t["priority"] or 0) >= 10 else ("task-priority-med" if int(t["priority"] or 0) >= 5 else "") ) hidden = " hidden-tasks" if i >= 5 else "" task_items.append( f"
" f"
{html.escape(t['title'] or '')}
" f"
" f"👤 {html.escape(t['assignee'] or '?')}" f"#{t['id']}" f"
" ) body = "".join(task_items[:show_initially]) if count > 5: remaining = count - 5 body += ( f"" ) body += f"
{''.join(task_items[5:])}
" cols_html.append( f"
" f"
{status_name}{count}
" f"
{body}
" ) return f""" Kanban OS - {html.escape(board_name)}

📋 Kanban OS — {html.escape(board_name)}

Board:{''.join(board_btns)}
Sắp xếp: 📁 Files
{''.join(cols_html)}
""" # ── TASK DETAIL ────────────────────────────────────────────────────── def _get_task_detail(board_name, task_id): db_path = _get_kanban_db_path(board_name) if not os.path.exists(db_path): return None try: con = sqlite3.connect("file:%s?mode=ro" % db_path, uri=True, timeout=2) con.row_factory = sqlite3.Row task = con.execute( "SELECT * FROM tasks WHERE id = ?", (task_id,) ).fetchone() con.close() if task: return dict(task) return None except Exception as e: print(f"Error reading task {task_id}: {e}") return None def _render_task_html(board_name, task): if not task: return f"""Not Found

❌ Task không tìm thấy

← Về Kanban OS""" def fmt(val): if val is None: return '' return html.escape(str(val)) def iso(ts): if not ts: return '' import datetime return datetime.datetime.fromtimestamp(int(ts)).strftime("%Y-%m-%d %H:%M") priority_label = f"{task.get('priority', 0)} {'⭐' * min(int(task.get('priority', 0) or 0), 5)}" status_map = {"todo": "📋 Cần làm", "ready": "📋 Cần làm", "running": "🔄 Đang chạy", "blocked": "🚫 Tắc", "done": "✅ Xong", "cancelled": "❌ Đã huỷ", "archived": "🗄️ Lưu trữ"} status_display = status_map.get(task.get("status", ""), task.get("status", "?")) fields = [ ("ID", fmt(task["id"])), ("Tiêu đề", fmt(task.get("title", ""))), ("Trạng thái", status_display), ("Người phụ trách", fmt(task.get("assignee", ""))), ("Độ ưu tiên", priority_label), ("Ngày tạo", iso(task.get("created_at"))), ("Ngày bắt đầu", iso(task.get("started_at"))), ("Ngày hoàn thành", iso(task.get("completed_at"))), ] optional_fields = [] if task.get("workspace_path"): optional_fields.append(("Workspace", f'{html.escape(task["workspace_path"])}')) if task.get("block_kind"): optional_fields.append(("Block Kind", fmt(task["block_kind"]))) if task.get("block_recurrences") and task["block_recurrences"] > 0: optional_fields.append(("Block Recurrences", str(task["block_recurrences"]))) if task.get("max_runtime_seconds"): optional_fields.append(("Max Runtime", f"{task['max_runtime_seconds']}s")) if task.get("consecutive_failures") and task["consecutive_failures"] > 0: optional_fields.append(("Lỗi liên tiếp", str(task["consecutive_failures"]))) rows = lambda items: "".join( f"{label}{value}" for label, value in items ) body_content = "" if task.get("body"): body_content = f"""

📝 Mô tả

{html.escape(task['body'])}
""" result_content = "" if task.get("result"): result_content = f"""

📊 Kết quả

{html.escape(task['result'])}
""" return f""" {html.escape(task.get('title','Task'))} — {html.escape(board_name)}
← Về Kanban OS

📌 {html.escape(task.get('title',''))}

{rows(fields)}
{("" if not optional_fields else f"
{rows(optional_fields)}
")} {body_content} {result_content} ← Về Kanban OS
""" # ── HTTP HANDLER ────────────────────────────────────────────────────── class KanbanHandler(http.server.BaseHTTPRequestHandler): def do_GET(self): parsed = urllib.parse.urlparse(self.path) path = parsed.path query = urllib.parse.parse_qs(parsed.query) if path == "/" or path == "/kanban": board_name = query.get("board", ["nv-office"])[0] sort_by = query.get("sort", ["priority"])[0] tasks = _get_kanban_data(board_name, sort_by) self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.end_headers() self.wfile.write(_render_kanban_html(board_name, tasks, sort_by).encode()) return if path == "/task": board_name = query.get("board", ["nv-office"])[0] task_id = query.get("id", [""])[0] if not task_id: self.send_response(302) self.send_header("Location", f"/?board={board_name}") self.end_headers() return task = _get_task_detail(board_name, task_id) self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.end_headers() self.wfile.write(_render_task_html(board_name, task).encode()) return if path.startswith("/files/"): self._serve_files(path) return self.send_response(404) self.end_headers() self.wfile.write(b"Not found") def _serve_files(self, path): rel = path[7:] target = os.path.abspath(os.path.join(FILES_ROOT, rel.lstrip("/"))) real_target = os.path.realpath(target) if _is_blocked(real_target): self.send_response(403) self.end_headers() self.wfile.write(b"Access denied") return if os.path.isdir(target): try: items = sorted(os.listdir(target)) except Exception as e: self.send_response(403) self.end_headers() self.wfile.write(str(e).encode()) return url_parts = rel.rstrip("/").split("/") if rel.strip("/") else [] bc = ['/'] acc = "" for p in url_parts: if not p: continue acc += "/" + p bc.append(f'{html.escape(p)}') crumbs = " / ".join(bc) rows = [] for item in items: full = os.path.join(target, item) is_dir = os.path.isdir(full) icon = "📁" if is_dir else "📄" lnk = f"/files/{rel.strip('/')}/{item}".replace("//", "/") rows.append(f'
  • {icon}{html.escape(item)}
  • ') html_page = f""" 📁 {html.escape(rel or '/')}
    ← Về Kanban OS
    📁 {crumbs}
    📍 {html.escape(real_target)}
    """ self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.end_headers() self.wfile.write(html_page.encode()) return if os.path.isfile(target): ext = os.path.splitext(target)[1].lower() if ext == ".md": try: with open(target, "rb") as f: md_text = f.read().decode("utf-8", errors="replace") if "[TOC]" not in md_text.upper(): md_text = "[TOC]\n\n" + md_text md = markdown.Markdown(extensions=['extra', 'tables', 'fenced_code', 'toc']) md_html = md.convert(md_text) # Support math formula rendering using MathJax mathjax_script = """ """ url_dir = os.path.dirname(rel).rstrip("/") url_parts = url_dir.split("/") if url_dir else [] bc = ['/'] acc = "" for p in url_parts: if not p: continue acc += "/" + p bc.append(f'{html.escape(p)}') crumbs = " / ".join(bc) + (" / " if bc else "") html_page = f""" {html.escape(os.path.basename(target))} {mathjax_script}
    ← Về thư mục
    📄 {crumbs}{html.escape(os.path.basename(target))}
    {md_html}
    """ self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.end_headers() self.wfile.write(html_page.encode()) except Exception as e: self.send_response(500) self.end_headers() self.wfile.write(str(e).encode()) return self.send_response(200) self.send_header("Content-Type", CTYPE.get(ext, "text/plain; charset=utf-8")) self.end_headers() with open(target, "rb") as f: self.wfile.write(f.read()) return self.send_response(404) self.end_headers() self.wfile.write(b"File not found") if __name__ == "__main__": print(f"Starting Kanban Server on port {PORT}...") socketserver.ThreadingTCPServer.allow_reuse_address = True with socketserver.ThreadingTCPServer(("0.0.0.0", PORT), KanbanHandler) as httpd: httpd.serve_forever()