diff --git a/backend/python/app/api/admin/im_bot_admin.py b/backend/python/app/api/admin/im_bot_admin.py index 3f3e699..4e0bd21 100644 --- a/backend/python/app/api/admin/im_bot_admin.py +++ b/backend/python/app/api/admin/im_bot_admin.py @@ -3,13 +3,17 @@ Phase 4: IM Bot Integration - Admin API Module im_bot_admin.py Provides admin management for IM bot channels, routing rules, -LangBot gateway config, statistics, and message monitoring. +statistics, message monitoring, and native IM platform adapters +(e.g. 个人微信 WeKnora/Clawbot) that connect directly to each IM +platform's official open API — no third-party gateway required. """ import uuid import json import hashlib import hmac import time +import threading +import requests from datetime import datetime, timedelta from typing import Optional @@ -167,10 +171,6 @@ async def get_im_bot_stats(user=Depends(get_current_user)): # Total routing rules routes_count = _execute_raw(db, text("SELECT COUNT(*) as cnt FROM im_bot_routes WHERE is_active=1"))[0]._mapping['cnt'] - # LangBot status - langbot_row = _execute_raw(db, text("SELECT status FROM im_bot_langbot_config ORDER BY id LIMIT 1")) - langbot_status = _row_to_dict(langbot_row[0])['status'] if langbot_row else 'inactive' - return { "success": True, "data": { @@ -182,7 +182,6 @@ async def get_im_bot_stats(user=Depends(get_current_user)): "today_unique_users": today_stats.get('users', 0), "today_errors": today_stats.get('errors', 0), "active_routes": routes_count, - "langbot_status": langbot_status, } } except Exception as e: @@ -570,102 +569,264 @@ async def delete_route(route_id: str, user=Depends(get_current_user)): # ============================================================ -# LangBot Gateway +# 个人微信 (iLink 协议 / 微信 ClawBot 插件) 原生适配器 +# 直接对接腾讯官方 iLink Bot HTTP API (ilinkai.weixin.qq.com) +# 扫码登录免 AppID；消息收发走 Bearer bot_token + X-WECHAT-UIN # ============================================================ -@router.get("/langbot") -async def get_langbot_config(user=Depends(get_current_user)): - """Get LangBot gateway configuration""" - db = SessionLocal() - try: - rows = _execute_raw(db, text("SELECT * FROM im_bot_langbot_config ORDER BY id")) - return {"success": True, "data": _rows_to_dict(rows)} - except Exception as e: - raise HTTPException(status_code=500, detail=str(e)) - finally: - db.close() +ILINK_BASE = "https://ilinkai.weixin.qq.com" +_wechat_receiver_threads = {} -@router.put("/langbot") -async def update_langbot_config( - config_id: int = Query(...), - name: Optional[str] = Query(None), - gateway_url: Optional[str] = Query(None), - api_key: Optional[str] = Query(None), - webhook_secret: Optional[str] = Query(None), - bot_uuid: Optional[str] = Query(None), - callback_url: Optional[str] = Query(None), - sync_channels: Optional[int] = Query(None), - status: Optional[str] = Query(None), - user=Depends(get_current_user) -): - """Update LangBot configuration""" - db = SessionLocal() - try: - updates = [] - params = {"id": config_id} - for field, val in [("name", name), ("gateway_url", gateway_url), - ("api_key", api_key), ("webhook_secret", webhook_secret), - ("bot_uuid", bot_uuid), ("callback_url", callback_url), - ("sync_channels", sync_channels), ("status", status)]: - if val is not None: - updates.append(f"{field} = :{field}") - params[field] = val - if not updates: - raise HTTPException(status_code=400, detail="No fields to update") - sql = f"UPDATE im_bot_langbot_config SET {', '.join(updates)} WHERE id = :id" - _execute_raw(db, text(sql), params) - return {"success": True, "message": "LangBot config updated"} - except HTTPException: - raise - except Exception as e: - raise HTTPException(status_code=500, detail=str(e)) - finally: - db.close() - - -@router.post("/langbot/test") -async def test_langbot( - gateway_url: str = Query(...), - api_key: Optional[str] = Query(None), - user=Depends(get_current_user) -): - """Test LangBot gateway connectivity""" - import urllib.request - import urllib.error - - result = { - "gateway_url": gateway_url, - "tested_at": datetime.now().isoformat(), - "reachable": False, - "api_version": None, - "error": None, +def _wechat_headers(bot_token: str) -> dict: + """iLink 协议请求头：Bearer bot_token + 随机 X-WECHAT-UIN 防重放""" + import base64, random + uin = base64.b64encode(str(random.randint(0, 2 ** 32 - 1)).encode("utf-8")).decode("utf-8") + return { + "Content-Type": "application/json", + "Authorization": f"Bearer {bot_token}", + "AuthorizationType": "ilink_bot_token", + "X-WECHAT-UIN": uin, } + +def _wechat_row(db): + rows = _execute_raw(db, text("SELECT * FROM im_bot_wechat_account WHERE user_id IS NULL ORDER BY id LIMIT 1")) + return rows[0] if rows else None + + +def _wechat_mask(row) -> dict: + d = _row_to_dict(row) + if d.get("bot_token"): + bt = d["bot_token"] + d["bot_token"] = ("***" + bt[-6:]) if len(bt) > 6 else "***" + return d + + +def _qihua_reply(text: str) -> str: + """调用七虎AI 生成回复（复用 channels 的真实 LLM 调用）。 + 自行创建 DB 会话，避免上层线程无可用 session 时崩溃。""" + db = None try: - url = gateway_url.rstrip('/') + '/api/v1/pipelines' - headers = {"Content-Type": "application/json"} - if api_key: - headers["Authorization"] = f"Bearer {api_key}" - - req = urllib.request.Request(url, headers=headers, method='GET') - with urllib.request.urlopen(req, timeout=10) as resp: - result['reachable'] = True - result['status_code'] = resp.status - body = resp.read().decode('utf-8', errors='replace') - try: - data = json.loads(body) - result['api_version'] = data.get('version', 'unknown') - except: - result['api_version'] = 'unknown' - except urllib.error.HTTPError as e: - result['reachable'] = True - result['status_code'] = e.code - result['error'] = f"HTTP {e.code}" + from app.api.channels import _call_real_llm + db = SessionLocal() + return _call_real_llm(db, None, text, "个人微信") except Exception as e: - result['error'] = str(e) + return f"（AI 回复生成失败：{e}）" + finally: + if db: + db.close() - return {"success": True, "data": result} + +def _wechat_get_updates(baseurl: str, bot_token: str, cursor: str): + """长轮询收消息，返回 (msgs, new_cursor)。最多 hold 35s。""" + try: + resp = requests.post( + baseurl.rstrip("/") + "/ilink/bot/getupdates", + headers=_wechat_headers(bot_token), + json={"get_updates_buf": cursor, "base_info": {"channel_version": "1.0.2"}}, + timeout=40, + ) + if not resp.ok: + return [], cursor + j = resp.json() if resp.content else {} + return j.get("msgs", []) or [], j.get("get_updates_buf", cursor) + except Exception: + return [], cursor + + +def _wechat_send_message(baseurl: str, bot_token: str, to_user_id: str, context_token: str, text: str) -> bool: + """发送文本回复（必须带 context_token 关联到正确的对话窗口）""" + try: + resp = requests.post( + baseurl.rstrip("/") + "/ilink/bot/sendmessage", + headers=_wechat_headers(bot_token), + json={"msg": { + "to_user_id": to_user_id, + "message_type": 2, + "message_state": 2, + "context_token": context_token, + "item_list": [{"type": 1, "text_item": {"text": text}}], + }}, + timeout=20, + ) + return resp.ok + except Exception: + return False + + +def _wechat_extract_text(msg: dict) -> str: + parts = [] + for it in (msg.get("item_list") or []): + if it.get("type") == 1 and it.get("text_item", {}).get("text"): + parts.append(it["text_item"]["text"]) + return "\n".join(parts) + + +def _wechat_receiver_loop(account_id: int): + """后台接收线程：长轮询个人微信消息 → 七虎AI 回复 → 发送。 + 底层 iLink 协议直连 ilinkai.weixin.qq.com，无需 AppID/SecretKey。""" + cursor = "" + while True: + db = SessionLocal() + try: + row = _wechat_row(db) + if not row: + return + d = _row_to_dict(row) + if d.get("status") != "connected" or not d.get("bot_token"): + return + bot_token = d["bot_token"] + baseurl = d.get("baseurl") or ILINK_BASE + msgs, cursor = _wechat_get_updates(baseurl, bot_token, cursor) + for m in msgs: + if m.get("message_type") not in (1, None): + continue # 只处理用户消息（2=bot 自己发出的） + content = _wechat_extract_text(m) + if not content: + continue + reply = _qihua_reply(content) + _wechat_send_message( + baseurl, bot_token, + m.get("from_user_id", ""), + m.get("context_token", ""), + reply, + ) + time.sleep(1) + except Exception: + time.sleep(5) + finally: + db.close() + + +def _start_wechat_receiver(account_id: int): + if account_id in _wechat_receiver_threads and _wechat_receiver_threads[account_id].is_alive(): + return + t = threading.Thread(target=_wechat_receiver_loop, args=(account_id,), daemon=True) + t.start() + _wechat_receiver_threads[account_id] = t + + +@router.get("/wechat") +async def get_wechat_config(user=Depends(get_current_user)): + """获取个人微信(Clawbot)适配器配置（敏感字段脱敏）""" + db = SessionLocal() + try: + row = _wechat_row(db) + if not row: + return {"success": True, "data": None} + return {"success": True, "data": _wechat_mask(row)} + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + finally: + db.close() + + +# （POST /wechat 凭证保存端点已移除：iLink 扫码登录无需 AppID/SecretKey） + + +@router.post("/wechat/qrcode") +async def wechat_qrcode(user=Depends(get_current_user)): + """获取个人微信扫码登录二维码（iLink 协议，免 AppID）""" + try: + resp = requests.get( + ILINK_BASE + "/ilink/bot/get_bot_qrcode", + params={"bot_type": 3}, + timeout=20, + ) + if not resp.ok: + return {"success": False, "error": f"获取二维码失败 HTTP {resp.status_code}"} + j = resp.json() if resp.content else {} + return {"success": True, "data": { + "qrcode": j.get("qrcode"), + "qrcode_img_content": j.get("qrcode_img_content"), + }} + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + +@router.post("/wechat/qrcode/status") +async def wechat_qrcode_status(qrcode: str = Query(...), user=Depends(get_current_user)): + """轮询扫码状态；confirmed 时保存 bot_token 并启动接收线程""" + try: + resp = requests.get( + ILINK_BASE + "/ilink/bot/get_qrcode_status", + params={"qrcode": qrcode}, + timeout=40, + ) + if not resp.ok: + return {"success": False, "error": f"查询状态失败 HTTP {resp.status_code}"} + j = resp.json() if resp.content else {} + data = j.get("data", j) if isinstance(j, dict) else j + status = data.get("status") + if status == "confirmed": + # 兼容两种返回结构：顶层 或 credentials 嵌套 + creds = data.get("credentials", {}) or {} + bot_token = data.get("bot_token") or creds.get("bot_token") + baseurl = data.get("baseurl") or creds.get("baseurl") or ILINK_BASE + ilink_bot_id = data.get("ilink_bot_id") or creds.get("ilink_bot_id") + db = SessionLocal() + try: + row = _wechat_row(db) + if not row: + _execute_raw(db, text("""INSERT INTO im_bot_wechat_account + (bot_token, ilink_bot_id, baseurl, status, connected_at, created_at, updated_at) + VALUES (:bot_token, :ilink_bot_id, :baseurl, 'connected', :connected_at, :created_at, :updated_at)"""), + {"bot_token": bot_token, "ilink_bot_id": ilink_bot_id, "baseurl": baseurl, + "connected_at": datetime.now(), "created_at": datetime.now(), "updated_at": datetime.now()}) + else: + _execute_raw(db, text("""UPDATE im_bot_wechat_account SET + bot_token=:bot_token, ilink_bot_id=:ilink_bot_id, baseurl=:baseurl, + status='connected', connected_at=:connected_at, updated_at=:updated_at WHERE id=:id"""), + {"bot_token": bot_token, "ilink_bot_id": ilink_bot_id, "baseurl": baseurl, + "connected_at": datetime.now(), "updated_at": datetime.now(), "id": _row_to_dict(row)["id"]}) + row = _wechat_row(db) + aid = _row_to_dict(row)["id"] if row else 1 + _start_wechat_receiver(aid) + finally: + db.close() + return {"success": True, "data": {"status": "confirmed", "connected": True}} + return {"success": True, "data": {"status": status, "connected": False}} + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + +@router.get("/wechat/status") +async def wechat_status(user=Depends(get_current_user)): + """个人微信连接状态""" + db = SessionLocal() + try: + row = _wechat_row(db) + if not row: + return {"success": True, "data": {"connected": False, "configured": False}} + d = _row_to_dict(row) + return {"success": True, "data": { + "connected": d.get("status") == "connected", + "configured": True, + "baseurl": d.get("baseurl"), + "ilink_bot_id": d.get("ilink_bot_id"), + "connected_at": str(d.get("connected_at")) if d.get("connected_at") else None, + }} + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + finally: + db.close() + + +@router.post("/wechat/disconnect") +async def wechat_disconnect(user=Depends(get_current_user)): + """断开个人微信连接（清除 bot_token）""" + db = SessionLocal() + try: + row = _wechat_row(db) + if row: + _execute_raw(db, text("UPDATE im_bot_wechat_account SET bot_token=NULL, status='disconnected', updated_at=:updated_at WHERE id=:id"), + {"updated_at": datetime.now(), "id": _row_to_dict(row)["id"]}) + return {"success": True, "message": "已断开个人微信连接"} + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + finally: + db.close() # ============================================================ diff --git a/backend/python/app/api/im_bot_user.py b/backend/python/app/api/im_bot_user.py new file mode 100644 index 0000000..d0e86b3 --- /dev/null +++ b/backend/python/app/api/im_bot_user.py @@ -0,0 +1,359 @@ +""" +Phase 4: 个人微信 (iLink 协议 / 微信 ClawBot 插件) —— 前台用户模块 +im_bot_user.py + +为【平台注册用户】提供"各自绑定自己的个人微信"能力： +- 每个用户独立一条 im_bot_wechat_account 记录（按 user_id 隔离） +- 扫码登录免 AppID；消息收发走 Bearer bot_token + X-WECHAT-UIN +- 绑定后由七虎AI 大脑自动应答（复用 channels 的真实 LLM 调用） + +与后台 im_bot_admin 的区别： +- 后台：平台级 IM Bot 接入中枢（飞书/企微/Telegram 等多平台、路由、统计），其个人微信 + 端点仅限 user_id IS NULL 的"全局 Bot"，与用户绑定互不干扰。 +- 前台：每个登录用户管理自己的个人微信，数据严格按 user_id 隔离。 +""" +import base64 +import json +import random +import threading +import time +from datetime import datetime +from typing import Optional + +import requests +from fastapi import APIRouter, Depends, HTTPException, Query +from sqlalchemy import text +from sqlalchemy.orm import Session + +from ..auth import get_current_user +from ..database import SessionLocal +from ..models import User + +router = APIRouter() + +# ============================================================ +# iLink 协议常量（腾讯官方 iLink Bot HTTP API，直连免 AppID） +# ============================================================ +ILINK_BASE = "https://ilinkai.weixin.qq.com" + +# 每个已连接账号一个后台接收线程，key = 账号表主键 id +_wechat_receiver_threads: dict[int, threading.Thread] = {} + + +# ============================================================ +# 通用 DB 工具 +# ============================================================ +def _execute_raw(db: Session, sql_text, params=None): + result = db.execute(sql_text, params or {}) + if result.returns_rows: + return result.fetchall() + db.commit() + return result + + +def _row_to_dict(row) -> dict: + if row is None: + return None + d = {} + for col in row._mapping.keys(): + val = row._mapping[col] + if isinstance(val, datetime): + val = val.isoformat() + d[col] = val + return d + + +# ============================================================ +# iLink 协议请求工具 +# ============================================================ +def _wechat_headers(bot_token: str) -> dict: + """iLink 协议请求头：Bearer bot_token + 随机 X-WECHAT-UIN 防重放""" + uin = base64.b64encode(str(random.randint(0, 2 ** 32 - 1)).encode("utf-8")).decode("utf-8") + return { + "Content-Type": "application/json", + "Authorization": f"Bearer {bot_token}", + "AuthorizationType": "ilink_bot_token", + "X-WECHAT-UIN": uin, + } + + +def _wechat_row_for_user(db: Session, user_id: str): + rows = _execute_raw( + db, + text("SELECT * FROM im_bot_wechat_account WHERE user_id = :uid ORDER BY id DESC LIMIT 1"), + {"uid": user_id}, + ) + return rows[0] if rows else None + + +def _wechat_mask(row) -> dict: + d = _row_to_dict(row) + if d.get("bot_token"): + bt = d["bot_token"] + d["bot_token"] = ("***" + bt[-6:]) if len(bt) > 6 else "***" + d.pop("appid", None) + d.pop("secret_key", None) + return d + + +# ============================================================ +# 七虎AI 大脑（复用 channels 的真实 LLM 调用） +# ============================================================ +def _qihua_reply(text: str) -> str: + """调用七虎AI 生成回复。自行创建 DB 会话，避免上层线程无可用 session。""" + db = None + try: + from app.api.channels import _call_real_llm + db = SessionLocal() + return _call_real_llm(db, None, text, "个人微信") + except Exception as e: + return f"（AI 回复生成失败：{e}）" + finally: + if db: + db.close() + + +# ============================================================ +# iLink 消息收发 +# ============================================================ +def _wechat_get_updates(baseurl: str, bot_token: str, cursor: str): + """长轮询收消息，返回 (msgs, new_cursor)。最多 hold 35s。""" + try: + resp = requests.post( + baseurl.rstrip("/") + "/ilink/bot/getupdates", + headers=_wechat_headers(bot_token), + json={"get_updates_buf": cursor, "base_info": {"channel_version": "1.0.2"}}, + timeout=40, + ) + if not resp.ok: + return [], cursor + j = resp.json() if resp.content else {} + return j.get("msgs", []) or [], j.get("get_updates_buf", cursor) + except Exception: + return [], cursor + + +def _wechat_send_message(baseurl: str, bot_token: str, to_user_id: str, context_token: str, text: str) -> bool: + """发送文本回复（必须带 context_token 关联到正确的对话窗口）""" + try: + resp = requests.post( + baseurl.rstrip("/") + "/ilink/bot/sendmessage", + headers=_wechat_headers(bot_token), + json={"msg": { + "to_user_id": to_user_id, + "message_type": 2, + "message_state": 2, + "context_token": context_token, + "item_list": [{"type": 1, "text_item": {"text": text}}], + }}, + timeout=20, + ) + return resp.ok + except Exception: + return False + + +def _wechat_extract_text(msg: dict) -> str: + parts = [] + for it in (msg.get("item_list") or []): + if it.get("type") == 1 and it.get("text_item", {}).get("text"): + parts.append(it["text_item"]["text"]) + return "\n".join(parts) + + +# ============================================================ +# 后台接收线程（每用户独立） +# ============================================================ +def _wechat_receiver_loop(account_id: int): + """长轮询个人微信消息 → 七虎AI 回复 → 发送。""" + cursor = "" + while True: + db = SessionLocal() + try: + row = _execute_raw( + db, + text("SELECT * FROM im_bot_wechat_account WHERE id = :aid"), + {"aid": account_id}, + ) + if not row: + return + d = _row_to_dict(row[0]) + if d.get("status") != "connected" or not d.get("bot_token"): + return + bot_token = d["bot_token"] + baseurl = d.get("baseurl") or ILINK_BASE + msgs, cursor = _wechat_get_updates(baseurl, bot_token, cursor) + for m in msgs: + if m.get("message_type") not in (1, None): + continue # 只处理用户消息（2=bot 自己发出的） + content = _wechat_extract_text(m) + if not content: + continue + reply = _qihua_reply(content) + _wechat_send_message( + baseurl, bot_token, + m.get("from_user_id", ""), + m.get("context_token", ""), + reply, + ) + time.sleep(1) + except Exception: + time.sleep(5) + finally: + db.close() + + +def _start_wechat_receiver(account_id: int): + if account_id in _wechat_receiver_threads and _wechat_receiver_threads[account_id].is_alive(): + return + t = threading.Thread(target=_wechat_receiver_loop, args=(account_id,), daemon=True) + t.start() + _wechat_receiver_threads[account_id] = t + + +# ============================================================ +# 前台用户接口（按 user_id 隔离） +# ============================================================ +@router.get("/status") +async def wechat_status(user: User = Depends(get_current_user)): + """当前登录用户的个人微信连接状态""" + db = SessionLocal() + try: + row = _wechat_row_for_user(db, user.id) + if not row: + return {"success": True, "data": {"connected": False, "configured": False}} + d = _row_to_dict(row) + return {"success": True, "data": { + "connected": d.get("status") == "connected", + "configured": True, + "baseurl": d.get("baseurl"), + "ilink_bot_id": d.get("ilink_bot_id"), + "username": d.get("username"), + "connected_at": str(d.get("connected_at")) if d.get("connected_at") else None, + }} + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + finally: + db.close() + + +@router.post("/qrcode") +async def wechat_qrcode(user: User = Depends(get_current_user)): + """获取个人微信扫码登录二维码（iLink 协议，免 AppID）""" + try: + resp = requests.get( + ILINK_BASE + "/ilink/bot/get_bot_qrcode", + params={"bot_type": 3}, + timeout=20, + ) + if not resp.ok: + return {"success": False, "error": f"获取二维码失败 HTTP {resp.status_code}"} + j = resp.json() if resp.content else {} + return {"success": True, "data": { + "qrcode": j.get("qrcode"), + "qrcode_img_content": j.get("qrcode_img_content"), + }} + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + +@router.post("/qrcode/status") +async def wechat_qrcode_status( + qrcode: str = Query(...), + user: User = Depends(get_current_user), +): + """轮询扫码状态；confirmed 时保存该用户的 bot_token 并启动接收线程""" + try: + resp = requests.get( + ILINK_BASE + "/ilink/bot/get_qrcode_status", + params={"qrcode": qrcode}, + timeout=40, + ) + if not resp.ok: + return {"success": False, "error": f"查询状态失败 HTTP {resp.status_code}"} + j = resp.json() if resp.content else {} + data = j.get("data", j) if isinstance(j, dict) else j + status = data.get("status") + if status == "confirmed": + creds = data.get("credentials", {}) or {} + bot_token = data.get("bot_token") or creds.get("bot_token") + baseurl = data.get("baseurl") or creds.get("baseurl") or ILINK_BASE + ilink_bot_id = data.get("ilink_bot_id") or creds.get("ilink_bot_id") + db = SessionLocal() + try: + row = _wechat_row_for_user(db, user.id) + if not row: + _execute_raw(db, text("""INSERT INTO im_bot_wechat_account + (user_id, username, bot_token, ilink_bot_id, baseurl, status, + connected_at, created_at, updated_at) + VALUES (:user_id, :username, :bot_token, :ilink_bot_id, :baseurl, + 'connected', :connected_at, :created_at, :updated_at)"""), + { + "user_id": user.id, "username": user.username, + "bot_token": bot_token, "ilink_bot_id": ilink_bot_id, + "baseurl": baseurl, + "connected_at": datetime.now(), + "created_at": datetime.now(), "updated_at": datetime.now(), + }) + else: + _execute_raw(db, text("""UPDATE im_bot_wechat_account SET + user_id=:user_id, username=:username, bot_token=:bot_token, + ilink_bot_id=:ilink_bot_id, baseurl=:baseurl, + status='connected', connected_at=:connected_at, updated_at=:updated_at + WHERE id=:id"""), + { + "user_id": user.id, "username": user.username, + "bot_token": bot_token, "ilink_bot_id": ilink_bot_id, + "baseurl": baseurl, "connected_at": datetime.now(), + "updated_at": datetime.now(), "id": _row_to_dict(row)["id"], + }) + row = _wechat_row_for_user(db, user.id) + aid = _row_to_dict(row)["id"] if row else 1 + _start_wechat_receiver(aid) + finally: + db.close() + return {"success": True, "data": {"status": "confirmed", "connected": True}} + return {"success": True, "data": {"status": status, "connected": False}} + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + +@router.post("/disconnect") +async def wechat_disconnect(user: User = Depends(get_current_user)): + """断开当前用户的个人微信连接（清除 bot_token）""" + db = SessionLocal() + try: + row = _wechat_row_for_user(db, user.id) + if row: + _execute_raw(db, text( + "UPDATE im_bot_wechat_account SET bot_token=NULL, status='disconnected', " + "updated_at=:updated_at WHERE id=:id" + ), {"updated_at": datetime.now(), "id": _row_to_dict(row)["id"]}) + return {"success": True, "message": "已断开个人微信连接"} + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + finally: + db.close() + + +# ============================================================ +# 启动恢复：为所有已连接用户账号拉起接收线程 +# ============================================================ +def start_all_wechat_receivers(): + """在应用启动时调用，恢复此前已连接用户的个人微信接收线程。""" + db = SessionLocal() + try: + rows = _execute_raw( + db, + text("SELECT id FROM im_bot_wechat_account WHERE status='connected' AND user_id IS NOT NULL"), + ) + for r in rows: + _start_wechat_receiver(r._mapping["id"]) + if rows: + import logging + logging.getLogger("aiflow").info(f"[IM个人微信] 已恢复 {len(rows)} 个用户微信接收线程") + except Exception as e: + import logging + logging.getLogger("aiflow").warning(f"[IM个人微信] 启动恢复失败(不影响服务): {e}") + finally: + db.close() diff --git a/backend/python/main.py b/backend/python/main.py index 2d4c997..1d39b6f 100644 --- a/backend/python/main.py +++ b/backend/python/main.py @@ -1,476 +1,486 @@ -""" - -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"} - +""" + +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.im_bot_user import router as im_bot_user_router +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}") + + + + # 恢复用户个人微信(iLink)接收线程 + try: + from app.api.im_bot_user import start_all_wechat_receivers + start_all_wechat_receivers() + logger.info("[IM个人微信] 用户微信接收线程已恢复") + except Exception as e: + logger.warning(f"[IM个人微信] 启动恢复异常(不影响服务): {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(im_bot_user_router, prefix="/app/api/im/wechat", tags=["im-wechat-user"]) +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/App.tsx b/frontend/src/App.tsx index 2b32509..258245e 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -22,6 +22,7 @@ import NotificationPage from '@/pages/front/NotificationPage'; import AssetsPage from '@/pages/front/AssetsPage'; import NovelPreviewPage from '@/pages/front/NovelPreviewPage'; import EvolutionDashboard from '@/pages/front/EvolutionDashboard'; +import WechatBot from '@/pages/front/WechatBot'; // Admin Pages import AdminDashboard from '@/pages/admin/Dashboard'; @@ -117,6 +118,7 @@ const App: React.FC = () => { } /> } /> } /> + } /> {/* Admin Routes */} diff --git a/frontend/src/api/index.ts b/frontend/src/api/index.ts index 6ccdf3d..8906c3d 100644 --- a/frontend/src/api/index.ts +++ b/frontend/src/api/index.ts @@ -110,7 +110,7 @@ export const chatAPI = { clearContext: (convId: string) => post(`/chat/conversations/${convId}/clear`), convertToKnowledgeGraph: (convId: string) => - post(`/chat/conversations/${convId}/to-knowledge-graph`), + post(`/chat/conversations/${convId}/to-knowledge-graph/`), getModels: () => get('/chat/models'), }; @@ -155,7 +155,7 @@ export const knowledgeAPI = { deleteBase: (id: string) => del(`/knowledge/bases/${id}`), getDocuments: (baseId: string) => get(`/knowledge/bases/${baseId}/documents`), getDocument: (baseId: string, docId: string) => get(`/knowledge/bases/${baseId}/documents/${docId}`), - uploadDocument: (baseId: string, data: any) => post(`/knowledge/bases/${baseId}/documents`, data), + uploadDocument: (baseId: string, data: any) => post(`/knowledge/bases/${baseId}/documents/`, data), deleteDocument: (baseId: string, docId: string) => del(`/knowledge/bases/${baseId}/documents/${docId}`), getDirectories: () => get('/knowledge/directories'), createDirectory: (data: any) => post('/knowledge/directories', data), @@ -501,6 +501,7 @@ export const adminAPI = { toggleAdminStorageConfig: (id: string) => put(`/admin/storage/${id}/toggle`), setDefaultAdminStorageConfig: (id: string) => put(`/admin/storage/${id}/set-default`), testAdminStorageConfig: (data: any) => post('/admin/storage/test', data), + getAdminStorageUsage: (id: string) => get(`/admin/storage/${id}/usage`), // Storage Quota APIs (Phase C5) getStorageQuotaOverview: () => get('/admin/storage/quota/overview'), getStorageQuotaUsers: (params?: any) => { @@ -859,9 +860,12 @@ export const evolutionAPI = { getImBotRoutes: (params?: string) => get(`/admin/im-bot/routes${params || ''}`), createImBotRoute: (params: string) => post(`/admin/im-bot/routes${params}`), deleteImBotRoute: (id: string) => del(`/admin/im-bot/routes/${id}`), - getLangbotConfig: () => get('/admin/im-bot/langbot'), - updateLangbotConfig: (params: string) => post(`/admin/im-bot/langbot${params}`), - testLangbot: (params: string) => post(`/admin/im-bot/langbot/test${params}`), + // 个人微信（iLink 协议）原生适配器 + getWechatConfig: () => get('/admin/im-bot/wechat'), + wechatQrcode: () => post('/admin/im-bot/wechat/qrcode'), + wechatQrcodeStatus: (params: string) => post(`/admin/im-bot/wechat/qrcode/status${params}`), + getWechatStatus: () => get('/admin/im-bot/wechat/status'), + disconnectWechat: () => post('/admin/im-bot/wechat/disconnect'), getImBotPlatforms: () => get('/admin/im-bot/platforms'), getImBotWebhookUrl: (platform: string) => get(`/admin/im-bot/webhook-url/${platform}`), getImBotMessages: (params?: string) => get(`/admin/im-bot/messages${params || ''}`), @@ -988,3 +992,12 @@ export const evolutionAPI = { }, }; + + +// ===== 前台用户个人微信 (iLink 协议 / 微信 ClawBot 插件) ===== +export const imUserAPI = { + getStatus: () => get('/im/wechat/status'), + qrcode: () => post('/im/wechat/qrcode'), + qrcodeStatus: (qrcode: string) => post(`/im/wechat/qrcode/status?qrcode=${encodeURIComponent(qrcode)}`), + disconnect: () => post('/im/wechat/disconnect'), +}; diff --git a/frontend/src/components/layout/FrontLayout.tsx b/frontend/src/components/layout/FrontLayout.tsx index 5de0ea2..885c921 100644 --- a/frontend/src/components/layout/FrontLayout.tsx +++ b/frontend/src/components/layout/FrontLayout.tsx @@ -12,6 +12,7 @@ const navItems = [ { path: '/pricing', icon: 'pricing', label: '付费中心' }, { path: '/assets', icon: 'assets', label: '资产文件' }, { path: '/evolution', icon: 'evolution', label: 'AI自进化' }, + { path: '/wechat-bot', icon: 'wechat', label: '微信机器人' }, ]; const bottomItems = [ @@ -31,6 +32,7 @@ const iconPaths: Record = { notify: , assets: , evolution: , + wechat: , }; const FrontLayout: React.FC = () => { diff --git a/frontend/src/pages/front/WechatBot.tsx b/frontend/src/pages/front/WechatBot.tsx new file mode 100644 index 0000000..e7b0d08 --- /dev/null +++ b/frontend/src/pages/front/WechatBot.tsx @@ -0,0 +1,214 @@ +// @ts-nocheck +import React, { useState, useEffect, useCallback } from 'react'; +import { imUserAPI } from '@/api'; + +const WechatBot: React.FC = () => { + const [loading, setLoading] = useState(false); + const [status, setStatus] = useState({ connected: false, configured: false }); + const [qrcode, setQrcode] = useState(null); + const [polling, setPolling] = useState(false); + 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); + }; + + // 加载当前用户的连接状态 + const loadStatus = useCallback(async () => { + try { + const res = await imUserAPI.getStatus(); + if ((res as any)?.success) setStatus(res.data || { connected: false, configured: false }); + } catch (e) { + // 静默失败 + } + }, []); + + useEffect(() => { + loadStatus(); + }, [loadStatus]); + + // 获取扫码二维码（iLink 协议免 AppID） + const handleQrcode = async () => { + setLoading(true); + try { + const res = await imUserAPI.qrcode(); + if ((res as any)?.success && (res as any).data) { + setQrcode((res as any).data); + setPolling(true); + showToast('二维码已生成，请用手机微信扫码'); + } else { + showToast((res as any).error || '获取二维码失败', 'error'); + } + } catch (e) { + showToast('获取二维码失败', 'error'); + } finally { + setLoading(false); + } + }; + + // 轮询扫码状态 + const handlePoll = useCallback(async () => { + if (!qrcode?.qrcode) return; + try { + const res = await imUserAPI.qrcodeStatus(qrcode.qrcode); + if ((res as any)?.success && (res as any).data) { + const d = (res as any).data; + if (d.status === 'confirmed' || d.connected) { + setPolling(false); + setStatus({ ...status, connected: true, configured: true }); + showToast('个人微信已连接！'); + loadStatus(); + } else if (d.status === 'expired') { + setPolling(false); + showToast('二维码已过期，请重新获取', 'error'); + } + } + } catch (e) { + // 忽略轮询错误 + } + }, [qrcode, status, loadStatus]); + + useEffect(() => { + if (polling && qrcode?.qrcode) { + const t = setInterval(() => handlePoll(), 3000); + return () => clearInterval(t); + } + }, [polling, qrcode, handlePoll]); + + // 断开连接 + const handleDisconnect = async () => { + try { + await imUserAPI.disconnect(); + setPolling(false); + setQrcode(null); + setStatus({ connected: false, configured: false }); + showToast('已断开个人微信连接'); + loadStatus(); + } catch (e) { + showToast('断开失败', 'error'); + } + }; + + const formatTime = (t: string) => (t ? new Date(t).toLocaleString('zh-CN') : '-'); + + return ( +
+
+ {/* Header */} +
+

