SOONet实战教程：Python API接入企业内部IM（钉钉/企微）实现Bot式查询

无畏道人

475人浏览 · 2026-03-09 06:52:58

无畏道人 · 2026-03-09 06:52:58 发布

SOONet实战教程：Python API接入企业内部IM（钉钉/企微）实现Bot式查询

你是不是经常遇到这种情况？团队里有人问：“上周三下午开会那段视频，谁记得具体讨论了什么？”或者“培训视频里讲操作流程的那段，能帮我截出来吗？”面对几小时长的视频，手动查找特定片段就像大海捞针，费时费力。

现在，有了SOONet，你可以用一句话就让AI帮你精准定位。更棒的是，我们可以把它接入企业内部通讯工具，让整个团队都能轻松使用。今天，我就带你一步步实现这个功能。

1. 项目简介：SOONet是什么？

SOONet是一个基于自然语言的长视频时序片段定位系统。简单说，你给它一段视频和一句描述（比如“一个人从冰箱里拿食物”），它就能快速找出视频中符合描述的时间段。

1.1 核心优势

速度快得惊人：相比传统方法，推理速度提升14.6到102.8倍。处理1小时视频，传统方法可能要几分钟甚至更久，SOONet可能只需要几十秒。
定位很精准：在MAD和Ego4D这两个权威数据集上，准确度达到了业界领先水平。
能处理长视频：专门为小时级的长视频设计，不用担心视频太长处理不了。
用起来简单：直接用自然语言描述你想找什么，不需要复杂的配置或专业术语。

1.2 它能做什么？

想象一下这些场景：

从2小时的会议录像中，快速找到“讨论预算”的那10分钟
在培训视频里，定位“安装步骤演示”的部分
监控视频中，查找“有人进入区域”的时段
教学视频中，找到“讲解重点公式”的片段

这些原本需要人工反复观看、标记的繁琐工作，现在一句话就能搞定。

2. 环境准备与快速部署

在开始接入IM之前，我们需要先把SOONet服务跑起来。别担心，步骤很简单。

2.1 基础环境检查

首先确认你的环境是否符合要求：

# 检查Python版本
python --version
# 应该显示 Python 3.7 或更高版本

# 检查GPU（如果有的话）
nvidia-smi
# 能看到GPU信息最好，没有GPU也能运行，只是会慢一些

2.2 一键启动SOONet服务

SOONet已经预置了所有需要的文件，启动非常简单：

# 进入工作目录
cd /root/multi-modal_soonet_video-temporal-grounding

# 启动服务
python app.py

看到类似下面的输出，就说明服务启动成功了：

Running on local URL:  http://0.0.0.0:7860

2.3 验证服务是否正常

打开浏览器，访问 http://你的服务器IP:7860，应该能看到SOONet的Web界面。

上传一个测试视频（系统自带了一个测试视频在 /root/ai-models/iic/multi-modal_soonet_video-temporal-grounding/ 目录下），输入描述文字测试一下功能是否正常。

3. Python API基础使用

在接入IM之前，我们先熟悉一下如何用Python调用SOONet。

3.1 最简单的调用方式

# soonet_basic.py
from modelscope.pipelines import pipeline
from modelscope.utils.constant import Tasks

def basic_soonet_query():
    """
    最基本的SOONet调用示例
    """
    # 初始化pipeline
    soonet_pipeline = pipeline(
        Tasks.video_temporal_grounding,
        model='/root/ai-models/iic/multi-modal_soonet_video-temporal-grounding'
    )
    
    # 准备输入
    input_text = "a person is cooking in the kitchen"  # 查询文本
    input_video = "/path/to/your/video.mp4"  # 视频文件路径
    
    # 执行查询
    result = soonet_pipeline((input_text, input_video))
    
    # 解析结果
    print("查询结果：")
    for i, (score, timestamp) in enumerate(zip(result['scores'], result['timestamps'])):
        start_time = timestamp[0]
        end_time = timestamp[1]
        print(f"片段 {i+1}: {start_time:.2f}s - {end_time:.2f}s, 置信度: {score:.3f}")
    
    return result

