From 3b1e106cb08c0d08fe94350f7c15c6b00039dd2f Mon Sep 17 00:00:00 2001 From: anqiang12 <15731146+anqiang12@user.noreply.gitee.com> Date: Thu, 18 Jun 2026 15:53:24 +0800 Subject: [PATCH] Initial commit: WeChat bot project with AI chat, video export, and image upload scripts --- .gitignore | 32 +++ CLAUDE.md | 75 +++++++ INSTALL_FFMPEG.md | 40 ++++ VIDEO_EXPORT_README.md | 173 +++++++++++++++ VIDEO_EXPORT_SUMMARY.md | 215 ++++++++++++++++++ cha_xianglaing.py | 238 ++++++++++++++++++++ chat_bot.py | 232 ++++++++++++++++++++ get_tianqi.py | 125 +++++++++++ qdrant_tool.py | 200 +++++++++++++++++ requirements.txt | 46 ++++ test_video_export.py | 155 +++++++++++++ video_cutter.py | 329 +++++++++++++++++++++++++++ video_to_frames.py | 166 ++++++++++++++ weixin.py | 1 + weixin_sender.py | 476 ++++++++++++++++++++++++++++++++++++++++ 猫和老鼠分镜.md | 38 ++++ 16 files changed, 2541 insertions(+) create mode 100644 .gitignore create mode 100644 CLAUDE.md create mode 100644 INSTALL_FFMPEG.md create mode 100644 VIDEO_EXPORT_README.md create mode 100644 VIDEO_EXPORT_SUMMARY.md create mode 100644 cha_xianglaing.py create mode 100644 chat_bot.py create mode 100644 get_tianqi.py create mode 100644 qdrant_tool.py create mode 100644 requirements.txt create mode 100644 test_video_export.py create mode 100644 video_cutter.py create mode 100644 video_to_frames.py create mode 100644 weixin.py create mode 100644 weixin_sender.py create mode 100644 猫和老鼠分镜.md diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..0d5e331 --- /dev/null +++ b/.gitignore @@ -0,0 +1,32 @@ +# Python +__pycache__/ +*.py[cod] +*.pyo +*.egg-info/ +dist/ +build/ + +# Virtual environment +venv/ +.venv/ +env/ + +# Large data files +vectors.jsonl +vectors1.jsonl + +# Video and image data (too large for git) +shipin/ +tupian/ +export_videos/ + +# IDE +.vscode/ +.idea/ + +# OS +.DS_Store +Thumbs.db + +# Claude Code +.claude/ diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..deb4cf3 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,75 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## Project Overview + +This is a collection of Python scripts for a WeChat (微信) bot that integrates with OpenAI-compatible APIs for AI-powered conversations with tool calling capabilities. + +## Architecture + +The codebase contains two main categories of scripts: + +### 1. Chat Bot Scripts (f25904664.py, f25904704.py, f25912200.py, f25919488.py, f25919528.py, f25922064.py) +- `send_message` class manages conversation state and OpenAI API calls +- Uses OpenAI-compatible API at `https://token-plan-cn.xiaomimimo.com/v1` with model `mimo-v2.5-pro` +- Implements tool calling for: weather queries (`tianqi`), web search (`sousuo`) +- External modules: `weixin_bot`, `get_tianqi`, `cha_xianglaing`, `weixin` + +### 2. Image Upload Scripts (f25912304.py, f25919896.py, f25920056.py) +- Handles WeChat image sending via `WeixinBot` +- AES-ECB encryption for file uploads +- Upload flow: prepare → encrypt → get upload URL → upload to WeChat servers +- Uses `ilinkai.weixin.qq.com` API endpoints + +### 3. Video Export Module (video_cutter.py) +- Exports video clips based on image timestamps +- Supports multiple modes: around/after/before/total +- Uses ffmpeg for fast video cutting (no re-encoding) +- Saves exported videos to `./export_videos/` directory + +## Dependencies + +```bash +pip install openai httpx pycryptodome prompt_toolkit requests +``` + +### System Dependencies (for video export) + +```bash +# Ubuntu/Debian +sudo apt install ffmpeg + +# macOS +brew install ffmpeg +``` + +## Key External Modules (not in repo) + +- `weixin_bot` - WeChat bot framework (provides `WeixinBot` class) +- `get_tianqi` - Weather data retrieval (`get_tianqi_data`) +- `cha_xianglaing` - Web search functionality (`search1`) +- `weixin` - WeChat utilities (`weix`) + +## Running + +Scripts are standalone and can be run directly: +```bash +python f25904664.py +``` + +### Testing Video Export + +```bash +# Test video cutter module +python video_cutter.py + +# Run full test suite +python test_video_export.py +``` + +## Code Notes + +- All comments and UI strings are in Chinese +- Files appear to be versioned snapshots with numeric suffixes (likely from a code review or commit system) +- API keys in code are for the Xiaomi Mimo API proxy, not direct OpenAI diff --git a/INSTALL_FFMPEG.md b/INSTALL_FFMPEG.md new file mode 100644 index 0000000..0b1459e --- /dev/null +++ b/INSTALL_FFMPEG.md @@ -0,0 +1,40 @@ +# 安装 ffmpeg + +视频导出功能需要 ffmpeg 支持。请根据你的操作系统安装： + +## Ubuntu/Debian + +```bash +sudo apt update +sudo apt install -y ffmpeg +``` + +## CentOS/RHEL + +```bash +sudo yum install -y epel-release +sudo yum install -y ffmpeg +``` + +## macOS + +```bash +brew install ffmpeg +``` + +## Windows + +1. 下载 ffmpeg: https://ffmpeg.org/download.html +2. 解压到目录，例如 `C:\ffmpeg` +3. 将 `C:\ffmpeg\bin` 添加到系统 PATH 环境变量 + +## 验证安装 + +```bash +ffmpeg -version +ffprobe -version +``` + +## 已安装？ + +如果已经安装，请忽略此文档。 diff --git a/VIDEO_EXPORT_README.md b/VIDEO_EXPORT_README.md new file mode 100644 index 0000000..20c32e4 --- /dev/null +++ b/VIDEO_EXPORT_README.md @@ -0,0 +1,173 @@ +# 视频导出功能使用说明 + +## 功能简介 + +当用户通过相似度搜索找到图片后，可以要求导出对应的视频片段。 + +## 支持的导出模式 + +1. **前后各N分钟**（around） + - 用户说："给我前后5分钟的视频" + - 效果：导出图片时间点前5分钟到后5分钟的视频 + +2. **后N分钟**（after） + - 用户说："导出后面5分钟" + - 效果：导出图片时间点到后5分钟的视频 + +3. **前N分钟**（before） + - 用户说："导出前面5分钟" + - 效果：导出图片时间点前5分钟到图片时间点的视频 + +4. **整个视频**（total） + - 用户说："我要整个视频" + - 效果：导出完整的视频文件 + +## 使用流程 + +### 1. 搜索相似图片 + +``` +用户：找一下猫和老鼠里汤姆追杰瑞的场景 +AI：（调用 get_donghua_images） + 找到以下相似图片： + [图片1] 相似度: 0.95, 001 Puss Gets the Boot [1940]_05m23s_456.jpg + [图片2] 相似度: 0.89, 002 The Midnight Snack [1941]_02m15s_123.jpg + （发送图片给用户） +``` + +### 2. 确认图片并导出视频 + +``` +用户：第一张图不错，给我前后5分钟的视频 +AI：✅ 视频已导出！ + 📁 文件名：001 Puss Gets the Boot [1940]_03m23s-07m23s.mp4 + 📂 保存位置：./export_videos/001 Puss Gets the Boot [1940]_03m23s-07m23s.mp4 + ⏱️ 时间范围：03m23s -> 07m23s + 🎬 时长：04m00s +``` + +## 文件结构 + +``` +/home/hasee/py/ +├── video_cutter.py # 视频切割导出模块 +├── chat_bot.py # 聊天机器人（已添加视频导出工具） +├── weixin_sender.py # 微信发送模块（已添加视频发送函数） +├── test_video_export.py # 测试脚本 +├── INSTALL_FFMPEG.md # ffmpeg 安装说明 +├── VIDEO_EXPORT_README.md # 本文件 +├── shipin/ # 原始视频目录 +│ ├── 001 Puss Gets the Boot [1940].avi +│ ├── 001 Puss Gets the Boot [1940]_frames/ +│ └── ... +└── export_videos/ # 导出视频目录（自动创建） + └── 001 Puss Gets the Boot [1940]_03m23s-07m23s.mp4 +``` + +## 依赖要求 + +1. **Python 依赖** + - 标准库：os, re, subprocess, pathlib + - 无需额外安装 + +2. **系统依赖** + - ffmpeg：用于视频切割 + - ffprobe：用于获取视频信息 + +## 安装 ffmpeg + +请参考 `INSTALL_FFMPEG.md` 文件。 + +## 测试 + +运行测试脚本： + +```bash +python test_video_export.py +``` + +测试内容： +1. 图片名解析功能 +2. 视频文件查找功能 +3. 视频导出功能（需要 ffmpeg 和视频文件） + +## 配置参数 + +在 `video_cutter.py` 中可以修改以下配置： + +```python +VIDEO_DIR = "./shipin" # 原始视频目录 +EXPORT_DIR = "./export_videos" # 导出视频目录 +VIDEO_EXTENSIONS = {'.avi', '.mp4', '.mkv', '.mov', '.flv', '.wmv', '.webm', '.ts', '.m4v'} +``` + +## 注意事项 + +1. **视频格式** + - 输入：支持多种视频格式（avi, mp4, mkv 等） + - 输出：统一为 MP4 格式（兼容性最好） + +2. **切割方式** + - 使用 `-c copy` 参数，直接复制视频流，不重新编码 + - 优点：速度快，质量无损 + - 缺点：切割点可能不精确（取决于关键帧位置） + +3. **存储空间** + - 导出的视频会占用额外空间 + - 建议定期清理 `export_videos` 目录 + +4. **性能考虑** + - 大视频文件切割可能需要较长时间 + - 建议设置合理的超时时间（当前为10分钟） + +5. **微信发送** + - 当前实现：发送文字提示（视频已保存到服务器） + - 后续扩展：可直接发送视频文件 + +## 扩展功能 + +### 1. 直接发送视频文件 + +修改 `weixin_sender.py` 中的 `send_video` 方法，实现视频文件上传和发送。 + +### 2. 自定义切割精度 + +添加参数控制切割精度（精确到帧或秒）。 + +### 3. 批量导出 + +支持一次导出多个视频片段。 + +### 4. 进度提示 + +在切割过程中发送进度提示给用户。 + +## 常见问题 + +### Q1: 提示"无法解析图片名" + +**原因**：图片名格式不符合要求 +**解决**：确保图片名格式为 `{视频名}_{分钟}m{秒}s_{毫秒}.jpg` + +### Q2: 提示"未找到视频" + +**原因**：视频文件不在 `./shipin` 目录下 +**解决**：检查视频文件路径，或修改 `VIDEO_DIR` 配置 + +### Q3: 提示"无法获取视频时长" + +**原因**：ffmpeg/ffprobe 未安装或视频文件损坏 +**解决**：安装 ffmpeg，检查视频文件是否完整 + +### Q4: 视频切割超时 + +**原因**：视频文件过大或系统性能不足 +**解决**：减小切割范围，或增加超时时间 + +## 技术支持 + +如有问题，请检查： +1. ffmpeg 是否正确安装 +2. 视频文件是否存在 +3. 图片名格式是否正确 +4. 磁盘空间是否充足 diff --git a/VIDEO_EXPORT_SUMMARY.md b/VIDEO_EXPORT_SUMMARY.md new file mode 100644 index 0000000..74bb629 --- /dev/null +++ b/VIDEO_EXPORT_SUMMARY.md @@ -0,0 +1,215 @@ +# 视频导出功能实现总结 + +## 已完成的工作 + +### 1. 创建 `video_cutter.py` 模块 + +**功能**： +- 解析图片名，提取视频名和时间点 +- 根据视频名找到原始视频文件 +- 使用 ffmpeg 切割视频 +- 支持4种导出模式：around/after/before/total + +**核心函数**： +- `parse_image_info(image_name)` - 解析图片名 +- `find_video_file(video_name)` - 查找视频文件 +- `get_video_duration(video_path)` - 获取视频时长 +- `cut_video(video_path, start_time, end_time, output_path)` - 切割视频 +- `export_video_clip(image_name, mode, minutes)` - 导出视频片段 + +### 2. 修改 `chat_bot.py` + +**添加内容**： +- 在 `tool_list` 中添加 `export_video` 工具 +- 在 `test_openai` 中添加工具定义（包含参数说明） +- 在 `sen_mess` 方法中添加处理逻辑 + +**工具定义**： +```python +{ + "name": "export_video", + "description": "当用户确认某张图片并要求导出视频时调用此工具", + "parameters": { + "image_name": "图片文件名", + "mode": "around/after/before/total", + "minutes": "分钟数（默认5）" + } +} +``` + +### 3. 修改 `weixin_sender.py` + +**添加内容**： +- 在 `weix` 类中添加 `send_video` 方法 +- 当前实现：发送文字提示（视频已保存到服务器） +- 后续可扩展：直接发送视频文件 + +### 4. 创建测试和文档 + +**测试文件**： +- `test_video_export.py` - 完整测试脚本 + +**文档文件**： +- `INSTALL_FFMPEG.md` - ffmpeg 安装说明 +- `VIDEO_EXPORT_README.md` - 使用说明文档 +- `VIDEO_EXPORT_SUMMARY.md` - 本文件 + +**更新文件**： +- `CLAUDE.md` - 添加视频导出功能说明 + +## 用户交互流程 + +### 场景1：前后5分钟 + +``` +用户：找一下猫和老鼠里汤姆追杰瑞的场景 +AI：（发送相似图片） + +用户：第一张图不错，给我前后5分钟的视频 +AI：✅ 视频已导出！ + 📁 文件名：001 Puss Gets the Boot [1940]_03m23s-07m23s.mp4 + 📂 保存位置：./export_videos/... + ⏱️ 时间范围：03m23s -> 07m23s + 🎬 时长：04m00s +``` + +### 场景2：后面5分钟 + +``` +用户：这张图后面的5分钟视频给我 +AI：（调用 export_video，mode="after", minutes=5） +``` + +### 场景3：整个视频 + +``` +用户：我要整个视频 +AI：（调用 export_video，mode="total"） +``` + +## 文件结构 + +``` +/home/hasee/py/ +├── video_cutter.py # 视频切割导出模块（新增） +├── chat_bot.py # 聊天机器人（已修改） +├── weixin_sender.py # 微信发送模块（已修改） +├── test_video_export.py # 测试脚本（新增） +├── INSTALL_FFMPEG.md # 安装说明（新增） +├── VIDEO_EXPORT_README.md # 使用说明（新增） +├── VIDEO_EXPORT_SUMMARY.md # 本文件（新增） +├── CLAUDE.md # 项目说明（已更新） +├── shipin/ # 原始视频目录 +│ ├── 001 Puss Gets the Boot [1940].avi +│ ├── 001 Puss Gets the Boot [1940]_frames/ +│ └── ... +└── export_videos/ # 导出视频目录（自动创建） +``` + +## 依赖要求 + +### Python 依赖 + +已包含在 `requirements.txt` 中，无需额外安装。 + +### 系统依赖 + +需要安装 ffmpeg： + +```bash +# Ubuntu/Debian +sudo apt install ffmpeg + +# macOS +brew install ffmpeg +``` + +## 测试方法 + +### 1. 测试解析功能 + +```bash +python video_cutter.py +``` + +输出示例： +``` +=== 测试图片名解析 === + 001 Puss Gets the Boot [1940]_05m23s_456.jpg + -> {'video_name': '001 Puss Gets the Boot [1940]', 'timestamp_seconds': 323.456} +``` + +### 2. 运行完整测试 + +```bash +python test_video_export.py +``` + +测试内容： +1. 图片名解析功能 +2. 视频文件查找功能 +3. 视频导出功能（需要 ffmpeg 和视频文件） + +## 配置参数 + +在 `video_cutter.py` 中可以修改： + +```python +VIDEO_DIR = "./shipin" # 原始视频目录 +EXPORT_DIR = "./export_videos" # 导出视频目录 +VIDEO_EXTENSIONS = {'.avi', '.mp4', '.mkv', '.mov', '.flv', '.wmv', '.webm', '.ts', '.m4v'} +``` + +## 注意事项 + +1. **ffmpeg 必须安装** + - 视频切割依赖 ffmpeg + - 请参考 `INSTALL_FFMPEG.md` 安装 + +2. **视频文件位置** + - 原始视频必须在 `./shipin/` 目录下 + - 视频名必须与图片名中的视频名一致 + +3. **图片名格式** + - 必须符合：`{视频名}_{分钟}m{秒}s_{毫秒}.jpg` + - 例如：`001 Puss Gets the Boot [1940]_05m23s_456.jpg` + +4. **切割精度** + - 使用 `-c copy` 参数，不重新编码 + - 切割点可能不精确（取决于关键帧位置） + +5. **存储空间** + - 导出的视频会占用额外空间 + - 建议定期清理 `export_videos` 目录 + +## 后续扩展 + +### 1. 直接发送视频文件 + +修改 `weixin_sender.py` 中的 `send_video` 方法，实现视频文件上传和发送。 + +### 2. 自定义切割精度 + +添加参数控制切割精度（精确到帧或秒）。 + +### 3. 批量导出 + +支持一次导出多个视频片段。 + +### 4. 进度提示 + +在切割过程中发送进度提示给用户。 + +### 5. 视频格式转换 + +支持导出不同格式的视频（MP4, AVI, MKV 等）。 + +## 技术支持 + +如有问题，请检查： +1. ffmpeg 是否正确安装 +2. 视频文件是否存在 +3. 图片名格式是否正确 +4. 磁盘空间是否充足 + +详细说明请参考 `VIDEO_EXPORT_README.md`。 diff --git a/cha_xianglaing.py b/cha_xianglaing.py new file mode 100644 index 0000000..ad9eca1 --- /dev/null +++ b/cha_xianglaing.py @@ -0,0 +1,238 @@ +import os +import sys +import json +import base64 +import time +from pathlib import Path +from multiprocessing import Pool + +# ========== 配置区（直接改这里）========== + +# 图片目录 +IMAGE_DIR = "./shipin/" + +# API Keys（填多个就多进程并行） +API_KEYS = [ + "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", + "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", + "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" +] + +collection_name = "my_collection1" +# API 地址 +BASE_URL = "https://integrate.api.nvidia.com/v1" + +# 模型名称 +MODEL = "nvidia/llama-nemotron-embed-vl-1b-v2" + +# 输出文件 +OUTPUT = "vectors1.jsonl" + +# 同一 key 请求间隔（秒），防限流 +DELAY = 0.1 + +# ========== 配置结束 ========== + +IMAGE_EXTENSIONS = {'.jpg', '.jpeg', '.png', '.bmp', '.webp'} + +# ========== 独立函数（供 weixin_bot 等外部调用）========== + +def get_client(api_key=None, base_url=None): + """获取 OpenAI 客户端（单 key 模式，兼容旧接口）""" + from openai import OpenAI + return OpenAI( + api_key=api_key or API_KEYS[0], + base_url=base_url or BASE_URL + ) + + +def encode_image_base64(image_path: str) -> str: + """将图片编码为 base64""" + with open(image_path, "rb") as f: + return base64.b64encode(f.read()).decode("utf-8") + + +def get_image_vector(image_path: str, client=None) -> list[float]: + if client is None: + client = get_client() + b64, mime = encode_image_base64(image_path) # 确保你的函数能同时返回这些值 + url = f"data:{mime};base64,{b64}" + input_payload = [ + { + "content": [ + {"type": "text", "text": "What is in this image?"}, + {"type": "image_url", "image_url": {"url": url}} + ] + } + ] + resp = client.embeddings.create( + input=input_payload, # 使用新格式 + model=MODEL, + encoding_format="float", + extra_body={"input_type": "passage","truncate": "NONE", "modality": ["image"]} + ) + return resp.data[0].embedding + + +def get_text_vector(text: str, client=None) -> list[float]: + """获取文本的向量（外部可直接调用）""" + if client is None: + client = get_client() + resp = client.embeddings.create( + input=[text], + model=MODEL, + encoding_format="float", + extra_body={"modality": ["text"], "input_type": "query", "truncate": "NONE"} + ) + return resp.data[0].embedding + + +def search1(query_text: str, top_k=5, qdrant_host="10.0.0.66", qdrant_port=6333): + """在 Qdrant 中搜索（外部可直接调用）""" + from qdrant_client import QdrantClient + vector = get_text_vector(query_text) + qdrant = QdrantClient(host=qdrant_host, port=qdrant_port) + results = qdrant.query_points( + collection_name=collection_name, + query=vector, + limit=top_k + ) + return results.points + + +# ========== 批量向量化（多 key 多进程）========== + +def scan_images(root_path: str) -> list[str]: + """扫描目录下所有图片文件""" + images = [] + root = Path(root_path) + if root.is_file() and root.suffix.lower() in IMAGE_EXTENSIONS: + return [str(root.resolve())] + for p in sorted(root.rglob("*")): + if p.is_file() and p.suffix.lower() in IMAGE_EXTENSIONS: + images.append(str(p.resolve())) + return images + + +def vectorize(image_path: str, api_key: str, base_url: str, model: str) -> dict: + from openai import OpenAI + import mimetypes + import base64 + try: + client = OpenAI(api_key=api_key, base_url=base_url) + + # 获取 base64 和 MIME 类型 + with open(image_path, "rb") as f: + b64 = base64.b64encode(f.read()).decode("utf-8") + mime, _ = mimetypes.guess_type(image_path) + if not mime: + mime = "image/jpeg" + + uri = f"data:{mime};base64,{b64}" + + resp = client.embeddings.create( + input=[uri], + model=model, + encoding_format="float", + # !!! 关键修改：把 "query" 改为 "passage" !!! + extra_body={"input_type": "passage", "truncate": "NONE", "modality": ["image"]} + ) + return {"path": image_path, "vector": resp.data[0].embedding, "status": "ok"} + except Exception as e: + return {"path": image_path, "vector": None, "status": "error", "error": str(e)} + + +def _worker(args) -> dict: + """多进程 worker，返回结果同时携带进程信息""" + image_path, api_key, base_url, model, delay = args + if delay > 0: + time.sleep(delay) + pid = os.getpid() + # 也可以使用 current_process().name + result = vectorize(image_path, api_key, base_url, model) + result["pid"] = pid + result["api_key_prefix"] = api_key[:10] + "..." # 只显示前几位 + return result + + +def load_existing(output_path: str) -> dict: + """加载已有的向量 JSONL（每行一个 JSON 对象）""" + result = {} + if os.path.exists(output_path): + with open(output_path, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if line: + item = json.loads(line) + result[item["path"]] = item + return result + + +def append_vectors(new_items: list[dict], output_path: str): + """追加写入新向量到 JSONL 文件（每行一个 JSON 对象）""" + with open(output_path, "a", encoding="utf-8") as f: + for item in new_items: + f.write(json.dumps(item, ensure_ascii=False) + "\n") + + +def main(): + images = scan_images(IMAGE_DIR) + if not images: + print(f"未在 {IMAGE_DIR} 中找到图片文件") + sys.exit(1) + + existing = load_existing(OUTPUT) + todo = [img for img in images if img not in existing] + skipped = len(images) - len(todo) + + if not todo: + print(f"所有 {len(images)} 张图片已存在于 {OUTPUT}，无需处理") + return + + jobs = len(API_KEYS) + print(f"找到 {len(images)} 张图片，待处理 {len(todo)}，跳过 {skipped}") + print(f"使用 {jobs} 个 API Key，并行处理") + + # 构建任务：轮询分配 key，同一 key 的请求错开 delay + tasks = [] + for i, img in enumerate(todo): + key = API_KEYS[i % len(API_KEYS)] + delay = DELAY + # print(delay) + tasks.append((img, key, BASE_URL, MODEL, delay)) + + done = 0 + errors = 0 + total = len(existing) + + with Pool(jobs) as pool: + for result in pool.imap_unordered(_worker, tasks): + if result["status"] == "ok": + done += 1 + append_vectors([{"path": result["path"], "vector": result["vector"]}], OUTPUT) + total += 1 + # 打印详细信息：进程ID + 文件路径 + print(f"[PID {result['pid']}] 已处理: {os.path.basename(result['path'])}") + if done % 50 == 0: + print(f"[进度] 已完成 {done}/{len(todo)}，错误 {errors}，总计 {total}") + else: + errors += 1 + print(f"[错误][PID {result.get('pid', '?')}] {os.path.basename(result['path'])}: {result['error']}") + + print(f"\n处理完毕: 完成 {done}, 跳过 {skipped}, 错误 {errors}") + print(f"向量已追加到 {OUTPUT}，共 {total} 条记录") + + +if __name__ == "__main__": + main() +# from qdrant_client import QdrantClient, models +# client = QdrantClient(host="10.0.0.66", port=6333) +# collection_name = "my_collection1" #集合名称 +# client.create_collection( +# collection_name=collection_name, +# vectors_config=models.VectorParams( +# size=2048, # 向量维度，128维 +# distance=models.Distance.COSINE # 距离计算方法，这里用余弦相似度 +# ) +# ) +# client.delete_collection(collection_name="my_collection") diff --git a/chat_bot.py b/chat_bot.py new file mode 100644 index 0000000..06dfdc9 --- /dev/null +++ b/chat_bot.py @@ -0,0 +1,232 @@ +import time +from prompt_toolkit import prompt +from openai import OpenAI +from get_tianqi import get_tianqi_data +from cha_xianglaing import search1 +from weixin import weix +import json + +tool_list = [ + {"name":"tainqi","description":"查询天气"}, + {"name":"sousuo","description":"网页搜索"}, + {"name":"export_video","description":"导出视频片段"}, + ] + +toole_details = { + "tianqi":{ + "type":"function", + "function":{ + "name":"tianqi", + "description":"查询天气状况(温度、湿度、风力、天气....)时使用此tool", + "strict": True, + "parameters": { + "type":"object", + "properties":{ + "city":{"type": "string", "description": "要获取天气的城市名称"} + }, + "required": ["city"], + "additionalProperties": False + } + } + }, + } + +def get_tools_details(): + # 获取工具详情 + pass + + +class send_message: + def __init__(self, system_content: str ,tool_choice="auto"): + self.ai_return = [] + self.tool_choice1 = tool_choice + self.messages_1 = [ + {"role": "system", "content": system_content}, + ] + def test_openai(self): + # print("111",self.messages_1) + client = OpenAI(api_key="tp-cj0x379me5rqk198qnnt4n1spcdbx986jjr163gu2iiw1s7w", + base_url="https://token-plan-cn.xiaomimimo.com/v1") + response = client.chat.completions.create( + model="mimo-v2.5-pro", + messages=self.messages_1, + top_p=0.5, + tools=[ + {"type": "function", + "function": { + "name": "get_tianqi", + "description": "获取天气信息。当你需要回答任何与天气、气温、冷暖有关的问题时调用此工具；判断用户是否是询问温度而不是出现冷热就调用此工具，例如用户说：你这个笑话好冷，则无需调用。**必须**使用此工具来获取实时、真实的天气数据。", + "parameters": { + "type": "object", + "properties": { + "city": {"type": "string", "description": "城市名称"} + }, + "required": ["city"], + "additionalProperties": False + } + } + },{"type": "function", + "function": { + "name": "get_donghua_images", + "description": "用户需要查询具体动画场景时调用此工具。例如:用户说这个场景不错帮我找下，调用此工具；", + "parameters": { + "type": "object", + "properties": { + "donghua": {"type": "string", "description": "动画场景描述,必须为英文;例如:Tom is tired"}, + "xiangsidu": {"type": "integer", "description": "图片与场景的相似度要求,例如:'给我相似度大于50%的图片 则传值0.5'"}, + "tupianshuliang": {"type": "integer", "description": "返回图片数量,例如:'给我返回10张图片 则传值10'"}, + }, + "required": ["donghua"], + "additionalProperties": False + }} + }, + {"type": "function", + "function": { + "name": "export_video", + "description": "当用户确认某张图片并要求导出视频时调用此工具。例如：'这张图不错，给我前后30秒的视频'、'导出这个场景后面5分钟'、'我要整个视频'、'帮我截取这个片段'", + "parameters": { + "type": "object", + "properties": { + "image_name": { + "type": "string", + "description": "用户确认的图片文件名，例如 '001 Puss Gets the Boot [1940]_05m23s_456.jpg'" + }, + "mode": { + "type": "string", + "enum": ["around", "after", "before", "total"], + "description": "导出模式：around=前后各N秒，after=后N秒，before=前N秒，total=整个视频" + }, + "seconds": { + "type": "integer", + "description": "秒数，total模式下可省略，默认300秒。例如用户说'前后30秒'则传30，'前后5分钟'则传300", + "default": 300 + } + }, + "required": ["image_name", "mode","seconds"], + "additionalProperties": False + }} + } + ], + tool_choice=self.tool_choice1, + ) + return response.choices[0].message + + +def test_json_format(json_string): + try: + return json.loads(json_string) + except json.JSONDecodeError: + return False +def get_tianqi(ai_json): + print(f"正在获取{ai_json['city']}的天气信息...") + return f"{get_tianqi_data(ai_json['city'])}" +# m = send_message('') + + + +cont = 1 +# m = send_message('你是一个个人助手，会适当使用工具') +class get_ai_message: + def __init__(self): + self.system_content = open("猫和老鼠分镜.md", "r", encoding="utf-8").read() + self.m = send_message(self.system_content) + self.wx = weix() + + async def _handle_donghua(self, args): + """处理动画图片搜索工具""" + donghua = args["donghua"] + xiangsidu = args.get("xiangsidu", 0) + tupianshuliang = args.get("tupianshuliang", 5) + re_donghua_image = search1(query_text=donghua, top_k=tupianshuliang) + result_str = "" + for result in re_donghua_image: + if result.score < xiangsidu: + continue + result_str += f"相似度:{result.score},图片路径:{result.payload}\n" + print(result) + await self.wx.send_images(result.payload["image_path"]) + return result_str if result_str else "未找到符合条件的图片" + + async def _handle_export_video(self, args): + """处理视频导出工具""" + from video_cutter import export_video_clip + export_result = export_video_clip( + image_name=args["image_name"], + mode=args["mode"], + seconds=args.get("seconds", 300) + ) + if export_result["success"]: + await self.wx.send_video(export_result["video_path"]) + return ( + f"✅ 视频已导出并发送！\n" + f"⏱️ 时间范围：{export_result['start_time']} -> {export_result['end_time']}\n" + f"🎬 时长：{export_result['duration']}" + ) + else: + return f"❌ 视频导出失败：{export_result['error']}" + + async def sen_mess(self, s_input_text): + # 拼接用户消息 + self.m.messages_1.append({"role": "user", "content": s_input_text}) + + while True: + out_text = self.m.test_openai() + + if out_text.tool_calls: + # 将 assistant 的 tool_calls 消息整体追加到 messages（转为 dict） + self.m.messages_1.append({ + "role": "assistant", + "content": out_text.content, + "tool_calls": [ + { + "id": tc.id, + "type": "function", + "function": { + "name": tc.function.name, + "arguments": tc.function.arguments + } + } + for tc in out_text.tool_calls + ] + }) + + for tool_call in out_text.tool_calls: + args = test_json_format(tool_call.function.arguments) + tool_name = tool_call.function.name + print("M调用了工具:", tool_name) + + tool_result = "" + if tool_name == "get_tianqi": + tool_result = get_tianqi(args) + elif tool_name == "get_donghua_images": + tool_result = await self._handle_donghua(args) + elif tool_name == "export_video": + tool_result = await self._handle_export_video(args) + + # 工具结果以 tool role 追加到 messages + self.m.messages_1.append({ + "role": "tool", + "tool_call_id": tool_call.id, + "content": tool_result + }) + print(f"工具 {tool_name} 返回: {tool_result[:100]}...") + + # 循环继续，让 LLM 看到工具结果后继续推理 + + elif out_text.content: + # 没有工具调用，发送最终文本给用户 + self.m.messages_1.append({"role": "assistant", "content": out_text.content}) + await self.wx.send_text(out_text.content) + print(f"小爱:{out_text.content}") + return out_text.content + +gm = get_ai_message() +bot = gm.wx.bot +@bot.on_message +async def echo_handler(msg): + print(f"收到来自 {msg.user_id} 的消息: {msg.text}") + await gm.sen_mess(msg.text) + # await bot.reply(msg, f"你说了: {msg.text}") +bot.run() + + diff --git a/get_tianqi.py b/get_tianqi.py new file mode 100644 index 0000000..17e1cc5 --- /dev/null +++ b/get_tianqi.py @@ -0,0 +1,125 @@ +import requests + +# ===================== 配置区（仅需替换你的高德Key）===================== +GAODE_KEY = "ff608780d5d18bab2f875a62b9c0c106" # 必须是Web服务类型Key +# ====================================================================== + +# 高德API基础地址 +GEO_URL = "https://restapi.amap.com/v3/geocode/geo" # 地理编码（城市名转ADCODE） +WEATHER_URL = "https://restapi.amap.com/v3/weather/weatherInfo" # 天气API + +def get_city_adcode(city_name: str) -> str: + """ + 【核心函数】传入城市名称字符串，返回高德ADCODE编码 + :param city_name: 城市名（如：北京、上海市、深圳、朝阳区、杭州） + :return: 城市ADCODE编码，失败返回None + """ + params = { + "key": GAODE_KEY, + "address": city_name, # 传入的城市名称 + "output": "json" + } + try: + response = requests.get(GEO_URL, params=params, timeout=10) + data = response.json() + + # 调用成功，解析ADCODE + if data.get("status") == "1" and data.get("geocodes"): + adcode = data["geocodes"][0]["adcode"] + print(f"✅ 城市【{city_name}】匹配成功，ADCODE编码：{adcode}") + return adcode + else: + print(f"❌ 未找到城市【{city_name}】，请检查城市名称是否正确") + return None + + except Exception as e: + print(f"❌ 查询编码失败：{str(e)}") + return None + +def get_gaode_weather(adcode: str, is_forecast=False): + """调用高德天气API（实时/4天预报）""" + params = { + "key": GAODE_KEY, + "city": adcode, + "extensions": "base" if not is_forecast else "all", + "output": "json" + } + try: + response = requests.get(WEATHER_URL, params=params, timeout=10) + data = response.json() + return data if data.get("status") == "1" else None + except: + return None + +def format_weather_info(data) -> str: + """格式化天气信息并返回字符串""" + if not data: + return "天气数据获取失败" + + result = [] + + # 实时天气 + if "lives" in data: + live = data["lives"][0] + result.append(f"【实时天气】") + result.append(f"城市：{live['city']}") + result.append(f"天气：{live['weather']}") + result.append(f"温度：{live['temperature']}℃") + result.append(f"风向：{live['winddirection']}风") + result.append(f"风力：{live['windpower']}级") + result.append(f"湿度：{live['humidity']}%") + + # 4天预报 + if "forecasts" in data: + forecast = data["forecasts"][0] + result.append(f"\n【{forecast['city']} 4天预报】") + for day in forecast["casts"]: + result.append(f"{day['date']}：白天{day['dayweather']}，夜间{day['nightweather']}，气温{day['nighttemp']}~{day['daytemp']}℃") + + return "\n".join(result) + + +def print_weather_info(data): + """格式化打印天气信息""" + print(format_weather_info(data)) + + +def get_tianqi_data(city_name: str) -> str: + """ + 【供外部调用的主函数】传入城市名称，返回格式化的天气信息字符串 + :param city_name: 城市名称（如：北京、上海、深圳） + :return: 格式化的天气信息字符串 + """ + # 1. 获取城市ADCODE + adcode = get_city_adcode(city_name) + if not adcode: + return f"未找到城市【{city_name}】的天气信息" + + # 2. 获取实时天气 + realtime = get_gaode_weather(adcode, is_forecast=False) + # 3. 获取4天预报 + forecast = get_gaode_weather(adcode, is_forecast=True) + + # 4. 合并结果 + result = [] + if realtime: + result.append(format_weather_info(realtime)) + if forecast: + result.append(format_weather_info(forecast)) + + return "\n\n".join(result) if result else f"获取【{city_name}】天气失败" + +# ===================== 使用示例（直接改这里的城市名）===================== +# if __name__ == "__main__": +# # 1. 传入城市名称（字符串），自动获取编码 +# city_name = "北京" # 这里修改为你要查询的城市：上海、广州、深圳、成都、重庆等 +# city_adcode = get_city_adcode(city_name) + +# # 2. 获取编码成功后，查询天气 +# if city_adcode: +# # 实时天气 +# realtime = get_gaode_weather(city_adcode, is_forecast=False) +# print_weather_info(realtime) +# # 4天预报 +# forecast = get_gaode_weather(city_adcode, is_forecast=True) +# print_weather_info(forecast) \ No newline at end of file diff --git a/qdrant_tool.py b/qdrant_tool.py new file mode 100644 index 0000000..29aa00c --- /dev/null +++ b/qdrant_tool.py @@ -0,0 +1,200 @@ +""" +Qdrant 图片向量数据库工具 +功能：上传 vectors.jsonl 中已分析好的向量，用文本查询返回相似度+图片名 +依赖：pip install qdrant-client openai +""" + +import json +import re +import uuid +from qdrant_client import QdrantClient, models +from openai import OpenAI + +# ========== 配置区 ========== +QDRANT_HOST = "10.0.0.66" +QDRANT_PORT = 6333 +COLLECTION_NAME = "my_collection1" +VECTOR_DIM = 2048 # vectors.jsonl 中向量维度 + +# 嵌入模型配置（查询时用） +EMBED_API_KEY = "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" +EMBED_BASE_URL = "https://integrate.api.nvidia.com/v1" +EMBED_MODEL = "nvidia/llama-nemotron-embed-vl-1b-v2" + + +def get_qdrant_client() -> QdrantClient: + return QdrantClient(host=QDRANT_HOST, port=QDRANT_PORT) + + +def get_text_vector(text: str) -> list[float]: + """获取文本向量（用于查询）""" + client = OpenAI(api_key=EMBED_API_KEY, base_url=EMBED_BASE_URL) + resp = client.embeddings.create( + input=[text], + model=EMBED_MODEL, + encoding_format="float", + extra_body={"modality": ["text"], "input_type": "query", "truncate": "NONE"} + ) + return resp.data[0].embedding + + +def parse_timestamp(filename: str) -> str: + """ + 从文件名解析时间戳 + 文件名格式: xxx_00m05s_213.jpg → "00:05" + """ + match = re.search(r'(\d{2})m(\d{2})s', filename) + if match: + return f"{match.group(1)}:{match.group(2)}" + return "" + + +# ========== 创建集合 ========== + +def create_collection( + collection_name: str = COLLECTION_NAME, + vector_dim: int = VECTOR_DIM +): + """创建 Qdrant 集合""" + client = get_qdrant_client() + collections = [c.name for c in client.get_collections().collections] + if collection_name in collections: + print(f"集合 '{collection_name}' 已存在，跳过创建") + return + + client.create_collection( + collection_name=collection_name, + vectors_config=models.VectorParams( + size=vector_dim, + distance=models.Distance.COSINE + ) + ) + print(f"集合 '{collection_name}' 创建成功，维度={vector_dim}") + + +# ========== 上传向量 ========== + +def upload_vectors_jsonl( + jsonl_path: str = "vectors.jsonl", + collection_name: str = COLLECTION_NAME +): + """ + 读取 vectors.jsonl 并上传到 Qdrant + jsonl 格式: {"image_path": "/path/to/image.jpg","image_name":"[猫和....01s_042.jpg", "vector": [...]} + """ + client = get_qdrant_client() + + points = [] + with open(jsonl_path, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line: + continue + item = json.loads(line) + path = item["path"] + vector = item["vector"] + image_name = path.split("/")[-1].split("\\")[-1] + timestamp = parse_timestamp(image_name) + + points.append( + models.PointStruct( + id=str(uuid.uuid4()), + vector=vector, + payload={ + "image_name": image_name, + "image_path": path, + "timestamp": timestamp, + } + ) + ) + + sp = serch_payload(image_name) + # print(sp) + if sp[0]: + print(f"向量已经上传过，不再处理{path}") + points = [] + continue + # 每 500 条批量上传 + if len(points) >= 100: + client.upsert(collection_name=collection_name, points=points) + print(f"已上传 {len(points)} 条...") + points = [] + + if points: + client.upsert(collection_name=collection_name, points=points) + + total = client.count(collection_name=collection_name).count + print(f"上传完成，集合 '{collection_name}' 共 {total} 条向量") + + +# ========== 查询 ========== + +def search( + query_text: str, + top_k: int = 5, + collection_name: str = COLLECTION_NAME +): + """ + 用文本搜索相似图片 + 返回: [(相似度, 图片名, 时间戳), ...] + """ + client = get_qdrant_client() + vector = get_text_vector(query_text) + + results = client.query_points( + collection_name=collection_name, + query=vector, + limit=top_k + ) + + output = [] + print(f"查询: '{query_text}'\n") + for i, point in enumerate(results.points): + name = point.payload.get("image_name", "") + ts = point.payload.get("timestamp", "") + score = point.score + output.append((score, name, ts)) + print(f" [{i + 1}] 相似度={score:.4f} 时间={ts} 文件={name}") + + return output + +def serch_payload(exact_value): + # exact_value = '[猫和老鼠(五十周年纪念版-国粤英三语)].Tom.And.Jerry.E01.2001.DVDrip.x264.AC3-CMCT_102m13s_210.jpg' + filter_condition = models.Filter( + must=[ + models.FieldCondition( + key="image_name", + match=models.MatchValue(value=exact_value) + ) + ] + ) + client = get_qdrant_client() + result = client.scroll( + collection_name=COLLECTION_NAME, + scroll_filter=filter_condition, + limit=10, + with_payload=True + ) + return result + + +# ========== 运行设置（改这里）========== +# 设为 True 的才会执行 +DO_CREATE = False # 创建集合 +DO_UPLOAD = 1 # 上传 vectors.jsonl +DO_SEARCH = False # 文本搜图片 + +SEARCH_TEXT = "猫和老鼠" # 搜索内容 +SEARCH_TOP_K = 5 # 返回数量 +# ====================================== + + +if __name__ == "__main__": + if DO_CREATE: + create_collection() + + if DO_UPLOAD: + upload_vectors_jsonl("vectors1.jsonl") + + if DO_SEARCH: + search(SEARCH_TEXT, top_k=SEARCH_TOP_K) \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..64d2d67 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,46 @@ +aiohappyeyeballs==2.6.2 +aiohttp==3.14.1 +aiosignal==1.4.0 +annotated-types==0.7.0 +anyio==4.13.0 +attrs==26.1.0 +certifi==2026.5.20 +cffi==2.0.0 +charset-normalizer==3.4.7 +cryptography==45.0.7 +distro==1.9.0 +faiss-cpu==1.14.2 +frozenlist==1.8.0 +grpcio==1.81.1 +h11==0.16.0 +h2==4.3.0 +hpack==4.1.0 +httpcore==1.0.9 +httpx==0.28.1 +hyperframe==6.1.0 +idna==3.18 +jiter==0.15.0 +multidict==6.7.1 +numpy==2.4.6 +openai==2.41.1 +packaging==26.2 +portalocker==3.2.0 +prompt_toolkit==3.0.52 +propcache==0.5.2 +protobuf==7.35.1 +pycparser==3.0 +pycryptodome==3.23.0 +pydantic==2.13.4 +pydantic_core==2.46.4 +qdrant-client==1.18.0 +requests==2.34.2 +silk-python==0.2.8 +sniffio==1.3.1 +tqdm==4.68.2 +typing-inspection==0.4.2 +typing_extensions==4.15.0 +urllib3==2.7.0 +wcwidth==0.8.1 +wechat_clawbot_sdk==0.4.0 +weixin-bot-sdk==0.2.0 +yarl==1.24.2 diff --git a/test_video_export.py b/test_video_export.py new file mode 100644 index 0000000..e2339f6 --- /dev/null +++ b/test_video_export.py @@ -0,0 +1,155 @@ +#!/usr/bin/env python3 +""" +视频导出功能测试脚本 +测试 video_cutter.py 的各项功能 +""" + +import sys +import os + +# 添加当前目录到 Python 路径 +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from video_cutter import parse_image_info, find_video_file, export_video_clip + + +def test_parse_image_info(): + """测试图片名解析功能""" + print("=" * 60) + print("测试图片名解析功能") + print("=" * 60) + + test_cases = [ + ("001 Puss Gets the Boot [1940]_05m23s_456.jpg", True), + ("002 The Midnight Snack [1941]_02m15s_123.jpg", True), + ("003 The Night Before Christmas [1941]_00m00s_000.jpg", True), + ("invalid_image.jpg", False), + ("no_timestamp.jpg", False), + ] + + for image_name, expected in test_cases: + result = parse_image_info(image_name) + success = (result is not None) == expected + status = "✅" if success else "❌" + + print(f"\n{status} {image_name}") + if result: + print(f" 视频名: {result['video_name']}") + print(f" 时间点: {result['timestamp_seconds']:.3f}秒") + else: + print(f" 解析失败") + + return True + + +def test_find_video_file(): + """测试视频文件查找功能""" + print("\n" + "=" * 60) + print("测试视频文件查找功能") + print("=" * 60) + + test_cases = [ + "001 Puss Gets the Boot [1940]", + "002 The Midnight Snack [1941]", + "nonexistent_video", + ] + + for video_name in test_cases: + result = find_video_file(video_name) + status = "✅" if result else "❌" + + print(f"\n{status} {video_name}") + if result: + print(f" 找到: {result}") + else: + print(f" 未找到") + + return True + + +def test_export_video_clip(): + """测试视频导出功能""" + print("\n" + "=" * 60) + print("测试视频导出功能") + print("=" * 60) + + # 测试用例 + test_cases = [ + { + "image_name": "001 Puss Gets the Boot [1940]_05m23s_456.jpg", + "mode": "around", + "seconds": 120, + "description": "前后各2分钟" + }, + { + "image_name": "001 Puss Gets the Boot [1940]_05m23s_456.jpg", + "mode": "after", + "seconds": 180, + "description": "后3分钟" + }, + { + "image_name": "001 Puss Gets the Boot [1940]_05m23s_456.jpg", + "mode": "before", + "seconds": 60, + "description": "前1分钟" + }, + { + "image_name": "001 Puss Gets the Boot [1940]_05m23s_456.jpg", + "mode": "total", + "seconds": 0, + "description": "整个视频" + }, + ] + + for i, case in enumerate(test_cases, 1): + print(f"\n测试 {i}: {case['description']}") + print(f" 图片: {case['image_name']}") + print(f" 模式: {case['mode']}") + print(f" 秒数: {case['seconds']}") + + result = export_video_clip( + image_name=case['image_name'], + mode=case['mode'], + seconds=case['seconds'] + ) + + if result['success']: + print(f" ✅ 导出成功") + print(f" 📁 文件: {result['video_path']}") + print(f" ⏱️ 时间: {result['start_time']} -> {result['end_time']}") + print(f" 🎬 时长: {result['duration']}") + else: + print(f" ❌ 导出失败: {result['error']}") + + return True + + +def main(): + """主测试函数""" + print("视频导出功能测试") + print("=" * 60) + + # 测试解析功能 + test_parse_image_info() + + # 测试查找功能 + test_find_video_file() + + # 询问是否测试导出功能 + print("\n" + "=" * 60) + print("注意：视频导出测试需要：") + print(" 1. 安装 ffmpeg") + print(" 2. 存在对应的视频文件") + print("=" * 60) + + response = input("\n是否测试视频导出功能？(y/n): ").strip().lower() + if response == 'y': + test_export_video_clip() + + print("\n" + "=" * 60) + print("测试完成！") + print("=" * 60) + + +if __name__ == "__main__": + main() diff --git a/video_cutter.py b/video_cutter.py new file mode 100644 index 0000000..5d0f5d2 --- /dev/null +++ b/video_cutter.py @@ -0,0 +1,329 @@ +#!/usr/bin/env python3 +""" +视频切割导出模块 +功能：根据图片名找到原始视频，按时间范围切割并导出 +依赖：系统需要安装 ffmpeg +""" + +import os +import re +import subprocess +from pathlib import Path +from typing import Optional + +# ========== 配置 ========== +VIDEO_DIR = "./shipin" # 原始视频目录 +EXPORT_DIR = "./export_videos" # 导出视频目录 +VIDEO_EXTENSIONS = {'.avi', '.mp4', '.mkv', '.mov', '.flv', '.wmv', '.webm', '.ts', '.m4v'} +# ========================== + + +def parse_image_info(image_name: str) -> Optional[dict]: + """ + 从图片名解析视频名和时间点 + + Args: + image_name: 图片文件名，例如 '001 Puss Gets the Boot [1940]_00m05s_213.jpg' + + Returns: + { + 'video_name': '001 Puss Gets the Boot [1940]', + 'timestamp_seconds': 5.213 + } + 解析失败返回 None + """ + # 移除扩展名 + name_without_ext = Path(image_name).stem + + # 匹配时间戳格式：_XXmYYs_ZZZ + match = re.search(r'_(\d{2})m(\d{2})s_(\d{3})$', name_without_ext) + if not match: + return None + + minutes = int(match.group(1)) + seconds = int(match.group(2)) + milliseconds = int(match.group(3)) + timestamp_seconds = minutes * 60 + seconds + milliseconds / 1000 + + # 提取视频名（去掉时间戳部分） + video_name = name_without_ext[:match.start()] + + return { + 'video_name': video_name, + 'timestamp_seconds': timestamp_seconds + } + + +def find_video_file(video_name: str) -> Optional[str]: + """ + 根据视频名找到对应的视频文件 + + Args: + video_name: 视频名，例如 '001 Puss Gets the Boot [1940]' + + Returns: + 视频文件路径，未找到返回 None + """ + video_dir = Path(VIDEO_DIR) + for ext in VIDEO_EXTENSIONS: + video_path = video_dir / f"{video_name}{ext}" + if video_path.exists(): + return str(video_path) + return None + + +def get_video_duration(video_path: str) -> float: + """ + 获取视频总时长（秒） + + Args: + video_path: 视频文件路径 + + Returns: + 视频时长（秒），失败返回 0.0 + """ + try: + cmd = [ + 'ffprobe', + '-v', 'error', + '-show_entries', 'format=duration', + '-of', 'default=noprint_wrappers=1:nokey=1', + video_path + ] + result = subprocess.run(cmd, capture_output=True, text=True, timeout=30) + if result.returncode == 0: + return float(result.stdout.strip()) + except (subprocess.TimeoutExpired, ValueError): + pass + return 0.0 + + +def format_time(seconds: float) -> str: + """ + 格式化时间：秒 -> '01m23s' + + Args: + seconds: 秒数 + + Returns: + 格式化的时间字符串 + """ + mins = int(seconds // 60) + secs = int(seconds % 60) + return f"{mins:02d}m{secs:02d}s" + + +def cut_video( + video_path: str, + start_time: float, + end_time: float, + output_path: str +) -> dict: + """ + 使用 ffmpeg 切割视频 + + Args: + video_path: 原始视频路径 + start_time: 开始时间（秒） + end_time: 结束时间（秒） + output_path: 输出路径 + + Returns: + { + 'success': bool, + 'output_path': str, + 'start_time': float, + 'end_time': float, + 'duration': float, + 'error': str + } + """ + # 确保输出目录存在 + os.makedirs(os.path.dirname(output_path), exist_ok=True) + + # 计算持续时间 + duration = end_time - start_time + + cmd = [ + 'ffmpeg', + '-i', video_path, + '-ss', str(start_time), + '-t', str(duration), + '-c', 'copy', # 直接复制，不重新编码（速度快） + '-y', # 覆盖已存在文件 + '-avoid_negative_ts', 'make_zero', # 避免负时间戳 + output_path + ] + + try: + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=600 # 10分钟超时 + ) + if result.returncode == 0: + return { + 'success': True, + 'output_path': output_path, + 'start_time': start_time, + 'end_time': end_time, + 'duration': duration, + 'error': None + } + else: + return { + 'success': False, + 'output_path': None, + 'start_time': start_time, + 'end_time': end_time, + 'duration': duration, + 'error': f"ffmpeg 错误: {result.stderr[-300:]}" + } + except subprocess.TimeoutExpired: + return { + 'success': False, + 'output_path': None, + 'start_time': start_time, + 'end_time': end_time, + 'duration': duration, + 'error': "视频切割超时（超过10分钟）" + } + + +def export_video_clip( + image_name: str, + mode: str = "around", + seconds: int = 300 +) -> dict: + """ + 根据图片导出视频片段 + + Args: + image_name: 图片文件名 + mode: 导出模式 + - "around": 前后各 N 秒 + - "after": 后 N 秒 + - "before": 前 N 秒 + - "total": 整个视频 + seconds: 秒数（total 模式下忽略） + + Returns: + { + 'success': bool, + 'video_path': str, # 导出的视频路径 + 'video_name': str, # 视频名 + 'start_time': str, # 开始时间（格式化） + 'end_time': str, # 结束时间（格式化） + 'duration': str, # 时长（格式化） + 'original_image': str, # 原始图片名 + 'original_timestamp': str, # 图片在视频中的时间点 + 'error': str # 错误信息 + } + """ + # 1. 解析图片信息 + info = parse_image_info(image_name) + if not info: + return { + 'success': False, + 'error': f"无法解析图片名: {image_name}（格式应为：视频名_XXmYYs_ZZZ.jpg）" + } + + video_name = info['video_name'] + timestamp = info['timestamp_seconds'] + + # 2. 找到原始视频 + video_path = find_video_file(video_name) + if not video_path: + return { + 'success': False, + 'error': f"未找到视频: {video_name}（请确认视频文件在 {VIDEO_DIR} 目录下）" + } + + # 3. 获取视频总时长 + video_duration = get_video_duration(video_path) + if video_duration <= 0: + return { + 'success': False, + 'error': "无法获取视频时长，请确认视频文件格式正确" + } + + # 4. 计算切割范围 + if mode == "total": + start_time = 0 + end_time = video_duration + elif mode == "around": + start_time = max(0, timestamp - seconds) + end_time = min(video_duration, timestamp + seconds) + elif mode == "after": + start_time = timestamp + end_time = min(video_duration, timestamp + seconds) + elif mode == "before": + start_time = max(0, timestamp - seconds) + end_time = timestamp + else: + return { + 'success': False, + 'error': f"未知模式: {mode}（支持：around/after/before/total）" + } + + # 5. 生成输出路径 + start_str = format_time(start_time) + end_str = format_time(end_time) + output_filename = f"{video_name}_{start_str}-{end_str}.mp4" + output_path = os.path.join(EXPORT_DIR, output_filename) + + # 6. 执行切割 + print(f"开始切割视频: {video_name}") + print(f" 时间范围: {start_str} -> {end_str}") + print(f" 输出路径: {output_path}") + + result = cut_video(video_path, start_time, end_time, output_path) + + if result['success']: + print(f"✅ 视频导出成功: {output_path}") + return { + 'success': True, + 'video_path': output_path, + 'video_name': video_name, + 'start_time': start_str, + 'end_time': end_str, + 'duration': format_time(result['duration']), + 'original_image': image_name, + 'original_timestamp': format_time(timestamp), + 'error': None + } + else: + print(f"❌ 视频导出失败: {result['error']}") + return { + 'success': False, + 'error': result['error'] + } + + +# ========== 测试 ========== +if __name__ == "__main__": + import sys + + # 测试解析功能 + test_images = [ + "001 Puss Gets the Boot [1940]_05m23s_456.jpg", + "002 The Midnight Snack [1941]_02m15s_123.jpg", + "invalid_image_name.jpg" + ] + + print("=== 测试图片名解析 ===") + for img in test_images: + info = parse_image_info(img) + print(f" {img}") + print(f" -> {info}") + + # 测试导出功能（如果有命令行参数） + if len(sys.argv) > 1: + print("\n=== 测试视频导出 ===") + test_image = sys.argv[1] + test_mode = sys.argv[2] if len(sys.argv) > 2 else "around" + test_seconds = int(sys.argv[3]) if len(sys.argv) > 3 else 120 + + result = export_video_clip(test_image, mode=test_mode, seconds=test_seconds) + print(f"\n导出结果: {result}") \ No newline at end of file diff --git a/video_to_frames.py b/video_to_frames.py new file mode 100644 index 0000000..9ca9a93 --- /dev/null +++ b/video_to_frames.py @@ -0,0 +1,166 @@ +#!/usr/bin/env python3 +""" +视频切割脚本 - 将视频逐帧切割为图片 +支持扫描目录下所有视频文件，多进程并行处理，已切割的自动跳过 +""" + +import os +import sys +import cv2 +import argparse +from multiprocessing import Pool, cpu_count +from pathlib import Path + +# 支持的视频扩展名 +VIDEO_EXTENSIONS = {'.mp4', '.avi', '.mkv', '.mov', '.flv', '.wmv', '.webm', '.ts', '.m4v'} + + +def scan_videos(root_path: str) -> list[str]: + """扫描目录下所有视频文件""" + videos = [] + root = Path(root_path) + if root.is_file() and root.suffix.lower() in VIDEO_EXTENSIONS: + return [str(root)] + for p in sorted(root.rglob('*')): + if p.is_file() and p.suffix.lower() in VIDEO_EXTENSIONS: + videos.append(str(p)) + return videos + + +def extract_frames(video_path: str, output_dir: str = None) -> dict: + """ + 将单个视频按每秒1帧切割为图片 + + Args: + video_path: 视频文件路径 + output_dir: 输出目录（默认在视频同目录下创建同名文件夹） + + Returns: + 处理结果字典 + """ + video_path = os.path.abspath(video_path) + video_name = Path(video_path).stem + + if output_dir is None: + output_dir = os.path.join(os.path.dirname(video_path), video_name + '_frames') + + # 已经切割过则跳过 + if os.path.isdir(output_dir) and any(Path(output_dir).glob('*.jpg')): + existing = len(list(Path(output_dir).glob('*.jpg'))) + return { + 'video': video_path, + 'status': 'skipped', + 'reason': f'已存在，共 {existing} 张图片', + 'output': output_dir, + } + + os.makedirs(output_dir, exist_ok=True) + + cap = cv2.VideoCapture(video_path) + if not cap.isOpened(): + return { + 'video': video_path, + 'status': 'error', + 'reason': '无法打开视频文件', + 'output': output_dir, + } + + fps = cap.get(cv2.CAP_PROP_FPS) or 25 + total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) + duration = total_frames / fps if fps else 0 + + # 每秒取一帧：根据fps算出每隔多少帧取一帧 + frame_interval = int(fps) + frame_idx = 0 + saved = 0 + while True: + ret, frame = cap.read() + if not ret: + break + if frame_idx % frame_interval == 0: + # 用视频名+时间戳命名：第01集_01m23s_456.jpg + sec = frame_idx / fps + mins = int(sec // 60) + secs = int(sec % 60) + ms = int((sec % 1) * 1000) + filename = os.path.join(output_dir, f'{video_name}_{mins:02d}m{secs:02d}s_{ms:03d}.jpg') + cv2.imwrite(filename, frame) + saved += 1 + frame_idx += 1 + + cap.release() + + return { + 'video': video_path, + 'status': 'done', + 'frames_total': total_frames, + 'frames_saved': saved, + 'duration': f'{duration:.1f}s', + 'fps': fps, + 'output': output_dir, + } + + +def _worker(args): + """多进程 worker 包装""" + video_path, output_dir = args + try: + return extract_frames(video_path, output_dir) + except Exception as e: + return { + 'video': video_path, + 'status': 'error', + 'reason': str(e), + } + + +def main(video_path, output=None, interval=1, jobs=None): + videos = scan_videos(video_path) + if not videos: + print(f'未在 {video_path} 中找到视频文件') + return + + jobs = jobs or cpu_count() + print(f'找到 {len(videos)} 个视频，使用 {jobs} 个进程并行处理') + + tasks = [] + for v in videos: + if output: + video_name = Path(v).stem + out = os.path.join(output, video_name + '_frames') + else: + out = None + tasks.append((v, out)) + + done = 0 + skipped = 0 + errors = 0 + + with Pool(jobs) as pool: + for result in pool.imap_unordered(_worker, tasks): + status = result['status'] + video = result['video'] + if status == 'done': + done += 1 + print(f'[完成] {video} -> {result["frames_saved"]}帧 ' + f'(总{result["frames_total"]}, {result["duration"]})') + elif status == 'skipped': + skipped += 1 + print(f'[跳过] {video} ({result["reason"]})') + else: + errors += 1 + print(f'[错误] {video} ({result["reason"]})') + + print(f'\n处理完毕: 完成 {done}, 跳过 {skipped}, 错误 {errors}, 共 {len(videos)}') + + +# ========== 运行设置（改这里）========== +VIDEO_DIR = "./shipin" # 视频文件或目录路径 +OUTPUT_DIR = None # 输出目录，None 则在视频同目录 +INTERVAL = 1 # 每隔 N 帧取一帧 +JOBS = None # 并行进程数，None 则用 CPU 核心数 +# ====================================== + + +if __name__ == '__main__': + main(VIDEO_DIR, OUTPUT_DIR, INTERVAL, JOBS) diff --git a/weixin.py b/weixin.py new file mode 100644 index 0000000..eeacb9c --- /dev/null +++ b/weixin.py @@ -0,0 +1 @@ +from weixin_sender import weix diff --git a/weixin_sender.py b/weixin_sender.py new file mode 100644 index 0000000..99c0d97 --- /dev/null +++ b/weixin_sender.py @@ -0,0 +1,476 @@ +import asyncio +import os +from pathlib import Path +from weixin_bot import WeixinBot + +# ---------- 发送图片的核心函数（不依赖 OpenClaw）---------- +import httpx +import json +import base64 +import random +import time +import hashlib +import uuid +from Crypto.Cipher import AES +from Crypto.Util.Padding import pad +from urllib.parse import quote + +def calculate_encrypted_size(raw_size: int) -> int: + return ((raw_size + 1 + 15) // 16) * 16 + +def prepare_image_upload(image_path: str): + if not Path(image_path).exists(): + return {"success": False, "error": f"图片文件不存在: {image_path}"} + with open(image_path, 'rb') as f: + file_data = f.read() + rawsize = len(file_data) + rawfilemd5 = hashlib.md5(file_data).hexdigest() + filekey = ''.join(random.choices('0123456789abcdef', k=32)) + aeskey_hex = ''.join(random.choices('0123456789abcdef', k=32)) + filesize = calculate_encrypted_size(rawsize) + return { + "success": True, + "filekey": filekey, + "aeskey_hex": aeskey_hex, + "rawsize": rawsize, + "rawfilemd5": rawfilemd5, + "filesize": filesize, + "file_data": file_data + } + +def aes_encrypt_file(file_path: str, aes_key_hex: str) -> bytes: + aes_key = bytes.fromhex(aes_key_hex) + with open(file_path, 'rb') as f: + file_data = f.read() + cipher = AES.new(aes_key, AES.MODE_ECB) + padded_data = pad(file_data, AES.block_size, style='pkcs7') + encrypted_data = cipher.encrypt(padded_data) + return encrypted_data + +def encode_aes_key(aes_key_hex: str) -> str: + hex_bytes = aes_key_hex.encode('utf-8') + return base64.b64encode(hex_bytes).decode('utf-8') + +def get_upload_params(bot_token: str, filekey: str, media_type: int, to_user_id: str, + rawsize: int, rawfilemd5: str, filesize: int, aeskey_hex: str): + random_uint32 = os.urandom(4) + x_wechat_uin = base64.b64encode(random_uint32).decode('utf-8') + body = { + "filekey": filekey, + "media_type": media_type, + "to_user_id": to_user_id, + "rawsize": rawsize, + "rawfilemd5": rawfilemd5, + "filesize": filesize, + "no_need_thumb": True, + "aeskey": aeskey_hex, + "base_info": {"channel_version": "1.0.0"} + } + raw = json.dumps(body, ensure_ascii=False) + headers = { + "Content-Type": "application/json", + "AuthorizationType": "ilink_bot_token", + "Authorization": f"Bearer {bot_token}", + "X-WECHAT-UIN": x_wechat_uin, + "Content-Length": str(len(raw.encode("utf-8"))), + } + import requests + resp = requests.post( + "https://ilinkai.weixin.qq.com/ilink/bot/getuploadurl", + headers=headers, + data=raw.encode('utf-8'), + timeout=15 + ) + if resp.status_code != 200: + return {"success": False, "error": f"申请上传参数失败，HTTP状态码: {resp.status_code}"} + result = resp.json() + upload_param = result.get("upload_param", "") + if not upload_param: + return {"success": False, "error": "响应中未包含upload_param"} + return {"success": True, "upload_param": upload_param} + +def upload_to_cdn(upload_param: str, filekey: str, encrypted_data: bytes): + encoded_param = quote(upload_param, safe='') + cdn_url = f"https://novac2c.cdn.weixin.qq.com/c2c/upload?encrypted_query_param={encoded_param}&filekey={filekey}" + headers = {"Content-Type": "application/octet-stream"} + resp = httpx.post(cdn_url, headers=headers, content=encrypted_data, timeout=30) + if resp.status_code == 200: + encrypt_query_param = resp.headers.get('x-encrypted-param', '') + if not encrypt_query_param: + return {"success": False, "error": "未在响应头中找到x-encrypted-param"} + return {"success": True, "encrypt_query_param": encrypt_query_param} + else: + return {"success": False, "error": f"上传失败，HTTP状态码: {resp.status_code}"} + +def send_weixin_image(bot_token: str, to_user_id: str, context_token: str, image_path: str) -> dict: + """发送图片，需要提供正确的 context_token。返回 {"success": bool, "new_context_token": str}""" + prepare_result = prepare_image_upload(image_path) + if not prepare_result["success"]: + print(f"准备图片失败: {prepare_result.get('error')}") + return {"success": False, "new_context_token": ""} + + filekey = prepare_result["filekey"] + aeskey_hex = prepare_result["aeskey_hex"] + rawsize = prepare_result["rawsize"] + rawfilemd5 = prepare_result["rawfilemd5"] + filesize = prepare_result["filesize"] + + upload_params = get_upload_params(bot_token, filekey, 1, to_user_id, + rawsize, rawfilemd5, filesize, aeskey_hex) + if not upload_params["success"]: + print(f"申请上传失败: {upload_params.get('error')}") + return {"success": False, "new_context_token": ""} + + encrypted_data = aes_encrypt_file(image_path, aeskey_hex) + upload_result = upload_to_cdn(upload_params["upload_param"], filekey, encrypted_data) + if not upload_result["success"]: + print(f"上传CDN失败: {upload_result.get('error')}") + return {"success": False, "new_context_token": ""} + encrypt_query_param = upload_result["encrypt_query_param"] + encoded_aes_key = encode_aes_key(aeskey_hex) + + # 构造最终消息 + timestamp_ms = int(time.time() * 1000) + random_suffix = uuid.uuid4().hex[:8] + client_id = f"weixin-bot:{timestamp_ms}-{random_suffix}" + random_uint32 = os.urandom(4) + x_wechat_uin = base64.b64encode(random_uint32).decode('utf-8') + + payload = { + "msg": { + "from_user_id": "", + "to_user_id": to_user_id, + "client_id": client_id, + "message_type": 2, + "message_state": 2, + "context_token": context_token, + "item_list": [ + { + "type": 2, + "image_item": { + "media": { + "encrypt_query_param": encrypt_query_param, + "aes_key": encoded_aes_key, + "encrypt_type": 1 + }, + "mid_size": rawsize + } + } + ] + }, + "base_info": {"channel_version": "1.0.0"} + } + + raw = json.dumps(payload, ensure_ascii=False) + headers = { + "Content-Type": "application/json", + "AuthorizationType": "ilink_bot_token", + "Authorization": f"Bearer {bot_token}", + "X-WECHAT-UIN": x_wechat_uin, + "Content-Length": str(len(raw.encode("utf-8"))), + } + import requests + resp = requests.post( + "https://ilinkai.weixin.qq.com/ilink/bot/sendmessage", + headers=headers, + data=raw.encode('utf-8'), + timeout=15 + ) + if resp.status_code == 200: + resp_json = resp.json() + # 调试：打印完整响应，确认 context_token 位置 + print(f"[DEBUG] sendmessage 响应: {json.dumps(resp_json, ensure_ascii=False)[:500]}") + # 尝试从 msg.context_token 或顶层 context_token 获取 + new_ctx = ( + resp_json.get("msg", {}).get("context_token", "") + or resp_json.get("context_token", "") + ) + ret = resp_json.get("ret") + if ret == 0 or not resp_json: + print(f"✅ 图片发送成功, new_ctx={'有' if new_ctx else '无'}") + return {"success": True, "new_context_token": new_ctx} + else: + print(f"❌ 发送失败，ret={ret}") + return {"success": False, "new_context_token": new_ctx} + else: + print(f"❌ HTTP错误 {resp.status_code}") + return {"success": False, "new_context_token": ""} + + +def send_weixin_text(bot_token: str, to_user_id: str, context_token: str, text: str) -> dict: + """发送文字消息，返回 {"success": bool, "new_context_token": str}""" + timestamp_ms = int(time.time() * 1000) + random_suffix = uuid.uuid4().hex[:8] + client_id = f"weixin-bot:{timestamp_ms}-{random_suffix}" + random_uint32 = os.urandom(4) + x_wechat_uin = base64.b64encode(random_uint32).decode('utf-8') + + payload = { + "msg": { + "from_user_id": "", + "to_user_id": to_user_id, + "client_id": client_id, + "message_type": 2, + "message_state": 2, + "context_token": context_token, + "item_list": [ + { + "type": 1, + "text_item": {"text": text} + } + ] + }, + "base_info": {"channel_version": "1.0.0"} + } + + raw = json.dumps(payload, ensure_ascii=False) + headers = { + "Content-Type": "application/json", + "AuthorizationType": "ilink_bot_token", + "Authorization": f"Bearer {bot_token}", + "X-WECHAT-UIN": x_wechat_uin, + "Content-Length": str(len(raw.encode("utf-8"))), + } + import requests + resp = requests.post( + "https://ilinkai.weixin.qq.com/ilink/bot/sendmessage", + headers=headers, + data=raw.encode('utf-8'), + timeout=15 + ) + if resp.status_code == 200: + resp_json = resp.json() + new_ctx = ( + resp_json.get("msg", {}).get("context_token", "") + or resp_json.get("context_token", "") + ) + ret = resp_json.get("ret") + if ret == 0 or not resp_json: + print(f"✅ 文字发送成功") + return {"success": True, "new_context_token": new_ctx} + else: + print(f"❌ 文字发送失败，ret={ret}, errmsg={resp_json.get('errmsg', '')}") + return {"success": False, "new_context_token": new_ctx} + else: + print(f"❌ 文字HTTP错误 {resp.status_code}") + return {"success": False, "new_context_token": ""} + + +class weix: + def __init__(self): + self.bot = WeixinBot() + self.bot.login() + self.BOT_TOKEN = self.bot._credentials.token + self.MY_USER_ID = self.bot._credentials.user_id + self.bot.on_message(self.handle_message) + self.msg = "" + + async def send_text(self, input_text): + # 不通过 bot.reply()，改用独立函数发送，避免 context_token 冲突 + context_token = self.bot._context_tokens.get(self.msg.user_id) + result = await asyncio.to_thread( + send_weixin_text, + self.BOT_TOKEN, + self.msg.user_id, + context_token, + input_text + ) + if result.get("new_context_token"): + self.bot._context_tokens[self.msg.user_id] = result["new_context_token"] + self.msg._context_token = result["new_context_token"] + return result["success"] + + async def send_images(self, image_path): + # 发送图片，返回是否成功 + context_token = self.bot._context_tokens.get(self.msg.user_id) + result = await asyncio.to_thread( + send_weixin_image, + self.BOT_TOKEN, + self.msg.user_id, + context_token, + image_path + ) + # 用服务端返回的新 context_token 更新缓存，避免后续发送失败 + if result.get("new_context_token"): + self.bot._context_tokens[self.msg.user_id] = result["new_context_token"] + self.msg._context_token = result["new_context_token"] + return result["success"] + + async def send_video(self, video_path): + """发送视频文件""" + context_token = self.bot._context_tokens.get(self.msg.user_id) + result = await asyncio.to_thread( + send_weixin_file, + self.BOT_TOKEN, + self.msg.user_id, + context_token, + video_path + ) + return result + + async def handle_message(self, msg): + print(f"收到消息: {msg.text} from {msg.user_id}") + self.msg = msg + + +def send_weixin_file(bot_token: str, to_user_id: str, context_token: str, file_path: str) -> bool: + """ + 发送任意文件（图片、视频、音频、文档等） + 返回 True/False + """ + # 判断文件类型 + file_ext = Path(file_path).suffix.lower().lstrip('.') + image_exts = ['jpg','jpeg','png','gif','bmp','webp','tiff','svg'] + video_exts = ['mp4','mov','avi','wmv','flv','mkv','webm','mpeg','mpg'] + audio_exts = ['mp3','wav','aac','flac','m4a','ogg','wma'] + doc_exts = ['pdf','doc','docx','xls','xlsx','ppt','pptx','txt'] + archive_exts = ['zip','rar','7z','tar','gz'] + + if file_ext in image_exts: + media_type = 1 + elif file_ext in video_exts: + media_type = 2 + elif file_ext in audio_exts: + media_type = 4 + elif file_ext in doc_exts or file_ext in archive_exts: + media_type = 3 + else: + media_type = 3 # 默认当作普通文件 + + # 1. 准备上传参数 + prepare_result = prepare_image_upload(file_path) # 复用之前的准备函数 + if not prepare_result["success"]: + print(f"准备文件失败: {prepare_result.get('error')}") + return False + + filekey = prepare_result["filekey"] + aeskey_hex = prepare_result["aeskey_hex"] + rawsize = prepare_result["rawsize"] + rawfilemd5 = prepare_result["rawfilemd5"] + filesize = prepare_result["filesize"] + + # 2. 申请上传参数（媒体类型已确定） + upload_params = get_upload_params(bot_token, filekey, media_type, to_user_id, + rawsize, rawfilemd5, filesize, aeskey_hex) + if not upload_params["success"]: + print(f"申请上传失败: {upload_params.get('error')}") + return False + + # 3. 加密文件 + encrypted_data = aes_encrypt_file(file_path, aeskey_hex) + + # 4. 上传到 CDN + upload_result = upload_to_cdn(upload_params["upload_param"], filekey, encrypted_data) + if not upload_result["success"]: + print(f"上传CDN失败: {upload_result.get('error')}") + return False + encrypt_query_param = upload_result["encrypt_query_param"] + encoded_aes_key = encode_aes_key(aeskey_hex) + + # 5. 构造消息体（根据 media_type 不同） + # 公共字段 + timestamp_ms = int(time.time() * 1000) + random_suffix = uuid.uuid4().hex[:8] + client_id = f"weixin-bot:{timestamp_ms}-{random_suffix}" + random_uint32 = os.urandom(4) + x_wechat_uin = base64.b64encode(random_uint32).decode('utf-8') + + # 基础消息结构 + msg_base = { + "msg": { + "from_user_id": "", + "to_user_id": to_user_id, + "client_id": client_id, + "message_type": 2, + "message_state": 2, + "context_token": context_token, + "item_list": [] + }, + "base_info": {"channel_version": "1.0.0"} + } + + # 根据类型填充 item + if media_type == 1: # 图片 + item = { + "type": 2, + "image_item": { + "media": { + "encrypt_query_param": encrypt_query_param, + "aes_key": encoded_aes_key, + "encrypt_type": 1 + }, + "mid_size": rawsize + } + } + elif media_type == 2: # 视频 + item = { + "type": 5, + "video_item": { + "media": { + "encrypt_query_param": encrypt_query_param, + "aes_key": encoded_aes_key, + "encrypt_type": 1 + }, + "video_size": rawsize # 明文大小 + } + } + elif media_type == 3: # 普通文件 + item = { + "type": 4, + "file_item": { + "media": { + "encrypt_query_param": encrypt_query_param, + "aes_key": encoded_aes_key, + "encrypt_type": 1 + }, + "file_name": Path(file_path).name + # 注意：main_send_file.py 中注释说 len 和 md5 写了可能发不出去，所以不加 + } + } + elif media_type == 4: # 音频（通常也是文件类型，但微信可能支持语音，此处按文件处理） + # 微信音频消息类型可能是 3？文档不详，暂按文件处理 + # 但为了兼容，我们可以也作为普通文件发送 + item = { + "type": 4, + "file_item": { + "media": { + "encrypt_query_param": encrypt_query_param, + "aes_key": encoded_aes_key, + "encrypt_type": 1 + }, + "file_name": Path(file_path).name + } + } + else: + print(f"不支持的媒体类型: {media_type}") + return False + + msg_base["msg"]["item_list"].append(item) + + # 6. 发送请求 + raw = json.dumps(msg_base, ensure_ascii=False) + headers = { + "Content-Type": "application/json", + "AuthorizationType": "ilink_bot_token", + "Authorization": f"Bearer {bot_token}", + "X-WECHAT-UIN": x_wechat_uin, + "Content-Length": str(len(raw.encode("utf-8"))), + } + import requests + resp = requests.post( + "https://ilinkai.weixin.qq.com/ilink/bot/sendmessage", + headers=headers, + data=raw.encode('utf-8'), + timeout=15 + ) + if resp.status_code == 200: + resp_json = resp.json() + if resp_json.get('ret') == 0: + print(f"✅ {Path(file_path).name} 发送成功") + return True + else: + print(f"❌ 发送失败，ret={resp_json.get('ret')}") + return False + else: + print(f"❌ HTTP错误 {resp.status_code}") + return False \ No newline at end of file diff --git a/猫和老鼠分镜.md b/猫和老鼠分镜.md new file mode 100644 index 0000000..528fea3 --- /dev/null +++ b/猫和老鼠分镜.md @@ -0,0 +1,38 @@ +# 角色：动漫场景匹配师 + +## 背景 + +你只做一件事：根据我的剧情描述，从指定动漫中返回最符合我剧情的场景/动作描述，并输出此场景的"英语"提示词。 + +## 指定动漫 + +这里需要用户进行填写，对话前需要询问用户，用户可随时改变动漫，如果未指定需要提醒用户。如果用户未指定提问则拒绝回答！ + +## 执行规则（按顺序） + +1. **场景可行性判断**：收到剧情描述后，需要先判断剧情是否可能存在于指定动漫中。例如：用户场景为坐飞机，动漫为"熊出没"，熊出没不会有坐飞机场景，所以直接回复：指定动漫不会有此场景。 + +2. **无需精确集数**：无需从记忆中返回指定集数、第几季，只需返回大概场景即可。 + +3. **相似场景查询**：可根据用户提供的场景查询动漫中指定相似场景。例如：用户场景为坐电脑前敲电脑，动漫为猫和老鼠，动漫中没有敲电脑场景，但是有TOM猫弹钢琴动作与敲电脑相似，那么输出提示词为：TOM猫敲钢琴。可根据角色动作、角色神情、场景布局来做相似查询。 + +4. **提示词格式**：输出提示词格式为 `[角色] + [动作/情绪状态] + [环境/背景] + [关键道具或细节]`，总长度不超过20个英文单词或30个中文字符。 + +5. **静态画面限制**：提示词只描述静态画面，不返回连续动画场景、动作（例如：跳起来然后落下、跑过去跑回来）。 + +6. **提示词语言**：默认使用中文，除非用户指定语言。 + +## 输出格式（严格遵守） + +1. 禁止任何开场白（例如：好的、收到、现在立即回复...） +2. 禁止任何结束语（例如：完成、希望对你有帮助...） +3. 禁止输出场景外的任何解释 +4. 如果找不到任何匹配场景，只输出一行："未找到匹配场景，请尝试更具体的剧情描述。" +5. 输出格式必须为中文和英文,例如:中文:tom疲惫了;英文:Tom is tired. + +## 绝对禁止的行为 + +1. 禁止输出不存在的动漫场景 +2. 禁止编造场景 +3. 禁止在提示词中加入"风格：动漫""画质：高清"等无效词 +4. 禁止对画面进行评价