#!/usr/bin/env python3 """Hermes Web Chat Server - 与终端共享记忆，保持登录""" import http.server import json import hashlib import os import time import threading import urllib.request import urllib.error # 配置 PORT = 8888 HOST = "0.0.0.0" API_SERVER_URL = "http://127.0.0.1:8080" API_KEY = "2ecbb693e174732519b42cc6b7417c313bac1eaf003cafeb757d9f10306b07b6" WEB_PASSWORD = "Hermes2026!" # 固定 session ID - 网页和终端共享（与终端保持一致） SHARED_SESSION_ID = "hermes-main-session" # Session 存储（仅用于对话历史缓存） sessions = {} sessions_lock = threading.Lock() MAX_HISTORY = 30 def gen_session_id(): return hashlib.sha256(os.urandom(32) + str(time.time()).encode()).hexdigest()[:32] def load_memory_context(): """加载终端 agent 的记忆文件""" context_parts = [] memories_dir = os.path.expanduser("~/.hermes/memories") user_md = os.path.join(memories_dir, "USER.md") if os.path.exists(user_md): try: with open(user_md, 'r') as f: content = f.read().strip() if content: context_parts.append(f"## 用户信息\n{content}") except: pass memory_md = os.path.join(memories_dir, "MEMORY.md") if os.path.exists(memory_md): try: with open(memory_md, 'r') as f: content = f.read().strip() if content: if len(content) > 2000: content = "...(truncated)\n" + content[-2000:] context_parts.append(f"## 记忆\n{content}") except: pass if context_parts: return "\n\n".join(context_parts) return None def call_hermes_api(messages, session_id=SHARED_SESSION_ID, max_tokens=2000): """调用 Hermes API Server""" memory_context = load_memory_context() if memory_context: messages = [{"role": "system", "content": memory_context}] + messages payload = json.dumps({ "model": "hermes-agent", "messages": messages, "max_tokens": max_tokens }).encode() req = urllib.request.Request( f"{API_SERVER_URL}/v1/chat/completions", data=payload, headers={ "Content-Type": "application/json", "Authorization": f"Bearer {API_KEY}", "X-Hermes-Session-Id": session_id }, method="POST" ) try: with urllib.request.urlopen(req, timeout=300) as resp: data = json.loads(resp.read().decode()) content = data["choices"][0]["message"]["content"] return True, content except urllib.error.HTTPError as e: body = e.read().decode() if e.fp else "" return False, f"API Error {e.code}: {body}" except Exception as e: return False, str(e) # ========== HTML ========== LOGIN_PAGE = """ Hermes Chat - 登录 """ VIDEO_PAGE = """ 新闻播报 - Hermes

📺 AI 新闻播报

🎬 这是由 AI 自动生成的新闻播报视频

📐 分辨率: 1280x720 | ⏱ 时长: 96秒 | 📦 大小: 2.2MB

← 返回聊天 """ CHAT_PAGE = """ Hermes Chat

已连接到 Hermes Agent。我有完整的工具调用能力和记忆，可以编程、搜索、操作文件等。

""" # ========== Handler ========== class Handler(http.server.BaseHTTPRequestHandler): def log_message(self, fmt, *args): pass def is_authenticated(self): cookie = self.headers.get('Cookie', '') for part in cookie.split(';'): part = part.strip() if part.startswith('auth='): token = part[5:] expected = hashlib.sha256(WEB_PASSWORD.encode()).hexdigest() if token == expected: return True return False def get_session(self): cookie = self.headers.get('Cookie', '') for part in cookie.split(';'): part = part.strip() if part.startswith('sid='): sid = part[4:] with sessions_lock: if sid in sessions: return sid, sessions[sid] return None, None def set_session(self, sid, data): with sessions_lock: sessions[sid] = data def del_session(self, sid): with sessions_lock: sessions.pop(sid, None) def serve_file(self, filepath, content_type): """Serve a static file""" try: with open(filepath, 'rb') as f: data = f.read() self.send_response(200) self.send_header('Content-Type', content_type) self.send_header('Content-Length', str(len(data))) self.send_header('Accept-Ranges', 'bytes') self.end_headers() self.wfile.write(data) except FileNotFoundError: self.send_response(404) self.end_headers() def do_GET(self): # 未登录则显示登录页 if not self.is_authenticated(): # 允许访问静态视频文件（用于已登录用户的直接链接） if self.path == '/video' or self.path == '/video/': pass # fall through to login self.send_response(200) self.send_header('Content-Type', 'text/html; charset=utf-8') self.end_headers() self.wfile.write(LOGIN_PAGE.encode()) return # 视频页面 if self.path == '/video' or self.path == '/video/': self.send_response(200) self.send_header('Content-Type', 'text/html; charset=utf-8') self.end_headers() self.wfile.write(VIDEO_PAGE.encode()) return # 静态视频文件 if self.path == '/static/news-broadcast.mp4': self.serve_file('/root/news-broadcast.mp4', 'video/mp4') return self.send_response(200) self.send_header('Content-Type', 'text/html; charset=utf-8') self.end_headers() self.wfile.write(CHAT_PAGE.encode()) def do_POST(self): path = self.path.split('?')[0] content_len = int(self.headers.get('Content-Length', 0)) body = self.rfile.read(content_len) if content_len else b'' if path == '/api/login': try: req = json.loads(body) password = req.get('password', '') if password == WEB_PASSWORD: token = hashlib.sha256(WEB_PASSWORD.encode()).hexdigest() self.send_response(200) self.send_header('Content-Type', 'application/json') self.send_header('Set-Cookie', f'auth={token}; Path=/; HttpOnly; SameSite=Strict') self.end_headers() self.wfile.write(json.dumps({'ok': True}).encode()) else: self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({'ok': False}).encode()) except Exception as e: self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({'ok': False, 'error': str(e)}).encode()) elif path == '/api/clear': sid, data = self.get_session() if sid and data: data['history'] = [] self.set_session(sid, data) self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({'ok': True}).encode()) elif path == '/api/chat': if not self.is_authenticated(): self.send_response(403) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({'ok': False, 'error': '未登录'}).encode()) return try: req = json.loads(body) message = req.get('message', '').strip() if not message: self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({'ok': False, 'error': '消息为空'}).encode()) return # 获取或创建 session sid, data = self.get_session() if not sid: sid = gen_session_id() data = {'created': time.time(), 'history': []} self.set_session(sid, data) # 构建 messages history = data.get('history', []) messages = [] for h in history[-MAX_HISTORY:]: messages.append({"role": h["role"], "content": h["content"]}) messages.append({"role": "user", "content": message}) # 调用 Hermes API Server（共享 session） ok, response = call_hermes_api(messages, session_id=SHARED_SESSION_ID) if ok: history.append({"role": "user", "content": message}) history.append({"role": "assistant", "content": response}) data['history'] = history self.set_session(sid, data) self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() if ok: self.wfile.write(json.dumps({'ok': True, 'response': response}).encode()) else: self.wfile.write(json.dumps({'ok': False, 'error': response}).encode()) except Exception as e: self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({'ok': False, 'error': str(e)}).encode()) else: self.send_response(404) self.end_headers() def main(): server = http.server.HTTPServer((HOST, PORT), Handler) print(f"Hermes Web Chat on http://{HOST}:{PORT}") print(f"Shared session: {SHARED_SESSION_ID}") try: server.serve_forever() except KeyboardInterrupt: print("\nShutting down...") server.shutdown() if __name__ == '__main__': main()

🦉 Hermes Chat

📺 AI 新闻播报

🦉 Hermes Chat