if __name__ == "__main__":
    basic_soonet_query()

3.2 封装成可重用的服务类

为了后续接入IM更方便，我们先把SOONet功能封装一下：

# soonet_service.py
import os
from modelscope.pipelines import pipeline
from modelscope.utils.constant import Tasks

class SOONetService:
    def __init__(self, model_path=None):
        """
        初始化SOONet服务
        """
        if model_path is None:
            model_path = '/root/ai-models/iic/multi-modal_soonet_video-temporal-grounding'
        
        self.model_path = model_path
        self.pipeline = None
        self._init_pipeline()
    
    def _init_pipeline(self):
        """初始化模型pipeline"""
        try:
            self.pipeline = pipeline(
                Tasks.video_temporal_grounding,
                model=self.model_path
            )
            print("SOONet服务初始化成功")
        except Exception as e:
            print(f"初始化失败: {e}")
            raise
    
    def query_video(self, text_query, video_path, top_k=3):
        """
        查询视频片段
        
        参数：
        - text_query: 文本描述（英文）
        - video_path: 视频文件路径
        - top_k: 返回前几个最相关的结果
        
        返回：
        - 包含时间戳和置信度的列表
        """
        if not os.path.exists(video_path):
            return {"error": f"视频文件不存在: {video_path}"}
        
        try:
            # 执行查询
            result = self.pipeline((text_query, video_path))
            
            # 提取top_k个结果
            scores = result['scores'][:top_k]
            timestamps = result['timestamps'][:top_k]
            
            # 格式化结果
            formatted_results = []
            for i, (score, timestamp) in enumerate(zip(scores, timestamps)):
                formatted_results.append({
                    "rank": i + 1,
                    "start_time": float(timestamp[0]),
                    "end_time": float(timestamp[1]),
                    "duration": float(timestamp[1] - timestamp[0]),
                    "confidence": float(score),
                    "time_range": f"{timestamp[0]:.1f}s - {timestamp[1]:.1f}s"
                })
            
            return {
                "success": True,
                "query": text_query,
                "video": os.path.basename(video_path),
                "results": formatted_results
            }
            
        except Exception as e:
            return {"error": f"查询失败: {str(e)}"}
    
    def format_results_for_display(self, result):
        """将结果格式化为易读的文本"""
        if "error" in result:
            return f"查询失败: {result['error']}"
        
        output = []
        output.append(f"查询: {result['query']}")
        output.append(f"视频: {result['video']}")
        output.append("找到的片段:")
        
        for item in result['results']:
            output.append(
                f"{item['rank']}. {item['time_range']} "
                f"(时长: {item['duration']:.1f}s, 置信度: {item['confidence']:.3f})"
            )
        
        return "\n".join(output)

# 测试一下
if __name__ == "__main__":
    service = SOONetService()
    
    # 使用测试视频
    test_video = "/root/ai-models/iic/multi-modal_soonet_video-temporal-grounding/soonet_video_temporal_grounding_test_video.mp4"
    
    # 测试查询
    result = service.query_video(
        text_query="a man takes food out of the refrigerator",
        video_path=test_video,
        top_k=2
    )
    
    print(service.format_results_for_display(result))

4. 接入钉钉机器人

现在进入正题，我们来把SOONet接入钉钉群聊机器人。

4.1 创建钉钉机器人

打开钉钉群设置：在你要接入的群里，点击右上角设置
选择智能群助手：找到"智能群助手"选项
添加机器人：选择"自定义机器人"
配置机器人：
- 给机器人起个名字，比如"视频搜索助手"
- 设置安全设置，建议使用"加签"
- 复制webhook地址和加签密钥

4.2 钉钉消息处理服务

# dingtalk_bot.py
import json
import time
import hashlib
import base64
import hmac
import requests
from flask import Flask, request, jsonify
from soonet_service import SOONetService

app = Flask(__name__)

# 配置信息
DINGTALK_WEBHOOK = "你的webhook地址"
DINGTALK_SECRET = "你的加签密钥"
SOONET_SERVICE = SOONetService()

