qihu-bot
ab11283d92
CI / gate (push) Has been cancelled
1.mirror_to_ai_chat: _uuid→uuid(NameError被except吞了, 对话从未同步到webclaw) 2.im_bot_admin加_beijing_now并替换5处datetime.now()(容器UTC存naive时间前端按+08:00展示错位8小时)
1122 lines
40 KiB
Python
1122 lines
40 KiB
Python
"""
|
||
Phase 4: IM Bot Integration - Admin API Module
|
||
im_bot_admin.py
|
||
|
||
Provides admin management for IM bot channels, routing rules,
|
||
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, timezone
|
||
|
||
# #976 全平台统一使用北京时间（容器 UTC，前端按 +08:00 展示）
|
||
_TZ_BEIJING = timezone(timedelta(hours=8))
|
||
|
||
|
||
def _beijing_now():
|
||
"""返回 naive 北京时间（容器 UTC 时区下，需显式转）"""
|
||
return datetime.now(_TZ_BEIJING).replace(tzinfo=None)
|
||
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=true"))[0]._mapping['cnt']
|
||
|
||
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,
|
||
}
|
||
}
|
||
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 >= CURRENT_DATE - 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 >= CURRENT_DATE - 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 >= CURRENT_DATE - 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=true, 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=false, 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()
|
||
|
||
|
||
# ============================================================
|
||
# 个人微信 (iLink 协议 / 微信 ClawBot 插件) 原生适配器
|
||
# 直接对接腾讯官方 iLink Bot HTTP API (ilinkai.weixin.qq.com)
|
||
# 扫码登录免 AppID；消息收发走 Bearer bot_token + X-WECHAT-UIN
|
||
# ============================================================
|
||
|
||
ILINK_BASE = "https://ilinkai.weixin.qq.com"
|
||
_wechat_receiver_threads = {}
|
||
|
||
|
||
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:
|
||
from app.api.im_llm 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()
|
||
|
||
|
||
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": _beijing_now(), "created_at": _beijing_now(), "updated_at": _beijing_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": _beijing_now(), "updated_at": _beijing_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": _beijing_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()
|
||
|
||
|
||
# ============================================================
|
||
# 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()
|