我的微信机器人

+

+ 绑定你自己的个人微信，由七虎AI 大脑自动应答。基于腾讯官方 iLink 协议，扫码授权即可，无需任何开放平台凭证。 +

+
+ + {/* Toast */} + {toast && ( +
{toast.msg}
+ )} + + {/* 状态条 */} +
+
+

个人微信（iLink 协议）

+

+ 直接对接腾讯官方 iLink Bot HTTP API，扫码授权后由七虎AI 自动回复。 +

+
+ + {status.connected ? '● 已连接' : status.configured ? '○ 已配置' : '○ 未绑定'} + +
+ + {/* 扫码绑定 */} +
+

扫码授权绑定

+ {status.connected ? ( +
已绑定，你的个人微信正在自动收发消息并由七虎AI 应答。
+ ) : !qrcode ? ( +
+
+ 点击「获取扫码二维码」，使用手机微信扫描二维码并确认授权，即可将你的个人微信接入七虎AI。整个过程无需申请任何开放平台凭证。 +
+ +
+ ) : ( +
+ {qrcode.qrcode_img_content ? ( + 微信扫码 + ) : ( +
+ 二维码数据缺失，请重新获取。 +
+ )} +
+ {polling ? '等待扫码中…（每 3 秒自动检测一次）' : '已停止轮询'} +
+
+ {polling && ( + + )} + +
+
+ )} +
+ + {/* 已连接信息 */} + {status.connected && ( +
+

连接信息

+
+ + + + +
+
+ )} + + {/* 说明 */} +
+

