an/agent_test
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit: WeChat bot project with AI chat, video export, and image upload scripts

Browse Source
This commit is contained in:
anqiang12
2026-06-18 15:53:24 +08:00
commit 3b1e106cb0
16 changed files with 2541 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
.gitignore vendored Normal file
View File

@@ -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/

75
CLAUDE.md Normal file
View File

@@ -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

40
INSTALL_FFMPEG.md Normal file
View File

@@ -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
```
## 已安装？
如果已经安装，请忽略此文档。

173
VIDEO_EXPORT_README.md Normal file
View File

@@ -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. 磁盘空间是否充足

215
VIDEO_EXPORT_SUMMARY.md Normal file
View File

@@ -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`。

238
cha_xianglaing.py Normal file
View File

@@ -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")

232
chat_bot.py Normal file
View File

@@ -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()

125
get_tianqi.py Normal file
View File

@@ -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)

200
qdrant_tool.py Normal file
View File

@@ -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)

46
requirements.txt Normal file
View File

@@ -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

155
test_video_export.py Normal file
View File

@@ -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()

329
video_cutter.py Normal file
View File

@@ -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}")

166
video_to_frames.py Normal file
View File

@@ -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)

1
weixin.py Normal file
View File

@@ -0,0 +1 @@
from weixin_sender import weix

476
weixin_sender.py Normal file
View File

@@ -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

38
猫和老鼠分镜.md Normal file
View File

@@ -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. 禁止对画面进行评价