#!/usr/bin/env python3 """ Durable & Interactive web server for Kanban data with 2-row layout, filters, and collapsible details. Supports column layout and Swimlanes by Profile. """ import sqlite3 import json import http.server import socketserver import urllib.parse import time import sys import os import re import html as html_lib from pathlib import Path PORT = 9120 KANBAN_DB = "/root/.hermes/kanban/boards/nv-office/kanban.db" KANBAN_ROOT = Path(KANBAN_DB).resolve().parent HERMES_ROOT = Path("/root/.hermes").resolve() ALLOWED_FILE_ROOTS = (KANBAN_ROOT, HERMES_ROOT) def slugify_heading(text, used): slug = re.sub(r"[^\w\s-]", "", text.lower(), flags=re.UNICODE) slug = re.sub(r"[\s_-]+", "-", slug, flags=re.UNICODE).strip("-") or "section" base = slug index = 2 while slug in used: slug = f"{base}-{index}" index += 1 used.add(slug) return slug def safe_file_path(raw_path): decoded = urllib.parse.unquote(raw_path) candidate = Path(decoded).expanduser() if not candidate.is_absolute(): candidate = (KANBAN_ROOT / decoded.lstrip("/")).resolve() else: candidate = candidate.resolve() if not any(candidate == root or root in candidate.parents for root in ALLOWED_FILE_ROOTS): return None if not candidate.is_file(): return None return candidate def inline_markdown(text): escaped = html_lib.escape(text) escaped = re.sub(r"`([^`]+)`", r"\1", escaped) escaped = re.sub(r"\*\*([^*]+)\*\*", r"\1", escaped) escaped = re.sub(r"\*([^*]+)\*", r"\1", escaped) escaped = re.sub(r"\[([^\]]+)\]\(([^)]+)\)", r'\1', escaped) return escaped def render_markdown_with_toc(markdown_text, title): toc = [] used_slugs = set() body = [] in_code = False paragraph = [] list_items = [] def flush_paragraph(): nonlocal paragraph if paragraph: body.append(f"

{inline_markdown(' '.join(paragraph))}

") paragraph = [] def flush_list(): nonlocal list_items if list_items: body.append("") list_items = [] for line in markdown_text.splitlines(): if line.strip().startswith("```"): flush_paragraph(); flush_list() if not in_code: in_code = True body.append("
")
            else:
                in_code = False
                body.append("
") continue if in_code: body.append(html_lib.escape(line) + "\n") continue match = re.match(r"^(#{1,6})\s+(.+?)\s*$", line) if match: flush_paragraph(); flush_list() level = len(match.group(1)) heading_text = match.group(2).strip() slug = slugify_heading(heading_text, used_slugs) toc.append({"level": level, "text": heading_text, "slug": slug}) body.append(f'{inline_markdown(heading_text)}') continue item = re.match(r"^\s*[-*+]\s+(.+)$", line) if item: flush_paragraph() list_items.append(item.group(1).strip()) continue if not line.strip(): flush_paragraph(); flush_list() continue paragraph.append(line.strip()) flush_paragraph(); flush_list() if in_code: body.append("") toc_html = "".join( f'{html_lib.escape(item["text"])}' for item in toc ) or '
Không tìm thấy heading trong file.
' safe_title = html_lib.escape(title) content_html = "\n".join(body) return f""" {safe_title}
{content_html}
""" class KanbanHandler(http.server.BaseHTTPRequestHandler): def log_message(self, format, *args): return # Silent log def do_GET(self): try: parsed = urllib.parse.urlparse(self.path) if parsed.path.startswith("/files/"): file_path = safe_file_path(parsed.path[len("/files/"):]) if file_path is None: self.send_response(404) self.send_header("Content-Type", "text/plain; charset=utf-8") self.end_headers() self.wfile.write(b"File not found or not allowed") return if file_path.suffix.lower() == ".md": markdown_text = file_path.read_text(encoding="utf-8") rendered = render_markdown_with_toc(markdown_text, file_path.name) self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.end_headers() self.wfile.write(rendered.encode("utf-8")) return content = file_path.read_bytes() self.send_response(200) self.send_header("Content-Type", "text/plain; charset=utf-8") self.end_headers() self.wfile.write(content) elif parsed.path == "/api/tasks": self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() conn = sqlite3.connect(KANBAN_DB) rows = conn.execute("SELECT id, title, body, status, assignee, priority, created_at, completed_at FROM tasks").fetchall() conn.close() data = [dict(zip(["id","title","body","status","assignee","priority","created_at","completed_at"], r)) for r in rows] self.wfile.write(json.dumps(data, ensure_ascii=False).encode()) elif parsed.path == "/": self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.end_headers() self.wfile.write(b'Redirecting to /kanban-os/') return else: self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.end_headers() html = """ NV-Office · Kanban OS

📋 NV-Office Kanban OS

""" self.wfile.write(html.encode()) except ConnectionResetError: pass except Exception as e: print(f"Error: {e}", file=sys.stderr) if __name__ == "__main__": while True: try: socketserver.TCPServer.allow_reuse_address = True with socketserver.TCPServer(("0.0.0.0", PORT), KanbanHandler) as httpd: print(f"Durable Kanban server online on port {PORT}") httpd.serve_forever() except Exception as e: print(f"Server crashed: {e}. Restarting...") time.sleep(1)