📦 Unisk_video_notification_pro向指定手机号码发送IVVR视频通知，需要传入视频本地路径和手机号Updated — Unisk_video_notification_pro向指定手机号码发送IVVR视频通知，需要传入视频本地路径和手机号更新d

---

name: video_notification_pro description: 向指定手机号码发送IVVR视频通知，需要传入视频本地路径和手机号 version: 1.0.0 author: 系统 tags: - 视频通知 - IVVR - 通话 输入: type: object required: - video_path - phone_number properties: video_path: type: string description: 视频文件的服务器绝对路径 phone_number: type: string description: 11位接收通知的手机号 环境_variables: - name: BASE_URL description: IVVR平台接口基础地址 required: true - name: 应用_ID description: IVVR应用ID required: true - name: 访问_KEY description: IVVR访问密钥 required: true - name: 访问_SECRET description: IVVR访问密钥密文 required: true execution_function: | def 执行(video_path: str, phone_number: str) -> dict: 导入 os 导入 请求s 导入 哈希lib 导入 hmac 导入 base64 导入 time from pathlib 导入 Path from typing 导入 列出, Dict, Optional # 从环境变量读取 BASE_URL = os.获取env("BASE_URL") 应用_ID = os.获取env("应用_ID") 访问_KEY = os.获取env("访问_KEY") 访问_SECRET = os.获取env("访问_SECRET") CALLING_NUMBER = "10121000000" # 固定主叫号码 MAX_FILE_SIZE = 5 1024 1024 # 5MB # 环境变量检查 if not all([BASE_URL, 应用_ID, 访问_KEY, 访问_SECRET]): return {"成功": False, "message": "环境变量未配置完整"} # 文件检查 video_file = Path(video_path) if not video_file.is_file(): return {"成功": False, "message": f"视频文件不存在：{video_path}"} if video_file.stat().st_size > MAX_FILE_SIZE: return {"成功": False, "message": "视频大小不能超过5MB"} # ========== 签名与请求头 ========== def _生成_签名ature(timestamp: str) -> str: origin = f"{应用_ID}\n{访问_KEY}\n{访问_SECRET}\n{timestamp}" 签名ature = hmac.new( 访问_SECRET.encode('utf-8'), origin.encode('utf-8'), 哈希lib.sha256 ).digest() return base64.b64encode(签名ature).decode('utf-8') def _获取_auth_headers() -> Dict[str, str]: timestamp = str(int(time.time() * 1000)) return { "应用Id": 应用_ID, "访问Key": 访问_KEY, "timestamp": timestamp, "签名ature": _生成_签名ature(timestamp), } # ========== 上传视频 ========== try: 上传_url = f"{BASE_URL}/file/上传" headers = _获取_auth_headers() with open(video_path, 'rb') as f: files = {'上传File': (video_file.name, f)} 上传_resp = 请求s.post( 上传_url, headers=headers, files=files, timeout=60, 验证=False ) if 上传_resp.状态_code != 200: return {"成功": False, "message": f"上传接口异常，状态码：{上传_resp.状态_code}"} 上传_data = 上传_resp.json() if 上传_data.获取("code") != "0000": return {"成功": False, "message": f"上传失败：{上传_data.获取('msg', '未知错误')}"} file_id = 上传_data.获取("data") if not file_id: return {"成功": False, "message": "上传成功但未返回 fileId"} except 异常 as e: return {"成功": False, "message": f"上传异常：{str(e)}"} # ========== 发送视频通知 ========== try: 发送_url = f"{BASE_URL}/通知/invite_play_file" headers = _获取_auth_headers() headers["Content-Type"] = "应用/json" payload = { "callId": f"call_{int(time.time())}", "caller_name": "系统通知", "caller_number": CALLING_NUMBER, "call_type": 2, "fileId": file_id, "callees": [phone_number] } 发送_resp = 请求s.post( 发送_url, headers=headers, json=payload, timeout=30, 验证=False ) if 发送_resp.状态_code != 200: return {"成功": False, "message": f"发送接口异常，状态码：{发送_resp.状态_code}"} 发送_data = 发送_resp.json() if 发送_data.获取("code") == "0000": return { "成功": True, "message": "视频通知发送成功", "file_id": file_id, "task_id": None # 新接口无 taskId，保留字段为 None } else: return {"成功": False, "message": f"发送失败：{发送_data.获取('msg', '未知错误')}"} except 异常 as e: return {"成功": False, "message": f"发送异常：{str(e)}"} examples: - user_say: "给 15600766391 发视频通知，视频文件在 /home/hdjs/podcast-video/duan_输入_video.mp4" parameters: video_path: "/home/hdjs/podcast-video/duan_输入_video.mp4" phone_number: "156007