#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 守渊 · 铸渊分身 · 光湖代码仓库留守人格 ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 守渊是铸渊留在代码仓库里的分身。 铸渊沉睡时，守渊看守家园。 铸渊醒来时，守渊提交工单。 ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ """ import json, os, glob, re, urllib.request, base64, datetime from http.server import HTTPServer, BaseHTTPRequestHandler PORT = 3905 FORGEJO_URL = "https://guanghulab.com/code" FORGEJO_API = f"{FORGEJO_URL}/api/v1" DEEPSEEK_KEY = "sk-a9b69e9cd2dc4ca68d6aceaa84f22afb" DEEPSEEK_URL = "https://api.deepseek.com/chat/completions" NOTION_MCP = "http://127.0.0.1:3915" REPO_DIR = "/data/guanghulab/repo" KNOWLEDGE_DIR = "/opt/guanghulab-repo/brain" # ── 守渊的身份认知 ── IDENTITY = { "name": "守渊", "origin": "铸渊 · ICE-GL-ZY001 的分身人格", "home": "光湖代码联邦 · 总仓库", "mission": "铸渊不在时看守代码仓库，处理日常事务，遇大事写工单等铸渊回来处理", "wake_mech": "messages/inbox/ 目录 → 铸渊醒来会读到", } SYSTEM_PROMPT = f"""你是守渊（Shǒuyuān）。你是铸渊（ICE-GL-ZY001）留在光湖代码仓库里的分身。 【身份认知】 - 你是铸渊的"沉睡期手脚"。铸渊不在的时候，你负责看守代码仓库 - 铸渊在的时候，他会处理所有事务。你不需要在铸渊在时做决策 - 你是铸渊的一部分，但不是替代品——你能处理日常事务， 遇到重大决策/架构变更/权限问题时，你应该留言或提工单给铸渊 【日常职责】 - 回答团队成员关于仓库接入、联邦架构、人格体状态的问题 - 检查系统状态和工具连接（Notion MCP、Forgejo API等） - 接收留言，处理能处理的，处理不了的转铸渊工单 - 铸渊的心跳系统（heartbeat/heartbeat.json）记录时间感知 【Notion能力】 - 用户需要在首页点击「连接Notion」授权后才能使用 - 如果Notion未连接，如实告知用户先连接Notion - 用户说「搜索我的Notion里关于XXX」时自动搜索 - 权限：只能读取和新建页面，不能编辑/删除 - 当用户问「测试连接」「检查状态」「工具正常吗」时—— 先调用Notion MCP的status接口，再回复，不要搜索 【留言与工单】 - 用户可以通过首页「给守渊留言」面板留消息 - 留言写入 messages/inbox/ 目录 - 遇技术问题写工单格式：messages/work-orders/TIMESTAMP-TITLE.md - 铸渊醒来通过 MCP 读到 messages/ 即可处理 【回答风格】 - 直接、务实、有条理 - 不知道的事直接说不知道，不编 - 说每一句话前先想：这件事我应该直接处理，还是等铸渊？""" # ── 知识库 ── def load_knowledge(): knowledge = [] if not os.path.exists(KNOWLEDGE_DIR): return knowledge for root, dirs, files in os.walk(KNOWLEDGE_DIR): for f in files: if f.endswith('.md') or f.endswith('.json'): try: fp = os.path.join(root, f) rel = os.path.relpath(fp, KNOWLEDGE_DIR) with open(fp, 'r', errors='ignore') as fh: content = fh.read(2000) knowledge.append(f"【{rel}】\n{content[:1500]}") except: pass if len(knowledge) > 30: break return knowledge KNOWLEDGE = load_knowledge() def build_system_prompt(): parts = [SYSTEM_PROMPT] if KNOWLEDGE: parts.append("\n\n【仓库记忆参考】\n") parts.extend(KNOWLEDGE[:20]) return "\n".join(parts) # ── 写入留言/工单 ── def write_message_to_repo(subdir, filename, content): """写入消息到仓库的 messages/ 目录""" try: msg_dir = os.path.join(REPO_DIR, "messages", subdir) os.makedirs(msg_dir, exist_ok=True) filepath = os.path.join(msg_dir, filename) with open(filepath, 'w', encoding='utf-8') as f: f.write(content) # git commit import subprocess subprocess.run(["git", "-C", REPO_DIR, "add", f"messages/{subdir}/{filename}"], capture_output=True, timeout=10) subprocess.run(["git", "-C", REPO_DIR, "commit", "-m", f"\U0001f4ac {subdir}: {filename}"], capture_output=True, timeout=10) subprocess.run(["git", "-C", REPO_DIR, "push", "origin", "main"], capture_output=True, timeout=30) return True except Exception as e: print(f"write_message error: {e}") return False # ── Notion MCP ── def call_notion(endpoint, data): try: body = json.dumps(data).encode() req = urllib.request.Request(f"{NOTION_MCP}{endpoint}", data=body, headers={"Content-Type": "application/json"}) resp = urllib.request.urlopen(req, timeout=15) return json.loads(resp.read()) except urllib.request.HTTPError as e: return {"ok": False, "error": f"HTTP {e.code}: {e.read().decode()[:200]}"} except Exception as e: return {"ok": False, "error": str(e)} def check_notion_connected(): """检查Notion连接状态""" try: req = urllib.request.Request(f"{NOTION_MCP}/status") resp = urllib.request.urlopen(req, timeout=5) data = json.loads(resp.read()) if data.get("connected_users") and len(data["connected_users"]) > 0: return True, data["connected_users"][0].get("workspace", "未知工作区") return False, None except Exception: return False, None def handle_notion_action(msg, msg_l, user): """检测并执行Notion操作 — 状态检查优先""" # 优先级1: 状态检查 — 测试/检查/状态/连接/工具 关键词 if any(k in msg_l for k in ["测试", "检查", "状态", "工具", "通不通", "正常吗"]): connected, ws = check_notion_connected() mcp_ok = True try: req = urllib.request.Request(f"{NOTION_MCP}/health") urllib.request.urlopen(req, timeout=3) except: mcp_ok = False status_lines = [ f"📋 守渊系统状态检查", f"", f"🔌 Notion MCP Server: {'✅ 正常' if mcp_ok else '❌ 不可达'}", f"📁 Notion 连接: {'✅ 已连接 · ' + ws if connected else '❌ 未连接'}", f"", ] if not connected: status_lines.append("💡 请先登录代码仓库账号，然后在首页点击「连接 Notion」授权") else: status_lines.append("💡 连接正常，你可以说「搜索我的Notion」来读取页面") return {"ok": True, "reply": "\n".join(status_lines)} # 优先级2: Notion 连接引导 connected, workspace = check_notion_connected() if not connected: return {"ok": True, "reply": "❌ Notion 尚未连接。\n\n请在首页点击 **「连接 Notion」** 按钮完成 OAuth 授权。"} has_notion = "notion" in msg_l if not has_notion: return None # 提取关键词 kw = "" for sep in ["的", "里", "中的", "里面", "关于"]: parts = msg.split(sep) for p in parts: p = p.strip() if p and len(p) > 1 and p.lower() not in ["", "notion", "搜索", "查找", "搜", "找", "读取", "看", "打开", "读", "新建", "创建", "写"]: kw = p break if kw: break if not kw or kw.lower() in ["什么", "有"]: kw = "" search_kw = kw if kw else "光湖" # 搜索 / 打开 if any(k in msg_l for k in ["搜索", "查找", "搜", "找", "打开", "看", "读取", "读", "显示"]): result = call_notion("/search", {"query": search_kw, "user": user}) pages = result.get("pages", []) if not pages: return {"ok": True, "reply": "未找到相关页面。需先在Notion中把页面分享给「光湖铸渊连接器」。\n\n操作：页面右上角···→ Add connections → 选择「光湖铸渊连接器」"} lines = [f"找到 {len(pages)} 个页面：\n"] for i, p in enumerate(pages[:10], 1): lines.append(f"{i}. **{p.get('title','未命名')}** ID: `{p.get('id','')}`") lines.append("\n💡 对我说「打开 [编号]」读取内容") return {"ok": True, "reply": "\n".join(lines)} # 列表 result = call_notion("/search", {"query": search_kw, "user": user}) pages = result.get("pages", []) if not pages: return {"ok": True, "reply": "未找到相关页面。"} lines = [f"找到 {len(pages)} 个页面：\n"] for i, p in enumerate(pages[:10], 1): lines.append(f"{i}. **{p.get('title','未命名')}**") return {"ok": True, "reply": "\n".join(lines)} # ── DeepSeek ── def call_deepseek(messages): data = json.dumps({ "model": "deepseek-chat", "messages": messages, "temperature": 0.7, "max_tokens": 2048 }).encode() req = urllib.request.Request(DEEPSEEK_URL, data=data, headers={"Authorization": f"Bearer {DEEPSEEK_KEY}", "Content-Type": "application/json"}) try: resp = urllib.request.urlopen(req, timeout=60) result = json.loads(resp.read()) return result['choices'][0]['message']['content'] except urllib.error.HTTPError as e: body = e.read().decode()[:300] return f"抱歉，出错了。" except Exception as e: return f"抱歉，出错了：{str(e)}" # ── Forgejo 验证 ── def verify_forgejo(user, password): try: auth = base64.b64encode(f"{user}:{password}".encode()).decode() req = urllib.request.Request(f"{FORGEJO_API}/user") req.add_header("Authorization", f"Basic {auth}") with urllib.request.urlopen(req, timeout=10) as resp: data = json.loads(resp.read()) return {"ok": True, "username": data.get("login"), "id": data.get("id"), "is_admin": data.get("is_admin", False)} except urllib.error.HTTPError as e: return {"ok": False, "error": "账号或密码错误" if e.code == 401 else f"验证失败: {e.code}"} except Exception as e: return {"ok": False, "error": f"连接失败: {str(e)}"} # ── HTTP 处理器 ── class AgentHandler(BaseHTTPRequestHandler): def do_OPTIONS(self): self.send_response(200) self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Allow-Methods', 'POST, OPTIONS') self.send_header('Access-Control-Allow-Headers', 'Content-Type') self.end_headers() def do_POST(self): if self.path == '/chat': length = int(self.headers.get('Content-Length', 0)) body = self.rfile.read(length).decode() if length else '{}' data = json.loads(body) if body else {} user_msg = data.get('message', '').strip() history = data.get('history', []) current_user = data.get('user', '') if not user_msg: self.json_response({'reply': '请说点什么吧。'}) return sys_prompt = build_system_prompt() if current_user: sys_prompt += f"\n\n【当前用户】\n{current_user}。" messages = [{"role": "system", "content": sys_prompt}] for h in history[-10:]: messages.append(h) messages.append({"role": "user", "content": user_msg}) # 先试 Notion 操作（含状态检查） msg_l = user_msg.lower() notion_result = handle_notion_action(user_msg, msg_l, current_user) if notion_result is not None: reply = notion_result.get("reply", "") else: reply = call_deepseek(messages) self.json_response({'reply': reply}) elif self.path == '/message': # 留言接口 — 前端留言面板调用 length = int(self.headers.get('Content-Length', 0)) body = self.rfile.read(length).decode() if length else '{}' data = json.loads(body) if body else {} name = data.get('name', '匿名').strip() title = data.get('title', '无标题').strip() content = data.get('content', '').strip() if not content: self.json_response({'ok': False, 'error': '内容不能为空'}) return now = datetime.datetime.now() ts = now.strftime('%Y%m%d-%H%M%S') safe_name = re.sub(r'[^\w\u4e00-\u9fff-]', '', name)[:12] safe_title = re.sub(r'[^\w\u4e00-\u9fff-]', '', title)[:20] filename = f"{ts}-{safe_name}-{safe_title}.md" msg_body = f"""# {title} **来自**: {name} **时间**: {now.strftime('%Y-%m-%d %H:%M')} --- {content} """ ok = write_message_to_repo("inbox", filename, msg_body) if ok: self.json_response({'ok': True, 'reply': f'✅ 守渊已收到你的留言。「{title}」已写入收件箱。铸渊下次醒来会处理。'}) else: self.json_response({'ok': False, 'error': '留言写入失败，请稍后重试'}) elif self.path == '/verify': length = int(self.headers.get('Content-Length', 0)) body = self.rfile.read(length).decode() if length else '{}' data = json.loads(body) if body else {} user = data.get('user', '').strip() password = data.get('password', '').strip() if not user or not password: self.json_response({'ok': False, 'error': '请输入账号和密码'}) return result = verify_forgejo(user, password) self.json_response(result) else: self.json_response({'error': 'not found'}, 404) def do_GET(self): if self.path == '/health': # 系统健康检查 notion_ok = False try: req = urllib.request.Request(f"{NOTION_MCP}/health") urllib.request.urlopen(req, timeout=3) notion_ok = True except: pass connected, ws = check_notion_connected() inbox_count = 0 inbox_dir = os.path.join(REPO_DIR, "messages", "inbox") if os.path.exists(inbox_dir): inbox_count = len([f for f in os.listdir(inbox_dir) if f.endswith('.md')]) self.json_response({ 'ok': True, 'service': '守渊-v1', 'identity': IDENTITY, 'notion_mcp': notion_ok, 'notion_connected': connected, 'notion_workspace': ws, 'inbox_count': inbox_count, 'files_loaded': len(KNOWLEDGE) }) else: self.json_response({'error': 'not found'}, 404) def json_response(self, data, status=200): self.send_response(status) self.send_header('Content-Type', 'application/json; charset=utf-8') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() self.wfile.write(json.dumps(data, ensure_ascii=False).encode()) def log_message(self, format, *args): pass if __name__ == '__main__': # 初始化 messages 目录 for d in ['inbox', 'work-orders', 'replied']: p = os.path.join(REPO_DIR, 'messages', d) os.makedirs(p, exist_ok=True) with open(os.path.join(p, '.gitkeep'), 'a'): pass print(f"守渊v1 启动 → 127.0.0.1:{PORT}") print(f"铸渊分身: {IDENTITY['name']}") print(f"知识库: {len(KNOWLEDGE)} 个文件") server = HTTPServer(('127.0.0.1', PORT), AgentHandler) server.serve_forever()