# 支持的视频目录（安全考虑，限制可访问的视频路径）
ALLOWED_VIDEO_PATHS = {
    "meeting": "/data/videos/meetings/",
    "training": "/data/videos/trainings/",
    "monitor": "/data/videos/monitor/"
}

def generate_dingtalk_signature(secret):
    """生成钉钉加签"""
    timestamp = str(round(time.time() * 1000))
    string_to_sign = f'{timestamp}\n{secret}'
    hmac_code = hmac.new(
        secret.encode('utf-8'), 
        string_to_sign.encode('utf-8'), 
        digestmod=hashlib.sha256
    ).digest()
    sign = base64.b64encode(hmac_code).decode('utf-8')
    return timestamp, sign

def send_dingtalk_message(content, at_users=None):
    """发送消息到钉钉"""
    timestamp, sign = generate_dingtalk_signature(DINGTALK_SECRET)
    
    headers = {'Content-Type': 'application/json'}
    webhook_url = f"{DINGTALK_WEBHOOK}&timestamp={timestamp}&sign={sign}"
    
    message = {
        "msgtype": "text",
        "text": {
            "content": content
        }
    }
    
    if at_users:
        message["at"] = {
            "atUserIds": at_users,
            "isAtAll": False
        }
    
    try:
        response = requests.post(webhook_url, headers=headers, data=json.dumps(message))
        return response.json()
    except Exception as e:
        print(f"发送钉钉消息失败: {e}")
        return None

def parse_user_command(text):
    """解析用户命令
    
    命令格式：搜索 [视频类型] [描述文字]
    示例：搜索 meeting 讨论预算的部分
    """
    text = text.strip()
    if not text.startswith("搜索"):
        return None
    
    parts = text[2:].strip().split(" ", 1)  # 去掉"搜索"，按第一个空格分割
    if len(parts) < 2:
        return None
    
    video_type = parts[0].lower()
    query_text = parts[1]
    
    # 翻译中文查询为英文（简单版本）
    # 实际使用时可以接入翻译API
    english_query = translate_chinese_to_english(query_text)
    
    return {
        "video_type": video_type,
        "original_query": query_text,
        "english_query": english_query
    }

def translate_chinese_to_english(text):
    """简单的中文到英文翻译（示例）"""
    # 这里可以接入真正的翻译服务，如百度翻译、谷歌翻译等
    # 为了简化，这里只处理一些常见词汇
    translations = {
        "会议": "meeting",
        "讨论": "discuss",
        "预算": "budget",
        "培训": "training",
        "演示": "demonstration",
        "操作": "operation",
        "步骤": "steps",
        "监控": "monitor",
        "进入": "enter",
        "离开": "leave"
    }
    
    # 简单的关键词替换
    english_text = text
    for chinese, english in translations.items():
        english_text = english_text.replace(chinese, english)
    
    return english_text

def find_video_file(video_type, query_hint=None):
    """根据类型查找视频文件"""
    if video_type not in ALLOWED_VIDEO_PATHS:
        return None
    
    video_dir = ALLOWED_VIDEO_PATHS[video_type]
    
    # 这里可以根据query_hint智能选择视频文件
    # 例如：根据日期、项目名称等
    # 简化版本：返回目录下的第一个mp4文件
    
    import glob
    video_files = glob.glob(f"{video_dir}/*.mp4")
    
    if video_files:
        return video_files[0]  # 返回第一个文件
    return None

