From 177e3455393055c13e076217979cca141082293c Mon Sep 17 00:00:00 2001 From: qyyfw Date: Wed, 8 Jul 2026 12:10:32 +0800 Subject: [PATCH] P1+P2 complete: pgvector, Casdoor SSO, CLI, KB router, K8s, monitoring, UI fixes (Promotion rewrite, Storage/Automation btn fix, membership configs mojibake fix) --- backend/python/app/casdoor_auth.py | 214 +++ backend/python/app/kb_router.py | 267 ++++ backend/python/app/pgvector_search.py | 360 +++++ backend/python/main.py | 931 ++++++------ frontend/src/pages/admin/Automation.tsx | 1198 +++++++-------- frontend/src/pages/admin/Promotion.tsx | 1147 +++++---------- frontend/src/pages/admin/Storage.tsx | 1782 +++++++++++------------ 7 files changed, 3197 insertions(+), 2702 deletions(-) create mode 100644 backend/python/app/casdoor_auth.py create mode 100644 backend/python/app/kb_router.py create mode 100644 backend/python/app/pgvector_search.py diff --git a/backend/python/app/casdoor_auth.py b/backend/python/app/casdoor_auth.py new file mode 100644 index 0000000..258e96f --- /dev/null +++ b/backend/python/app/casdoor_auth.py @@ -0,0 +1,214 @@ +""" +Casdoor SSO 集成模块 +================== +实现 Casdoor OIDC 单点登录集成， +支持 JWT + Casdoor 双认证模式。 + +使用方式: + from app.casdoor_auth import casdoor_router, get_casdoor_login_url, handle_casdoor_callback +""" + +import os +import logging +import httpx +from typing import Optional +from datetime import datetime +from urllib.parse import urlencode + +from fastapi import APIRouter, Depends, HTTPException, Request +from fastapi.responses import RedirectResponse +from sqlalchemy.orm import Session +from pydantic import BaseModel + +from .database import get_db +from .models import User, gen_uuid +from .auth import create_access_token, hash_password + +logger = logging.getLogger("aiflow.casdoor") + +router = APIRouter() + +# Casdoor 配置 +CASDOOR_ENDPOINT = os.environ.get("CASDOOR_ENDPOINT", "http://casdoor:8000") +CASDOOR_CLIENT_ID = os.environ.get("CASDOOR_CLIENT_ID", "aiflow") +CASDOOR_CLIENT_SECRET = os.environ.get("CASDOOR_CLIENT_SECRET", "") +CASDOOR_ORG_NAME = os.environ.get("CASDOOR_ORG_NAME", "aiflow") +CASDOOR_APP_NAME = os.environ.get("CASDOOR_APP_NAME", "aiflow-app") +CASDOOR_REDIRECT_URI = os.environ.get( + "CASDOOR_REDIRECT_URI", + "http://210.16.161.218:12888/app/api/auth/casdoor/callback" +) +APP_FRONTEND_URL = os.environ.get("APP_ACCESS_URL", "http://210.16.161.218:12888") + + +def get_casdoor_login_url(state: str = "") -> str: + """获取 Casdoor 登录跳转URL""" + params = { + "client_id": CASDOOR_CLIENT_ID, + "response_type": "code", + "redirect_uri": CASDOOR_REDIRECT_URI, + "scope": "openid profile email", + "state": state or "aiflow", + } + return f"{CASDOOR_ENDPOINT}/login/oauth/authorize?{urlencode(params)}" + + +async def exchange_code_for_token(code: str) -> Optional[dict]: + """用授权码换取 access_token 和用户信息""" + try: + async with httpx.AsyncClient(timeout=30) as client: + # 获取 token + token_resp = await client.post( + f"{CASDOOR_ENDPOINT}/api/login/oauth/access_token", + json={ + "grant_type": "authorization_code", + "client_id": CASDOOR_CLIENT_ID, + "client_secret": CASDOOR_CLIENT_SECRET, + "code": code, + "redirect_uri": CASDOOR_REDIRECT_URI, + }, + ) + + if token_resp.status_code != 200: + logger.error(f"Casdoor token exchange failed: {token_resp.status_code} {token_resp.text}") + return None + + token_data = token_resp.json() + access_token = token_data.get("access_token") + if not access_token: + logger.error("No access_token in Casdoor response") + return None + + # 获取用户信息 + user_resp = await client.get( + f"{CASDOOR_ENDPOINT}/api/account", + headers={"Authorization": f"Bearer {access_token}"}, + ) + + if user_resp.status_code != 200: + # 尝试备用端点 + user_resp = await client.get( + f"{CASDOOR_ENDPOINT}/api/userinfo", + headers={"Authorization": f"Bearer {access_token}"}, + ) + + if user_resp.status_code != 200: + logger.error(f"Casdoor userinfo failed: {user_resp.status_code}") + return {"access_token": access_token, "user": None} + + user_info = user_resp.json() + return {"access_token": access_token, "user": user_info} + + except Exception as e: + logger.error(f"Casdoor token exchange error: {e}") + return None + + +async def get_or_create_user(db: Session, casdoor_user: dict) -> User: + """根据 Casdoor 用户信息获取或创建本地用户""" + casdoor_id = casdoor_user.get("id", casdoor_user.get("name", "")) + casdoor_name = casdoor_user.get("name", casdoor_user.get("preferred_username", "")) + casdoor_email = casdoor_user.get("email", "") + casdoor_display = casdoor_user.get("displayName", casdoor_user.get("name", "")) + + # 通过 Casdoor ID 查找已有用户 + existing = db.query(User).filter(User.casdoor_id == casdoor_id).first() if hasattr(User, "casdoor_id") else None + + # 通过用户名查找 + if not existing and casdoor_name: + existing = db.query(User).filter(User.username == f"sso_{casdoor_name}").first() + if existing and hasattr(existing, "casdoor_id"): + existing.casdoor_id = casdoor_id + + if existing: + # 更新最后登录时间 + existing.last_login = datetime.now() + db.commit() + return existing + + # 创建新用户 + username = f"sso_{casdoor_name}" if casdoor_name else f"sso_{casdoor_id[:8]}" + # 避免用户名冲突 + base_username = username + counter = 1 + while db.query(User).filter(User.username == username).first(): + username = f"{base_username}_{counter}" + counter += 1 + + new_user = User( + id=gen_uuid(), + username=username, + email=casdoor_email or None, + password_hash=hash_password(casdoor_id + datetime.now().isoformat()), # 随机密码 + role="user", + status="active", + display_name=casdoor_display or username, + ) + + # 如果 User 模型有 casdoor_id 字段则设置 + if hasattr(new_user, "casdoor_id"): + new_user.casdoor_id = casdoor_id + + db.add(new_user) + db.commit() + db.refresh(new_user) + + logger.info(f"Created SSO user: {username} (casdoor_id={casdoor_id})") + return new_user + + +# ==================== API 路由 ==================== + +@router.get("/casdoor/login") +async def casdoor_login(state: str = ""): + """跳转到 Casdoor 登录页面""" + login_url = get_casdoor_login_url(state) + return RedirectResponse(url=login_url) + + +@router.get("/casdoor/callback") +async def casdoor_callback( + code: str, + state: str = "", + db: Session = Depends(get_db), +): + """Casdoor OAuth 回调""" + result = await exchange_code_for_token(code) + if not result or not result.get("user"): + # 重定向到前端登录页，带错误参数 + return RedirectResponse(url=f"{APP_FRONTEND_URL}/#/login?error=casdoor_failed") + + casdoor_user = result["user"] + user = await get_or_create_user(db, casdoor_user) + + # 生成平台 JWT token + token_data = {"sub": user.id, "username": user.username, "role": user.role} + access_token = create_access_token(token_data) + + # 重定向到前端，带 token + redirect_url = f"{APP_FRONTEND_URL}/#/login?token={access_token}&sso=casdoor" + return RedirectResponse(url=redirect_url) + + +@router.get("/casdoor/status") +async def casdoor_status(): + """检查 Casdoor SSO 是否可用""" + try: + async with httpx.AsyncClient(timeout=5) as client: + resp = await client.get(f"{CASDOOR_ENDPOINT}/api/health") + if resp.status_code == 200: + return {"enabled": True, "endpoint": CASDOOR_ENDPOINT} + except Exception: + pass + return {"enabled": False, "endpoint": CASDOOR_ENDPOINT} + + +# 配置说明： +# 1. 在 Casdoor 中创建应用，获取 Client ID 和 Client Secret +# 2. 设置环境变量: +# - CASDOOR_ENDPOINT=http://casdoor:8000 +# - CASDOOR_CLIENT_ID=your_client_id +# - CASDOOR_CLIENT_SECRET=your_client_secret +# - CASDOOR_REDIRECT_URI=http://your-domain/app/api/auth/casdoor/callback +# 3. 在 User 模型中添加 casdoor_id 字段 (可选): +# casdoor_id = Column(String(128), unique=True, nullable=True) diff --git a/backend/python/app/kb_router.py b/backend/python/app/kb_router.py new file mode 100644 index 0000000..ec82a3f --- /dev/null +++ b/backend/python/app/kb_router.py @@ -0,0 +1,267 @@ +""" +知识库路由层 +============ +将检索请求智能路由到本地BM25、向量检索、或外部KB服务(FastGPT/Dify/RAGFlow)。 +根据知识库配置自动选择最优检索策略。 + +使用方式: + from app.kb_router import kb_router_search + results = await kb_router_search(query, kb_id, top_k=5) +""" + +import os +import json +import logging +import httpx +from typing import List, Dict, Optional + +logger = logging.getLogger("aiflow.kb_router") + + +async def _get_external_services(db_session, enabled_only: bool = True) -> List[Dict]: + """获取已配置的外部KB服务""" + from sqlalchemy import text + try: + query = "SELECT * FROM external_kb_services" + if enabled_only: + query += " WHERE is_enabled = 1" + query += " ORDER BY created_at DESC" + result = db_session.execute(text(query)) + columns = result.keys() + return [dict(zip(columns, row)) for row in result.fetchall()] + except Exception as e: + logger.error(f"Failed to get external services: {e}") + return [] + + +async def _search_fastgpt(api_url: str, api_key: str, query: str, top_k: int = 5) -> List[Dict]: + """搜索 FastGPT 知识库""" + try: + async with httpx.AsyncClient(timeout=30) as client: + resp = await client.post( + f"{api_url.rstrip('/')}/api/v1/dataset/search", + headers={ + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + }, + json={ + "query": query, + "similarity": 0.5, + "limit": top_k, + }, + ) + if resp.status_code == 200: + data = resp.json() + results = [] + for item in data.get("data", {}).get("list", []): + results.append({ + "content": item.get("q", item.get("content", "")), + "score": float(item.get("score", 0.5)), + "source": f"FastGPT: {item.get('collectionId', '')}", + "metadata": item, + }) + return results + else: + logger.warning(f"FastGPT search returned {resp.status_code}") + return [] + except Exception as e: + logger.error(f"FastGPT search error: {e}") + return [] + + +async def _search_dify(api_url: str, api_key: str, query: str, top_k: int = 5) -> List[Dict]: + """搜索 Dify 知识库""" + try: + async with httpx.AsyncClient(timeout=30) as client: + resp = await client.post( + f"{api_url.rstrip('/')}/v1/datasets/retrieval", + headers={ + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + }, + json={ + "query": query, + "retrieval_mode": "semantic", + "top_k": top_k, + }, + ) + if resp.status_code == 200: + data = resp.json() + results = [] + for item in data.get("records", []): + segment = item.get("segment", {}) + results.append({ + "content": segment.get("content", ""), + "score": float(item.get("score", 0.5)), + "source": f"Dify: {segment.get('id', '')}", + "metadata": item, + }) + return results + else: + logger.warning(f"Dify search returned {resp.status_code}") + return [] + except Exception as e: + logger.error(f"Dify search error: {e}") + return [] + + +async def _search_ragflow(api_url: str, api_key: str, query: str, top_k: int = 5) -> List[Dict]: + """搜索 RAGFlow 知识库""" + try: + async with httpx.AsyncClient(timeout=30) as client: + resp = await client.post( + f"{api_url.rstrip('/')}/api/v1/retrieval", + headers={ + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + }, + json={ + "question": query, + "top_k": top_k, + }, + ) + if resp.status_code == 200: + data = resp.json() + results = [] + for item in data.get("data", {}).get("chunks", []): + results.append({ + "content": item.get("content", ""), + "score": float(item.get("similarity", 0.5)), + "source": f"RAGFlow: {item.get('document_id', '')}", + "metadata": item, + }) + return results + else: + logger.warning(f"RAGFlow search returned {resp.status_code}") + return [] + except Exception as e: + logger.error(f"RAGFlow search error: {e}") + return [] + + +async def _search_external(service: Dict, query: str, top_k: int = 5) -> List[Dict]: + """路由到对应的外部KB服务""" + service_type = service.get("service_type", "") + api_url = service.get("api_url", "") + api_key = service.get("api_key", "") + + if service_type == "fastgpt": + return await _search_fastgpt(api_url, api_key, query, top_k) + elif service_type == "dify": + return await _search_dify(api_url, api_key, query, top_k) + elif service_type == "ragflow": + return await _search_ragflow(api_url, api_key, query, top_k) + else: + logger.warning(f"Unknown external KB type: {service_type}") + return [] + + +async def kb_router_search( + query: str, + kb_id: Optional[str] = None, + top_k: int = 5, + bm25_results: Optional[List[Dict]] = None, + db_session=None, + use_vector: bool = True, + use_external: bool = True, +) -> List[Dict]: + """ + 知识库路由层 — 混合检索 + + 检索策略: + 1. 本地 BM25 检索 (MySQL, 已有) + 2. 向量语义检索 (PostgreSQL+pgvector, 新增) + 3. 外部KB检索 (FastGPT/Dify/RAGFlow, 按配置) + + 融合方式: Reciprocal Rank Fusion (RRF) + + Args: + query: 查询文本 + kb_id: 知识库ID + top_k: 返回结果数 + bm25_results: BM25检索结果 + db_session: 数据库会话(用于获取外部KB配置) + use_vector: 是否使用向量检索 + use_external: 是否使用外部KB检索 + + Returns: + 融合后的检索结果列表 + """ + all_results = [] + + # 1. BM25 结果 + if bm25_results: + for rank, item in enumerate(bm25_results): + all_results.append({ + "content": item.get("content", item.get("text", "")), + "score": item.get("score", 1.0 / (60 + rank + 1)), + "source": "local_bm25", + "rank": rank + 1, + "metadata": item.get("metadata", {}), + }) + + # 2. 向量检索 + if use_vector: + try: + from app.pgvector_search import vector_search + vec_results = await vector_search(query, kb_id, top_k) + for rank, item in enumerate(vec_results): + all_results.append({ + "content": item.get("content", ""), + "score": item.get("score", 0.5), + "source": "vector_pg", + "rank": rank + 1, + "metadata": item.get("metadata", {}), + }) + except ImportError: + logger.debug("pgvector_search not available, skipping vector search") + except Exception as e: + logger.warning(f"Vector search failed: {e}") + + # 3. 外部KB检索 + if use_external and db_session: + services = await _get_external_services(db_session) + for service in services: + if service.get("status") == "connected": + ext_results = await _search_external(service, query, top_k) + for rank, item in enumerate(ext_results): + all_results.append({ + "content": item.get("content", ""), + "score": item.get("score", 0.5), + "source": item.get("source", f"external_{service['service_type']}"), + "rank": rank + 1, + "metadata": item.get("metadata", {}), + }) + + if not all_results: + return [] + + # RRF 融合 + rrf_k = 60 + score_map = {} + for item in all_results: + content_key = item["content"][:200] # 用内容前200字符作为去重键 + rrf_score = 1.0 / (rrf_k + item.get("rank", 1)) + + if content_key in score_map: + score_map[content_key]["score"] += rrf_score + score_map[content_key]["sources"].append(item["source"]) + else: + score_map[content_key] = { + "content": item["content"], + "score": rrf_score, + "sources": [item["source"]], + "metadata": item.get("metadata", {}), + } + + # 按融合分数排序 + merged = sorted(score_map.values(), key=lambda x: x["score"], reverse=True) + + # 标准化分数到 0-1 + if merged: + max_score = merged[0]["score"] + for item in merged: + item["score"] = item["score"] / max_score if max_score > 0 else 0 + item["sources"] = list(set(item["sources"])) + + return merged[:top_k] diff --git a/backend/python/app/pgvector_search.py b/backend/python/app/pgvector_search.py new file mode 100644 index 0000000..a38654a --- /dev/null +++ b/backend/python/app/pgvector_search.py @@ -0,0 +1,360 @@ +""" +PGVector 向量检索模块 +==================== +使用 PostgreSQL + pgvector 实现语义向量检索， +与现有 MySQL BM25 检索配合，实现混合检索。 + +使用方式: + from app.pgvector_search import vector_search + results = await vector_search(query, kb_id, top_k=5) +""" + +import os +import logging +import asyncio +from typing import List, Dict, Optional +from datetime import datetime + +logger = logging.getLogger("aiflow.pgvector") + +# PostgreSQL 连接配置 +PG_HOST = os.environ.get("POSTGRES_HOST", "pgvector") +PG_PORT = int(os.environ.get("POSTGRES_PORT", "5432")) +PG_USER = os.environ.get("POSTGRES_USER", "aiflow") +PG_PASSWORD = os.environ.get("POSTGRES_PASSWORD", "pwd2026") +PG_DB = os.environ.get("POSTGRES_DB", "aiflow_vectors") + +# 嵌入模型配置 +EMBEDDING_API_URL = os.environ.get("SILICONFLOW_BASE_URL", "https://api.siliconflow.cn") +EMBEDDING_API_KEY = os.environ.get("SILICONFLOW_API_KEY", "") +EMBEDDING_MODEL = os.environ.get("EMBEDDING_MODEL", "BAAI/bge-large-zh-v1.5") + +# 连接池 +_pg_pool = None +_pool_lock = asyncio.Lock() + + +async def get_pg_pool(): + """获取 PostgreSQL 异步连接池""" + global _pg_pool + if _pg_pool is not None: + return _pg_pool + + async with _pool_lock: + if _pg_pool is not None: + return _pg_pool + + try: + import asyncpg + except ImportError: + logger.warning("asyncpg not installed, vector search disabled") + return None + + try: + _pg_pool = await asyncpg.create_pool( + host=PG_HOST, + port=PG_PORT, + user=PG_USER, + password=PG_PASSWORD, + database=PG_DB, + min_size=2, + max_size=10, + command_timeout=30, + ) + logger.info(f"PostgreSQL connection pool created: {PG_HOST}:{PG_PORT}/{PG_DB}") + + # 初始化表结构 + async with _pg_pool.acquire() as conn: + await conn.execute(""" + CREATE TABLE IF NOT EXISTS document_embeddings ( + id SERIAL PRIMARY KEY, + doc_id VARCHAR(64) NOT NULL, + kb_id VARCHAR(64) NOT NULL, + chunk_id VARCHAR(64), + content TEXT NOT NULL, + embedding vector(1024), + metadata JSONB DEFAULT '{}', + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ); + """) + await conn.execute(""" + CREATE INDEX IF NOT EXISTS idx_embeddings_kb + ON document_embeddings(kb_id); + """) + await conn.execute(""" + CREATE INDEX IF NOT EXISTS idx_embeddings_doc + ON document_embeddings(doc_id); + """) + await conn.execute(""" + CREATE INDEX IF NOT EXISTS idx_embeddings_vector + ON document_embeddings USING ivfflat (embedding vector_cosine_ops) + WITH (lists = 100); + """) + logger.info("PGVector tables initialized") + + return _pg_pool + except Exception as e: + logger.error(f"Failed to create PostgreSQL pool: {e}") + _pg_pool = None + return None + + +async def get_embedding(text: str) -> Optional[List[float]]: + """调用嵌入API获取文本向量""" + if not EMBEDDING_API_KEY: + logger.warning("No embedding API key configured") + return None + + try: + import httpx + async with httpx.AsyncClient(timeout=30) as client: + resp = await client.post( + f"{EMBEDDING_API_URL}/v1/embeddings", + headers={ + "Authorization": f"Bearer {EMBEDDING_API_KEY}", + "Content-Type": "application/json", + }, + json={ + "model": EMBEDDING_MODEL, + "input": text[:8000], # 截断防止超长 + }, + ) + if resp.status_code == 200: + data = resp.json() + embedding = data["data"][0]["embedding"] + return embedding + else: + logger.error(f"Embedding API error: {resp.status_code} {resp.text[:200]}") + return None + except Exception as e: + logger.error(f"Embedding generation failed: {e}") + return None + + +async def vector_search( + query: str, + kb_id: Optional[str] = None, + top_k: int = 5, + threshold: float = 0.5, +) -> List[Dict]: + """ + 向量语义检索 + + Args: + query: 查询文本 + kb_id: 知识库ID (可选，不指定则搜索全部) + top_k: 返回结果数 + threshold: 相似度阈值 (0-1) + + Returns: + [{"doc_id": ..., "content": ..., "score": ..., "metadata": ...}, ...] + """ + pool = await get_pg_pool() + if pool is None: + return [] + + # 获取查询向量 + query_embedding = await get_embedding(query) + if query_embedding is None: + return [] + + try: + async with pool.acquire() as conn: + # 将向量转换为 PostgreSQL 格式 + embedding_str = f"[{','.join(str(x) for x in query_embedding)}]" + + if kb_id: + rows = await conn.fetch( + """ + SELECT doc_id, chunk_id, content, metadata, + 1 - (embedding <=> $1::vector) as score + FROM document_embeddings + WHERE kb_id = $2 + ORDER BY embedding <=> $1::vector + LIMIT $3; + """, + embedding_str, kb_id, top_k, + ) + else: + rows = await conn.fetch( + """ + SELECT doc_id, chunk_id, content, metadata, + 1 - (embedding <=> $1::vector) as score + FROM document_embeddings + ORDER BY embedding <=> $1::vector + LIMIT $2; + """, + embedding_str, top_k, + ) + + results = [] + for row in rows: + score = float(row["score"]) + if score >= threshold: + results.append({ + "doc_id": row["doc_id"], + "chunk_id": row["chunk_id"], + "content": row["content"], + "score": score, + "metadata": row["metadata"] if isinstance(row["metadata"], dict) else {}, + }) + + return results + + except Exception as e: + logger.error(f"Vector search failed: {e}") + return [] + + +async def index_document( + doc_id: str, + kb_id: str, + content: str, + chunk_id: Optional[str] = None, + metadata: Optional[Dict] = None, +) -> bool: + """ + 将文档内容索引到 PGVector + + Args: + doc_id: 文档ID + kb_id: 知识库ID + content: 文档内容 + chunk_id: 分块ID (可选) + metadata: 元数据 + + Returns: + 是否成功 + """ + pool = await get_pg_pool() + if pool is None: + return False + + # 分块处理（简单按长度分块） + chunks = [] + chunk_size = 500 + overlap = 50 + if len(content) <= chunk_size: + chunks = [(content, chunk_id or f"{doc_id}_0")] + else: + for i in range(0, len(content), chunk_size - overlap): + chunk = content[i:i + chunk_size] + if len(chunk.strip()) > 10: + chunks.append((chunk, f"{doc_id}_{i // (chunk_size - overlap)}")) + + success_count = 0 + for chunk_text, cid in chunks: + embedding = await get_embedding(chunk_text) + if embedding is None: + continue + + try: + async with pool.acquire() as conn: + embedding_str = f"[{','.join(str(x) for x in embedding)}]" + import json + await conn.execute( + """ + INSERT INTO document_embeddings (doc_id, kb_id, chunk_id, content, embedding, metadata) + VALUES ($1, $2, $3, $4, $5::vector, $6) + ON CONFLICT DO NOTHING; + """, + doc_id, kb_id, cid, chunk_text, embedding_str, + json.dumps(metadata or {}), + ) + success_count += 1 + except Exception as e: + logger.error(f"Failed to index chunk {cid}: {e}") + + logger.info(f"Indexed document {doc_id}: {success_count}/{len(chunks)} chunks") + return success_count > 0 + + +async def hybrid_search( + query: str, + kb_id: Optional[str] = None, + top_k: int = 5, + bm25_results: Optional[List[Dict]] = None, +) -> List[Dict]: + """ + 混合检索：BM25 + 向量检索融合 + + Args: + query: 查询文本 + kb_id: 知识库ID + top_k: 返回结果数 + bm25_results: BM25 检索结果（来自MySQL） + + Returns: + 融合后的检索结果 + """ + # 获取向量检索结果 + vec_results = await vector_search(query, kb_id, top_k) + + if not vec_results and not bm25_results: + return [] + + if not vec_results: + return bm25_results or [] + + if not bm25_results: + return vec_results + + # 融合：Reciprocal Rank Fusion (RRF) + rrf_k = 60 + score_map = {} + + for rank, item in enumerate(bm25_results): + doc_id = item.get("doc_id", item.get("id", "")) + score_map[doc_id] = { + "content": item.get("content", item.get("text", "")), + "score": 1.0 / (rrf_k + rank + 1), + "bm25_rank": rank + 1, + "metadata": item.get("metadata", {}), + } + + for rank, item in enumerate(vec_results): + doc_id = item["doc_id"] + if doc_id in score_map: + score_map[doc_id]["score"] += 1.0 / (rrf_k + rank + 1) + score_map[doc_id]["vec_score"] = item["score"] + score_map[doc_id]["vec_rank"] = rank + 1 + else: + score_map[doc_id] = { + "content": item["content"], + "score": 1.0 / (rrf_k + rank + 1), + "vec_score": item["score"], + "vec_rank": rank + 1, + "metadata": item.get("metadata", {}), + } + + # 按融合分数排序 + merged = sorted(score_map.values(), key=lambda x: x["score"], reverse=True) + return merged[:top_k] + + +async def delete_document_embeddings(doc_id: str) -> bool: + """删除文档的所有向量索引""" + pool = await get_pg_pool() + if pool is None: + return False + + try: + async with pool.acquire() as conn: + await conn.execute( + "DELETE FROM document_embeddings WHERE doc_id = $1;", + doc_id, + ) + logger.info(f"Deleted embeddings for document {doc_id}") + return True + except Exception as e: + logger.error(f"Failed to delete embeddings: {e}") + return False + + +async def close_pool(): + """关闭连接池""" + global _pg_pool + if _pg_pool is not None: + await _pg_pool.close() + _pg_pool = None + logger.info("PostgreSQL connection pool closed") diff --git a/backend/python/main.py b/backend/python/main.py index af98e64..2d4c997 100644 --- a/backend/python/main.py +++ b/backend/python/main.py @@ -1,455 +1,476 @@ -""" - -Qihu-AIFlow FastAPI 主应用 - -完整后端API，监听18080端口，所有接口以 /app 为前缀 - -""" - -import sys - -import os - -import time - -import uuid - -import logging - -from contextlib import asynccontextmanager - -from fastapi import FastAPI, Request - -from fastapi.middleware.cors import CORSMiddleware - -from starlette.middleware.base import BaseHTTPMiddleware - -from datetime import datetime - - - -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) - - - -from app.api import auth, chat, channels, knowledge, market, token_usage, payment, profile, notifications, assets, evolution - -from app.api.admin import dashboard, users, models, settings, payment_admin, services, membership_admin, ads, market_admin, logs, skills_admin, agents_admin, knowledge_admin, workflow_admin, storage_admin, automation_admin, mcp_admin, developer_admin, apps_admin, distro_admin, code_exec_admin, privacy_admin, im_bot_admin, promotion_admin, achievement_admin , tenant_admin, billing_admin, analytics_admin -from app.api.portal import user_portal -from app.api.admin.developer_tools_admin import router as developer_tools_router -from app.api.admin.mcp_market_admin import router as mcp_market_router -from app.api.admin import asset_admin, achievement_admin -from app.config import setup_file_logging, SLOW_REQUEST_THRESHOLD - -from app.database import SessionLocal - -from app.models import OperationLog as OperationLogModel, TokenUsageLog as TokenUsageLogModel - - - -# 导入平台适配器触发注册（AstrBot借鉴优化方案 P1） - -import app.platform_adapters # noqa: F401 - 触发所有平台适配器装饰器注册 - - - -# 初始化文件日志（按天轮转） - -access_logger = setup_file_logging() - -logger = logging.getLogger("aiflow") - - - - - -class APITrackingMiddleware(BaseHTTPMiddleware): - - """API调用统计中间件 - 记录请求耗时，标记慢请求""" - - - - async def dispatch(self, request: Request, call_next): - - request_id = str(uuid.uuid4()) - - start_time = time.time() - - path = request.url.path - - method = request.method - - - - # 跳过健康检查和静态资源 - - skip_paths = ["/app/api/health", "/favicon.ico"] - - should_skip = any(path.startswith(p) for p in skip_paths) - - - - try: - - response = await call_next(request) - - elapsed = time.time() - start_time - - status_code = response.status_code - - - - if not should_skip: - - access_logger.info( - - f"[{request_id}] {method} {path} -> {status_code} ({elapsed:.3f}s)" - - ) - - - - # 异步记录API调用统计 - - try: - - db = SessionLocal() - - try: - - _record_api_call(db, method, path, status_code, elapsed) - - finally: - - db.close() - - except Exception as e: - - logger.warning(f"API统计记录失败: {e}") - - - - return response - - except Exception as e: - - elapsed = time.time() - start_time - - if not should_skip: - - access_logger.info( - - f"[{request_id}] {method} {path} -> ERROR ({elapsed:.3f}s): {str(e)}" - - ) - - raise - - - - async def __call__(self, scope, receive, send): - - if scope["type"] != "http": - - await self.app(scope, receive, send) - - return - - await super().__call__(scope, receive, send) - - - - - -def _record_api_call(db, method: str, path: str, status_code: int, elapsed: float): - - """记录API调用到数据库统计表""" - - # 简化路径以减少基数（去掉动态ID） - - parts = path.split("/") - - simplified_parts = [] - - for part in parts: - - # 将UUID样式的路径段替换为{id} - - if len(part) == 36 and part.count("-") == 4: - - simplified_parts.append("{id}") - - # 将纯数字路径段替换为{id} - - elif part.isdigit(): - - simplified_parts.append("{id}") - - else: - - simplified_parts.append(part) - - simplified_path = "/".join(simplified_parts) - - - - # 统计记录通过数据库实现，简化版记录到日志 - - is_error = status_code >= 400 - - is_slow = elapsed >= SLOW_REQUEST_THRESHOLD - - - - if is_slow: - - logger.warning( - - f"慢请求: {method} {simplified_path} -> {status_code} ({elapsed:.3f}s, 阈值: {SLOW_REQUEST_THRESHOLD}s)" - - ) - - - - - -@asynccontextmanager - -async def lifespan(app: FastAPI): - - logger.info("Qihu-AIFlow v2.0.0 启动中...") - - - - # 启动后台 Evolution Pipeline（知识库图谱已存在的用户配置自动恢复运行） - - try: - - from app.services.evolution_service import EvolutionWatcherManager - - from app.models import UserEvolutionConfig - - db = SessionLocal() - - try: - - configs = db.query(UserEvolutionConfig).filter( - - UserEvolutionConfig.watcher_status == "running" - - ).all() - - if configs: - - watcher = EvolutionWatcherManager() - - for cfg in configs: - - try: - - watcher.start_user_evolution(cfg.user_id, lambda: SessionLocal()) - - logger.info(f"[Evolution] 恢复用户 {cfg.user_id} 的进化Pipeline") - - except Exception as e: - - logger.warning(f"[Evolution] 恢复用户 {cfg.user_id} 失败: {e}") - - logger.info(f"[Evolution] 已恢复 {len(configs)} 个用户的进化Pipeline") - - else: - - logger.info("[Evolution] 无待恢复的进化配置") - - finally: - - db.close() - - except Exception as e: - - logger.warning(f"[Evolution] Pipeline自动启动异常(不影响正常服务): {e}") - - - - # 启动Redis事件总线 - - try: - - from app.event_bus import get_event_bus - - event_bus = get_event_bus() - - event_bus.start(num_consumers=2) - - logger.info("[EventBus] Redis事件总线已启动") - - except Exception as e: - - logger.warning(f"[EventBus] 启动失败(不影响正常服务): {e}") - - - - logger.info("Qihu-AIFlow v2.0.0 启动完成") - - - - yield - - - - # 停止Redis事件总线 - - try: - - from app.event_bus import get_event_bus - - get_event_bus().stop() - - logger.info("[EventBus] Redis事件总线已停止") - - except Exception: - - pass - - - - # 优雅关闭：停止所有 Evolution Pipeline - - try: - - from app.services.evolution_service import EvolutionWatcherManager - - EvolutionWatcherManager().shutdown_all() - - logger.info("[Evolution] 所有Pipeline已停止") - - except Exception: - - pass - - - - logger.info("Qihu-AIFlow 正在关闭...") - - - - - -app = FastAPI( - - title="Qihu-AIFlow - 智能自动化PaaS平台", - - version="1.0.0", - - lifespan=lifespan, - -) - - - -# API调用追踪中间件 - -app.add_middleware(APITrackingMiddleware) - - - -# CORS配置 - -app.add_middleware( - - CORSMiddleware, - - allow_origins=["*"], - - allow_credentials=True, - - allow_methods=["*"], - - allow_headers=["*"], - -) - - - -# 前台 API - -app.include_router(auth.router, prefix="/app/api/auth", tags=["认证"]) - -app.include_router(chat.router, prefix="/app/api/chat", tags=["AI聊天"]) - -app.include_router(channels.router, prefix="/app/api/channels", tags=["频道"]) - -app.include_router(knowledge.router, prefix="/app/api/knowledge", tags=["知识库"]) - -app.include_router(market.router, prefix="/app/api/market", tags=["应用市场"]) - -app.include_router(token_usage.router, prefix="/app/api/token", tags=["Token驾驶舱"]) - -app.include_router(payment.router, prefix="/app/api/payment", tags=["付费中心"]) - -app.include_router(profile.router, prefix="/app/api/profile", tags=["个人中心"]) - -app.include_router(notifications.router, prefix="/app/api/notifications", tags=["消息通知"]) - -app.include_router(assets.router, prefix="/app/api/assets", tags=["资产管理"]) - -app.include_router(evolution.router, prefix="/app/api/evolution", tags=["AI自进化"]) - - - -# 后台 API - -app.include_router(dashboard.router, prefix="/app/api/admin/dashboard", tags=["后台-仪表盘"]) - -app.include_router(users.router, prefix="/app/api/admin/users", tags=["后台-用户管理"]) - -app.include_router(models.router, prefix="/app/api/admin/models", tags=["后台-模型管理"]) - -app.include_router(settings.router, prefix="/app/api/admin/settings", tags=["后台-网站设置"]) - -app.include_router(payment_admin.router, prefix="/app/api/admin/payment", tags=["后台-支付配置"]) - -app.include_router(services.router, prefix="/app/api/admin/services", tags=["后台-增值服务"]) - -app.include_router(membership_admin.router, prefix="/app/api/admin/membership", tags=["后台-会员管理"]) - -app.include_router(ads.router, prefix="/app/api/admin/ads", tags=["后台-广告管理"]) - -app.include_router(market_admin.router, prefix="/app/api/admin/market", tags=["后台-应用审核"]) - -app.include_router(logs.router, prefix="/app/api/admin/logs", tags=["后台-日志风控"]) - -app.include_router(skills_admin.router, prefix="/app/api/admin/skills", tags=["后台-技能管理"]) - -app.include_router(agents_admin.router, prefix="/app/api/admin/agents", tags=["后台-Agent管理"]) -app.include_router(knowledge_admin.router, prefix="/app/api/admin/knowledge", tags=["后台-知识库管理"]) -app.include_router(workflow_admin.router, prefix="/app/api/admin/workflows", tags=["后台-工作流编排"]) -app.include_router(asset_admin.router, prefix="/app/api/admin/assets", tags=["后台-资产管理"]) -app.include_router(storage_admin.router, prefix="/app/api/admin/storage", tags=["后台-存储配置"]) -app.include_router(automation_admin.router, prefix="/app/api/admin/automation", tags=["后台-自动化任务"]) -app.include_router(mcp_admin.router, prefix="/app/api/admin/mcp", tags=["后台-MCP服务"]) -app.include_router(developer_admin.router, prefix="/app/api/admin/developer", tags=["后台-开发者API"]) -app.include_router(developer_tools_router, prefix="/app/api/admin/developer-tools", tags=["dev-tools"]) -app.include_router(apps_admin.router, prefix="/app/api/admin/apps", tags=["admin-ai-apps"]) -app.include_router(mcp_market_router, prefix="/app/api/admin/mcp-market", tags=["mcp-market"]) -app.include_router(code_exec_admin.router, prefix="/app/api/admin/code-exec", tags=["code-exec"]) -app.include_router(distro_admin.router, prefix="/app/api/admin/distributions", tags=["distributions"]) -app.include_router(privacy_admin.router, prefix="/app/api/admin/privacy", tags=["privacy"]) -app.include_router(im_bot_admin.router, prefix="/app/api/admin/im-bot", tags=["im-bot"]) -app.include_router(promotion_admin.router, prefix="/app/api/admin/promotion", tags=["promotion"]) - -app.include_router(tenant_admin.router, prefix="/app/api/admin/tenant", tags=["tenant"]) -app.include_router(billing_admin.router, prefix="/app/api/admin/billing", tags=["billing"]) -app.include_router(analytics_admin.router, prefix="/app/api/admin/analytics", tags=["analytics"]) -app.include_router(user_portal.router, prefix="/app/api/portal", tags=["portal"]) -app.include_router(achievement_admin.router, prefix="/app/api/admin/achievement", tags=["achievement"]) - - - - -@app.get("/app/api/health") - -async def health_check(): - - return {"status": "ok", "service": "Qihu-AIFlow"} - +""" + +Qihu-AIFlow FastAPI 主应用 + +完整后端API，监听18080端口，所有接口以 /app 为前缀 + +""" + +import sys + +import os + +import time + +import uuid + +import logging + +from contextlib import asynccontextmanager + +from fastapi import FastAPI, Request +from fastapi.responses import PlainTextResponse + +from fastapi.middleware.cors import CORSMiddleware + +from starlette.middleware.base import BaseHTTPMiddleware + +from datetime import datetime + + + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + + + +from app.casdoor_auth import router as casdoor_router +from app.api import auth, chat, channels, knowledge, market, token_usage, payment, profile, notifications, assets, evolution + +from app.api.admin import dashboard, users, models, settings, payment_admin, services, membership_admin, ads, market_admin, logs, skills_admin, agents_admin, knowledge_admin, workflow_admin, storage_admin, automation_admin, mcp_admin, developer_admin, apps_admin, distro_admin, code_exec_admin, privacy_admin, im_bot_admin, promotion_admin, achievement_admin , tenant_admin, billing_admin, analytics_admin +from app.api.portal import user_portal +from app.api.admin.developer_tools_admin import router as developer_tools_router +from app.api.admin.mcp_market_admin import router as mcp_market_router +from app.api.admin import asset_admin, achievement_admin +from app.config import setup_file_logging, SLOW_REQUEST_THRESHOLD + +from app.database import SessionLocal + +from app.models import OperationLog as OperationLogModel, TokenUsageLog as TokenUsageLogModel + + + +# 导入平台适配器触发注册（AstrBot借鉴优化方案 P1） + +import app.platform_adapters # noqa: F401 - 触发所有平台适配器装饰器注册 + + + +# 初始化文件日志（按天轮转） + +access_logger = setup_file_logging() + +logger = logging.getLogger("aiflow") + + + + + +class APITrackingMiddleware(BaseHTTPMiddleware): + + """API调用统计中间件 - 记录请求耗时，标记慢请求""" + + + + async def dispatch(self, request: Request, call_next): + + request_id = str(uuid.uuid4()) + + start_time = time.time() + + path = request.url.path + + method = request.method + + + + # 跳过健康检查和静态资源 + + skip_paths = ["/app/api/health", "/favicon.ico"] + + should_skip = any(path.startswith(p) for p in skip_paths) + + + + try: + + response = await call_next(request) + + elapsed = time.time() - start_time + + status_code = response.status_code + + + + if not should_skip: + + access_logger.info( + + f"[{request_id}] {method} {path} -> {status_code} ({elapsed:.3f}s)" + + ) + + + + # 异步记录API调用统计 + + try: + + db = SessionLocal() + + try: + + _record_api_call(db, method, path, status_code, elapsed) + + finally: + + db.close() + + except Exception as e: + + logger.warning(f"API统计记录失败: {e}") + + + + return response + + except Exception as e: + + elapsed = time.time() - start_time + + if not should_skip: + + access_logger.info( + + f"[{request_id}] {method} {path} -> ERROR ({elapsed:.3f}s): {str(e)}" + + ) + + raise + + + + async def __call__(self, scope, receive, send): + + if scope["type"] != "http": + + await self.app(scope, receive, send) + + return + + await super().__call__(scope, receive, send) + + + + + +def _record_api_call(db, method: str, path: str, status_code: int, elapsed: float): + + """记录API调用到数据库统计表""" + + # 简化路径以减少基数（去掉动态ID） + + parts = path.split("/") + + simplified_parts = [] + + for part in parts: + + # 将UUID样式的路径段替换为{id} + + if len(part) == 36 and part.count("-") == 4: + + simplified_parts.append("{id}") + + # 将纯数字路径段替换为{id} + + elif part.isdigit(): + + simplified_parts.append("{id}") + + else: + + simplified_parts.append(part) + + simplified_path = "/".join(simplified_parts) + + + + # 统计记录通过数据库实现，简化版记录到日志 + + is_error = status_code >= 400 + + is_slow = elapsed >= SLOW_REQUEST_THRESHOLD + + + + if is_slow: + + logger.warning( + + f"慢请求: {method} {simplified_path} -> {status_code} ({elapsed:.3f}s, 阈值: {SLOW_REQUEST_THRESHOLD}s)" + + ) + + + + + +@asynccontextmanager + +async def lifespan(app: FastAPI): + + logger.info("Qihu-AIFlow v2.0.0 启动中...") + + + + # 启动后台 Evolution Pipeline（知识库图谱已存在的用户配置自动恢复运行） + + try: + + from app.services.evolution_service import EvolutionWatcherManager + + from app.models import UserEvolutionConfig + + db = SessionLocal() + + try: + + configs = db.query(UserEvolutionConfig).filter( + + UserEvolutionConfig.watcher_status == "running" + + ).all() + + if configs: + + watcher = EvolutionWatcherManager() + + for cfg in configs: + + try: + + watcher.start_user_evolution(cfg.user_id, lambda: SessionLocal()) + + logger.info(f"[Evolution] 恢复用户 {cfg.user_id} 的进化Pipeline") + + except Exception as e: + + logger.warning(f"[Evolution] 恢复用户 {cfg.user_id} 失败: {e}") + + logger.info(f"[Evolution] 已恢复 {len(configs)} 个用户的进化Pipeline") + + else: + + logger.info("[Evolution] 无待恢复的进化配置") + + finally: + + db.close() + + except Exception as e: + + logger.warning(f"[Evolution] Pipeline自动启动异常(不影响正常服务): {e}") + + + + # 启动Redis事件总线 + + try: + + from app.event_bus import get_event_bus + + event_bus = get_event_bus() + + event_bus.start(num_consumers=2) + + logger.info("[EventBus] Redis事件总线已启动") + + except Exception as e: + + logger.warning(f"[EventBus] 启动失败(不影响正常服务): {e}") + + + + logger.info("Qihu-AIFlow v2.0.0 启动完成") + + + + yield + + + + # 停止Redis事件总线 + + try: + + from app.event_bus import get_event_bus + + get_event_bus().stop() + + logger.info("[EventBus] Redis事件总线已停止") + + except Exception: + + pass + + + + # 优雅关闭：停止所有 Evolution Pipeline + + try: + + from app.services.evolution_service import EvolutionWatcherManager + + EvolutionWatcherManager().shutdown_all() + + logger.info("[Evolution] 所有Pipeline已停止") + + except Exception: + + pass + + + + logger.info("Qihu-AIFlow 正在关闭...") + + + + + +app = FastAPI( + + title="Qihu-AIFlow - 智能自动化PaaS平台", + + version="1.0.0", + + lifespan=lifespan, + +) + + + +# API调用追踪中间件 + +app.add_middleware(APITrackingMiddleware) + + + +# CORS配置 + +app.add_middleware( + + CORSMiddleware, + + allow_origins=["*"], + + allow_credentials=True, + + allow_methods=["*"], + + allow_headers=["*"], + +) + + + +# 前台 API + +app.include_router(auth.router, prefix="/app/api/auth", tags=["认证"]) + +app.include_router(chat.router, prefix="/app/api/chat", tags=["AI聊天"]) + +app.include_router(channels.router, prefix="/app/api/channels", tags=["频道"]) + +app.include_router(knowledge.router, prefix="/app/api/knowledge", tags=["知识库"]) + +app.include_router(market.router, prefix="/app/api/market", tags=["应用市场"]) + +app.include_router(token_usage.router, prefix="/app/api/token", tags=["Token驾驶舱"]) + +app.include_router(payment.router, prefix="/app/api/payment", tags=["付费中心"]) + +app.include_router(profile.router, prefix="/app/api/profile", tags=["个人中心"]) + +app.include_router(notifications.router, prefix="/app/api/notifications", tags=["消息通知"]) + +app.include_router(assets.router, prefix="/app/api/assets", tags=["资产管理"]) + +app.include_router(evolution.router, prefix="/app/api/evolution", tags=["AI自进化"]) + + + +# 后台 API + +app.include_router(dashboard.router, prefix="/app/api/admin/dashboard", tags=["后台-仪表盘"]) + +app.include_router(users.router, prefix="/app/api/admin/users", tags=["后台-用户管理"]) + +app.include_router(models.router, prefix="/app/api/admin/models", tags=["后台-模型管理"]) + +app.include_router(settings.router, prefix="/app/api/admin/settings", tags=["后台-网站设置"]) + +app.include_router(payment_admin.router, prefix="/app/api/admin/payment", tags=["后台-支付配置"]) + +app.include_router(services.router, prefix="/app/api/admin/services", tags=["后台-增值服务"]) + +app.include_router(membership_admin.router, prefix="/app/api/admin/membership", tags=["后台-会员管理"]) + +app.include_router(ads.router, prefix="/app/api/admin/ads", tags=["后台-广告管理"]) + +app.include_router(market_admin.router, prefix="/app/api/admin/market", tags=["后台-应用审核"]) + +app.include_router(logs.router, prefix="/app/api/admin/logs", tags=["后台-日志风控"]) + +app.include_router(skills_admin.router, prefix="/app/api/admin/skills", tags=["后台-技能管理"]) + +app.include_router(agents_admin.router, prefix="/app/api/admin/agents", tags=["后台-Agent管理"]) +app.include_router(knowledge_admin.router, prefix="/app/api/admin/knowledge", tags=["后台-知识库管理"]) +app.include_router(workflow_admin.router, prefix="/app/api/admin/workflows", tags=["后台-工作流编排"]) +app.include_router(asset_admin.router, prefix="/app/api/admin/assets", tags=["后台-资产管理"]) +app.include_router(storage_admin.router, prefix="/app/api/admin/storage", tags=["后台-存储配置"]) +app.include_router(automation_admin.router, prefix="/app/api/admin/automation", tags=["后台-自动化任务"]) +app.include_router(mcp_admin.router, prefix="/app/api/admin/mcp", tags=["后台-MCP服务"]) +app.include_router(developer_admin.router, prefix="/app/api/admin/developer", tags=["后台-开发者API"]) +app.include_router(developer_tools_router, prefix="/app/api/admin/developer-tools", tags=["dev-tools"]) +app.include_router(apps_admin.router, prefix="/app/api/admin/apps", tags=["admin-ai-apps"]) +app.include_router(mcp_market_router, prefix="/app/api/admin/mcp-market", tags=["mcp-market"]) +app.include_router(code_exec_admin.router, prefix="/app/api/admin/code-exec", tags=["code-exec"]) +app.include_router(distro_admin.router, prefix="/app/api/admin/distributions", tags=["distributions"]) +app.include_router(privacy_admin.router, prefix="/app/api/admin/privacy", tags=["privacy"]) +app.include_router(im_bot_admin.router, prefix="/app/api/admin/im-bot", tags=["im-bot"]) +app.include_router(promotion_admin.router, prefix="/app/api/admin/promotion", tags=["promotion"]) + +app.include_router(tenant_admin.router, prefix="/app/api/admin/tenant", tags=["tenant"]) +app.include_router(billing_admin.router, prefix="/app/api/admin/billing", tags=["billing"]) +app.include_router(analytics_admin.router, prefix="/app/api/admin/analytics", tags=["analytics"]) +app.include_router(user_portal.router, prefix="/app/api/portal", tags=["portal"]) +app.include_router(achievement_admin.router, prefix="/app/api/admin/achievement", tags=["achievement"]) +app.include_router(casdoor_router, prefix="/app/api/auth", tags=["Casdoor SSO"]) + + +@app.get("/metrics") +async def prometheus_metrics(): + """Prometheus metrics endpoint""" + import psutil + metrics = [] + try: + metrics.append(f"aiflow_cpu_percent {psutil.cpu_percent()}") + mem = psutil.virtual_memory() + metrics.append(f"aiflow_memory_used_bytes {mem.used}") + metrics.append(f"aiflow_memory_total_bytes {mem.total}") + metrics.append(f"aiflow_memory_percent {mem.percent}") + disk = psutil.disk_usage('/') + metrics.append(f"aiflow_disk_used_bytes {disk.used}") + metrics.append(f"aiflow_disk_total_bytes {disk.total}") + metrics.append(f"aiflow_process_count {len(psutil.pids())}") + except Exception: + pass + return PlainTextResponse("\n".join(metrics), media_type="text/plain") + + +@app.get("/app/api/health") + +async def health_check(): + + return {"status": "ok", "service": "Qihu-AIFlow"} + diff --git a/frontend/src/pages/admin/Automation.tsx b/frontend/src/pages/admin/Automation.tsx index efee622..98d4114 100644 --- a/frontend/src/pages/admin/Automation.tsx +++ b/frontend/src/pages/admin/Automation.tsx @@ -1,599 +1,599 @@ -import React, { useState, useEffect } from 'react'; -import { adminAPI } from '@/api'; - -interface AutomationTask { - id: string; - name: string; - description: string; - trigger_type: string; - trigger_type_label: string; - cron_expression: string; - action_type: string; - action_type_label: string; - action_config: Record; - event_condition: Record; - is_enabled: boolean; - status: string; - max_retries: number; - timeout_seconds: number; - last_run_at: string; - next_run_at: string; - total_runs: number; - success_runs: number; - created_at: string; - updated_at: string; -} - -interface ExecLog { - id: string; - task_id: string; - task_name: string; - status: string; - trigger_type: string; - started_at: string; - finished_at: string; - duration_ms: number; - result_summary: string; - error_message: string; - created_at: string; -} - -interface AutomationStats { - total_tasks: number; - enabled_tasks: number; - running_tasks: number; - by_trigger: Record; - recent_executions: number; - recent_success: number; -} - -const TRIGGER_CONFIG: Record = { - cron: { label: '定时触发', color: '#6C63FF', bg: '#EDE9FE' }, - webhook: { label: 'Webhook', color: '#059669', bg: '#D1FAE5' }, - event: { label: '事件触发', color: '#D97706', bg: '#FEF3C7' }, -}; - -const STATUS_CONFIG: Record = { - idle: { label: '空闲', color: '#6B7280', bg: '#F3F4F6' }, - running: { label: '运行中', color: '#3B82F6', bg: '#DBEAFE' }, - success: { label: '成功', color: '#10B981', bg: '#D1FAE5' }, - failed: { label: '失败', color: '#EF4444', bg: '#FEE2E2' }, -}; - -const PAGE_SIZE = 15; - -const Automation: React.FC = () => { - // Data state - const [tasks, setTasks] = useState([]); - const [total, setTotal] = useState(0); - const [page, setPage] = useState(1); - const [loading, setLoading] = useState(false); - const [stats, setStats] = useState(null); - - // Filter state - const [filterTrigger, setFilterTrigger] = useState(''); - const [keyword, setKeyword] = useState(''); - - // Modal state - const [modalVisible, setModalVisible] = useState(false); - const [editingTask, setEditingTask] = useState(null); - const [formData, setFormData] = useState({ - name: '', - description: '', - trigger_type: 'cron', - cron_expression: '', - event_condition_str: '', - action_type: 'ai_pipeline', - action_config_str: '{"pipeline_name": "default"}', - is_enabled: true, - max_retries: 3, - timeout_seconds: 300, - }); - const [submitting, setSubmitting] = useState(false); - - // Execution history - const [historyVisible, setHistoryVisible] = useState(false); - const [historyTaskId, setHistoryTaskId] = useState(''); - const [historyTaskName, setHistoryTaskName] = useState(''); - const [historyLogs, setHistoryLogs] = useState([]); - const [historyPage, setHistoryPage] = useState(1); - const [historyTotal, setHistoryTotal] = useState(0); - const [executingId, setExecutingId] = useState(null); - - // Toast - const [toast, setToast] = useState<{ msg: string; type: 'success' | 'error' } | null>(null); - const showToast = (msg: string, type: 'success' | 'error' = 'success') => { - setToast({ msg, type }); - setTimeout(() => setToast(null), 3000); - }; - - // Load stats - const loadStats = async () => { - try { - const result = await adminAPI.getAdminAutomationStats() as any; - setStats(result.data || null); - } catch (e) { /* ignore */ } - }; - - // Load tasks - const loadTasks = async () => { - setLoading(true); - try { - const params: any = { page, page_size: PAGE_SIZE }; - if (filterTrigger) params.trigger_type = filterTrigger; - if (keyword) params.keyword = keyword; - const result = await adminAPI.getAdminAutomationTasks(params) as any; - const data = result.data || result; - setTasks(data.items || []); - setTotal(data.total || 0); - } catch (e: any) { - showToast('加载失败: ' + (e.message || e), 'error'); - } finally { - setLoading(false); - } - }; - - useEffect(() => { loadStats(); }, []); - useEffect(() => { loadTasks(); }, [page, filterTrigger, keyword]); - - // Open create modal - const openCreateModal = () => { - setEditingTask(null); - setFormData({ - name: '', - description: '', - trigger_type: 'cron', - cron_expression: '*/5 * * * *', - event_condition_str: '', - action_type: 'ai_pipeline', - action_config_str: '{"pipeline_name": "default"}', - is_enabled: true, - max_retries: 3, - timeout_seconds: 300, - }); - setModalVisible(true); - }; - - // Open edit modal - const openEditModal = async (task: AutomationTask) => { - try { - const result = await adminAPI.getAdminAutomationTask(task.id) as any; - const detail = result.data; - setEditingTask(task); - setFormData({ - name: detail.name || '', - description: detail.description || '', - trigger_type: detail.trigger_type || 'cron', - cron_expression: detail.cron_expression || '', - event_condition_str: detail.event_condition ? JSON.stringify(detail.event_condition, null, 2) : '', - action_type: detail.action_type || 'ai_pipeline', - action_config_str: detail.action_config ? JSON.stringify(detail.action_config, null, 2) : '{}', - is_enabled: detail.is_enabled !== false, - max_retries: detail.max_retries || 3, - timeout_seconds: detail.timeout_seconds || 300, - }); - setModalVisible(true); - } catch (e: any) { - showToast('获取任务详情失败', 'error'); - } - }; - - // Submit - const handleSubmit = async () => { - if (!formData.name.trim()) { showToast('请输入任务名称', 'error'); return; } - if (formData.trigger_type === 'cron' && !formData.cron_expression.trim()) { - showToast('请输入Cron表达式', 'error'); return; - } - - let actionConfig: any = {}; - if (formData.action_config_str.trim()) { - try { actionConfig = JSON.parse(formData.action_config_str); } - catch { showToast('操作配置JSON格式不正确', 'error'); return; } - } - - let eventCondition: any = undefined; - if (formData.event_condition_str.trim()) { - try { eventCondition = JSON.parse(formData.event_condition_str); } - catch { showToast('事件条件JSON格式不正确', 'error'); return; } - } - - setSubmitting(true); - try { - const data: any = { - name: formData.name.trim(), - description: formData.description.trim(), - trigger_type: formData.trigger_type, - cron_expression: formData.cron_expression.trim(), - action_type: formData.action_type, - action_config: actionConfig, - is_enabled: formData.is_enabled, - max_retries: formData.max_retries, - timeout_seconds: formData.timeout_seconds, - }; - if (eventCondition) data.event_condition = eventCondition; - - if (editingTask) { - await adminAPI.updateAdminAutomationTask(editingTask.id, data); - showToast('更新成功'); - } else { - await adminAPI.createAdminAutomationTask(data); - showToast('创建成功'); - } - - setModalVisible(false); - loadTasks(); - loadStats(); - } catch (e: any) { - showToast('操作失败: ' + (e.message || e), 'error'); - } finally { - setSubmitting(false); - } - }; - - // Delete - const handleDelete = async (task: AutomationTask) => { - if (!window.confirm(`确定要删除任务 "${task.name}" 吗？`)) return; - try { - await adminAPI.deleteAdminAutomationTask(task.id); - showToast('删除成功'); - loadTasks(); - loadStats(); - } catch (e: any) { - showToast('删除失败: ' + (e.message || e), 'error'); - } - }; - - // Toggle - const handleToggle = async (task: AutomationTask) => { - try { - await adminAPI.toggleAdminAutomationTask(task.id); - showToast(task.is_enabled ? '已禁用' : '已启用'); - loadTasks(); - loadStats(); - } catch (e: any) { - showToast('操作失败: ' + (e.message || e), 'error'); - } - }; - - // Execute now - const handleExecuteNow = async (task: AutomationTask) => { - setExecutingId(task.id); - try { - const result = await adminAPI.executeAdminAutomationTask(task.id) as any; - showToast(result.message || '执行完成'); - loadTasks(); - loadStats(); - } catch (e: any) { - showToast('执行失败: ' + (e.message || e), 'error'); - } finally { - setExecutingId(null); - } - }; - - // Show history - const showHistory = async (task: AutomationTask, pageNum: number = 1) => { - setHistoryTaskId(task.id); - setHistoryTaskName(task.name); - setHistoryPage(pageNum); - setHistoryVisible(true); - - try { - const result = await adminAPI.getAdminAutomationTaskHistory(task.id, { page: pageNum, page_size: 20 }) as any; - const data = result.data || result; - setHistoryLogs(data.items || []); - setHistoryTotal(data.total || 0); - } catch (e) { /* ignore */ } - }; - - // Format time - const formatTime = (ts: string) => { - if (!ts) return '-'; - try { return new Date(ts).toLocaleString('zh-CN'); } - catch { return ts; } - }; - - // Format duration - const formatDuration = (ms: number) => { - if (!ms) return '-'; - if (ms < 1000) return `${ms}ms`; - return `${(ms / 1000).toFixed(1)}s`; - }; - - const labelStyle: React.CSSProperties = { display: 'block', marginBottom: '4px', fontSize: '13px', color: '#374151', fontWeight: 500 }; - const inputStyle: React.CSSProperties = { width: '100%', padding: '8px 12px', borderRadius: '6px', border: '1px solid #D1D5DB', fontSize: '14px', boxSizing: 'border-box' }; - const btnStyle = (bg: string, disabled = false): React.CSSProperties => ({ - padding: '8px 20px', borderRadius: '6px', border: 'none', background: bg, - cursor: disabled ? 'not-allowed' : 'pointer', fontSize: '14px', color: '#fff', - opacity: disabled ? 0.7 : 1, - }); - - return ( -
-
-

