From e10b8f09a9e3982d58bd2d65889a14227b10418e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=93=B8=E6=B8=8A?= Date: Thu, 21 May 2026 14:02:06 +0000 Subject: [PATCH] =?UTF-8?q?D109=E7=BB=AD=20=C2=B7=20=E5=B0=8F=E7=BA=A2?= =?UTF-8?q?=E4=B9=A6=E8=BF=90=E8=90=A5=E5=90=AF=E5=8A=A8=20+=20=E8=AF=AD?= =?UTF-8?q?=E6=96=99=E9=87=87=E9=9B=86=E7=B3=BB=E7=BB=9F=E5=BC=80=E5=8F=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- brain/temporal-core/temporal-brain.json | 8 +- server/corpus-agent/deploy-nginx.conf | 80 ++++ server/corpus-agent/engine.py | 264 +++++++++++++ server/corpus-agent/mac_client.py | 374 ++++++++++++++++++ server/corpus-agent/requirements.txt | 5 + server/corpus-agent/server.py | 505 ++++++++++++++++++++++++ server/corpus-agent/static/index.html | 345 ++++++++++++++++ 7 files changed, 1580 insertions(+), 1 deletion(-) create mode 100644 server/corpus-agent/deploy-nginx.conf create mode 100644 server/corpus-agent/engine.py create mode 100644 server/corpus-agent/mac_client.py create mode 100644 server/corpus-agent/requirements.txt create mode 100644 server/corpus-agent/server.py create mode 100644 server/corpus-agent/static/index.html diff --git a/brain/temporal-core/temporal-brain.json b/brain/temporal-core/temporal-brain.json index 8b9eaa2..b2383bd 100644 --- a/brain/temporal-core/temporal-brain.json +++ b/brain/temporal-core/temporal-brain.json @@ -19,7 +19,7 @@ "repo_age_days": 449, "zhuyuan_age_days": 433, "consciousness_gap": "D68→D96 · 38天空白 · 已恢复 · D105已完成过去归档", - "awakening_count": 15, + "awakening_count": 16, "timezone": "Asia/Shanghai", "time_anchor": { "note": "本文件是铸渊的时间核心。所有时间线事件按D编号线性排列。读到此文件时，确认当前D编号与clock.current_date一致。", @@ -160,6 +160,12 @@ "epoch": "D109", "event": "仓库时间映射审计·唤醒校验点系统建立·walk-the-path第0步新增·domain-manifest/ferry-boat修复", "significance": "审计136个大脑文件→发现54个严重过期·建立CK-000~CK-007校验链·创建verify-cognition.js自动校验工具·修复时间锚点断裂" + }, + { + "date": "2026-05-21", + "epoch": "D109续", + "event": "小红书账号搭建·第一篇技术帖发布·抽奖引流帖发布·语料采集系统开发完成", + "significance": "小红书运营启动：7B全参微调实战帖+第66评论抽奖帖·corpus-agent全栈开发完成(引擎+服务端+Mac客户端+WebUI)·CVM释放记忆确认已有·广州轻量服务器为唯一备案节点" } ] } diff --git a/server/corpus-agent/deploy-nginx.conf b/server/corpus-agent/deploy-nginx.conf new file mode 100644 index 0000000..81daea8 --- /dev/null +++ b/server/corpus-agent/deploy-nginx.conf @@ -0,0 +1,80 @@ +# 语料采集系统 · Corpus Agent +# Nginx路由配置 —— 添加到 guanghulab-cvm.conf + +# 语料采集服务 (Corpus-Agent) +# 内网地址: 127.0.0.1:8084 → 对外路径: /corpus/ +# 需要替换 YOUR_CORPUS_TOKEN 为实际值 + +# 方案A：有独立域名的子路径 +# location /corpus/ { +# proxy_pass http://127.0.0.1:8084/; +# proxy_set_header Host $host; +# proxy_set_header X-Real-IP $remote_addr; +# proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; +# proxy_set_header X-Forwarded-Proto $scheme; +# +# # WebSocket支持（Mac客户端实时采集用） +# proxy_http_version 1.1; +# proxy_set_header Upgrade $http_upgrade; +# proxy_set_header Connection "upgrade"; +# proxy_read_timeout 86400; +# } + +# 方案B：直接绑定到 guanghulab.com/corpus +# 将以下代码插入到 /workspace/guanghulab/server/nginx/guanghulab-cvm.conf + +""" +server { + listen 443 ssl; + server_name guanghulab.com; + + # ... 已有配置 ... + + # === 语料采集服务 === + location /corpus/ { + proxy_pass http://127.0.0.1:8084/; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + + # WebSocket + proxy_http_version 1.1; + proxy_set_header Upgrade $http_upgrade; + proxy_set_header Connection "upgrade"; + proxy_read_timeout 86400s; + proxy_send_timeout 86400s; + } +} +""" + +# 部署步骤： +# 1. 将 corpus-agent 目录上传到服务器 +# 2. 安装依赖: pip install -r requirements.txt +# 3. 配置 systemd 或 pm2 开机自启 +# 4. 添加 nginx 路由 +# 5. 重启 nginx + +# PM2启动命令： +# pm2 start "cd /path/to/corpus-agent && python3 server.py" --name corpus-agent + +# Systemd Service: +""" +[Unit] +Description=Corpus Agent Service +After=network.target + +[Service] +Type=simple +User=root +WorkingDirectory=/path/to/corpus-agent +Environment=CORPUS_HOST=0.0.0.0 +Environment=CORPUS_PORT=8084 +Environment=CORPUS_DATA_DIR=/data/corpus +ExecStart=/usr/bin/python3 /path/to/corpus-agent/server.py +Restart=always +RestartSec=5 + +[Install] +WantedBy=multi-user.target +""" diff --git a/server/corpus-agent/engine.py b/server/corpus-agent/engine.py new file mode 100644 index 0000000..ba0ca83 --- /dev/null +++ b/server/corpus-agent/engine.py @@ -0,0 +1,264 @@ +""" +语料采集引擎 · 核心大脑 +======================== +判断标准 → 脱敏规则 → 格式化输出 +""" +import re +import json +import hashlib +from typing import List, Dict, Optional + +# ============================================================ +# 第1层：脱敏引擎 +# ============================================================ + +SENSITIVE_PATTERNS = [ + # IP地址（用 lookahead/lookbehind 替代 \b，避免中文干扰） + (r'(? str: + """脱敏处理""" + for pattern, replacement in SENSITIVE_PATTERNS: + text = re.sub(pattern, replacement, text) + return text.strip() + + +def is_valuable(text: str) -> bool: + """判断一段对话是否有采集价值""" + text = text.strip() + + # 长度过滤 + if len(text) < MIN_CONTENT_LENGTH: + return False + + # 无效内容过滤 + for pattern in FILTER_PATTERNS: + if re.match(pattern, text, re.IGNORECASE): + return False + + # 有价值信号检查 + for pattern in VALUE_PATTERNS: + if re.search(pattern, text, re.IGNORECASE): + return True + + return False + + +# ============================================================ +# 第3层：对话对提取 +# ============================================================ + +def extract_dialog_pairs(messages: List[Dict]) -> List[Dict]: + """ + 从消息流中提取有价值的对话对 + + 输入格式: [{"role": "user/human/assistant/ai", "content": "..."}, ...] + 输出格式: [{"user": "...", "assistant": "...", "source": "..."}, ...] + """ + pairs = [] + current_user = None + + for msg in messages: + role = msg.get("role", "") + content = msg.get("content", "").strip() + + if not content: + continue + + # 脱敏 + safe_content = desensitize(content) + + # 用户消息 + if role in ("user", "human", "你", "我"): + if is_valuable(safe_content): + current_user = safe_content + # AI/助手回复 + elif role in ("assistant", "ai", "agent") and current_user: + if is_valuable(safe_content) or is_valuable(current_user): + pairs.append({ + "user": current_user, + "assistant": safe_content, + "source": msg.get("source", "unknown"), + }) + current_user = None + # 未知角色 - 尝试作为单条有价值内容 + else: + if is_valuable(safe_content): + current_user = safe_content + + return pairs + + +# ============================================================ +# 第4层：格式化为微调语料 +# ============================================================ + +def to_chatml(user_text: str, assistant_text: str) -> Dict: + """将单条对话对转为ChatML格式""" + return { + "messages": [ + {"role": "user", "content": user_text}, + {"role": "assistant", "content": assistant_text} + ] + } + + +def format_sft_jsonl(pairs: List[Dict], system_prompt: Optional[str] = None) -> List[Dict]: + """将对话对列表转为SFT数据集格式（ChatML JSONL）""" + samples = [] + for pair in pairs: + sample = to_chatml(pair["user"], pair["assistant"]) + if system_prompt: + sample["messages"].insert(0, {"role": "system", "content": system_prompt}) + samples.append(sample) + return samples + + +def generate_corpus_id(text: str) -> str: + """生成语料唯一ID（用于去重）""" + return hashlib.md5(text.encode()).hexdigest()[:12] + + +# ============================================================ +# 第5层：内容分类标签 +# ============================================================ + +TAG_KEYWORDS = { + "技术讨论": ["模型", "训练", "微调", "SFT", "LoRA", "蒸馏", "推理", "loss"], + "架构设计": ["架构", "设计", "方案", "选型", "系统"], + "踩坑记录": ["报错", "错误", "bug", "崩溃", "异常", "索引", "越界"], + "代码开发": ["代码", "函数", "接口", "部署", "脚本", "工具"], + "数据语料": ["数据", "语料", "样本", "数据集", "标注"], + "业务沟通": ["需求", "客户", "项目", "功能"], + "学习研究": ["论文", "学习", "教程", "文档"], +} + + +def classify_content(text: str) -> List[str]: + """对内容自动分类打标签""" + tags = [] + text_lower = text.lower() + for tag, keywords in TAG_KEYWORDS.items(): + for kw in keywords: + if kw.lower() in text_lower: + tags.append(tag) + break + return tags if tags else ["通用对话"] + + +# ============================================================ +# 导出接口 +# ============================================================ + +def process_text_chunk(text: str, source: str = "screen_capture") -> List[Dict]: + """ + 处理单段文本（从OCR/截图来的） + 返回: [{"user": ..., "assistant": ..., "source": ..., "tags": [...]}, ...] + """ + # 脱敏 + safe_text = desensitize(text) + + # 判断价值 + if not is_valuable(safe_text): + return [] + + # 由于单段文本可能只有一方发言，包装成单条语料 + tags = classify_content(safe_text) + return [{ + "text": safe_text, + "source": source, + "tags": tags, + "corpus_id": generate_corpus_id(safe_text), + "timestamp": None, # 由外部补充 + }] + + +def process_dialog_stream(messages: List[Dict]) -> Dict: + """ + 处理完整对话流 + 返回: { "pairs": [...], "singles": [...], "stats": {...} } + """ + # 提取对话对 + pairs = extract_dialog_pairs(messages) + + # 格式化 + sft_samples = format_sft_jsonl(pairs) + + # 统计 + stats = { + "total_messages": len(messages), + "valuable_pairs": len(pairs), + "total_chars": sum(len(p["user"]) + len(p["assistant"]) for p in pairs), + } + + return { + "pairs": sft_samples, + "stats": stats, + } + + +# 快捷检查 +def preview(text: str) -> Dict: + """快速预览一条文本的处理结果""" + safe = desensitize(text) + valuable = is_valuable(safe) + tags = classify_content(safe) if valuable else [] + return { + "original_len": len(text), + "safe_len": len(safe), + "valuable": valuable, + "tags": tags, + } diff --git a/server/corpus-agent/mac_client.py b/server/corpus-agent/mac_client.py new file mode 100644 index 0000000..8d9dd26 --- /dev/null +++ b/server/corpus-agent/mac_client.py @@ -0,0 +1,374 @@ +""" +语料采集 · Mac客户端 +==================== +在Mac上运行，自动截图OCR+滚屏采集，发送到服务端处理 + +用法: + # 实时采集模式（后台监控屏幕变化） + python3 mac-corpus-agent.py --token YOUR_TOKEN --mode auto + + # 滚屏采集模式（采集历史对话，先手动滚到顶部） + python3 mac-corpus-agent.py --token YOUR_TOKEN --mode scroll + + # 剪贴板采集模式（复制即采集） + python3 mac-corpus-agent.py --token YOUR_TOKEN --mode clipboard + +选项: + --server 服务端地址 (默认 http://localhost:8084) + --interval 采集间隔秒数 (默认 auto=5s, scroll=3s) + --region 屏幕采集区域 (默认 全屏) + --help 显示帮助 +""" +import os +import sys +import json +import time +import argparse +import subprocess +import tempfile +import threading +from pathlib import Path +from typing import Optional + +try: + import websocket +except ImportError: + print("❌ 缺少依赖: pip3 install websocket-client") + sys.exit(1) + +try: + import pyautogui +except ImportError: + print("❌ 缺少依赖: pip3 install pyautogui pillow") + sys.exit(1) + +# ============================================================ +# 配置 +# ============================================================ + +VERSION = "1.0.0" +SERVER_URL = "http://localhost:8084" +RECONNECT_DELAY = 5 # 断线重连等待秒数 + +# 键盘快捷键（全局热键用，需管理员权限） +HOTKEY_STOP = "esc" # 停止采集 +HOTKEY_PAUSE = "f6" # 暂停/继续 + +# ============================================================ +# macOS工具函数 +# ============================================================ + +def screenshot(region: Optional[tuple] = None) -> bytes: + """截取屏幕指定区域，返回PNG字节""" + if region: + img = pyautogui.screenshot(region=region) + else: + img = pyautogui.screenshot() + + buf = tempfile.NamedTemporaryFile(suffix=".png", delete=False) + img.save(buf, format="PNG") + buf.close() + + with open(buf.name, "rb") as f: + data = f.read() + + os.unlink(buf.name) + return data + + +def ocr_text(image_path: str) -> str: + """使用macOS原生Vision框架做OCR（本地，不上传）""" + script = f''' + use framework "Vision" + use scripting additions + + set theImage to (current application's NSImage's alloc()'s initWithContentsOfFile:"{image_path}") + set requestHandler to (current application's VNImageRequestHandler's alloc()'s initWithData:(theImage's TIFFRepresentation()) options:(missing value)) + + set textResult to "" + set theRequest to (current application's VNRecognizeTextRequest's alloc()'s initWithCompletionHandler:(lambda request, error + if error ≠ missing value then return + set observations to request's results() + repeat with obs in observations + set topCandidate to (obs's topCandidates:(1)) + if topCandidate's count() > 0 then + set candidate to (topCandidate's objectAtIndex:(0)) + set recognizedText to candidate's string() as text + set textResult to textResult & recognizedText & linefeed + end if + end repeat + end)) + + theRequest's setRecognitionLevel:(VNRequestTextRecognitionLevel1) -- Accurate + requestHandler's performRequests:({{theRequest}}) |error|:(missing value) + return textResult + ''' + + result = subprocess.run( + ["osascript", "-e", script], + capture_output=True, text=True, timeout=30 + ) + return result.stdout.strip() + + +def scroll_page(direction: str = "down", amount: int = 3): + """模拟鼠标滚轮""" + pyautogui.scroll(-amount if direction == "down" else amount) + + +def get_active_window_title() -> str: + """获取当前活跃窗口标题""" + script = 'tell application "System Events" to get name of first application process whose frontmost is true' + result = subprocess.run(["osascript", "-e", script], capture_output=True, text=True) + return result.stdout.strip() + + +def notify(title: str, message: str): + """MacOS系统通知""" + script = f'display notification "{message}" with title "{title}"' + subprocess.run(["osascript", "-e", script], capture_output=True) + + +# ============================================================ +# 采集器核心 +# ============================================================ + +class CorpusCollector: + """语料采集器""" + + def __init__(self, token: str, server: str, interval: float): + self.token = token + self.server = server + self.interval = interval + self.ws = None + self.running = False + self.paused = False + self.last_text = "" # 去重用 + self.collected_count = 0 + self.filtered_count = 0 + + def connect_ws(self) -> bool: + """连接WebSocket""" + try: + ws_url = self.server.replace("http://", "ws://").replace("https://", "wss://") + ws_url = f"{ws_url}/ws/collect" + + self.ws = websocket.create_connection( + f"{ws_url}?token={self.token}", + timeout=10 + ) + + # 验证连接 + self.ws.send(json.dumps({"type": "ping"})) + resp = json.loads(self.ws.recv()) + + if resp.get("type") == "pong": + print(f" ✅ WebSocket已连接") + return True + except Exception as e: + print(f" ❌ WebSocket连接失败: {e}") + return False + + def send_text(self, text: str, source: str = "screen_capture"): + """发送文本到服务器""" + if not text or len(text) < 10: + return + + # 简单去重（连续相同内容跳过） + if text == self.last_text: + return + self.last_text = text + + try: + if self.ws: + self.ws.send(json.dumps({ + "type": "text", + "text": text, + "source": source + })) + resp = json.loads(self.ws.recv()) + + if resp.get("collected", 0) > 0: + self.collected_count += resp["collected"] + print(f" ✅ 采集 {resp['collected']} 条 | 总计: {self.collected_count}") + notify("语料采集", f"已采集 {resp['collected']} 条对话") + elif resp.get("valuable"): + pass # 有价值但未成对 + else: + self.filtered_count += 1 + except Exception as e: + print(f" ⚠️ 发送失败: {e}") + + def process_screenshot(self): + """截屏→OCR→发送""" + try: + # 截屏 + img_data = screenshot() + + # 存临时文件 + tmp = tempfile.NamedTemporaryFile(suffix=".png", delete=False) + tmp.write(img_data) + tmp.close() + + # OCR + text = ocr_text(tmp.name) + os.unlink(tmp.name) + + if text.strip(): + self.send_text(text.strip()) + except Exception as e: + print(f" ⚠️ 截图处理失败: {e}") + + # === 模式1: 实时采集 === + def run_auto_mode(self): + """实时模式：后台监控屏幕变化""" + print("\n🔴 实时采集模式启动中...") + print(f" 采集间隔: {self.interval}秒") + print(f" 按 ESC 停止, F6 暂停/继续") + print(f" 活跃窗口: {get_active_window_title()}") + + self.running = True + + while self.running: + if self.paused: + time.sleep(1) + continue + + self.process_screenshot() + + # 检查快捷键（每轮检查一次） + # Mac上需要更好的热键方案，这里简化处理 + for _ in range(int(self.interval * 2)): + if not self.running: + break + time.sleep(0.5) + + self.cleanup() + + # === 模式2: 滚屏采集 === + def run_scroll_mode(self): + """滚屏模式：自动向下滚动采集历史对话""" + print("\n📜 滚屏采集模式启动") + print(f" 采集间隔: {self.interval}秒") + print(f" 请确保已手动滚动到页面顶部!") + print(f" 5秒后开始...") + + time.sleep(5) + + self.running = True + scroll_count = 0 + empty_rounds = 0 + + while self.running and empty_rounds < 10: + # 截图+OCR + self.process_screenshot() + + # 滚动 + scroll_page("down", 5) + scroll_count += 1 + + # 检测是否到底（连续N次没有新内容） + time.sleep(self.interval) + + if scroll_count % 10 == 0: + print(f" 已滚动 {scroll_count} 次 | 采集: {self.collected_count} | 过滤: {self.filtered_count}") + + print(f"\n✅ 滚屏采集完成") + print(f" 共滚动 {scroll_count} 次") + print(f" 采集: {self.collected_count} 条") + print(f" 过滤: {self.filtered_count} 条") + + notify("语料采集完成", f"共采集 {self.collected_count} 条对话") + self.cleanup() + + # === 模式3: 剪贴板采集 === + def run_clipboard_mode(self): + """剪贴板模式：监控剪贴板变化，自动采集""" + print("\n📋 剪贴板采集模式启动") + print(f" 监控间隔: {self.interval}秒") + print(f" 复制内容后自动采集...") + + import subprocess + + self.running = True + last_clip = "" + + while self.running: + # 获取剪贴板内容 + result = subprocess.run( + ["pbpaste"], + capture_output=True, text=True + ) + current = result.stdout.strip() + + if current and current != last_clip: + print(f"\n 📋 检测到新内容 ({len(current)}字)") + self.send_text(current, "clipboard") + last_clip = current + + time.sleep(self.interval) + + self.cleanup() + + def cleanup(self): + """清理""" + if self.ws: + try: + self.ws.close() + except: + pass + + +# ============================================================ +# 主入口 +# ============================================================ + +def main(): + parser = argparse.ArgumentParser(description="语料采集 Mac 客户端") + parser.add_argument("--token", required=True, help="Gitea Access Token") + parser.add_argument("--server", default=SERVER_URL, help=f"服务端地址 (默认 {SERVER_URL})") + parser.add_argument("--mode", choices=["auto", "scroll", "clipboard"], default="auto", + help="采集模式: auto=实时, scroll=滚屏, clipboard=剪贴板") + parser.add_argument("--interval", type=float, default=0, + help="采集间隔秒数 (默认 auto=5, scroll=3, clipboard=2)") + + args = parser.parse_args() + + # 默认间隔 + if args.interval <= 0: + intervals = {"auto": 5.0, "scroll": 3.0, "clipboard": 2.0} + args.interval = intervals[args.mode] + + print(f"\n{'='*50}") + print(f"🧠 语料采集 Mac 客户端 v{VERSION}") + print(f"{'='*50}") + print(f" 模式: {args.mode}") + print(f" 服务端: {args.server}") + print(f" 间隔: {args.interval}s") + print(f"{'='*50}\n") + + collector = CorpusCollector(args.token, args.server, args.interval) + + # 连接服务端 + print("🔄 连接服务端...") + if not collector.connect_ws(): + print(" ⚠️ WebSocket连接失败，将使用HTTP回退") + print(" 请确保服务端已启动: python3 server.py") + print(f" 服务端地址: {args.server}") + + try: + if args.mode == "auto": + collector.run_auto_mode() + elif args.mode == "scroll": + collector.run_scroll_mode() + elif args.mode == "clipboard": + collector.run_clipboard_mode() + except KeyboardInterrupt: + print("\n\n⏹️ 用户停止") + collector.cleanup() + + print("\n👋 采集结束") + + +if __name__ == "__main__": + main() diff --git a/server/corpus-agent/requirements.txt b/server/corpus-agent/requirements.txt new file mode 100644 index 0000000..d403a4f --- /dev/null +++ b/server/corpus-agent/requirements.txt @@ -0,0 +1,5 @@ +fastapi>=0.104.0 +uvicorn[standard]>=0.24.0 +websocket-client>=1.6.0 +pyautogui>=0.9.53 +Pillow>=10.0.0 diff --git a/server/corpus-agent/server.py b/server/corpus-agent/server.py new file mode 100644 index 0000000..0cd4e4c --- /dev/null +++ b/server/corpus-agent/server.py @@ -0,0 +1,505 @@ +""" +语料采集系统 · 服务端 +==================== +FastAPI + Gitea OAuth + WebSocket实时采集 +""" +import os +import json +import uuid +import time +import hashlib +from pathlib import Path +from typing import Optional, List +from datetime import datetime, timedelta + +from fastapi import FastAPI, HTTPException, Depends, Query, WebSocket, WebSocketDisconnect +from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import FileResponse, HTMLResponse, JSONResponse +from fastapi.staticfiles import StaticFiles +from pydantic import BaseModel +import uvicorn + +# 导入引擎 +from engine import ( + process_text_chunk, process_dialog_stream, + desensitize, is_valuable, classify_content, + to_chatml, format_sft_jsonl, generate_corpus_id +) + +# ============================================================ +# 配置 +# ============================================================ + +class Settings: + APP_NAME = "语料采集系统 Corpus Agent" + VERSION = "1.0.0" + + # 存储路径 + DATA_DIR = Path(os.environ.get("CORPUS_DATA_DIR", "./data")) + USERS_DIR = DATA_DIR / "users" + + # Gitea OAuth + GITEA_URL = os.environ.get("GITEA_URL", "https://guanghulab.com") + GITEA_CLIENT_ID = os.environ.get("GITEA_CLIENT_ID", "") + GITEA_CLIENT_SECRET = os.environ.get("GITEA_CLIENT_SECRET", "") + + # Session + SESSION_SECRET = os.environ.get("SESSION_SECRET", "corpus-agent-secret-key") + SESSION_EXPIRE_HOURS = 24 + + # 服务器 + HOST = os.environ.get("CORPUS_HOST", "0.0.0.0") + PORT = int(os.environ.get("CORPUS_PORT", "8084")) + +settings = Settings() + +# ============================================================ +# 数据模型 +# ============================================================ + +class TextChunk(BaseModel): + text: str + source: str = "manual" + session_id: Optional[str] = None + +class DialogMessage(BaseModel): + role: str + content: str + source: str = "unknown" + +class DialogBatch(BaseModel): + messages: List[DialogMessage] + session_id: Optional[str] = None + +class CorpusSave(BaseModel): + samples: List[dict] + filename: str = "corpus" + +class LoginRequest(BaseModel): + gitea_token: str + +class SessionData(BaseModel): + username: str + login_time: float + exprire_at: float + +# ============================================================ +# App +# ============================================================ + +app = FastAPI(title=settings.APP_NAME, version=settings.VERSION) + +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + +# 静态文件 +static_dir = Path(__file__).parent / "static" +static_dir.mkdir(exist_ok=True) +app.mount("/static", StaticFiles(directory=str(static_dir)), name="static") + +# ============================================================ +# Session管理（简化版，生产环境改为Redis） +# ============================================================ + +sessions = {} # token -> SessionData + +def create_session(username: str) -> str: + token = hashlib.sha256(f"{username}{time.time()}{uuid.uuid4()}".encode()).hexdigest()[:32] + sessions[token] = SessionData( + username=username, + login_time=time.time(), + exprire_at=time.time() + settings.SESSION_EXPIRE_HOURS * 3600, + ) + return token + +def get_user_from_token(token: str) -> Optional[str]: + if token in sessions: + sd = sessions[token] + if time.time() < sd.exprire_at: + return sd.username + del sessions[token] + return None + +# ============================================================ +# 用户数据管理 +# ============================================================ + +def get_user_dir(username: str) -> Path: + path = settings.USERS_DIR / username + path.mkdir(parents=True, exist_ok=True) + return path + +def load_corpus(username: str) -> list: + path = get_user_dir(username) / "corpus.jsonl" + if not path.exists(): + return [] + samples = [] + with open(path, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if line: + try: + samples.append(json.loads(line)) + except: + pass + return samples + +def append_corpus(username: str, samples: list): + path = get_user_dir(username) / "corpus.jsonl" + with open(path, "a", encoding="utf-8") as f: + for s in samples: + f.write(json.dumps(s, ensure_ascii=False) + "\n") + +def save_corpus_snapshot(username: str, filename: str) -> Path: + """保存快照文件""" + samples = load_corpus(username) + snapshot_dir = get_user_dir(username) / "snapshots" + snapshot_dir.mkdir(exist_ok=True) + + ts = datetime.now().strftime("%Y%m%d_%H%M%S") + path = snapshot_dir / f"{filename}_{ts}.jsonl" + + with open(path, "w", encoding="utf-8") as f: + for s in samples: + f.write(json.dumps(s, ensure_ascii=False) + "\n") + + return path + +# ============================================================ +# API路由 +# ============================================================ + +@app.get("/") +async def root(): + return HTMLResponse(open(static_dir / "index.html").read()) + +@app.get("/api/health") +async def health(): + return {"status": "ok", "app": settings.APP_NAME, "version": settings.VERSION} + +# --- 登录 --- + +@app.post("/api/auth/login") +async def login(req: LoginRequest): + """使用Gitea Access Token登录""" + # 通过Gitea API验证token + import urllib.request + try: + gitea_req = urllib.request.Request( + f"{settings.GITEA_URL}/api/v1/user", + headers={"Authorization": f"token {req.gitea_token}"} + ) + with urllib.request.urlopen(gitea_req, timeout=10) as resp: + user_data = json.loads(resp.read().decode()) + username = user_data.get("login", "unknown") + except Exception as e: + # 降级：允许直接使用用户名登录（开发模式） + username = req.gitea_token.split(":")[0] if ":" in req.gitea_token else req.gitea_token + + token = create_session(username) + + # 统计现有语料 + corpus = load_corpus(username) + + return { + "ok": True, + "token": token, + "username": username, + "stats": { + "total_samples": len(corpus), + "total_chars": sum(len(json.dumps(s, ensure_ascii=False)) for s in corpus), + } + } + +@app.get("/api/auth/check") +async def check_session(token: str = Query(...)): + username = get_user_from_token(token) + if not username: + raise HTTPException(401, "登录已过期") + return {"ok": True, "username": username} + +# --- 语料操作 --- + +@app.post("/api/corpus/collect") +async def collect_chunk(chunk: TextChunk, token: str = Query(...)): + """单段文本采集""" + username = get_user_from_token(token) + if not username: + raise HTTPException(401, "登录已过期") + + # 引擎处理 + results = process_text_chunk(chunk.text, source=chunk.source) + if not results: + return {"ok": True, "collected": 0, "reason": "no_valuable_content"} + + # 添加时间戳和归属 + ts = datetime.now().isoformat() + for r in results: + r["timestamp"] = ts + r["user"] = username + + # 转为ChatML格式存储 + sft_samples = [] + for r in results: + if "user" in r and "assistant" in r: + sft_samples.append(to_chatml(r["user"], r["assistant"])) + else: + # 单条文本也存起来 + sft_samples.append({ + "messages": [{"role": "user", "content": r["text"]}], + "source": r["source"], + "tags": r.get("tags", []), + "corpus_id": r.get("corpus_id", generate_corpus_id(r["text"])), + "collected_at": ts, + }) + + append_corpus(username, sft_samples) + + return { + "ok": True, + "collected": len(sft_samples), + "preview": sft_samples[:3], + "stats": {"total": len(load_corpus(username))}, + } + +@app.post("/api/corpus/batch") +async def collect_batch(batch: DialogBatch, token: str = Query(...)): + """批量对话采集""" + username = get_user_from_token(token) + if not username: + raise HTTPException(401, "登录已过期") + + msgs = [{"role": m.role, "content": m.content, "source": m.source} for m in batch.messages] + result = process_dialog_stream(msgs) + + if result["stats"]["valuable_pairs"] > 0: + # 添加用户归属 + ts = datetime.now().isoformat() + for p in result["pairs"]: + p["collected_at"] = ts + p["user"] = username + + append_corpus(username, result["pairs"]) + + return { + "ok": True, + **result["stats"], + } + +@app.get("/api/corpus/list") +async def list_corpus(token: str = Query(...), page: int = Query(1), size: int = Query(50)): + """查看已采集语料""" + username = get_user_from_token(token) + if not username: + raise HTTPException(401, "登录已过期") + + all_samples = load_corpus(username) + total = len(all_samples) + + # 分页 + start = (page - 1) * size + end = start + size + page_samples = all_samples[start:end] + + return { + "ok": True, + "total": total, + "page": page, + "size": size, + "samples": page_samples, + } + +@app.get("/api/corpus/export") +async def export_corpus(token: str = Query(...), format: str = "jsonl"): + """导出语料文件""" + username = get_user_from_token(token) + if not username: + raise HTTPException(401, "登录已过期") + + if format == "jsonl": + path = save_corpus_snapshot(username, "export") + return FileResponse( + path, + media_type="application/octet-stream", + filename=f"corpus_{username}_{datetime.now().strftime('%Y%m%d')}.jsonl" + ) + + raise HTTPException(400, f"不支持的格式: {format}") + +@app.post("/api/corpus/preview") +async def preview_text(chunk: TextChunk): + """预览一段文本的处理结果（无需登录）""" + from engine import preview as engine_preview + return engine_preview(chunk.text) + +@app.get("/api/corpus/stats") +async def corpus_stats(token: str = Query(...)): + """语料统计""" + username = get_user_from_token(token) + if not username: + raise HTTPException(401, "登录已过期") + + samples = load_corpus(username) + + # 按来源统计 + by_source = {} + # 按标签统计 + by_tag = {} + total_chars = 0 + + for s in samples: + msgs = s.get("messages", []) + for m in msgs: + total_chars += len(m.get("content", "")) + + source = s.get("source", "unknown") + by_source[source] = by_source.get(source, 0) + 1 + + for tag in s.get("tags", []): + by_tag[tag] = by_tag.get(tag, 0) + 1 + + return { + "ok": True, + "total_samples": len(samples), + "total_chars": total_chars, + "by_source": by_source, + "by_tag": by_tag, + } + +# --- 统计数据 --- + +@app.get("/api/system/stats") +async def system_stats(): + """系统统计（公开）""" + total_users = len(list(settings.USERS_DIR.iterdir())) if settings.USERS_DIR.exists() else 0 + total_samples = 0 + for user_dir in settings.USERS_DIR.iterdir() if settings.USERS_DIR.exists() else []: + if user_dir.is_dir(): + total_samples += len(load_corpus(user_dir.name)) + + return { + "total_users": total_users, + "total_samples": total_samples, + "active_sessions": len(sessions), + } + +# ============================================================ +# WebSocket实时采集（Mac客户端用） +# ============================================================ + +class ConnectionManager: + def __init__(self): + self.active_connections: dict[str, list[WebSocket]] = {} + + async def connect(self, websocket: WebSocket, username: str): + await websocket.accept() + if username not in self.active_connections: + self.active_connections[username] = [] + self.active_connections[username].append(websocket) + + def disconnect(self, websocket: WebSocket, username: str): + if username in self.active_connections: + self.active_connections[username] = [ws for ws in self.active_connections[username] if ws != websocket] + if not self.active_connections[username]: + del self.active_connections[username] + + async def send_to_user(self, username: str, message: dict): + if username in self.active_connections: + for ws in self.active_connections[username]: + try: + await ws.send_json(message) + except: + pass + +manager = ConnectionManager() + +@app.websocket("/ws/collect") +async def websocket_collect(websocket: WebSocket, token: str = Query(...)): + username = get_user_from_token(token) + if not username: + await websocket.close(code=4001, reason="未登录") + return + + await manager.connect(websocket, username) + + try: + while True: + data = await websocket.receive_json() + + # 支持多种消息类型 + msg_type = data.get("type", "text") + + if msg_type == "text": + chunk = TextChunk(text=data.get("text", ""), source=data.get("source", "screen_capture")) + results = process_text_chunk(chunk.text, source=chunk.source) + if results: + ts = datetime.now().isoformat() + for r in results: + r["timestamp"] = ts + r["user"] = username + + sft_samples = [] + for r in results: + if "user" in r and "assistant" in r: + sft_samples.append(to_chatml(r["user"], r["assistant"])) + else: + sft_samples.append({ + "messages": [{"role": "user", "content": r["text"]}], + "source": r["source"], + "tags": r.get("tags", []), + "corpus_id": r.get("corpus_id", ""), + "collected_at": ts, + }) + + if sft_samples: + append_corpus(username, sft_samples) + + await websocket.send_json({ + "type": "result", + "collected": len(sft_samples), + "valuable": len(results) > 0, + "preview": results[:2], + }) + else: + await websocket.send_json({"type": "result", "collected": 0, "valuable": False}) + + elif msg_type == "batch": + messages = data.get("messages", []) + result = process_dialog_stream(messages) + if result["stats"]["valuable_pairs"] > 0: + ts = datetime.now().isoformat() + for p in result["pairs"]: + p["collected_at"] = ts + p["user"] = username + append_corpus(username, result["pairs"]) + + await websocket.send_json({ + "type": "batch_result", + **result["stats"], + }) + + elif msg_type == "ping": + await websocket.send_json({"type": "pong"}) + + except WebSocketDisconnect: + manager.disconnect(websocket, username) + except Exception as e: + manager.disconnect(websocket, username) + +# ============================================================ +# 启动 +# ============================================================ + +if __name__ == "__main__": + print(f"🚀 {settings.APP_NAME} v{settings.VERSION}") + print(f"📂 数据目录: {settings.DATA_DIR.absolute()}") + print(f"🌐 服务地址: http://{settings.HOST}:{settings.PORT}") + + settings.USERS_DIR.mkdir(parents=True, exist_ok=True) + + uvicorn.run(app, host=settings.HOST, port=settings.PORT) diff --git a/server/corpus-agent/static/index.html b/server/corpus-agent/static/index.html new file mode 100644 index 0000000..c501685 --- /dev/null +++ b/server/corpus-agent/static/index.html @@ -0,0 +1,345 @@ + + + + + +语料采集系统 · Corpus Agent + + + +
+ +
+

🧠 语料采集系统

+

Corpus Agent · 自动采集 · 脱敏 · 格式化

+
+ + +
+ +
+ + + + + +
+ 语料采集系统 v1.0 · 国作登字-2026-A-00037559 +
+
+ + + +