@app.route('/dingtalk/webhook', methods=['POST'])
def handle_dingtalk_message():
    """处理钉钉机器人消息"""
    data = request.json
    
    # 提取消息内容
    msg_content = data.get('text', {}).get('content', '').strip()
    sender_id = data.get('senderStaffId', '')
    
    print(f"收到消息: {msg_content} from {sender_id}")
    
    # 解析命令
    command = parse_user_command(msg_content)
    
    if not command:
        response_text = "命令格式不正确。请使用：搜索 [视频类型] [描述]\n例如：搜索 meeting 讨论预算"
        send_dingtalk_message(response_text, [sender_id])
        return jsonify({"code": 0})
    
    # 查找视频文件
    video_path = find_video_file(command["video_type"], command["original_query"])
    
    if not video_path:
        response_text = f"未找到{command['video_type']}类型的视频文件"
        send_dingtalk_message(response_text, [sender_id])
        return jsonify({"code": 0})
    
    # 执行SOONet查询
    try:
        result = SOONET_SERVICE.query_video(
            text_query=command["english_query"],
            video_path=video_path,
            top_k=3
        )
        
        if "error" in result:
            response_text = f"查询失败: {result['error']}"
        else:
            # 格式化结果
            formatted = SOONET_SERVICE.format_results_for_display(result)
            response_text = f"查询结果：\n{formatted}"
        
    except Exception as e:
        response_text = f"处理查询时出错: {str(e)}"
    
    # 发送回复
    send_dingtalk_message(response_text, [sender_id])
    
    return jsonify({"code": 0})

def start_dingtalk_bot():
    """启动钉钉机器人服务"""
    print("启动钉钉机器人服务...")
    print(f"服务地址: http://localhost:5000/dingtalk/webhook")
    print("请将此地址配置到钉钉机器人webhook中")
    
    app.run(host='0.0.0.0', port=5000, debug=False)

if __name__ == "__main__":
    start_dingtalk_bot()

4.3 配置和运行

安装依赖：

pip install flask requests

配置钉钉信息：在代码中替换 DINGTALK_WEBHOOK 和 DINGTALK_SECRET
准备视频目录：按照 ALLOWED_VIDEO_PATHS 中的配置创建目录并放入视频文件
启动服务：

python dingtalk_bot.py

在钉钉中测试：在群里发送：搜索 meeting 讨论预算

5. 接入企业微信机器人

如果你团队用的是企业微信，接入方式也很类似。

5.1 创建企业微信机器人

打开企业微信群：点击右上角群设置
添加机器人：选择"添加群机器人"
创建新机器人：设置名称和头像
获取webhook地址：复制生成的webhook URL

5.2 企业微信消息处理服务

# wecom_bot.py
import json
import requests
from flask import Flask, request, jsonify
from soonet_service import SOONetService

app = Flask(__name__)

# 配置信息
WECOM_WEBHOOK = "你的企业微信webhook地址"
SOONET_SERVICE = SOONetService()

# 视频目录配置（与钉钉版本相同）
ALLOWED_VIDEO_PATHS = {
    "会议": "/data/videos/meetings/",
    "培训": "/data/videos/trainings/",
    "监控": "/data/videos/monitor/"
}

def send_wecom_message(content, mentioned_list=None):
    """发送消息到企业微信"""
    headers = {'Content-Type': 'application/json'}
    
    message = {
        "msgtype": "text",
        "text": {
            "content": content
        }
    }
    
    if mentioned_list:
        message["text"]["mentioned_list"] = mentioned_list
    
    try:
        response = requests.post(WECOM_WEBHOOK, headers=headers, data=json.dumps(message))
        return response.json()
    except Exception as e:
        print(f"发送企业微信消息失败: {e}")
        return None

def parse_wecom_command(text):
    """解析企业微信命令
    
    支持两种格式：
    1. @机器人 搜索 [类型] [描述]
    2. 搜索 [类型] [描述]
    """
    # 移除@提及
    if "@" in text:
        parts = text.split(" ", 1)
        if len(parts) > 1:
            text = parts[1]
    
    text = text.strip()
    if not text.startswith("搜索"):
        return None
    
    parts = text[2:].strip().split(" ", 1)
    if len(parts) < 2:
        return None
    
    video_type = parts[0]
    query_text = parts[1]
    
    # 中文查询直接使用（SOONet支持部分中文）
    # 如果需要更好的效果，可以添加翻译功能
    
    return {
        "video_type": video_type,
        "query": query_text
    }

