Phase 4: IM Bot Integration - admin module, frontend, LangBot gateway, routing rules, quick setup wizard

backend/python/app/api/admin/im_bot_admin.py

@@ -0,0 +1,952 @@
+"""
+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.
+"""
+import uuid
+import json
+import hashlib
+import hmac
+import time
+from datetime import datetime, timedelta
+from typing import Optional
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy.orm import Session
+from sqlalchemy import text
+
+from ...database import SessionLocal
+from ...auth import get_current_user
+
+router = APIRouter()
+
+
+# ============================================================
+# Helper Functions
+# ============================================================
+
+def _execute_raw(db: Session, sql_text, params=None):
+    """Execute raw SQL and return results"""
+    result = db.execute(sql_text, params or {})
+    if result.returns_rows:
+        return result.fetchall()
+    db.commit()
+    return result
+
+
+def _get_platform_info(platform: str) -> dict:
+    """Get platform display info and config schema"""
+    platforms = {
+        "feishu": {
+            "name": "飞书",
+            "icon": "feishu",
+            "color": "#3370FF",
+            "credentials": ["app_id", "app_secret", "verification_token"],
+            "webhook_fields": ["encrypt_key"],
+            "docs_url": "https://open.feishu.cn/document/home/index",
+        },
+        "wechat": {
+            "name": "微信公众号",
+            "icon": "wechat",
+            "color": "#07C160",
+            "credentials": ["app_id", "app_secret", "token"],
+            "webhook_fields": ["encoding_aes_key"],
+            "docs_url": "https://developers.weixin.qq.com/doc/offiaccount/Getting_Started/Overview.html",
+        },
+        "personal_wechat": {
+            "name": "个人微信",
+            "icon": "wechat-personal",
+            "color": "#07C160",
+            "credentials": ["qr_code", "clawbot_token"],
+            "webhook_fields": [],
+            "docs_url": "",
+            "scan_auth": True,
+        },
+        "wecom": {
+            "name": "企业微信",
+            "icon": "wecom",
+            "color": "#2F80ED",
+            "credentials": ["corp_id", "agent_id", "secret", "token"],
+            "webhook_fields": ["encoding_aes_key"],
+            "docs_url": "https://developer.work.weixin.qq.com/document/",
+        },
+        "dingtalk": {
+            "name": "钉钉",
+            "icon": "dingtalk",
+            "color": "#0089FF",
+            "credentials": ["app_key", "app_secret"],
+            "webhook_fields": ["aes_key"],
+            "docs_url": "https://open.dingtalk.com/document/",
+        },
+        "telegram": {
+            "name": "Telegram",
+            "icon": "telegram",
+            "color": "#0088CC",
+            "credentials": ["bot_token"],
+            "webhook_fields": [],
+            "docs_url": "https://core.telegram.org/bots/api",
+        },
+        "whatsapp": {
+            "name": "WhatsApp",
+            "icon": "whatsapp",
+            "color": "#25D366",
+            "credentials": ["phone_number_id", "access_token"],
+            "webhook_fields": ["verify_token"],
+            "docs_url": "https://developers.facebook.com/docs/whatsapp",
+        },
+        "slack": {
+            "name": "Slack",
+            "icon": "slack",
+            "color": "#4A154B",
+            "credentials": ["bot_token", "signing_secret"],
+            "webhook_fields": [],
+            "docs_url": "https://api.slack.com/docs",
+        },
+    }
+    return platforms.get(platform, {"name": platform, "icon": "default", "color": "#666"})
+
+
+def _row_to_dict(row) -> dict:
+    """Convert SQLAlchemy row to 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
+
+
+def _rows_to_dict(rows) -> list:
+    """Convert multiple rows to list of dicts"""
+    return [_row_to_dict(r) for r in rows] if rows else []
+
+
+# ============================================================
+# Dashboard & Stats
+# ============================================================
+
+@router.get("/stats")
+async def get_im_bot_stats(user=Depends(get_current_user)):
+    """Get overall IM bot statistics"""
+    db = SessionLocal()
+    try:
+        # Total channels by platform
+        rows = _execute_raw(db, text("""
+            SELECT platform, COUNT(*) as count,
+                   SUM(CASE WHEN status='connected' THEN 1 ELSE 0 END) as active_count
+            FROM channels
+            WHERE is_active = 1
+            GROUP BY platform
+        """))
+        platform_stats = {}
+        total_channels = 0
+        total_active = 0
+        for r in rows:
+            d = _row_to_dict(r)
+            platform_stats[d['platform']] = d
+            total_channels += d['count']
+            total_active += d['active_count']
+
+        # Today's messages
+        today = datetime.now().strftime('%Y-%m-%d')
+        rows = _execute_raw(db, text("""
+            SELECT COALESCE(SUM(messages_in), 0) as msg_in,
+                   COALESCE(SUM(messages_out), 0) as msg_out,
+                   COALESCE(SUM(unique_users), 0) as users,
+                   COALESCE(SUM(error_count), 0) as errors
+            FROM im_bot_daily_stats
+            WHERE stat_date = :today
+        """), {"today": today})
+        today_stats = _row_to_dict(rows[0]) if rows else {"msg_in": 0, "msg_out": 0, "users": 0, "errors": 0}
+
+        # 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": {
+                "total_channels": total_channels,
+                "active_channels": total_active,
+                "platform_breakdown": platform_stats,
+                "today_messages_in": today_stats.get('msg_in', 0),
+                "today_messages_out": today_stats.get('msg_out', 0),
+                "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:
+        raise HTTPException(status_code=500, detail=f"Stats error: {str(e)}")
+    finally:
+        db.close()
+
+
+@router.get("/dashboard")
+async def get_dashboard(user=Depends(get_current_user)):
+    """Get dashboard data with charts"""
+    db = SessionLocal()
+    try:
+        # 7-day message trend
+        rows = _execute_raw(db, text("""
+            SELECT stat_date,
+                   SUM(messages_in) as msg_in,
+                   SUM(messages_out) as msg_out,
+                   SUM(unique_users) as users
+            FROM im_bot_daily_stats
+            WHERE stat_date >= DATE_SUB(CURDATE(), INTERVAL 7 DAY)
+            GROUP BY stat_date
+            ORDER BY stat_date
+        """))
+        trend = _rows_to_dict(rows)
+
+        # Channel list with latest stats
+        rows = _execute_raw(db, text("""
+            SELECT c.id, c.name, c.platform, c.status, c.is_active,
+                   c.last_active, c.default_agent_id,
+                   COALESCE(s.msg_count, 0) as msg_count
+            FROM channels c
+            LEFT JOIN (
+                SELECT channel_id, SUM(messages_in + messages_out) as msg_count
+                FROM im_bot_daily_stats
+                WHERE stat_date >= DATE_SUB(CURDATE(), INTERVAL 30 DAY)
+                GROUP BY channel_id
+            ) s ON c.id = s.channel_id
+            ORDER BY c.created_at DESC
+            LIMIT 20
+        """))
+        channels = []
+        for r in rows:
+            d = _row_to_dict(r)
+            d['platform_info'] = _get_platform_info(d['platform'])
+            channels.append(d)
+
+        # Platform distribution
+        rows = _execute_raw(db, text("""
+            SELECT platform, COUNT(*) as count
+            FROM channels WHERE is_active = 1
+            GROUP BY platform
+        """))
+        distribution = _rows_to_dict(rows)
+
+        return {
+            "success": True,
+            "data": {
+                "message_trend": trend,
+                "recent_channels": channels,
+                "platform_distribution": distribution,
+            }
+        }
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Dashboard error: {str(e)}")
+    finally:
+        db.close()
+
+
+# ============================================================
+# Channel Management
+# ============================================================
+
+@router.get("/channels")
+async def list_channels(
+    platform: Optional[str] = Query(None),
+    status: Optional[str] = Query(None),
+    page: int = Query(1, ge=1),
+    page_size: int = Query(20, ge=1, le=100),
+    user=Depends(get_current_user)
+):
+    """List all IM bot channels (admin view)"""
+    db = SessionLocal()
+    try:
+        where_clauses = ["1=1"]
+        params = {}
+        if platform:
+            where_clauses.append("c.platform = :platform")
+            params['platform'] = platform
+        if status:
+            where_clauses.append("c.status = :status")
+            params['status'] = status
+
+        where_sql = " AND ".join(where_clauses)
+        offset = (page - 1) * page_size
+
+        # Count
+        count_sql = f"SELECT COUNT(*) as cnt FROM channels c WHERE {where_sql}"
+        total = _execute_raw(db, text(count_sql), params)[0]._mapping['cnt']
+
+        # Query
+        sql = f"""
+            SELECT c.*, m.name as agent_name
+            FROM channels c
+            LEFT JOIN market_apps m ON c.default_agent_id = m.id
+            WHERE {where_sql}
+            ORDER BY c.created_at DESC
+            LIMIT :limit OFFSET :offset
+        """
+        params['limit'] = page_size
+        params['offset'] = offset
+        rows = _execute_raw(db, text(sql), params)
+
+        channels = []
+        for r in rows:
+            d = _row_to_dict(r)
+            d['platform_info'] = _get_platform_info(d['platform'])
+            channels.append(d)
+
+        return {
+            "success": True,
+            "data": {
+                "items": channels,
+                "total": total,
+                "page": page,
+                "page_size": page_size,
+            }
+        }
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"List error: {str(e)}")
+    finally:
+        db.close()
+
+
+@router.get("/channels/{channel_id}")
+async def get_channel_detail(channel_id: str, user=Depends(get_current_user)):
+    """Get channel detail with routing rules and recent messages"""
+    db = SessionLocal()
+    try:
+        # Channel
+        rows = _execute_raw(db, text("""
+            SELECT c.*, m.name as agent_name
+            FROM channels c
+            LEFT JOIN market_apps m ON c.default_agent_id = m.id
+            WHERE c.id = :cid
+        """), {"cid": channel_id})
+        if not rows:
+            raise HTTPException(status_code=404, detail="Channel not found")
+        channel = _row_to_dict(rows[0])
+        channel['platform_info'] = _get_platform_info(channel['platform'])
+
+        # Routing rules
+        routes = _execute_raw(db, text("""
+            SELECT r.*, m.name as agent_name
+            FROM im_bot_routes r
+            LEFT JOIN market_apps m ON r.agent_id = m.id
+            WHERE r.channel_id = :cid
+            ORDER BY r.priority DESC
+        """), {"cid": channel_id})
+        channel['routes'] = _rows_to_dict(routes)
+
+        # Recent messages (from channel_messages table)
+        msgs = _execute_raw(db, text("""
+            SELECT * FROM channel_messages
+            WHERE channel_id = :cid
+            ORDER BY created_at DESC
+            LIMIT 20
+        """), {"cid": channel_id})
+        channel['recent_messages'] = _rows_to_dict(msgs)
+
+        # 7-day stats
+        stats = _execute_raw(db, text("""
+            SELECT stat_date, messages_in, messages_out, unique_users, error_count
+            FROM im_bot_daily_stats
+            WHERE channel_id = :cid
+            AND stat_date >= DATE_SUB(CURDATE(), INTERVAL 7 DAY)
+            ORDER BY stat_date
+        """), {"cid": channel_id})
+        channel['stats_7d'] = _rows_to_dict(stats)
+
+        return {"success": True, "data": channel}
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Detail error: {str(e)}")
+    finally:
+        db.close()
+
+
+@router.post("/channels/{channel_id}/test")
+async def test_channel(channel_id: str, user=Depends(get_current_user)):
+    """Test channel connectivity"""
+    db = SessionLocal()
+    try:
+        rows = _execute_raw(db, text("SELECT * FROM channels WHERE id = :cid"), {"cid": channel_id})
+        if not rows:
+            raise HTTPException(status_code=404, detail="Channel not found")
+        channel = _row_to_dict(rows[0])
+
+        # Simulate connectivity test based on platform
+        platform = channel['platform']
+        creds = channel.get('credential', {})
+        if isinstance(creds, str):
+            creds = json.loads(creds) if creds else {}
+
+        result = {
+            "channel_id": channel_id,
+            "platform": platform,
+            "tested_at": datetime.now().isoformat(),
+            "checks": [],
+            "passed": True,
+        }
+
+        if platform == 'feishu':
+            result['checks'].append({"name": "App ID", "passed": bool(creds.get('app_id'))})
+            result['checks'].append({"name": "App Secret", "passed": bool(creds.get('app_secret'))})
+            result['checks'].append({"name": "Verification Token", "passed": bool(creds.get('verification_token'))})
+        elif platform in ('wechat', 'wecom'):
+            result['checks'].append({"name": "App ID", "passed": bool(creds.get('app_id') or creds.get('corp_id'))})
+            result['checks'].append({"name": "Secret", "passed": bool(creds.get('app_secret') or creds.get('secret'))})
+        elif platform == 'dingtalk':
+            result['checks'].append({"name": "App Key", "passed": bool(creds.get('app_key'))})
+            result['checks'].append({"name": "App Secret", "passed": bool(creds.get('app_secret'))})
+        elif platform == 'telegram':
+            result['checks'].append({"name": "Bot Token", "passed": bool(creds.get('bot_token'))})
+        else:
+            result['checks'].append({"name": "Credentials", "passed": len(creds) > 0})
+
+        result['passed'] = all(c['passed'] for c in result['checks'])
+        result['webhook_configured'] = bool(channel.get('webhook_url'))
+
+        return {"success": True, "data": result}
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Test error: {str(e)}")
+    finally:
+        db.close()
+
+
+@router.post("/channels/{channel_id}/activate")
+async def activate_channel(channel_id: str, user=Depends(get_current_user)):
+    """Activate a channel"""
+    db = SessionLocal()
+    try:
+        _execute_raw(db, text(
+            "UPDATE channels SET is_active=1, status='connected' WHERE id=:cid"
+        ), {"cid": channel_id})
+        return {"success": True, "message": "Channel activated"}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+    finally:
+        db.close()
+
+
+@router.post("/channels/{channel_id}/deactivate")
+async def deactivate_channel(channel_id: str, user=Depends(get_current_user)):
+    """Deactivate a channel"""
+    db = SessionLocal()
+    try:
+        _execute_raw(db, text(
+            "UPDATE channels SET is_active=0, status='disconnected' WHERE id=:cid"
+        ), {"cid": channel_id})
+        return {"success": True, "message": "Channel deactivated"}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+    finally:
+        db.close()
+
+
+# ============================================================
+# Routing Rules
+# ============================================================
+
+@router.get("/routes")
+async def list_routes(
+    channel_id: Optional[str] = Query(None),
+    user=Depends(get_current_user)
+):
+    """List routing rules"""
+    db = SessionLocal()
+    try:
+        where = "1=1"
+        params = {}
+        if channel_id:
+            where = "r.channel_id = :cid"
+            params['cid'] = channel_id
+
+        rows = _execute_raw(db, text(f"""
+            SELECT r.*, c.name as channel_name, c.platform,
+                   m.name as agent_name, m2.name as fallback_agent_name
+            FROM im_bot_routes r
+            LEFT JOIN channels c ON r.channel_id = c.id
+            LEFT JOIN market_apps m ON r.agent_id = m.id
+            LEFT JOIN market_apps m2 ON r.fallback_agent_id = m2.id
+            WHERE {where}
+            ORDER BY r.priority DESC, r.created_at DESC
+        """), params)
+        return {"success": True, "data": _rows_to_dict(rows)}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+    finally:
+        db.close()
+
+
+@router.post("/routes")
+async def create_route(
+    channel_id: str = Query(...),
+    agent_id: str = Query(...),
+    route_name: str = Query("default"),
+    match_type: str = Query("all"),
+    match_pattern: Optional[str] = Query(None),
+    priority: int = Query(0),
+    fallback_agent_id: Optional[str] = Query(None),
+    user=Depends(get_current_user)
+):
+    """Create a routing rule"""
+    db = SessionLocal()
+    try:
+        route_id = str(uuid.uuid4())
+        _execute_raw(db, text("""
+            INSERT INTO im_bot_routes
+            (id, channel_id, agent_id, route_name, match_type, match_pattern, priority, fallback_agent_id)
+            VALUES (:id, :cid, :aid, :name, :mt, :mp, :pri, :faid)
+        """), {
+            "id": route_id, "cid": channel_id, "aid": agent_id,
+            "name": route_name, "mt": match_type, "mp": match_pattern,
+            "pri": priority, "faid": fallback_agent_id,
+        })
+        return {"success": True, "data": {"id": route_id}}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+    finally:
+        db.close()
+
+
+@router.put("/routes/{route_id}")
+async def update_route(
+    route_id: str,
+    agent_id: Optional[str] = Query(None),
+    route_name: Optional[str] = Query(None),
+    match_type: Optional[str] = Query(None),
+    match_pattern: Optional[str] = Query(None),
+    priority: Optional[int] = Query(None),
+    fallback_agent_id: Optional[str] = Query(None),
+    is_active: Optional[int] = Query(None),
+    user=Depends(get_current_user)
+):
+    """Update a routing rule"""
+    db = SessionLocal()
+    try:
+        updates = []
+        params = {"id": route_id}
+        for field, val in [("agent_id", agent_id), ("route_name", route_name),
+                           ("match_type", match_type), ("match_pattern", match_pattern),
+                           ("priority", priority), ("fallback_agent_id", fallback_agent_id),
+                           ("is_active", is_active)]:
+            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_routes SET {', '.join(updates)} WHERE id = :id"
+        _execute_raw(db, text(sql), params)
+        return {"success": True, "message": "Route updated"}
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+    finally:
+        db.close()
+
+
+@router.delete("/routes/{route_id}")
+async def delete_route(route_id: str, user=Depends(get_current_user)):
+    """Delete a routing rule"""
+    db = SessionLocal()
+    try:
+        _execute_raw(db, text("DELETE FROM im_bot_routes WHERE id = :id"), {"id": route_id})
+        return {"success": True, "message": "Route deleted"}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+    finally:
+        db.close()
+
+
+# ============================================================
+# LangBot Gateway
+# ============================================================
+
+@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()
+
+
+@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,
+    }
+
+    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}"
+    except Exception as e:
+        result['error'] = str(e)
+
+    return {"success": True, "data": result}
+
+
+# ============================================================
+# Platforms & Webhooks
+# ============================================================
+
+@router.get("/platforms")
+async def list_platforms(user=Depends(get_current_user)):
+    """List all supported IM platforms with config schemas"""
+    platforms = []
+    for key in ["feishu", "wechat", "personal_wechat", "wecom", "dingtalk",
+                "telegram", "whatsapp", "slack"]:
+        info = _get_platform_info(key)
+        info['key'] = key
+        platforms.append(info)
+
+    # Count channels per platform
+    db = SessionLocal()
+    try:
+        rows = _execute_raw(db, text("""
+            SELECT platform, COUNT(*) as count,
+                   SUM(CASE WHEN status='connected' THEN 1 ELSE 0 END) as active
+            FROM channels
+            GROUP BY platform
+        """))
+        counts = {r._mapping['platform']: {"count": r._mapping['count'], "active": r._mapping['active']} for r in rows}
+        for p in platforms:
+            c = counts.get(p['key'], {"count": 0, "active": 0})
+            p['channel_count'] = c['count']
+            p['active_count'] = c['active']
+
+        return {"success": True, "data": platforms}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+    finally:
+        db.close()
+
+
+@router.get("/webhook-url/{platform}")
+async def get_webhook_url(platform: str, user=Depends(get_current_user)):
+    """Generate webhook URL for a platform"""
+    if platform not in ["feishu", "wechat", "personal_wechat", "wecom", "dingtalk",
+                         "telegram", "whatsapp", "slack"]:
+        raise HTTPException(status_code=400, detail=f"Unsupported platform: {platform}")
+
+    # Generate a webhook secret for verification
+    secret = hashlib.sha256(f"{platform}:{time.time()}".encode()).hexdigest()[:32]
+    webhook_url = f"/app/api/channels/webhook/{platform}"
+
+    return {
+        "success": True,
+        "data": {
+            "platform": platform,
+            "platform_name": _get_platform_info(platform)['name'],
+            "webhook_url": webhook_url,
+            "full_url": f"http://210.16.161.218:12888{webhook_url}",
+            "secret": secret,
+            "instructions": _get_webhook_instructions(platform),
+        }
+    }
+
+
+def _get_webhook_instructions(platform: str) -> list:
+    """Get setup instructions for each platform"""
+    guides = {
+        "feishu": [
+            "1. 登录飞书开放平台，创建企业自建应用",
+            "2. 在「事件订阅」页面配置请求地址为上述 Webhook URL",
+            "3. 填入 App ID、App Secret 和 Verification Token",
+            "4. 订阅消息事件: im.message.receive_v1",
+            "5. 发布应用版本并等待管理员审批",
+        ],
+        "wechat": [
+            "1. 登录微信公众平台，进入「设置与开发」",
+            "2. 在「基本配置」中填写服务器配置",
+            "3. URL 填入上述 Webhook URL",
+            "4. 填入 Token 和 EncodingAESKey",
+            "5. 启用服务器配置",
+        ],
+        "personal_wechat": [
+            "1. 点击「扫码授权」按钮，使用手机微信扫码",
+            "2. 等待授权完成后，频道状态将变为「已连接」",
+            "3. 个人微信通过 ClawBot 协议接入，无需 Webhook 配置",
+            "4. 注意：个人微信接入存在封号风险，请谨慎使用",
+        ],
+        "wecom": [
+            "1. 登录企业微信管理后台，创建自建应用",
+            "2. 在「接收消息」页面设置 API 接收",
+            "3. URL 填入上述 Webhook URL",
+            "4. 填入 Corp ID、Agent ID、Secret、Token 和 EncodingAESKey",
+            "5. 设置可信 IP 白名单",
+        ],
+        "dingtalk": [
+            "1. 登录钉钉开放平台，创建企业内部应用",
+            "2. 在「事件订阅」中配置 HTTP 模式",
+            "3. 请求地址填入上述 Webhook URL",
+            "4. 填入 App Key 和 App Secret",
+            "5. 订阅消息接收事件",
+        ],
+        "telegram": [
+            "1. 通过 @BotFather 创建 Bot，获取 Bot Token",
+            "2. 调用 Telegram API 设置 Webhook: setWebhook?url=<上述URL>",
+            "3. 填入 Bot Token",
+        ],
+        "whatsapp": [
+            "1. 在 Meta for Developers 创建 WhatsApp Business 应用",
+            "2. 配置 Webhook 验证 Token",
+            "3. Webhook URL 填入上述地址",
+            "4. 订阅 messages 事件",
+            "5. 填入 Phone Number ID 和 Access Token",
+        ],
+        "slack": [
+            "1. 在 api.slack.com 创建 Slack App",
+            "2. 启用 Event Subscriptions",
+            "3. Request URL 填入上述 Webhook URL",
+            "4. 订阅 message.im 事件",
+            "5. 填入 Bot Token 和 Signing Secret",
+        ],
+    }
+    return guides.get(platform, ["请参考平台官方文档进行配置"])
+
+
+# ============================================================
+# Messages & Logs
+# ============================================================
+
+@router.get("/messages")
+async def list_messages(
+    channel_id: Optional[str] = Query(None),
+    direction: Optional[str] = Query(None),
+    page: int = Query(1, ge=1),
+    page_size: int = Query(50, ge=1, le=200),
+    user=Depends(get_current_user)
+):
+    """List messages across all channels"""
+    db = SessionLocal()
+    try:
+        where = ["1=1"]
+        params = {}
+        if channel_id:
+            where.append("channel_id = :cid")
+            params['cid'] = channel_id
+        if direction:
+            where.append("direction = :dir")
+            params['dir'] = direction
+
+        where_sql = " AND ".join(where)
+        offset = (page - 1) * page_size
+
+        count = _execute_raw(db, text(f"SELECT COUNT(*) as cnt FROM channel_messages WHERE {where_sql}"), params)[0]._mapping['cnt']
+
+        rows = _execute_raw(db, text(f"""
+            SELECT cm.*, c.name as channel_name, c.platform
+            FROM channel_messages cm
+            LEFT JOIN channels c ON cm.channel_id = c.id
+            WHERE {where_sql}
+            ORDER BY cm.created_at DESC
+            LIMIT :limit OFFSET :offset
+        """), {**params, "limit": page_size, "offset": offset})
+
+        messages = []
+        for r in rows:
+            d = _row_to_dict(r)
+            d['platform_info'] = _get_platform_info(d.get('platform', '')) if d.get('platform') else None
+            messages.append(d)
+
+        return {
+            "success": True,
+            "data": {
+                "items": messages,
+                "total": count,
+                "page": page,
+                "page_size": page_size,
+            }
+        }
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+    finally:
+        db.close()
+
+
+@router.get("/logs")
+async def get_activity_logs(
+    channel_id: Optional[str] = Query(None),
+    limit: int = Query(100, ge=1, le=500),
+    user=Depends(get_current_user)
+):
+    """Get recent activity logs"""
+    db = SessionLocal()
+    try:
+        where = ""
+        params = {"limit": limit}
+        if channel_id:
+            where = "WHERE channel_id = :cid"
+            params['cid'] = channel_id
+
+        rows = _execute_raw(db, text(f"""
+            SELECT stat_date, channel_id,
+                   messages_in, messages_out, unique_users,
+                   avg_response_time_ms, error_count, tokens_used
+            FROM im_bot_daily_stats
+            {where}
+            ORDER BY stat_date DESC, channel_id
+            LIMIT :limit
+        """), params)
+
+        return {"success": True, "data": _rows_to_dict(rows)}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+    finally:
+        db.close()
+
+
+# ============================================================
+# Quick Setup Wizard
+# ============================================================
+
+@router.post("/quick-setup")
+async def quick_setup(
+    platform: str = Query(...),
+    name: str = Query(...),
+    agent_id: Optional[str] = Query(None),
+    credentials: str = Query(None),
+    user=Depends(get_current_user)
+):
+    """Quick setup wizard - create and configure a channel in one step"""
+    db = SessionLocal()
+    try:
+        if platform not in ["feishu", "wechat", "personal_wechat", "wecom", "dingtalk",
+                            "telegram", "whatsapp", "slack"]:
+            raise HTTPException(status_code=400, detail=f"Unsupported platform: {platform}")
+
+        channel_id = str(uuid.uuid4())
+        webhook_url = f"/app/api/channels/webhook/{platform}"
+
+        cred_json = None
+        if credentials:
+            try:
+                cred_json = json.loads(credentials)
+            except:
+                cred_json = {"raw": credentials}
+
+        _execute_raw(db, text("""
+            INSERT INTO channels (id, owner_id, name, platform, webhook_url,
+                                  credential, default_agent_id, status, is_active)
+            VALUES (:id, :owner, :name, :platform, :webhook,
+                    :cred, :agent, 'pending', 1)
+        """), {
+            "id": channel_id,
+            "owner": "admin-001",
+            "name": name,
+            "platform": platform,
+            "webhook": webhook_url,
+            "cred": json.dumps(cred_json) if cred_json else None,
+            "agent": agent_id,
+        })
+
+        # Create default routing rule if agent specified
+        if agent_id:
+            route_id = str(uuid.uuid4())
+            _execute_raw(db, text("""
+                INSERT INTO im_bot_routes (id, channel_id, agent_id, route_name, match_type, priority)
+                VALUES (:rid, :cid, :aid, 'default', 'all', 0)
+            """), {"rid": route_id, "cid": channel_id, "aid": agent_id})
+
+        platform_info = _get_platform_info(platform)
+
+        return {
+            "success": True,
+            "data": {
+                "channel_id": channel_id,
+                "webhook_url": f"http://210.16.161.218:12888{webhook_url}",
+                "platform": platform,
+                "platform_name": platform_info['name'],
+                "instructions": _get_webhook_instructions(platform),
+                "next_steps": "请按照上述说明在平台配置 Webhook，然后返回此页面点击「测试连接」",
+            }
+        }
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Quick setup error: {str(e)}")
+    finally:
+        db.close()

backend/python/main.py

@@ -34,7 +34,7 @@ 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
+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
 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
@@ -434,6 +434,7 @@ app.include_router(mcp_market_router, prefix="/app/api/admin/mcp-market", tags=[
 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"])
 
 
 

frontend/src/App.tsx

@@ -39,6 +39,7 @@ import AdminAgents from '@/pages/admin/Agents';
 import AdminCodeExec from '@/pages/admin/CodeExec';;
 import DistributionBuilder from '@/pages/admin/DistributionBuilder';
 import PrivacySettings from '@/pages/admin/PrivacySettings';
+import ImBots from '@/pages/admin/ImBots';
 
 const App: React.FC = () => {
   const [user, setUser] = useState<User | null>(null);
@@ -120,6 +121,7 @@ const App: React.FC = () => {
           <Route path="agents" element={<AdminAgents />} />
           <Route path="code-exec" element={<AdminCodeExec />} />
           <Route path="distributions" element={<DistributionBuilder />} />
+          <Route path="im-bots" element={<ImBots />} />
           <Route path="privacy" element={<PrivacySettings />} />
         </Route>
 

frontend/src/api/index.ts

@@ -814,4 +814,23 @@ export const evolutionAPI = {
   getPrivacyUsageStats: () => get<any>('/admin/privacy/usage/stats'),
   getBridgeConfig: () => get<any>('/admin/privacy/bridge-config'),
 
+  // IM Bot Management (Phase 4)
+  getImBotStats: () => get('/admin/im-bot/stats'),
+  getImBotDashboard: () => get('/admin/im-bot/dashboard'),
+  getImBotChannels: (params?: string) => get(`/admin/im-bot/channels${params || ''}`),
+  getImBotChannelDetail: (id: string) => get(`/admin/im-bot/channels/${id}`),
+  testImBotChannel: (id: string) => post(`/admin/im-bot/channels/${id}/test`),
+  activateImBotChannel: (id: string) => post(`/admin/im-bot/channels/${id}/activate`),
+  deactivateImBotChannel: (id: string) => post(`/admin/im-bot/channels/${id}/deactivate`),
+  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}`),
+  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 || ''}`),
+  imBotQuickSetup: (params: string) => post(`/admin/im-bot/quick-setup${params}`),
+
 };

frontend/src/components/layout/AdminLayout.tsx

@@ -20,6 +20,7 @@ const adminNavItems = [
   { path: '/admin/code-exec', icon: 'code', label: '代码执行沙箱' },
   { path: '/admin/distributions', icon: 'distro', label: '发行版构建' },
   { path: '/admin/privacy', icon: 'privacy', label: '隐私与本地代理' },
+    { path: '/admin/im-bots', icon: 'imbot', label: 'IM 机器人' },
 ];
 
 const adminIconPaths: Record<string, JSX.Element> = {

migrations/mig_p4_imbot.sql

@@ -0,0 +1,66 @@
+-- Phase 4: IM Bot Integration - Migration SQL
+-- Creates: im_bot_routes, im_bot_langbot_config, im_bot_daily_stats
+-- Alters: channels.platform ENUM to add personal_wechat, wecom
+
+-- 1. Extend channels platform ENUM to include personal_wechat and wecom
+ALTER TABLE channels
+MODIFY COLUMN platform ENUM(
+    'feishu', 'wechat', 'dingtalk', 'whatsapp', 'telegram', 'slack',
+    'personal_wechat', 'wecom'
+) NOT NULL;
+
+-- 2. IM Bot Routing Rules - map channels to agents with filters
+CREATE TABLE IF NOT EXISTS im_bot_routes (
+    id VARCHAR(36) NOT NULL PRIMARY KEY,
+    channel_id VARCHAR(36) NOT NULL COMMENT 'FK to channels.id',
+    agent_id VARCHAR(64) DEFAULT NULL COMMENT 'FK to market_apps.id (VARCHAR(36))',
+    route_name VARCHAR(100) NOT NULL DEFAULT 'default',
+    match_type ENUM('all', 'keyword', 'regex', 'sender') NOT NULL DEFAULT 'all',
+    match_pattern TEXT DEFAULT NULL COMMENT 'Keyword list or regex pattern',
+    priority INT NOT NULL DEFAULT 0 COMMENT 'Higher = matched first',
+    fallback_agent_id VARCHAR(64) DEFAULT NULL,
+    is_active TINYINT(1) NOT NULL DEFAULT 1,
+    created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+    INDEX idx_route_channel (channel_id),
+    INDEX idx_route_agent (agent_id),
+    INDEX idx_route_active (is_active, priority)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
+
+-- 3. LangBot Gateway Configuration
+CREATE TABLE IF NOT EXISTS im_bot_langbot_config (
+    id INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
+    name VARCHAR(100) NOT NULL DEFAULT 'default',
+    gateway_url VARCHAR(500) NOT NULL COMMENT 'LangBot API base URL',
+    api_key VARCHAR(255) DEFAULT NULL COMMENT 'LangBot REST API key',
+    webhook_secret VARCHAR(255) DEFAULT NULL COMMENT 'HMAC-SHA256 signing secret',
+    bot_uuid VARCHAR(100) DEFAULT NULL COMMENT 'LangBot HTTP Bot UUID',
+    callback_url VARCHAR(500) DEFAULT NULL COMMENT 'Our callback URL for LangBot responses',
+    sync_channels TINYINT(1) NOT NULL DEFAULT 0 COMMENT 'Auto-sync channels to LangBot',
+    status ENUM('active', 'inactive', 'error') NOT NULL DEFAULT 'inactive',
+    last_sync_at DATETIME DEFAULT NULL,
+    last_error TEXT DEFAULT NULL,
+    config_json JSON DEFAULT NULL COMMENT 'Additional LangBot configuration',
+    created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
+
+-- 4. Daily Statistics per Channel
+CREATE TABLE IF NOT EXISTS im_bot_daily_stats (
+    id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
+    channel_id VARCHAR(36) NOT NULL,
+    stat_date DATE NOT NULL,
+    messages_in INT NOT NULL DEFAULT 0,
+    messages_out INT NOT NULL DEFAULT 0,
+    unique_users INT NOT NULL DEFAULT 0,
+    avg_response_time_ms INT NOT NULL DEFAULT 0,
+    error_count INT NOT NULL DEFAULT 0,
+    tokens_used INT NOT NULL DEFAULT 0,
+    UNIQUE KEY uk_channel_date (channel_id, stat_date),
+    INDEX idx_stat_date (stat_date)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
+
+-- 5. Insert default LangBot config
+INSERT INTO im_bot_langbot_config (name, gateway_url, status)
+VALUES ('default', 'http://localhost:5300', 'inactive')
+ON DUPLICATE KEY UPDATE name = name;