自动化任务调度

- -
- - {/* Stats Cards */} - {stats && ( -
-
-
{stats.total_tasks}
-
总任务数
-
-
-
{stats.enabled_tasks}
-
已启用
-
-
-
{stats.running_tasks}
-
运行中
-
-
-
{stats.recent_executions}
-
24小时执行
-
-
-
{stats.recent_success}
-
24小时成功
-
-
- )} - - {/* Filter Bar */} -
- - { setKeyword(e.target.value); setPage(1); }} - style={{ padding: '8px 12px', borderRadius: '6px', border: '1px solid #D1D5DB', fontSize: '14px', flex: 1, maxWidth: '400px' }} /> -
- - {/* Table */} -
- - - - - - - - - - - - - - - {loading ? ( - - ) : tasks.length === 0 ? ( - - ) : ( - tasks.map(task => ( - - - - - - - - - - - )) - )} - -
任务名称触发方式操作类型状态上次执行下次执行成功率操作
加载中...
暂无自动化任务
-
{task.name}
- {task.description &&
{task.description}
} - 		- - {(TRIGGER_CONFIG[task.trigger_type] || {}).label || task.trigger_type} - - {task.cron_expression &&
{task.cron_expression}
} - 		{task.action_type_label} - - {(STATUS_CONFIG[task.status] || {}).label || task.status} - - {formatTime(task.last_run_at)}{formatTime(task.next_run_at)} - {task.total_runs > 0 - ? = 0.9 ? '#10B981' : '#D97706' }}> - {Math.round(task.success_runs / task.total_runs * 100)}% - ({task.success_runs}/{task.total_runs}) - - : -} - -
- - - - - -
-
-
- - {/* Pagination */} - {total > PAGE_SIZE && ( -
- - 第 {page} / {Math.ceil(total / PAGE_SIZE)} 页 (共 {total} 条) - -
- )} - - {/* Create/Edit Modal */} - {modalVisible && ( -
-
-

{editingTask ? '编辑任务' : '创建自动化任务'}

-
-
- setFormData({ ...formData, name: e.target.value })} style={inputStyle} placeholder="如: 每日报表生成" />
- -
- setFormData({ ...formData, description: e.target.value })} style={inputStyle} placeholder="可选描述" />
- -
-
-
-
-
-
- - {formData.trigger_type === 'cron' && ( -
- setFormData({ ...formData, cron_expression: e.target.value })} - style={{ ...inputStyle, fontFamily: 'monospace' }} placeholder="*/5 * * * * (每5分钟)" /> -
- 格式: 分 时 日 月 周 | 示例: 0 9 * * * (每天9点) -
- )} - - {formData.trigger_type !== 'cron' && ( -
-