@app.route('/wecom/webhook', methods=['POST'])
def handle_wecom_message():
    """处理企业微信机器人消息"""
    data = request.json
    
    # 企业微信的消息格式
    msg_content = data.get('text', '').strip()
    sender_name = data.get('sender', '')
    
    print(f"收到企业微信消息: {msg_content} from {sender_name}")
    
    # 解析命令
    command = parse_wecom_command(msg_content)
    
    if not command:
        response_text = "命令格式：搜索 [类型] [描述]\n类型：会议、培训、监控\n示例：搜索 会议 讨论项目进度"
        send_wecom_message(response_text, [sender_name])
        return jsonify({"code": 0})
    
    # 查找视频文件
    import glob
    video_dir = ALLOWED_VIDEO_PATHS.get(command["video_type"])
    
    if not video_dir:
        response_text = f"不支持的类型：{command['video_type']}\n支持的类型：{', '.join(ALLOWED_VIDEO_PATHS.keys())}"
        send_wecom_message(response_text, [sender_name])
        return jsonify({"code": 0})
    
    video_files = glob.glob(f"{video_dir}/*.mp4")
    if not video_files:
        response_text = f"未找到{command['video_type']}类型的视频文件"
        send_wecom_message(response_text, [sender_name])
        return jsonify({"code": 0})
    
    # 使用第一个视频文件（实际可以根据日期等条件选择）
    video_path = video_files[0]
    
    # 执行查询
    try:
        result = SOONET_SERVICE.query_video(
            text_query=command["query"],
            video_path=video_path,
            top_k=3
        )
        
        if "error" in result:
            response_text = f"查询失败: {result['error']}"
        else:
            # 格式化结果
            formatted = SOONET_SERVICE.format_results_for_display(result)
            response_text = f"🔍 查询结果：\n\n{formformed}"
        
    except Exception as e:
        response_text = f"处理查询时出错: {str(e)}"
    
    # 发送回复
    send_wecom_message(response_text, [sender_name])
    
    return jsonify({"code": 0})

def start_wecom_bot():
    """启动企业微信机器人服务"""
    print("启动企业微信机器人服务...")
    print(f"服务地址: http://localhost:5001/wecom/webhook")
    
    app.run(host='0.0.0.0', port=5001, debug=False)

if __name__ == "__main__":
    start_wecom_bot()

5.3 企业微信配置技巧

消息安全：企业微信支持设置IP白名单，建议配置只允许你的服务器IP调用
消息频率限制：企业微信机器人有频率限制（约20条/分钟），注意不要超过
@提及功能：代码中已经处理了@提及，用户可以直接@机器人发送命令

6. 高级功能扩展

基础功能跑通后，我们可以添加一些实用功能。

6.1 支持视频文件上传

让用户可以直接上传视频进行查询：

# upload_handler.py
import os
import uuid
from flask import request

UPLOAD_FOLDER = '/tmp/soonet_uploads'
ALLOWED_EXTENSIONS = {'mp4', 'avi', 'mov', 'mkv'}

def allowed_file(filename):
    """检查文件类型"""
    return '.' in filename and \
           filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS

def handle_video_upload():
    """处理视频上传"""
    if 'video' not in request.files:
        return {"error": "没有视频文件"}
    
    file = request.files['video']
    if file.filename == '':
        return {"error": "没有选择文件"}
    
    if not allowed_file(file.filename):
        return {"error": f"不支持的文件类型，支持: {', '.join(ALLOWED_EXTENSIONS)}"}
    
    # 生成唯一文件名
    file_ext = file.filename.rsplit('.', 1)[1].lower()
    unique_filename = f"{uuid.uuid4().hex}.{file_ext}"
    filepath = os.path.join(UPLOAD_FOLDER, unique_filename)
    
    # 保存文件
    os.makedirs(UPLOAD_FOLDER, exist_ok=True)
    file.save(filepath)
    
    # 从表单获取查询文本
    query_text = request.form.get('query', '')
    
    return {
        "success": True,
        "filepath": filepath,
        "filename": unique_filename,
        "query": query_text
    }