接入说明

+
    +
  • 每个账号独立绑定自己的个人微信，数据严格按账号隔离，互不影响
    • +
  • 个人微信通过腾讯官方 iLink Bot HTTP API（微信 ClawBot 插件）接入，消息经七虎AI 大脑生成回复
    • +
  • 采用官方合法协议，无需 Hook / 逆向，封号风险远低于历史方案，但仍请遵守微信平台规范
    • +
  • 消息收发基于长轮询（getupdates）+ 官方 sendmessage，回复自动关联对话窗口（context_token）
    • +
+
+
+
+ ); +}; + +const DetailItem: React.FC<{ label: string; value: any }> = ({ label, value }) => ( +
+
{label}
+
{value || '-'}
+
+); + +const primaryBtnStyle: React.CSSProperties = { + padding: '8px 20px', background: '#6C63FF', border: 'none', borderRadius: '6px', + color: '#fff', fontSize: '14px', fontWeight: 500, cursor: 'pointer', +}; + +const secondaryBtnStyle: React.CSSProperties = { + padding: '8px 20px', background: '#F0F2F8', border: '1px solid #F0F2F8', borderRadius: '6px', + color: '#1A1F36', fontSize: '14px', fontWeight: 500, cursor: 'pointer', +}; + +export default WechatBot;