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"{l} "
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"
"
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""
f"▼ Xem thêm ({remaining}) "
)
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)}
{''.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)}
"""
# ── 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 '/')}
"""
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()