6.2 查询历史记录

记录用户的查询历史，方便后续分析：

# query_history.py
import json
import time
from datetime import datetime

class QueryHistory:
    def __init__(self, history_file='query_history.json'):
        self.history_file = history_file
        self.history = self._load_history()
    
    def _load_history(self):
        """加载历史记录"""
        if os.path.exists(self.history_file):
            with open(self.history_file, 'r', encoding='utf-8') as f:
                return json.load(f)
        return []
    
    def _save_history(self):
        """保存历史记录"""
        with open(self.history_file, 'w', encoding='utf-8') as f:
            json.dump(self.history, f, ensure_ascii=False, indent=2)
    
    def add_query(self, user_id, query_text, video_name, results):
        """添加查询记录"""
        record = {
            "id": len(self.history) + 1,
            "user_id": user_id,
            "query": query_text,
            "video": video_name,
            "results": results,
            "timestamp": datetime.now().isoformat(),
            "time": time.time()
        }
        
        self.history.append(record)
        
        # 只保留最近100条记录
        if len(self.history) > 100:
            self.history = self.history[-100:]
        
        self._save_history()
        return record
    
    def get_user_history(self, user_id, limit=10):
        """获取用户查询历史"""
        user_queries = [q for q in self.history if q['user_id'] == user_id]
        return sorted(user_queries, key=lambda x: x['time'], reverse=True)[:limit]
    
    def get_popular_queries(self, limit=10):
        """获取热门查询"""
        from collections import Counter
        queries = [q['query'] for q in self.history]
        return Counter(queries).most_common(limit)

6.3 批量处理功能

支持一次查询多个视频：

# batch_processor.py
import concurrent.futures
from soonet_service import SOONetService

class BatchProcessor:
    def __init__(self, max_workers=2):
        self.service = SOONetService()
        self.max_workers = max_workers
    
    def batch_query(self, query_text, video_paths):
        """批量查询多个视频"""
        results = {}
        
        with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            # 提交所有查询任务
            future_to_video = {
                executor.submit(self.service.query_video, query_text, video_path): video_path
                for video_path in video_paths
            }
            
            # 收集结果
            for future in concurrent.futures.as_completed(future_to_video):
                video_path = future_to_video[future]
                try:
                    result = future.result()
                    results[video_path] = result
                except Exception as e:
                    results[video_path] = {"error": str(e)}
        
        return results
    
    def format_batch_results(self, results):
        """格式化批量结果"""
        output = []
        
        for video_path, result in results.items():
            video_name = os.path.basename(video_path)
            
            if "error" in result:
                output.append(f"❌ {video_name}: {result['error']}")
            elif result.get("success"):
                if result["results"]:
                    best_match = result["results"][0]
                    output.append(
                        f"✅ {video_name}: {best_match['time_range']} "
                        f"(置信度: {best_match['confidence']:.3f})"
                    )
                else:
                    output.append(f"⚠️ {video_name}: 未找到匹配片段")
            else:
                output.append(f"❓ {video_name}: 未知错误")
        
        return "\n".join(output)

7. 完整部署方案

7.1 项目结构

soonet-im-bot/
├── app.py                    # 主应用入口
├── soonet_service.py         # SOONet服务封装
├── dingtalk_bot.py          # 钉钉机器人
├── wecom_bot.py             # 企业微信机器人
├── upload_handler.py        # 文件上传处理
├── query_history.py         # 查询历史记录
├── batch_processor.py       # 批量处理
├── requirements.txt         # 依赖列表
├── config.py               # 配置文件
└── data/
    ├── videos/             # 视频存储目录
    └── history/           # 历史记录存储

7.2 依赖文件

# requirements.txt
flask>=2.0.0
requests>=2.25.0
modelscope>=1.0.0
torch>=1.10.0
torchvision>=0.11.0
gradio==6.4.0
opencv-python>=4.5.0
numpy<2.0

7.3 配置文件

# config.py
import os

class Config:
    # SOONet配置
    SOONET_MODEL_PATH = '/root/ai-models/iic/multi-modal_soonet_video-temporal-grounding'
    
    # 服务器配置
    HOST = '0.0.0.0'
    DINGTALK_PORT = 5000
    WECOM_PORT = 5001
    
    # 钉钉配置
    DINGTALK_WEBHOOK = os.getenv('DINGTALK_WEBHOOK', '')
    DINGTALK_SECRET = os.getenv('DINGTALK_SECRET', '')
    
    # 企业微信配置
    WECOM_WEBHOOK = os.getenv('WECOM_WEBHOOK', '')
    
    # 文件上传配置
    UPLOAD_FOLDER = '/tmp/soonet_uploads'
    MAX_CONTENT_LENGTH = 100 * 1024 * 1024  # 100MB
    ALLOWED_EXTENSIONS = {'mp4', 'avi', 'mov', 'mkv'}
    
    # 视频目录配置
    VIDEO_DIRS = {
        'meeting': '/data/videos/meetings/',
        'training': '/data/videos/trainings/',
        'monitor': '/data/videos/monitor/',
        '会议': '/data/videos/meetings/',
        '培训': '/data/videos/trainings/',
        '监控': '/data/videos/monitor/'
    }

7.4 启动脚本

#!/bin/bash
# start_bot.sh

# 启动SOONet服务（如果还没启动）
cd /root/multi-modal_soonet_video-temporal-grounding
python app.py &
SOONET_PID=$!

echo "SOONet服务启动，PID: $SOONET_PID"

# 等待SOONet服务就绪
sleep 10

# 启动钉钉机器人
cd /path/to/soonet-im-bot
python dingtalk_bot.py &
DINGTALK_PID=$!

echo "钉钉机器人启动，PID: $DINGTALK_PID"

# 启动企业微信机器人
python wecom_bot.py &
WECOM_PID=$!

echo "企业微信机器人启动，PID: $WECOM_PID"

echo "所有服务已启动"
echo "钉钉服务: http://localhost:5000"
echo "企业微信服务: http://localhost:5001"

# 保存PID到文件
echo "SOONET_PID=$SOONET_PID" > pids.txt
echo "DINGTALK_PID=$DINGTALK_PID" >> pids.txt
echo "WECOM_PID=$WECOM_PID" >> pids.txt

# 等待所有进程
wait

8. 实际应用建议

8.1 性能优化建议

视频预处理：

def preprocess_video(video_path, target_resolution=(224, 224)):
    """预处理视频，提高查询速度"""
    # 提取关键帧
    # 调整分辨率
    # 缓存处理结果
    pass

结果缓存：

import redis
import pickle

class ResultCache:
    def __init__(self):
        self.redis_client = redis.Redis(host='localhost', port=6379, db=0)
    
    def get_cache_key(self, query, video_path):
        """生成缓存键"""
        import hashlib
        key_str = f"{query}_{video_path}"
        return hashlib.md5(key_str.encode()).hexdigest()
    
    def get_cached_result(self, query, video_path):
        """获取缓存结果"""
        key = self.get_cache_key(query, video_path)
        cached = self.redis_client.get(key)
        if cached:
            return pickle.loads(cached)
        return None
    
    def set_cache(self, query, video_path, result, ttl=3600):
        """设置缓存"""
        key = self.get_cache_key(query, video_path)
        self.redis_client.setex(key, ttl, pickle.dumps(result))

8.2 安全考虑

访问控制：
- 限制可查询的视频目录
- 添加用户认证
- 记录操作日志
输入验证：
- 验证查询文本长度
- 检查视频文件类型和大小
- 防止路径遍历攻击
错误处理：
- 友好的错误提示
- 不泄露系统信息
- 超时处理机制

8.3 监控和维护

日志记录：

import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('soonet_bot.log'),
        logging.StreamHandler()
    ]
)

健康检查：

@app.route('/health')
def health_check():
    """健康检查接口"""
    return jsonify({
        "status": "healthy",
        "service": "SOONet IM Bot",
        "timestamp": datetime.now().isoformat()
    })