lizhengwei / aigc-embedding-service

Go to a project
Toggle navigation Toggle navigation pinning
lizhengwei
5468e082 1 parent a4dfad4d

jira:NYJ-1540 desc: init

Inline Side-by-side
Showing 33 changed files with 2313 additions and 2313 deletions
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
*.egg-info/
dist/
build/
# Virtual environments
.venv/
venv/
# IDE
.idea/
.vscode/
*.swp
*.swo
# Git
.git/
.gitignore
# Docker
Dockerfile
.dockerignore
# Logs
*.log
# Local configs
cfg/config.local.yaml
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
*.egg-info/
dist/
build/
# Virtual environments
.venv/
venv/
# IDE
.idea/
.vscode/
*.swp
*.swo
# Git
.git/
.gitignore
# Docker
Dockerfile
.dockerignore
# Logs
*.log
# Local configs
cfg/config.local.yaml
... ...
/live-stream-ai-task-workline/modules/wtt_highlight/model_weights/*
*mov
/.externalToolBuilders
/.settings
.svn
/.project
/target
./target
/.classpath
.idea/
*.iml
*.iws
*.ipr
*.ids
*.orig
classes/
/bin/
*/.classpath
*/.project
*/.settings/*
######################
# Project Specific
######################
/target/www/**
/src/test/javascript/coverage/
/src/test/javascript/PhantomJS*/
######################
# Node
######################
/node/
node_tmp/
node_modules/
npm-debug.log.*
/.awcache/*
######################
# SASS
######################
.sass-cache/
######################
# Eclipse
######################
*.pydevproject
.project
.metadata
tmp/
tmp/**/*
*.tmp
*.bak
*.swp
*~.nib
local.properties
.classpath
.settings/
.loadpath
.factorypath
/src/main/resources/rebel.xml
# External tool builders
.externalToolBuilders/**
# Locally stored "Eclipse launch configurations"
*.launch
# CDT-specific
.cproject
# PDT-specific
.buildpath
######################
# Intellij
######################
.idea/
*.iml
*.iws
*.ipr
*.ids
*.orig
classes/
######################
# Visual Studio Code
######################
.vscode/
######################
# Maven
######################
/log/
/target/
######################
# Gradle
######################
.gradle/
/build/
######################
# Package Files
######################
*.jar
*.war
*.ear
*.db
######################
# Windows
######################
# Windows image file caches
Thumbs.db
# Folder com.migu.oes.power.config file
Desktop.ini
######################
# Mac OSX
######################
.DS_Store
.svn
# Thumbnails
._*
# Files that might appear on external disk
.Spotlight-V100
.Trashes
######################
# Directories
######################
/bin/
/deploy/
######################
# Logs
######################
*.log*
######################
# Others
######################
*.class
*.*~
*~
.merge_file*
######################
# Gradle Wrapper
######################
!gradle/wrapper/gradle-wrapper.jar
######################
# Maven Wrapper
######################
!.mvn/wrapper/maven-wrapper.jar
######################
# ESLint
######################
.eslintcache
*.lst
oes-nryy-vo/target/maven-archiver/pom.properties
oes-nryy-dataservice/target/maven-archiver/pom.properties
oes-nryy-base/target/maven-archiver/pom.properties
oes-nryy-copyright/bin/.gitkeep
oes-nryy-copyright/bin/cbak
*.original
*.gz
oes-nryy-tag/target/oes-nryy-tag-1.0.0/label/bin/service.sh
oes-nryy-tag/target/maven-archiver/pom.properties
oes-nryy-data-trans/target/maven-archiver/pom.properties
oes-nryy-tag/target/oes-nryy-tag-1.0.0/label/bin/cbak
######################
# pp-scene-det
######################
__pycache__/
######################
# live-stream-ai-task-workline
######################
modules/basketball/weights
modules/singer/weights
weights/
weights1/
files/
*.log
*.log.*
cache_data/
data/
test/
tests/
# Media files
*.jpg
*.jpeg
*.png
*.gif
*.mp4
*.avi
*.mov
*.mkv
# OS
.DS_Store
Thumbs.db
logs/
/live-stream-ai-task-workline/modules/wtt_highlight/model_weights/*
*mov
/.externalToolBuilders
/.settings
.svn
/.project
/target
./target
/.classpath
.idea/
*.iml
*.iws
*.ipr
*.ids
*.orig
classes/
/bin/
*/.classpath
*/.project
*/.settings/*
######################
# Project Specific
######################
/target/www/**
/src/test/javascript/coverage/
/src/test/javascript/PhantomJS*/
######################
# Node
######################
/node/
node_tmp/
node_modules/
npm-debug.log.*
/.awcache/*
######################
# SASS
######################
.sass-cache/
######################
# Eclipse
######################
*.pydevproject
.project
.metadata
tmp/
tmp/**/*
*.tmp
*.bak
*.swp
*~.nib
local.properties
.classpath
.settings/
.loadpath
.factorypath
/src/main/resources/rebel.xml
# External tool builders
.externalToolBuilders/**
# Locally stored "Eclipse launch configurations"
*.launch
# CDT-specific
.cproject
# PDT-specific
.buildpath
######################
# Intellij
######################
.idea/
*.iml
*.iws
*.ipr
*.ids
*.orig
classes/
######################
# Visual Studio Code
######################
.vscode/
######################
# Maven
######################
/log/
/target/
######################
# Gradle
######################
.gradle/
/build/
######################
# Package Files
######################
*.jar
*.war
*.ear
*.db
######################
# Windows
######################
# Windows image file caches
Thumbs.db
# Folder com.migu.oes.power.config file
Desktop.ini
######################
# Mac OSX
######################
.DS_Store
.svn
# Thumbnails
._*
# Files that might appear on external disk
.Spotlight-V100
.Trashes
######################
# Directories
######################
/bin/
/deploy/
######################
# Logs
######################
*.log*
######################
# Others
######################
*.class
*.*~
*~
.merge_file*
######################
# Gradle Wrapper
######################
!gradle/wrapper/gradle-wrapper.jar
######################
# Maven Wrapper
######################
!.mvn/wrapper/maven-wrapper.jar
######################
# ESLint
######################
.eslintcache
*.lst
oes-nryy-vo/target/maven-archiver/pom.properties
oes-nryy-dataservice/target/maven-archiver/pom.properties
oes-nryy-base/target/maven-archiver/pom.properties
oes-nryy-copyright/bin/.gitkeep
oes-nryy-copyright/bin/cbak
*.original
*.gz
oes-nryy-tag/target/oes-nryy-tag-1.0.0/label/bin/service.sh
oes-nryy-tag/target/maven-archiver/pom.properties
oes-nryy-data-trans/target/maven-archiver/pom.properties
oes-nryy-tag/target/oes-nryy-tag-1.0.0/label/bin/cbak
######################
# pp-scene-det
######################
__pycache__/
######################
# live-stream-ai-task-workline
######################
modules/basketball/weights
modules/singer/weights
weights/
weights1/
files/
*.log
*.log.*
cache_data/
data/
test/
tests/
# Media files
*.jpg
*.jpeg
*.png
*.gif
*.mp4
*.avi
*.mov
*.mkv
# OS
.DS_Store
Thumbs.db
logs/
cache/
\ No newline at end of file
... ...
# ============================================================================
# embedding-server Dockerfile
# ============================================================================
# 构建命令:
# docker build -f Dockerfile.embedding -t embedding-server:latest .
#
# 运行命令:
# docker run -d -p 8000:8000 --name embedding-server embedding-server:latest
# ============================================================================
ARG PYTHON_VERSION=3.12
#ARG UV_VERSION=0.6.0
# ----------------------------------------------------------------------------
# 构建阶段
# ----------------------------------------------------------------------------
FROM python:${PYTHON_VERSION}-slim AS builder
# 使用 Debian 的阿里云镜像
RUN echo "deb http://mirrors.aliyun.com/debian/ bookworm main contrib non-free" > /etc/apt/sources.list && \
echo "deb http://mirrors.aliyun.com/debian/ bookworm-updates main contrib non-free" >> /etc/apt/sources.list && \
echo "deb http://mirrors.aliyun.com/debian-security/ bookworm-security main contrib non-free" >> /etc/apt/sources.list
# 安装编译依赖和 uv
RUN apt-get update \
&& apt-get install -y --no-install-recommends gcc curl vim \
&& rm -rf /var/lib/apt/lists/*
RUN pip install --no-cache-dir uv
WORKDIR /app
# 复制项目文件（仅 embedding-server 所需）
COPY pyproject.toml uv.lock ./
COPY src/embedding_sever/ ./src/embedding_sever/
# 安装依赖：核心依赖 + embedding 专用依赖
RUN uv sync --no-dev --no-editable --extra embedding
# ----------------------------------------------------------------------------
# 运行阶段
# ----------------------------------------------------------------------------
FROM python:${PYTHON_VERSION}-slim
WORKDIR /app
# 从构建阶段复制虚拟环境和源码
COPY --from=builder /app/.venv /app/.venv
COPY --from=builder /app/src/embedding_sever /app/src/embedding_sever
# 环境变量
ENV PATH="/app/.venv/bin:$PATH"
ENV PYTHONPATH="/app/src"
ENV PROJECT_ROOT="/app/src"
ENV PYTHONUNBUFFERED=1
ENV ENV=prod
ENV APP_LOG_TYPE=file,console
EXPOSE 8000
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/')" || exit 1
CMD ["embedding-server"]
# ============================================================================
# embedding-server Dockerfile
# ============================================================================
# 构建命令:
# docker build -f Dockerfile.embedding -t embedding-server:latest .
#
# 运行命令:
# docker run -d -p 8000:8000 --name embedding-server embedding-server:latest
# ============================================================================
ARG PYTHON_VERSION=3.12
#ARG UV_VERSION=0.6.0
# ----------------------------------------------------------------------------
# 构建阶段
# ----------------------------------------------------------------------------
FROM python:${PYTHON_VERSION}-slim AS builder
# 使用 Debian 的阿里云镜像
RUN echo "deb http://mirrors.aliyun.com/debian/ bookworm main contrib non-free" > /etc/apt/sources.list && \
echo "deb http://mirrors.aliyun.com/debian/ bookworm-updates main contrib non-free" >> /etc/apt/sources.list && \
echo "deb http://mirrors.aliyun.com/debian-security/ bookworm-security main contrib non-free" >> /etc/apt/sources.list
# 安装编译依赖和 uv
RUN apt-get update \
&& apt-get install -y --no-install-recommends gcc curl vim \
&& rm -rf /var/lib/apt/lists/*
RUN pip install --no-cache-dir uv
WORKDIR /app
# 复制项目文件（仅 embedding-server 所需）
COPY pyproject.toml uv.lock ./
COPY src/embedding_sever/ ./src/embedding_sever/
# 安装依赖：核心依赖 + embedding 专用依赖
RUN uv sync --no-dev --no-editable --extra embedding
# ----------------------------------------------------------------------------
# 运行阶段
# ----------------------------------------------------------------------------
FROM python:${PYTHON_VERSION}-slim
WORKDIR /app
# 从构建阶段复制虚拟环境和源码
COPY --from=builder /app/.venv /app/.venv
COPY --from=builder /app/src/embedding_sever /app/src/embedding_sever
# 环境变量
ENV PATH="/app/.venv/bin:$PATH"
ENV PYTHONPATH="/app/src"
ENV PROJECT_ROOT="/app/src"
ENV PYTHONUNBUFFERED=1
ENV ENV=prod
ENV APP_LOG_TYPE=file,console
EXPOSE 8000
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/')" || exit 1
CMD ["embedding-server"]
... ...
### 一些启动命令
1. 启动 embedding-server
默认
```bash
uv run embedding-server
```
window 开发
```bash
$env:ENV="dev";uv run embedding-server
$env:ENV="dev";uv run football-repaly-match-server
```
安装依赖
```shell
uv sync --no-dev --no-editable --extra football-replay-match --extra embedding
### 一些启动命令
1. 启动 embedding-server
默认
```bash
uv run embedding-server
```
window 开发
```bash
$env:ENV="dev";uv run embedding-server
$env:ENV="dev";uv run football-repaly-match-server
```
安装依赖
```shell
uv sync --no-dev --no-editable --extra football-replay-match --extra embedding
```
\ No newline at end of file
... ...
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "ai_server"
version = "0.1.0"
description = "Add your description here"
requires-python = ">=3.12"
dependencies = [
"fastapi==0.136.1",
"uvicorn[standard]==0.46.0",
"pydantic==2.13.4",
"numpy>=1.24.0",
"aiohttp>=3.8.0",
"aabd==0.4.8",
"path==17.1.1",
"omegaconf==2.3.0",
"packaging==26.2",
]
[project.optional-dependencies]
# embedding-server 专用依赖
embedding = [
"elasticsearch==9.3.0",
]
football-replay-match = [
"Pillow==12.2.0",
"av==13.1.0",
"confluent_kafka==2.6.1",
"langchain-openai>=1.1.15",
"opencv-contrib-python==4.13.0.92"
]
# task-server 专用依赖
task = [
]
[tool.uv]
package = true
[tool.hatch.build.targets.wheel]
packages = ["src/embedding_sever"]
[project.scripts]
embedding-server = "embedding_sever.main:main"
#task-server = "task_server.main:main"
football-repaly-match-server = "football_replay_match.main:main"
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "ai_server"
version = "0.1.0"
description = "Add your description here"
requires-python = ">=3.12"
dependencies = [
"fastapi==0.136.1",
"uvicorn[standard]==0.46.0",
"pydantic==2.13.4",
"numpy>=1.24.0",
"aiohttp>=3.8.0",
"aabd==0.4.8",
"path==17.1.1",
"omegaconf==2.3.0",
"packaging==26.2",
]
[project.optional-dependencies]
# embedding-server 专用依赖
embedding = [
"elasticsearch==9.3.0",
]
football-replay-match = [
"Pillow==12.2.0",
"av==13.1.0",
"confluent_kafka==2.6.1",
"langchain-openai>=1.1.15",
"opencv-contrib-python==4.13.0.92"
]
# task-server 专用依赖
task = [
]
[tool.uv]
package = true
[tool.hatch.build.targets.wheel]
packages = ["src/embedding_sever"]
[project.scripts]
embedding-server = "embedding_sever.main:main"
#task-server = "task_server.main:main"
football-repaly-match-server = "football_replay_match.main:main"
... ...
from typing import List, Optional, Literal, Union, Dict, Any
from fastapi import APIRouter, Depends
import numpy
from pydantic import BaseModel, Field, TypeAdapter
from embedding_sever.api.resp_bean import RespBean, success
router = APIRouter()
# ============== 请求/响应模型定义 ==============
class FaceKwargs(BaseModel):
"""人脸类型扩展字段"""
person_id: Optional[str] = Field(None, description="人物id")
person_name: Optional[str] = Field(None, description="人物姓名")
class SportShotKwargs(BaseModel):
"""体育镜头类型扩展字段"""
match_name: Optional[str] = Field(None, description="比赛名称")
person_name: Optional[str] = Field(None, description="人名")
video_desc: Optional[str] = Field(None, description="视频描述")
shot_cls: Optional[str] = Field(None, description="镜头分类")
match_id: Optional[str] = Field(None, description="比赛id")
program_id: Optional[str] = Field(None, description="节目id")
# 定义类型映射，用于根据type自动转换kwargs
KwargsType = Union[FaceKwargs, SportShotKwargs]
class PutDataRequest(BaseModel):
"""写入数据请求体"""
id: str = Field(..., description="唯一键id")
embedding: List[float] = Field(..., description="向量")
embedding_version: str = Field(..., description="向量版本 小写数字下划线组成")
kwargs: Optional[Dict[str, Any]] = Field(None, description="扩展字段，根据type不同字段不同")
def get_typed_kwargs(self, type: str) -> Optional[KwargsType]:
"""根据type自动转换kwargs为具体类型"""
if self.kwargs is None:
return None
type_map = {
'face': FaceKwargs,
'sport_shot': SportShotKwargs
}
kwargs_class = type_map.get(type)
if kwargs_class:
return kwargs_class(**self.kwargs)
return None
class DelByIdRequest(BaseModel):
"""删除数据请求体"""
ids: List[str] = Field(..., description="id 集合")
embedding_version: str = Field(..., description="向量版本 小写数字下划线组成")
class FilterItem(BaseModel):
"""过滤条件项"""
name: str = Field(..., description="字段名")
value: List = Field(..., description="字段值")
opt: Optional[Literal["eq", "neq", "lt", "gt", "lte", "gte", "like", "in"]] = Field(
None, description="操作类型: eq相等, neq不等, lt小于, gt大于, lte小于等于, gte大于等于, like模糊匹配, in内"
)
class SearchRequest(BaseModel):
"""检索数据请求体"""
embedding: List[float] = Field(..., description="向量")
embedding_version: str = Field(..., description="向量版本 小写数字下划线组成")
topk: int = Field(..., description="top k")
filters: Optional[List[FilterItem]] = Field(None, description="过滤字段")
class SearchResultItem(BaseModel):
"""检索结果单项"""
id: str = Field(..., description="id")
kwargs: dict = Field(..., description="扩展字段")
score: float = Field(..., description="分值")
class SearchResponseData(BaseModel):
"""检索响应数据"""
type: str = Field(..., description="type")
data_list: List[SearchResultItem] = Field(..., description="topk结果列表")
# ============== 接口定义 ==============
from embedding_sever.service.data_service import DataService
from embedding_sever.service import get_data_service
@router.put("/{type}/put", response_model=RespBean, tags=["数据服务"], summary="写入数据")
@router.post("/{type}/put", response_model=RespBean, tags=["数据服务"], summary="写入数据")
async def put_data(
type: Literal["face", "sport_shot"],
request: PutDataRequest,
data_service: DataService = Depends(get_data_service)
):
"""
写入数据
"""
request.embedding = numpy.array(request.embedding)
embedding_version = request.embedding_version
tb_name = f'{type}_{embedding_version}'
upserted_id = await data_service.upsert(tb_name, request.id, request.embedding,
request.get_typed_kwargs(type).model_dump())
return success(data={'id': upserted_id})
@router.delete("/{type}/del_by_id", response_model=RespBean, tags=["数据服务"], summary="删除数据")
@router.post("/{type}/del_by_id", response_model=RespBean, tags=["数据服务"], summary="删除数据")
async def del_by_id(
type: Literal["face", "sport_shot"],
request: DelByIdRequest,
data_service: DataService = Depends(get_data_service)
):
"""
删除数据
- **type**: 数据类型 (face-人脸, sport_shot-体育镜头)
- **ids**: id集合
"""
embedding_version = request.embedding_version
tb_name = f'{type}_{embedding_version}'
await data_service.delete_by_pks(tb_name, request.ids)
return success()
@router.post("/{type}/search", response_model=RespBean[SearchResponseData], tags=["数据服务"], summary="检索数据")
async def search_data(
type: Literal["face", "sport_shot"],
request: SearchRequest,
data_service: DataService = Depends(get_data_service),
adapter=TypeAdapter(List[SearchResultItem])
):
"""
检索数据
- **type**: 数据类型 (face-人脸, sport_shot-体育镜头)
- **embedding**: 查询向量
- **embedding_version**: 向量版本
- **topk**: 返回top k条结果
- **filters**: 过滤条件列表
"""
#
embedding_version = request.embedding_version
tb_name = f'{type}_{embedding_version}'
embedding = request.embedding
filters = request.filters
from_, size = 0, request.topk
result_data = await data_service.search(tb_name, embedding, filters, from_, size)
search_results: List[SearchResultItem] = adapter.validate_python(result_data)
return success(data=SearchResponseData(type=type, data_list=search_results))
from typing import List, Optional, Literal, Union, Dict, Any
from fastapi import APIRouter, Depends
import numpy
from pydantic import BaseModel, Field, TypeAdapter
from embedding_sever.api.resp_bean import RespBean, success
router = APIRouter()
# ============== 请求/响应模型定义 ==============
class FaceKwargs(BaseModel):
"""人脸类型扩展字段"""
person_id: Optional[str] = Field(None, description="人物id")
person_name: Optional[str] = Field(None, description="人物姓名")
class SportShotKwargs(BaseModel):
"""体育镜头类型扩展字段"""
match_name: Optional[str] = Field(None, description="比赛名称")
person_name: Optional[str] = Field(None, description="人名")
video_desc: Optional[str] = Field(None, description="视频描述")
shot_cls: Optional[str] = Field(None, description="镜头分类")
match_id: Optional[str] = Field(None, description="比赛id")
program_id: Optional[str] = Field(None, description="节目id")
# 定义类型映射，用于根据type自动转换kwargs
KwargsType = Union[FaceKwargs, SportShotKwargs]
class PutDataRequest(BaseModel):
"""写入数据请求体"""
id: str = Field(..., description="唯一键id")
embedding: List[float] = Field(..., description="向量")
embedding_version: str = Field(..., description="向量版本 小写数字下划线组成")
kwargs: Optional[Dict[str, Any]] = Field(None, description="扩展字段，根据type不同字段不同")
def get_typed_kwargs(self, type: str) -> Optional[KwargsType]:
"""根据type自动转换kwargs为具体类型"""
if self.kwargs is None:
return None
type_map = {
'face': FaceKwargs,
'sport_shot': SportShotKwargs
}
kwargs_class = type_map.get(type)
if kwargs_class:
return kwargs_class(**self.kwargs)
return None
class DelByIdRequest(BaseModel):
"""删除数据请求体"""
ids: List[str] = Field(..., description="id 集合")
embedding_version: str = Field(..., description="向量版本 小写数字下划线组成")
class FilterItem(BaseModel):
"""过滤条件项"""
name: str = Field(..., description="字段名")
value: List = Field(..., description="字段值")
opt: Optional[Literal["eq", "neq", "lt", "gt", "lte", "gte", "like", "in"]] = Field(
None, description="操作类型: eq相等, neq不等, lt小于, gt大于, lte小于等于, gte大于等于, like模糊匹配, in内"
)
class SearchRequest(BaseModel):
"""检索数据请求体"""
embedding: List[float] = Field(..., description="向量")
embedding_version: str = Field(..., description="向量版本 小写数字下划线组成")
topk: int = Field(..., description="top k")
filters: Optional[List[FilterItem]] = Field(None, description="过滤字段")
class SearchResultItem(BaseModel):
"""检索结果单项"""
id: str = Field(..., description="id")
kwargs: dict = Field(..., description="扩展字段")
score: float = Field(..., description="分值")
class SearchResponseData(BaseModel):
"""检索响应数据"""
type: str = Field(..., description="type")
data_list: List[SearchResultItem] = Field(..., description="topk结果列表")
# ============== 接口定义 ==============
from embedding_sever.service.data_service import DataService
from embedding_sever.service import get_data_service
@router.put("/{type}/put", response_model=RespBean, tags=["数据服务"], summary="写入数据")
@router.post("/{type}/put", response_model=RespBean, tags=["数据服务"], summary="写入数据")
async def put_data(
type: Literal["face", "sport_shot"],
request: PutDataRequest,
data_service: DataService = Depends(get_data_service)
):
"""
写入数据
"""
request.embedding = numpy.array(request.embedding)
embedding_version = request.embedding_version
tb_name = f'{type}_{embedding_version}'
upserted_id = await data_service.upsert(tb_name, request.id, request.embedding,
request.get_typed_kwargs(type).model_dump())
return success(data={'id': upserted_id})
@router.delete("/{type}/del_by_id", response_model=RespBean, tags=["数据服务"], summary="删除数据")
@router.post("/{type}/del_by_id", response_model=RespBean, tags=["数据服务"], summary="删除数据")
async def del_by_id(
type: Literal["face", "sport_shot"],
request: DelByIdRequest,
data_service: DataService = Depends(get_data_service)
):
"""
删除数据
- **type**: 数据类型 (face-人脸, sport_shot-体育镜头)
- **ids**: id集合
"""
embedding_version = request.embedding_version
tb_name = f'{type}_{embedding_version}'
await data_service.delete_by_pks(tb_name, request.ids)
return success()
@router.post("/{type}/search", response_model=RespBean[SearchResponseData], tags=["数据服务"], summary="检索数据")
async def search_data(
type: Literal["face", "sport_shot"],
request: SearchRequest,
data_service: DataService = Depends(get_data_service),
adapter=TypeAdapter(List[SearchResultItem])
):
"""
检索数据
- **type**: 数据类型 (face-人脸, sport_shot-体育镜头)
- **embedding**: 查询向量
- **embedding_version**: 向量版本
- **topk**: 返回top k条结果
- **filters**: 过滤条件列表
"""
#
embedding_version = request.embedding_version
tb_name = f'{type}_{embedding_version}'
embedding = request.embedding
filters = request.filters
from_, size = 0, request.topk
result_data = await data_service.search(tb_name, embedding, filters, from_, size)
search_results: List[SearchResultItem] = adapter.validate_python(result_data)
return success(data=SearchResponseData(type=type, data_list=search_results))
... ...
from fastapi.middleware.cors import CORSMiddleware
from starlette.responses import JSONResponse
from embedding_sever.api.http_log import LoggingMiddleware
from fastapi import FastAPI
from embedding_sever.api.router import api_router
from embedding_sever.config import settings
def config_fastapi(fastapi_app: FastAPI):
# 配置请求/响应日志中间件（需在CORS之前添加）
fastapi_app.add_middleware(LoggingMiddleware)
# 配置CORS
fastapi_app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 注册路由
fastapi_app.include_router(api_router)
@fastapi_app.exception_handler(Exception)
async def generic_exception_handler(request, exc: Exception):
# 返回自定义的错误响应
error_detail = str(exc)
return JSONResponse(
status_code=200,
content={"code": "error", "message": error_detail},
)
@fastapi_app.get("/", tags=["根路径"])
async def root():
"""根路径."""
return {
"name": settings.app_name,
"version": settings.app_version,
"docs": "/docs",
}
return fastapi_app
from fastapi.middleware.cors import CORSMiddleware
from starlette.responses import JSONResponse
from embedding_sever.api.http_log import LoggingMiddleware
from fastapi import FastAPI
from embedding_sever.api.router import api_router
from embedding_sever.config import settings
def config_fastapi(fastapi_app: FastAPI):
# 配置请求/响应日志中间件（需在CORS之前添加）
fastapi_app.add_middleware(LoggingMiddleware)
# 配置CORS
fastapi_app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 注册路由
fastapi_app.include_router(api_router)
@fastapi_app.exception_handler(Exception)
async def generic_exception_handler(request, exc: Exception):
# 返回自定义的错误响应
error_detail = str(exc)
return JSONResponse(
status_code=200,
content={"code": "error", "message": error_detail},
)
@fastapi_app.get("/", tags=["根路径"])
async def root():
"""根路径."""
return {
"name": settings.app_name,
"version": settings.app_version,
"docs": "/docs",
}
return fastapi_app
... ...
import json
import time
import logging
logger = logging.getLogger(__name__)
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import StreamingResponse
from embedding_sever.config import settings
class LoggingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
# 获取请求信息
method = request.method
url = str(request.url)
client_host = request.client.host if request.client else "unknown"
# 判断是否需要记录日志
should_log = request.url.path.startswith(settings.url_prefix)
if should_log:
# 记录请求开始时间
start_time = time.time()
# 读取请求体
request_body = None
if method in ("POST", "PUT", "PATCH"):
try:
body = await request.body()
if body:
request_body = body.decode("utf-8", errors="replace")
# 尝试格式化为JSON
try:
request_body = json.loads(request_body)
except json.JSONDecodeError:
pass # 保持原始字符串
except Exception:
pass
# 处理请求
response = await call_next(request)
# 计算处理时间
process_time = time.time() - start_time
# 读取响应体（仅处理非流式响应）
response_body = None
if not isinstance(response, StreamingResponse):
try:
response_body_bytes = b""
async for chunk in response.body_iterator:
response_body_bytes += chunk
# 重新构建响应
response_body = response_body_bytes.decode("utf-8", errors="replace")
# 尝试格式化为JSON
try:
response_body = json.loads(response_body)
except json.JSONDecodeError:
pass # 保持原始字符串
# 重建响应对象
response = StreamingResponse(
iter([response_body_bytes]),
status_code=response.status_code,
headers=dict(response.headers),
media_type=response.media_type,
)
except Exception:
pass
# 构建单条日志记录
status_code = response.status_code
log_data = {
"method": method,
"url": url,
"client": client_host,
"status": status_code,
"time": f"{process_time:.3f}s"
}
if request_body:
log_data["request_body"] = request_body
if response_body:
log_data["response_body"] = response_body
logger.info(json.dumps(log_data, ensure_ascii=False))
return response
else:
# 不需要记录日志的请求，直接处理
return await call_next(request)
import json
import time
import logging
logger = logging.getLogger(__name__)
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import StreamingResponse
from embedding_sever.config import settings
class LoggingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
# 获取请求信息
method = request.method
url = str(request.url)
client_host = request.client.host if request.client else "unknown"
# 判断是否需要记录日志
should_log = request.url.path.startswith(settings.url_prefix)
if should_log:
# 记录请求开始时间
start_time = time.time()
# 读取请求体
request_body = None
if method in ("POST", "PUT", "PATCH"):
try:
body = await request.body()
if body:
request_body = body.decode("utf-8", errors="replace")
# 尝试格式化为JSON
try:
request_body = json.loads(request_body)
except json.JSONDecodeError:
pass # 保持原始字符串
except Exception:
pass
# 处理请求
response = await call_next(request)
# 计算处理时间
process_time = time.time() - start_time
# 读取响应体（仅处理非流式响应）
response_body = None
if not isinstance(response, StreamingResponse):
try:
response_body_bytes = b""
async for chunk in response.body_iterator:
response_body_bytes += chunk
# 重新构建响应
response_body = response_body_bytes.decode("utf-8", errors="replace")
# 尝试格式化为JSON
try:
response_body = json.loads(response_body)
except json.JSONDecodeError:
pass # 保持原始字符串
# 重建响应对象
response = StreamingResponse(
iter([response_body_bytes]),
status_code=response.status_code,
headers=dict(response.headers),
media_type=response.media_type,
)
except Exception:
pass
# 构建单条日志记录
status_code = response.status_code
log_data = {
"method": method,
"url": url,
"client": client_host,
"status": status_code,
"time": f"{process_time:.3f}s"
}
if request_body:
log_data["request_body"] = request_body
if response_body:
log_data["response_body"] = response_body
logger.info(json.dumps(log_data, ensure_ascii=False))
return response
else:
# 不需要记录日志的请求，直接处理
return await call_next(request)
... ...
from pydantic import BaseModel, Field
from typing import Generic, TypeVar, Optional
T = TypeVar('T')
class RespBean(BaseModel, Generic[T]):
code: str = Field(default="success", description="code")
message: str = Field(default="success", description='message')
data: Optional[T] = Field(description="数据", default=None)
def error(message):
return RespBean(code="error", message=message, data=None)
def success(data=None):
return RespBean(data=data)
from pydantic import BaseModel, Field
from typing import Generic, TypeVar, Optional
T = TypeVar('T')
class RespBean(BaseModel, Generic[T]):
code: str = Field(default="success", description="code")
message: str = Field(default="success", description='message')
data: Optional[T] = Field(description="数据", default=None)
def error(message):
return RespBean(code="error", message=message, data=None)
def success(data=None):
return RespBean(data=data)
... ...
from fastapi import APIRouter
from embedding_sever.config import settings
from embedding_sever.api.data_router import router as data_router
api_router = APIRouter(prefix=settings.url_prefix)
def register_router(router: APIRouter, prefix: str):
api_router.include_router(router, prefix=prefix)
# 注册数据路由
register_router(data_router, prefix="/data")
from fastapi import APIRouter
from embedding_sever.config import settings
from embedding_sever.api.data_router import router as data_router
api_router = APIRouter(prefix=settings.url_prefix)
def register_router(router: APIRouter, prefix: str):
api_router.include_router(router, prefix=prefix)
# 注册数据路由
register_router(data_router, prefix="/data")
... ...
app_name: "Embedding Service"
app_version: "1.0.0"
port: 8000
debug: false
url_prefix: "/aigc-embedding/api"
db_es_enable: true
db_es_url: "http://192.168.0.14:9200"
#db_postgres_enable: false
app_name: "Embedding Service"
app_version: "1.0.0"
port: 8000
debug: false
url_prefix: "/aigc-embedding/api"
db_es_enable: true
db_es_url: "http://192.168.0.14:9200"
#db_postgres_enable: false
#db_postgres_url: "postgresql://postgres:postgres@localhost:5432/postgres"
\ No newline at end of file
... ...
app_name: "Embedding Service"
app_version: "1.0.0"
port: 8000
debug: false
url_prefix: "/aigc-embedding/api"
db_es_enable: true
db_es_url: "http://localhost:9200"
#db_postgres_enable: false
app_name: "Embedding Service"
app_version: "1.0.0"
port: 8000
debug: false
url_prefix: "/aigc-embedding/api"
db_es_enable: true
db_es_url: "http://localhost:9200"
#db_postgres_enable: false
#db_postgres_url: "postgresql://postgres:postgres@localhost:5432/postgres"
\ No newline at end of file
... ...
app_name: "Embedding Service"
app_version: "1.0.0"
port: 8000
debug: false
url_prefix: "/aigc-embedding/api"
db_es_enable: true
db_es_url: "http://localhost:9200"
#db_postgres_enable: false
app_name: "Embedding Service"
app_version: "1.0.0"
port: 8000
debug: false
url_prefix: "/aigc-embedding/api"
db_es_enable: true
db_es_url: "http://localhost:9200"
#db_postgres_enable: false
#db_postgres_url: "postgresql://postgres:postgres@localhost:5432/postgres"
\ No newline at end of file
... ...
from aabd.base.cfg_loader import load_yaml_by_file_with_env
from aabd.base.enhance_dict import EnhanceDict, read_prefixed_env_vars
from pathlib import Path
from logging import getLogger
logger = getLogger(__name__)
cfg_dir = Path(__file__).parent.absolute() / 'cfg'
settings = EnhanceDict(load_yaml_by_file_with_env((cfg_dir / 'config.yaml').as_posix()))
settings.update(read_prefixed_env_vars('EMB_'))
from aabd.base.cfg_loader import load_yaml_by_file_with_env
from aabd.base.enhance_dict import EnhanceDict, read_prefixed_env_vars
from pathlib import Path
from logging import getLogger
logger = getLogger(__name__)
cfg_dir = Path(__file__).parent.absolute() / 'cfg'
settings = EnhanceDict(load_yaml_by_file_with_env((cfg_dir / 'config.yaml').as_posix()))
settings.update(read_prefixed_env_vars('EMB_'))
logger.info(f'Settings: {settings}')
\ No newline at end of file
... ...
from functools import lru_cache
from elasticsearch import AsyncElasticsearch
from embedding_sever.config import settings
from .data_dao import DataDao
es_client = None
pg_client = None
async def connect():
if settings.db_es_enable:
global es_client
es_client = AsyncElasticsearch([settings.db_es_url])
await es_client.ping()
if settings.db_postgres_enable:
from .postgres_client import get_pg_client
global pg_client
pg_client = get_pg_client()
await pg_client.connect()
async def disconnect():
global es_client, pg_client
if es_client is not None:
await es_client.disconnect()
es_client = None
if pg_client is not None:
await pg_client.disconnect()
@lru_cache()
def get_data_dao():
return DataDao(es_client)
from functools import lru_cache
from elasticsearch import AsyncElasticsearch
from embedding_sever.config import settings
from .data_dao import DataDao
es_client = None
pg_client = None
async def connect():
if settings.db_es_enable:
global es_client
es_client = AsyncElasticsearch([settings.db_es_url])
await es_client.ping()
if settings.db_postgres_enable:
from .postgres_client import get_pg_client
global pg_client
pg_client = get_pg_client()
await pg_client.connect()
async def disconnect():
global es_client, pg_client
if es_client is not None:
await es_client.disconnect()
es_client = None
if pg_client is not None:
await pg_client.disconnect()
@lru_cache()
def get_data_dao():
return DataDao(es_client)
... ...
"""数据库抽象基类，定义统一的CRUD接口."""
from abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional
class VectorDBClient(ABC):
"""向量数据库客户端抽象基类.
定义了统一的向量数据CRUD接口，支持Elasticsearch和PostgreSQL等后端。
"""
@abstractmethod
async def connect(self) -> None:
"""建立数据库连接."""
pass
@abstractmethod
async def disconnect(self) -> None:
"""断开数据库连接."""
pass
@abstractmethod
async def health_check(self) -> bool:
"""检查数据库连接健康状态.
Returns:
bool: 连接正常返回True，否则返回False
"""
pass
class TableInit(ABC):
@abstractmethod
async def create_table(self, name, embedding_version, embedding_dim) -> None:
pass
"""数据库抽象基类，定义统一的CRUD接口."""
from abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional
class VectorDBClient(ABC):
"""向量数据库客户端抽象基类.
定义了统一的向量数据CRUD接口，支持Elasticsearch和PostgreSQL等后端。
"""
@abstractmethod
async def connect(self) -> None:
"""建立数据库连接."""
pass
@abstractmethod
async def disconnect(self) -> None:
"""断开数据库连接."""
pass
@abstractmethod
async def health_check(self) -> bool:
"""检查数据库连接健康状态.
Returns:
bool: 连接正常返回True，否则返回False
"""
pass
class TableInit(ABC):
@abstractmethod
async def create_table(self, name, embedding_version, embedding_dim) -> None:
pass
... ...
import json
import logging
import os
import threading
import typing
from elasticsearch import AsyncElasticsearch
import numpy
logger = logging.getLogger(__name__)
class DataDao:
def __init__(self, es_client: AsyncElasticsearch):
self.es_client = es_client
self._lock = threading.RLock()
async def search(self, tb_name: str,
embedding=typing.Optional[typing.Union[numpy.ndarray, list]],
filters=None,
from_=typing.Optional[int],
size=typing.Optional[int],
):
search_body = dict()
from_ = from_ if isinstance(from_, int) and from_ >= 0 else 0
size = size if isinstance(size, int) and size > 0 else 10
search_body.update({
'from': from_,
'size': size,
})
return_embedding = os.getenv('RETURN_EMBEDDING', 'false').lower().strip()
if return_embedding != 'true':
search_body.update({
"_source": {
"excludes": ["embedding"]
}
})
condition_query = {
"match_all": {}
}
if isinstance(filters, typing.Collection) and len(filters) > 0:
tb_mapping = await self._get_mapping(tb_name)
exist_properties = tb_mapping.get('mappings', {}).get('properties', {})
clauses = []
for filter in filters:
k = filter.name
v = filter.value
opt = filter.opt
if k not in exist_properties:
continue
clause = get_filter_clause(k, v, opt, exist_properties[k])
if clause:
clauses.append(clause)
condition_query = {
"bool": {
"filter": clauses
}
}
if embedding is not None:
use_knn = os.getenv('USE_KNN', 'false').lower().strip()
if use_knn != 'true':
query_body = {
"script_score": {
"query": condition_query,
"script": {
"source": """
// 1. 计算原始分数 (范围 [0, 2])
double rawScore = cosineSimilarity(params.query_vector, 'embedding') + 1.0;
// 2. 归一化到 [0, 1]
double normalizedScore = rawScore / 2.0;
// 3. 强制截断：确保最小值为 0.0，最大值为 1.0
// Math.max 防止出现负数，Math.min 防止出现 > 1.0 的数
double clampedScore = Math.max(0.0, Math.min(1.0, normalizedScore));
// 4. 四舍五入到 4 位小数
// 原理：乘以 10000 -> 四舍五入取整 -> 除以 10000
// clampedScore = Math.round(clampedScore * 10000.0) / 10000.0;
return clampedScore;
""",
"params": {
"query_vector": embedding.tolist() if isinstance(embedding, numpy.ndarray) else list(embedding)
}
}
}
}
search_body.update({
"query": query_body
})
else:
search_body.update({
"knn": {
"field": "embedding",
"query_vector": embedding.tolist() if isinstance(embedding, numpy.ndarray) else list(embedding),
"k": from_ + size,
"num_candidates": (from_ + size) * 20,
"filter": condition_query,
}
})
else:
query_body = condition_query
search_body.update({
"query": query_body
})
logger.info(f'index {tb_name} search body: {json.dumps(search_body, ensure_ascii=False, indent=4)}')
resp = await self.es_client.search(index=tb_name, body=search_body)
logger.info(f'index {tb_name} search response: {resp}')
body = resp.body
hits = body.get('hits', {}).get('hits', [])
result_data = [
{
'id': hit['_id'],
'score': round(hit['_score'], 4),
'kwargs': hit['_source'],
}
for hit in hits
]
return result_data
async def delete_by_pks(self, tb_name: str, pks: typing.List[str]):
query_body = {
"query": {
"terms": {
"_id": pks
}
}
}
resp = await self.es_client.delete_by_query(index=tb_name, body=query_body)
logger.info(f'index {tb_name} delete_by_query response: {resp}')
body = resp.body
# return body.get('deleted')
async def _get_mapping(self, tb_name: str):
resp = await self.es_client.indices.get_mapping(index=tb_name)
logger.info(f'index {tb_name} get_mapping response: {resp}')
return resp.body.get(tb_name, {})
async def _tb_exist(self, tb_name, mapping=None):
resp = await self.es_client.indices.exists(index=tb_name)
result = resp.body
if result and mapping is not None:
tb_mapping = await self._get_mapping(tb_name)
exist_properties = tb_mapping.get('mappings', {}).get('properties', {})
expect_properties = mapping.get('mappings', {}).get('properties', {})
if exist_properties and expect_properties:
shared_keys = set(exist_properties.keys()).intersection(set(expect_properties.keys()))
for k in shared_keys or []:
if exist_properties[k]['type'] != expect_properties[k]['type']:
raise Exception(
f'index {tb_name} `{k}` type not match, expect: {expect_properties[k]["type"]}, actual: {exist_properties[k]["type"]}')
else:
if exist_properties[k]['type'] == 'dense_vector':
if exist_properties[k].get('dims') != expect_properties[k].get('dims'):
raise Exception(
f'index {tb_name} dims not match, expect: {expect_properties[k].get("dims")}, actual: {exist_properties[k].get("dims")}')
return result
async def _create_index(self, tb_name, mapping=typing.Optional[dict]) -> bool:
result = await self._tb_exist(tb_name=tb_name, mapping=mapping)
if result:
return True
with self._lock:
result = await self._tb_exist(tb_name=tb_name, mapping=mapping)
if result:
return True
resp = await self.es_client.indices.create(index=tb_name, body=mapping)
logger.info(f'index {tb_name} create response: {resp}')
return resp.body.get('acknowledged') is True
async def upsert(self, tb_name, id, embedding, params: dict):
doc = {
'embedding': embedding,
**params
}
auto_create_index = os.getenv('AUTO_CREATE_INDEX', 'true').lower().strip()
if auto_create_index == 'true':
mapping = generate_mapping(doc)
# logger.info(f'Possible mapping: {json.dumps(mapping, indent=4)}')
result = await self._create_index(tb_name=tb_name, mapping=mapping)
if isinstance(embedding, numpy.ndarray):
embedding = embedding.tolist()
# elif isinstance(embedding, typing.Iterable):
# if isinstance(embedding, typing.Mapping):
# pass
# else:
# embedding = list(embedding)
doc['embedding'] = embedding
resp = await self.es_client.index(index=tb_name, id=id, body=doc)
logger.info(f'index {tb_name} index response: {resp}')
body = resp.body
assert body.get('result') in ['created', 'updated'], f'数据插入失败, id: {id}'
return body.get('_id')
def get_filter_clause(k, v, opt, type_property):
property_type = type_property.get('type')
if property_type == 'text':
property_fields = type_property.get('fields') or {}
if 'keyword' in property_fields:
k = f'{k}.keyword'
opt = opt.lower()
# "eq", "neq", "lt", "gt", "lte", "gte", "like", "in"
if opt in ['eq', 'in']:
v = list(v) if isinstance(v, typing.Collection) else [v]
should_list = []
if None in v or len(v) == 0:
should_list.append({
"bool": {
"must_not": [
{
"exists": {
"field": k
}
}
]
}
})
no_null_v = [item for item in v if item is not None]
if len(no_null_v) > 0:
should_list.append({
"terms": {
k: no_null_v
}
})
return should_list[0] if len(should_list) == 1 else {
"bool": {
"should": should_list,
"minimum_should_match": 1
}
}
elif opt == 'neq':
must_list = []
v = list(v) if isinstance(v, typing.Collection) else [v]
if None in v or len(v) == 0:
must_list.append({
"bool": {
"must": [
{
"exists": {
"field": k
}
}
]
}
})
no_null_v = [item for item in v if item is not None]
if len(no_null_v) > 0:
must_list.append({
"bool": {
"must_not": {
"terms": {
k: no_null_v
}
}
}
})
return must_list[0] if len(must_list) == 1 else {
"bool": {
"must": must_list
}
}
elif opt in ['lt', 'lte', 'gt', 'gte']:
v = list(v) if isinstance(v, typing.Collection) else [v]
if len(v) > 0 and v[0] is not None:
return {
"range": {
k: {opt: list(v)[0]}
}
}
else:
return {
"range": {
k: {}
}
}
elif opt == 'like':
v = list(v) if isinstance(v, typing.Collection) else [v]
if len(v) > 0 and v[0] is not None:
return {
"wildcard": {
k: f"*{list(v)[0]}*"
}
}
else:
return {
"bool": {
"must_not": [
{
"exists": {
"field": k
}
}
]
}
}
else:
raise Exception(f'opt {opt} not support')
def generate_mapping(doc):
def _inner(v):
if isinstance(v, numpy.ndarray):
return 'dense_vector'
elif isinstance(v, str):
return 'text'
elif isinstance(v, int):
return 'long'
elif isinstance(v, float):
return 'float'
elif isinstance(v, typing.Mapping):
return None
elif isinstance(v, typing.Collection):
if len(v) == 0:
return None
else:
return _inner(list(v)[0])
else:
return None
properties = dict()
for key, value in (doc or {}).items():
key_type = _inner(value)
if key_type:
properties[key] = {
'type': key_type,
}
if key_type == 'dense_vector':
properties[key]['dims'] = len(value)
properties[key].update({
'dims': len(value),
'index': True,
'similarity': 'cosine'
})
elif key_type == 'text':
properties[key]['fields'] = {
"keyword": {
"type": "keyword",
# "ignore_above": 256
}
}
es_mapping = {
"settings": {
"number_of_replicas": int(os.getenv('ES_NUMBER_OF_REPLICAS', 0)),
},
"mappings": {
"properties": properties
}
}
return es_mapping
import json
import logging
import os
import threading
import typing
from elasticsearch import AsyncElasticsearch
import numpy
logger = logging.getLogger(__name__)
class DataDao:
def __init__(self, es_client: AsyncElasticsearch):
self.es_client = es_client
self._lock = threading.RLock()
async def search(self, tb_name: str,
embedding=typing.Optional[typing.Union[numpy.ndarray, list]],
filters=None,
from_=typing.Optional[int],
size=typing.Optional[int],
):
search_body = dict()
from_ = from_ if isinstance(from_, int) and from_ >= 0 else 0
size = size if isinstance(size, int) and size > 0 else 10
search_body.update({
'from': from_,
'size': size,
})
return_embedding = os.getenv('RETURN_EMBEDDING', 'false').lower().strip()
if return_embedding != 'true':
search_body.update({
"_source": {
"excludes": ["embedding"]
}
})
condition_query = {
"match_all": {}
}
if isinstance(filters, typing.Collection) and len(filters) > 0:
tb_mapping = await self._get_mapping(tb_name)
exist_properties = tb_mapping.get('mappings', {}).get('properties', {})
clauses = []
for filter in filters:
k = filter.name
v = filter.value
opt = filter.opt
if k not in exist_properties:
continue
clause = get_filter_clause(k, v, opt, exist_properties[k])
if clause:
clauses.append(clause)
condition_query = {
"bool": {
"filter": clauses
}
}
if embedding is not None:
use_knn = os.getenv('USE_KNN', 'false').lower().strip()
if use_knn != 'true':
query_body = {
"script_score": {
"query": condition_query,
"script": {
"source": """
// 1. 计算原始分数 (范围 [0, 2])
double rawScore = cosineSimilarity(params.query_vector, 'embedding') + 1.0;
// 2. 归一化到 [0, 1]
double normalizedScore = rawScore / 2.0;
// 3. 强制截断：确保最小值为 0.0，最大值为 1.0
// Math.max 防止出现负数，Math.min 防止出现 > 1.0 的数
double clampedScore = Math.max(0.0, Math.min(1.0, normalizedScore));
// 4. 四舍五入到 4 位小数
// 原理：乘以 10000 -> 四舍五入取整 -> 除以 10000
// clampedScore = Math.round(clampedScore * 10000.0) / 10000.0;
return clampedScore;
""",
"params": {
"query_vector": embedding.tolist() if isinstance(embedding, numpy.ndarray) else list(embedding)
}
}
}
}
search_body.update({
"query": query_body
})
else:
search_body.update({
"knn": {
"field": "embedding",
"query_vector": embedding.tolist() if isinstance(embedding, numpy.ndarray) else list(embedding),
"k": from_ + size,
"num_candidates": (from_ + size) * 20,
"filter": condition_query,
}
})
else:
query_body = condition_query
search_body.update({
"query": query_body
})
logger.info(f'index {tb_name} search body: {json.dumps(search_body, ensure_ascii=False, indent=4)}')
resp = await self.es_client.search(index=tb_name, body=search_body)
logger.info(f'index {tb_name} search response: {resp}')
body = resp.body
hits = body.get('hits', {}).get('hits', [])
result_data = [
{
'id': hit['_id'],
'score': round(hit['_score'], 4),
'kwargs': hit['_source'],
}
for hit in hits
]
return result_data
async def delete_by_pks(self, tb_name: str, pks: typing.List[str]):
query_body = {
"query": {
"terms": {
"_id": pks
}
}
}
resp = await self.es_client.delete_by_query(index=tb_name, body=query_body)
logger.info(f'index {tb_name} delete_by_query response: {resp}')
body = resp.body
# return body.get('deleted')
async def _get_mapping(self, tb_name: str):
resp = await self.es_client.indices.get_mapping(index=tb_name)
logger.info(f'index {tb_name} get_mapping response: {resp}')
return resp.body.get(tb_name, {})
async def _tb_exist(self, tb_name, mapping=None):
resp = await self.es_client.indices.exists(index=tb_name)
result = resp.body
if result and mapping is not None:
tb_mapping = await self._get_mapping(tb_name)
exist_properties = tb_mapping.get('mappings', {}).get('properties', {})
expect_properties = mapping.get('mappings', {}).get('properties', {})
if exist_properties and expect_properties:
shared_keys = set(exist_properties.keys()).intersection(set(expect_properties.keys()))
for k in shared_keys or []:
if exist_properties[k]['type'] != expect_properties[k]['type']:
raise Exception(
f'index {tb_name} `{k}` type not match, expect: {expect_properties[k]["type"]}, actual: {exist_properties[k]["type"]}')
else:
if exist_properties[k]['type'] == 'dense_vector':
if exist_properties[k].get('dims') != expect_properties[k].get('dims'):
raise Exception(
f'index {tb_name} dims not match, expect: {expect_properties[k].get("dims")}, actual: {exist_properties[k].get("dims")}')
return result
async def _create_index(self, tb_name, mapping=typing.Optional[dict]) -> bool:
result = await self._tb_exist(tb_name=tb_name, mapping=mapping)
if result:
return True
with self._lock:
result = await self._tb_exist(tb_name=tb_name, mapping=mapping)
if result:
return True
resp = await self.es_client.indices.create(index=tb_name, body=mapping)
logger.info(f'index {tb_name} create response: {resp}')
return resp.body.get('acknowledged') is True
async def upsert(self, tb_name, id, embedding, params: dict):
doc = {
'embedding': embedding,
**params
}
auto_create_index = os.getenv('AUTO_CREATE_INDEX', 'true').lower().strip()
if auto_create_index == 'true':
mapping = generate_mapping(doc)
# logger.info(f'Possible mapping: {json.dumps(mapping, indent=4)}')
result = await self._create_index(tb_name=tb_name, mapping=mapping)
if isinstance(embedding, numpy.ndarray):
embedding = embedding.tolist()
# elif isinstance(embedding, typing.Iterable):
# if isinstance(embedding, typing.Mapping):
# pass
# else:
# embedding = list(embedding)
doc['embedding'] = embedding
resp = await self.es_client.index(index=tb_name, id=id, body=doc)
logger.info(f'index {tb_name} index response: {resp}')
body = resp.body
assert body.get('result') in ['created', 'updated'], f'数据插入失败, id: {id}'
return body.get('_id')
def get_filter_clause(k, v, opt, type_property):
property_type = type_property.get('type')
if property_type == 'text':
property_fields = type_property.get('fields') or {}
if 'keyword' in property_fields:
k = f'{k}.keyword'
opt = opt.lower()
# "eq", "neq", "lt", "gt", "lte", "gte", "like", "in"
if opt in ['eq', 'in']:
v = list(v) if isinstance(v, typing.Collection) else [v]
should_list = []
if None in v or len(v) == 0:
should_list.append({
"bool": {
"must_not": [
{
"exists": {
"field": k
}
}
]
}
})
no_null_v = [item for item in v if item is not None]
if len(no_null_v) > 0:
should_list.append({
"terms": {
k: no_null_v
}
})
return should_list[0] if len(should_list) == 1 else {
"bool": {
"should": should_list,
"minimum_should_match": 1
}
}
elif opt == 'neq':
must_list = []
v = list(v) if isinstance(v, typing.Collection) else [v]
if None in v or len(v) == 0:
must_list.append({
"bool": {
"must": [
{
"exists": {
"field": k
}
}
]
}
})
no_null_v = [item for item in v if item is not None]
if len(no_null_v) > 0:
must_list.append({
"bool": {
"must_not": {
"terms": {
k: no_null_v
}
}
}
})
return must_list[0] if len(must_list) == 1 else {
"bool": {
"must": must_list
}
}
elif opt in ['lt', 'lte', 'gt', 'gte']:
v = list(v) if isinstance(v, typing.Collection) else [v]
if len(v) > 0 and v[0] is not None:
return {
"range": {
k: {opt: list(v)[0]}
}
}
else:
return {
"range": {
k: {}
}
}
elif opt == 'like':
v = list(v) if isinstance(v, typing.Collection) else [v]
if len(v) > 0 and v[0] is not None:
return {
"wildcard": {
k: f"*{list(v)[0]}*"
}
}
else:
return {
"bool": {
"must_not": [
{
"exists": {
"field": k
}
}
]
}
}
else:
raise Exception(f'opt {opt} not support')
def generate_mapping(doc):
def _inner(v):
if isinstance(v, numpy.ndarray):
return 'dense_vector'
elif isinstance(v, str):
return 'text'
elif isinstance(v, int):
return 'long'
elif isinstance(v, float):
return 'float'
elif isinstance(v, typing.Mapping):
return None
elif isinstance(v, typing.Collection):
if len(v) == 0:
return None
else:
return _inner(list(v)[0])
else:
return None
properties = dict()
for key, value in (doc or {}).items():
key_type = _inner(value)
if key_type:
properties[key] = {
'type': key_type,
}
if key_type == 'dense_vector':
properties[key]['dims'] = len(value)
properties[key].update({
'dims': len(value),
'index': True,
'similarity': 'cosine'
})
elif key_type == 'text':
properties[key]['fields'] = {
"keyword": {
"type": "keyword",
# "ignore_above": 256
}
}
es_mapping = {
"settings": {
"number_of_replicas": int(os.getenv('ES_NUMBER_OF_REPLICAS', 0)),
},
"mappings": {
"properties": properties
}
}
return es_mapping
... ...
"""Elasticsearch向量数据库客户端实现."""
from typing import Any, Optional
from embedding_sever.db.base import VectorDBClient
class ElasticsearchClient(VectorDBClient):
"""Elasticsearch向量数据库客户端.
使用Elasticsearch的dense_vector类型存储向量，支持向量相似度搜索。
"""
def __init__(self, es_url: Optional[str] = None):
"""初始化ES客户端.
Args:
es_url: Elasticsearch连接URL，默认从配置读取
"""
self.es_url = es_url
self._client: Optional[Any] = None
async def connect(self) -> None:
"""建立ES连接."""
from elasticsearch import AsyncElasticsearch
self._client = AsyncElasticsearch([self.es_url])
# 验证连接
await self._client.ping()
async def disconnect(self) -> None:
"""断开ES连接."""
if self._client:
await self._client.close()
self._client = None
async def health_check(self) -> bool:
"""检查ES连接健康状态."""
if not self._client:
return False
try:
return await self._client.ping()
except Exception:
return False
_es_client: Optional[ElasticsearchClient] = None
def get_es_client() -> ElasticsearchClient:
"""获取全局ES客户端实例（单例模式）.
Returns:
ElasticsearchClient: ES客户端实例
"""
global _es_client
if _es_client is None:
_es_client = ElasticsearchClient()
return _es_client
"""Elasticsearch向量数据库客户端实现."""
from typing import Any, Optional
from embedding_sever.db.base import VectorDBClient
class ElasticsearchClient(VectorDBClient):
"""Elasticsearch向量数据库客户端.
使用Elasticsearch的dense_vector类型存储向量，支持向量相似度搜索。
"""
def __init__(self, es_url: Optional[str] = None):
"""初始化ES客户端.
Args:
es_url: Elasticsearch连接URL，默认从配置读取
"""
self.es_url = es_url
self._client: Optional[Any] = None
async def connect(self) -> None:
"""建立ES连接."""
from elasticsearch import AsyncElasticsearch
self._client = AsyncElasticsearch([self.es_url])
# 验证连接
await self._client.ping()
async def disconnect(self) -> None:
"""断开ES连接."""
if self._client:
await self._client.close()
self._client = None
async def health_check(self) -> bool:
"""检查ES连接健康状态."""
if not self._client:
return False
try:
return await self._client.ping()
except Exception:
return False
_es_client: Optional[ElasticsearchClient] = None
def get_es_client() -> ElasticsearchClient:
"""获取全局ES客户端实例（单例模式）.
Returns:
ElasticsearchClient: ES客户端实例
"""
global _es_client
if _es_client is None:
_es_client = ElasticsearchClient()
return _es_client
... ...
from .base import TableInit
from . import es_client
class ESTableInit(TableInit):
def __init__(self):
self.es_client = es_client
async def create_table(self, name, embedding_version, embedding_dim) -> None:
pass
from .base import TableInit
from . import es_client
class ESTableInit(TableInit):
def __init__(self):
self.es_client = es_client
async def create_table(self, name, embedding_version, embedding_dim) -> None:
pass
... ...
"""PostgreSQL向量数据库客户端实现."""
from typing import Any, Optional
from embedding_sever.db.base import VectorDBClient
class PostgresClient(VectorDBClient):
"""PostgreSQL向量数据库客户端.
使用pgvector扩展存储向量，支持向量相似度搜索。
需要安装pgvector扩展: CREATE EXTENSION IF NOT EXISTS vector;
"""
def __init__(self, pg_url: Optional[str] = None):
"""初始化PG客户端.
Args:
pg_url: PostgreSQL连接URL，默认从配置读取
"""
self.pg_url = pg_url
self._pool: Optional[Any] = None
async def connect(self) -> None:
"""建立PG连接池."""
from psycopg_pool import AsyncConnectionPool
self._pool = AsyncConnectionPool(
conninfo=self.pg_url,
min_size=1,
max_size=10,
)
# 验证连接并启用pgvector
async with self._pool.connection() as conn:
await conn.execute("CREATE EXTENSION IF NOT EXISTS vector")
await conn.commit()
async def disconnect(self) -> None:
"""断开PG连接."""
if self._pool:
await self._pool.close()
self._pool = None
async def health_check(self) -> bool:
"""检查PG连接健康状态."""
if not self._pool:
return False
try:
async with self._pool.connection() as conn:
cursor = await conn.execute("SELECT 1")
result = await cursor.fetchone()
return result is not None and result[0] == 1
except Exception:
return False
# 全局PG客户端实例
_pg_client: Optional[PostgresClient] = None
def get_pg_client() -> PostgresClient:
"""获取全局PG客户端实例（单例模式）.
Returns:
PostgresClient: PG客户端实例
"""
global _pg_client
if _pg_client is None:
_pg_client = PostgresClient()
return _pg_client
"""PostgreSQL向量数据库客户端实现."""
from typing import Any, Optional
from embedding_sever.db.base import VectorDBClient
class PostgresClient(VectorDBClient):
"""PostgreSQL向量数据库客户端.
使用pgvector扩展存储向量，支持向量相似度搜索。
需要安装pgvector扩展: CREATE EXTENSION IF NOT EXISTS vector;
"""
def __init__(self, pg_url: Optional[str] = None):
"""初始化PG客户端.
Args:
pg_url: PostgreSQL连接URL，默认从配置读取
"""
self.pg_url = pg_url
self._pool: Optional[Any] = None
async def connect(self) -> None:
"""建立PG连接池."""
from psycopg_pool import AsyncConnectionPool
self._pool = AsyncConnectionPool(
conninfo=self.pg_url,
min_size=1,
max_size=10,
)
# 验证连接并启用pgvector
async with self._pool.connection() as conn:
await conn.execute("CREATE EXTENSION IF NOT EXISTS vector")
await conn.commit()
async def disconnect(self) -> None:
"""断开PG连接."""
if self._pool:
await self._pool.close()
self._pool = None
async def health_check(self) -> bool:
"""检查PG连接健康状态."""
if not self._pool:
return False
try:
async with self._pool.connection() as conn:
cursor = await conn.execute("SELECT 1")
result = await cursor.fetchone()
return result is not None and result[0] == 1
except Exception:
return False
# 全局PG客户端实例
_pg_client: Optional[PostgresClient] = None
def get_pg_client() -> PostgresClient:
"""获取全局PG客户端实例（单例模式）.
Returns:
PostgresClient: PG客户端实例
"""
global _pg_client
if _pg_client is None:
_pg_client = PostgresClient()
return _pg_client
... ...
import os
from pathlib import Path
if not os.environ.get("PROJECT_ROOT"):
os.environ['PROJECT_ROOT'] = Path(__file__).parent.absolute().as_posix()
from aabd.base.patched_logging import init_logging, get_logger
init_logging()
logger = get_logger(__name__)
from contextlib import asynccontextmanager
from typing import AsyncGenerator
from fastapi import FastAPI
from embedding_sever.api.http_config import config_fastapi
from embedding_sever.config import settings
from embedding_sever.db import connect as db_client_connect, disconnect as db_client_disconnect
@asynccontextmanager
async def lifespan(fastapi_app: FastAPI) -> AsyncGenerator[None, None]:
"""应用生命周期管理."""
# 启动时建立数据库连接
try:
await db_client_connect()
logger.info(f"✓ database connected")
except Exception as e:
logger.error(f"✗ Failed to connect to database: {e}")
raise
yield
# 关闭时断开数据库连接
try:
await db_client_disconnect()
logger.info(f"✓ database disconnected")
except Exception as e:
logger.error(f"✗ Error disconnecting from database: {e}")
# 创建应用实例
app = FastAPI(
title=settings.app_name,
version=settings.app_version,
description="基于向量的CRUD服务，支持多种Embedding场景",
lifespan=lifespan,
docs_url="/docs",
redoc_url="/redoc",
)
config_fastapi(app)
def main():
"""应用入口函数."""
import uvicorn
logger.info(f'docs_url: http://127.0.0.1:{settings.port}/docs')
uvicorn.run(
app,
host='0.0.0.0',
port=settings.port,
reload=settings.debug,
log_config=None
)
if __name__ == "__main__":
main()
import os
from pathlib import Path
if not os.environ.get("PROJECT_ROOT"):
os.environ['PROJECT_ROOT'] = Path(__file__).parent.absolute().as_posix()
from aabd.base.patched_logging import init_logging, get_logger
init_logging()
logger = get_logger(__name__)
from contextlib import asynccontextmanager
from typing import AsyncGenerator
from fastapi import FastAPI
from embedding_sever.api.http_config import config_fastapi
from embedding_sever.config import settings
from embedding_sever.db import connect as db_client_connect, disconnect as db_client_disconnect
@asynccontextmanager
async def lifespan(fastapi_app: FastAPI) -> AsyncGenerator[None, None]:
"""应用生命周期管理."""
# 启动时建立数据库连接
try:
await db_client_connect()
logger.info(f"✓ database connected")
except Exception as e:
logger.error(f"✗ Failed to connect to database: {e}")
raise
yield
# 关闭时断开数据库连接
try:
await db_client_disconnect()
logger.info(f"✓ database disconnected")
except Exception as e:
logger.error(f"✗ Error disconnecting from database: {e}")
# 创建应用实例
app = FastAPI(
title=settings.app_name,
version=settings.app_version,
description="基于向量的CRUD服务，支持多种Embedding场景",
lifespan=lifespan,
docs_url="/docs",
redoc_url="/redoc",
)
config_fastapi(app)
def main():
"""应用入口函数."""
import uvicorn
logger.info(f'docs_url: http://127.0.0.1:{settings.port}/docs')
uvicorn.run(
app,
host='0.0.0.0',
port=settings.port,
reload=settings.debug,
log_config=None
)
if __name__ == "__main__":
main()
... ...
from functools import lru_cache
from embedding_sever.db import get_data_dao
from .data_service import DataService
@lru_cache()
def get_data_service():
data_dao = get_data_dao()
return DataService(data_dao)
from functools import lru_cache
from embedding_sever.db import get_data_dao
from .data_service import DataService
@lru_cache()
def get_data_service():
data_dao = get_data_dao()
return DataService(data_dao)
... ...
import logging
from embedding_sever.db import DataDao
logger = logging.getLogger(__name__)
class DataService:
def __init__(self, data_dao: DataDao):
self.data_dao = data_dao
async def upsert(self, tb_name, id, embedding, params):
return await self.data_dao.upsert(tb_name, id, embedding, params)
async def delete_by_pks(self, tb_name, ids):
return await self.data_dao.delete_by_pks(tb_name, ids)
async def search(self, tb_name, embedding, filters, from_, size):
return await self.data_dao.search(tb_name, embedding, filters, from_, size)
import logging
from embedding_sever.db import DataDao
logger = logging.getLogger(__name__)
class DataService:
def __init__(self, data_dao: DataDao):
self.data_dao = data_dao
async def upsert(self, tb_name, id, embedding, params):
return await self.data_dao.upsert(tb_name, id, embedding, params)
async def delete_by_pks(self, tb_name, ids):
return await self.data_dao.delete_by_pks(tb_name, ids)
async def search(self, tb_name, embedding, filters, from_, size):
return await self.data_dao.search(tb_name, embedding, filters, from_, size)
... ...
kafka_servers: 192.168.0.14:9092
llm:
base_url: http://192.168.1.59:11434/v1
model_name: 'Qwen3.6-35B-A3B-UD-Q8_K_XL.gguf'
common:
kafka_servers: 192.168.0.14:9092
llm:
base_url: http://192.168.1.59:11434/v1
model_name: 'Qwen3.6-35B-A3B-UD-Q8_K_XL.gguf'
common:
cache_dir: ./cache
\ No newline at end of file
... ...
app_name: "Football Replay Service"
app_version: "1.0.0"
match_by_time_threshold: 30 # 通过时间匹配进球事件的阈值（秒）
#kafka_servers: 192.168.0.14:9092
kafka_username:
kafka_password:
input_kafka:
servers: ${kafka_servers}
group_id: ai_match_service
topic: football_replay_match_req
username: ${kafka_username}
password: ${kafka_password}
output_kafka:
servers: ${kafka_servers}
topic: football_replay_match_resp
username: ${kafka_username}
password: ${kafka_password}
llm:
base_url: http://192.168.1.59:11434/v1
model_name: Qwen3.6-35B-A3B-UD-Q8_K_XL.gguf
temperature: 0.7
common:
app_name: "Football Replay Service"
app_version: "1.0.0"
match_by_time_threshold: 30 # 通过时间匹配进球事件的阈值（秒）
#kafka_servers: 192.168.0.14:9092
kafka_username:
kafka_password:
input_kafka:
servers: ${kafka_servers}
group_id: ai_match_service
topic: football_replay_match_req
username: ${kafka_username}
password: ${kafka_password}
output_kafka:
servers: ${kafka_servers}
topic: football_replay_match_resp
username: ${kafka_username}
password: ${kafka_password}
llm:
base_url: http://192.168.1.59:11434/v1
model_name: Qwen3.6-35B-A3B-UD-Q8_K_XL.gguf
temperature: 0.7
common:
cache_dir: ./cache
\ No newline at end of file
... ...
from aabd.base.cfg_loader import load_yaml_by_file_with_env
from aabd.base.enhance_dict import EnhanceDict, read_prefixed_env_vars
from pathlib import Path
from logging import getLogger
logger = getLogger(__name__)
cfg_dir = Path(__file__).parent.absolute() / 'cfg'
settings = EnhanceDict(load_yaml_by_file_with_env((cfg_dir / 'config.yaml').as_posix()))
settings.update(read_prefixed_env_vars('EMB_'))
logger.info(f'Settings: {settings}')
from aabd.base.cfg_loader import load_yaml_by_file_with_env
from aabd.base.enhance_dict import EnhanceDict, read_prefixed_env_vars
from pathlib import Path
from logging import getLogger
logger = getLogger(__name__)
cfg_dir = Path(__file__).parent.absolute() / 'cfg'
settings = EnhanceDict(load_yaml_by_file_with_env((cfg_dir / 'config.yaml').as_posix()))
settings.update(read_prefixed_env_vars('EMB_'))
logger.info(f'Settings: {settings}')
... ...
import os
import logging
from aabd.base.enhance_dict import value_or_default
logger = logging.getLogger(__name__)
from utils.football_replay_video_event_by_llm import FootballReplayVideoEvent
from utils.football_replay_match_live import FootballReplayMatchLive
class FootballReplayMatch:
def __init__(self, settings):
self.settings = settings
self.match_by_time_threshold = value_or_default(settings.match_by_time_threshold, 30) * 1000
llm_base_url = value_or_default(settings.llm.base_url, None)
model_name = value_or_default(settings.llm.model_name, None)
temperature = value_or_default(settings.llm.temperature, 0.7)
self.cache_dir = value_or_default(settings.common.cache_dir, None)
save_frames_enable = value_or_default(settings.save_frames_enable, False)
self.videoEventRecognition = FootballReplayVideoEvent(llm_base_url, model_name, temperature, 'no_key', self.cache_dir, save_frames_enable)
self.videoMatchLive = FootballReplayMatchLive(llm_base_url, model_name, temperature, 'no_key', self.cache_dir, save_frames_enable)
def match_by_time(self, replay, events):
start_utc = replay.get('start_utc')
events = [(start_utc - e.get('event_utc'), e) for e in events]
events.sort(key=lambda x: x[0])
target_index = next((i for i, e in enumerate(events) if e[0] > 0), -1) # 找到第一个 event_utc 早于 start_utc 的事件
if target_index == -1:
return None
time_diff, event = events[target_index]
if time_diff < self.match_by_time_threshold:
return event
else:
return None
def det_goal_replay(self, replay, task_id):
return self.videoEventRecognition.video_event(replay, cache_dir=os.path.join(self.cache_dir, task_id, "replay"))
def match_by_llm(self, replay, events, task_id):
return self.videoMatchLive.match_batch(replay, events, max_parallel=2, cache_dir=os.path.join(self.cache_dir, task_id, "envents"))
def replay_match_event(self, data):
"""
:param data:
{
id:xx,
match_id: 比赛id，
replay:{
id:,
url:,
start_utc:,
end_utc:,
}
events:[{
id:xx,
type:xx,
url:xx,
event_utc,
}]
}
:return:
"""
task_id = data.get("id")
match_id = data.get("match_id")
matched_event_id = None
# 该时间段附近是否有进球事件
replay_info = data['replay']
replay_id = replay_info.get('id')
live_events = data['events']
goal_live_events = [e for e in live_events if str(e.get('type', '')) == '1']
matched_event = self.match_by_time(replay_info, goal_live_events)
if matched_event is None:
replay_det_info = self.det_goal_replay(replay_info, task_id)
replay_event_name = replay_det_info.get('event_name', '')
if replay_event_name == '进球':
matched_event = self.match_by_llm(replay_info, goal_live_events, task_id)
if matched_event is None:
logger.info(f"{task_id}_LLM认为视频{replay_info['url']}_{replay_info['start_utc']}_{replay_info['end_utc']} 是进球但是未能找到匹配的事件")
# print(f"{task_id}_LLM认为视频{replay_info['url']}_{replay_info['start_utc']}_{replay_info['end_utc']} 是进球但是未能找到匹配的事件")
else:
matched_event_id = matched_event.get('video_id')
logger.info(f"{task_id}_LLM认为视频{replay_info['url']}_{replay_info['start_utc']}_{replay_info['end_utc']} 对应的进球的事件视频是{matched_event_id}")
# print(f"{task_id}_LLM认为视频{replay_info['url']}_{replay_info['start_utc']}_{replay_info['end_utc']} 对应的进球的事件视频是{matched_event_id}")
else:
logger.info(f"{task_id}_LLM判断{replay_info['url']}_{replay_info['start_utc']}_{replay_info['end_utc']} 不是进球，无需匹配")
# print(f"{task_id}_LLM判断{replay_info['url']}_{replay_info['start_utc']}_{replay_info['end_utc']} 不是进球，无需匹配")
else:
matched_event_id = matched_event.get('id')
logger.info(f"{task_id}_通过时间找到匹配认为视频{replay_info['url']}_{replay_info['start_utc']}_{replay_info['end_utc']} 对应的进球的事件视频是{matched_event_id}")
# print(f"{task_id}_通过时间找到匹配认为视频{replay_info['url']}_{replay_info['start_utc']}_{replay_info['end_utc']} 对应的进球的事件视频是{matched_event_id}")
return {
"id": task_id,
"match_id": match_id,
"replay_id": replay_id,
"event_id": matched_event_id
import os
import logging
from aabd.base.enhance_dict import value_or_default
logger = logging.getLogger(__name__)
from utils.football_replay_video_event_by_llm import FootballReplayVideoEvent
from utils.football_replay_match_live import FootballReplayMatchLive
class FootballReplayMatch:
def __init__(self, settings):
self.settings = settings
self.match_by_time_threshold = value_or_default(settings.match_by_time_threshold, 30) * 1000
llm_base_url = value_or_default(settings.llm.base_url, None)
model_name = value_or_default(settings.llm.model_name, None)
temperature = value_or_default(settings.llm.temperature, 0.7)
self.cache_dir = value_or_default(settings.common.cache_dir, None)
save_frames_enable = value_or_default(settings.save_frames_enable, False)
self.videoEventRecognition = FootballReplayVideoEvent(llm_base_url, model_name, temperature, 'no_key', self.cache_dir, save_frames_enable)
self.videoMatchLive = FootballReplayMatchLive(llm_base_url, model_name, temperature, 'no_key', self.cache_dir, save_frames_enable)
def match_by_time(self, replay, events):
start_utc = replay.get('start_utc')
events = [(start_utc - e.get('event_utc'), e) for e in events]
events.sort(key=lambda x: x[0])
target_index = next((i for i, e in enumerate(events) if e[0] > 0), -1) # 找到第一个 event_utc 早于 start_utc 的事件
if target_index == -1:
return None
time_diff, event = events[target_index]
if time_diff < self.match_by_time_threshold:
return event
else:
return None
def det_goal_replay(self, replay, task_id):
return self.videoEventRecognition.video_event(replay, cache_dir=os.path.join(self.cache_dir, task_id, "replay"))
def match_by_llm(self, replay, events, task_id):
return self.videoMatchLive.match_batch(replay, events, max_parallel=2, cache_dir=os.path.join(self.cache_dir, task_id, "envents"))
def replay_match_event(self, data):
"""
:param data:
{
id:xx,
match_id: 比赛id，
replay:{
id:,
url:,
start_utc:,
end_utc:,
}
events:[{
id:xx,
type:xx,
url:xx,
event_utc,
}]
}
:return:
"""
task_id = data.get("id")
match_id = data.get("match_id")
matched_event_id = None
# 该时间段附近是否有进球事件
replay_info = data['replay']
replay_id = replay_info.get('id')
live_events = data['events']
goal_live_events = [e for e in live_events if str(e.get('type', '')) == '1']
matched_event = self.match_by_time(replay_info, goal_live_events)
if matched_event is None:
replay_det_info = self.det_goal_replay(replay_info, task_id)
replay_event_name = replay_det_info.get('event_name', '')
if replay_event_name == '进球':
matched_event = self.match_by_llm(replay_info, goal_live_events, task_id)
if matched_event is None:
logger.info(f"{task_id}_LLM认为视频{replay_info['url']}_{replay_info['start_utc']}_{replay_info['end_utc']} 是进球但是未能找到匹配的事件")
# print(f"{task_id}_LLM认为视频{replay_info['url']}_{replay_info['start_utc']}_{replay_info['end_utc']} 是进球但是未能找到匹配的事件")
else:
matched_event_id = matched_event.get('video_id')
logger.info(f"{task_id}_LLM认为视频{replay_info['url']}_{replay_info['start_utc']}_{replay_info['end_utc']} 对应的进球的事件视频是{matched_event_id}")
# print(f"{task_id}_LLM认为视频{replay_info['url']}_{replay_info['start_utc']}_{replay_info['end_utc']} 对应的进球的事件视频是{matched_event_id}")
else:
logger.info(f"{task_id}_LLM判断{replay_info['url']}_{replay_info['start_utc']}_{replay_info['end_utc']} 不是进球，无需匹配")
# print(f"{task_id}_LLM判断{replay_info['url']}_{replay_info['start_utc']}_{replay_info['end_utc']} 不是进球，无需匹配")
else:
matched_event_id = matched_event.get('id')
logger.info(f"{task_id}_通过时间找到匹配认为视频{replay_info['url']}_{replay_info['start_utc']}_{replay_info['end_utc']} 对应的进球的事件视频是{matched_event_id}")
# print(f"{task_id}_通过时间找到匹配认为视频{replay_info['url']}_{replay_info['start_utc']}_{replay_info['end_utc']} 对应的进球的事件视频是{matched_event_id}")
return {
"id": task_id,
"match_id": match_id,
"replay_id": replay_id,
"event_id": matched_event_id
}
\ No newline at end of file
... ...
import json
import os.path
from pathlib import Path
from datetime import timedelta
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_openai import ChatOpenAI
try:
from .llm_image import Video2Frame
except:
from llm_image import Video2Frame
req_prompt = """
# Role
你是一名拥有20年经验的足球视频技术分析师，擅长结合**视觉画面**与**解说音频（ASR）**进行跨镜头的事件匹配。你能通过解说员的描述锁定时间背景，并通过视觉特征确认物理细节。
# Task
输入包含：
1. 【回放片段】（Replay）：包含视频帧 + **对应的解说文本**。
2. 【直播进球片段列表】（Live Candidates）：包含视频帧 + **对应的解说文本**。
目标：在【直播片段】中找到与【回放片段】属于**同一个进球事件**的片段。如果没有任何片段匹配，返回 null。
# Analysis Workflow (多模态分析流程)
请按照以下步骤进行推理：
### 第一步：解说词元数据提取 (听觉线索)
分析【回放片段】的解说文本，寻找以下关键信息：
- **时间指代**：解说员是否提到了具体时间？（如“上半场”、“第10分钟”、“开场不久”）。
- **事件描述**：解说员如何描述这个进球？（如“世界波”、“点球”、“补射”、“乌龙球”）。
- **球员提及**：解说员念出了谁的名字？
### 第二步：视觉物理指纹提取 (视觉线索)
忽略镜头语言（慢放、特写），提取核心物理特征：
- **进攻与射门**：进攻方式（传中/直塞）、射门部位（头/脚）、射门位置。
- **球路与防守**：球的轨迹（高/低/折射）、门将扑救动作（侧扑/倒地/未动）。
- **庆祝**：进球后的庆祝动作（仅作为辅助验证）。
### 第三步：跨模态匹配与验证
遍历【直播片段】，结合视觉和听觉进行判断：
- **视觉一致性**：直播片段的动作、轨迹、门将反应是否与回放完全吻合？
- **听觉一致性**：
- 如果回放解说提到“这是第5分钟的进球”，而直播片段的时间戳是90分钟，**不要直接排除**，而是检查直播片段的解说是否也提到了“回顾第5分钟”或者直播片段的视觉内容确实是第5分钟的动作。
- **核心原则**：视觉物理特征 > 解说词描述 > 时间戳元数据。
- **特殊情况**：如果视觉特征高度相似（如同一个球员、同一个角度），但解说词明确说“这是另一个进球（例如：这是他的第二个进球，而回放说是第一个）”，则判定为不匹配。
# Logic Constraints (逻辑约束 - 必须严格遵守)
- **时间单向性原则（铁律）**：
- **回看中的比赛时间一定在比赛片段画面时间之后**。
- 逻辑：回放是对过去发生事件的回顾。如果【回放片段】中解说提到的比赛时间（或画面显示的比赛时钟）**早于**【直播片段】中解说提到的比赛时间（或画面时钟），则**绝对不可能**是同一个事件。
- *示例*：回放解说在描述“第10分钟的进球”，而直播片段明确发生在“第5分钟”，则该直播片段**一定不是**目标。
- **时间陷阱**：回放可能是赛后集锦。如果解说员说“让我们看看**刚才**那个球”或者“**上半场**那个球”，即使当前比赛时间是90分钟，也要去匹配对应时间段的直播片段（或视觉特征）。
- **同名陷阱**：如果解说提到“又是**凯恩**进球了”，不能只看凯恩，必须看**怎么进的**（头球还是点球）。
- **无匹配处理**：如果所有候选片段在视觉动作（如射门方式、进球位置）或关键事件逻辑上与回放明显不符，必须判定为无匹配，将 `video_id` 设为 `null`。
### 输出要求
请仅输出一个JSON格式的结果，不要输出任何分析过程。不要包含 markdown 标记（如 ```json ... ```），不要包含任何解释或额外文本。
格式如下：
{
"replay_summary": {
"audio_cues": "解说提到的关键信息（如：'第15分钟', '远射', '德布劳内'）",
"visual_cues": "视觉关键特征（如：'禁区外右脚', '球挂死角', '门将飞身扑救'）"
},
"reasoning": "综合分析：回放解说提到是'上半场的远射'，视觉显示'17号球员禁区外起脚'。Candidate_1 视觉是'近距离推射'，排除。Candidate_2 视觉是'禁区外远射'，且门将动作一致，虽然直播时间显示是下半场（可能是集锦回顾），但解说也提到了'回顾上半场'，确认为同一事件。",
"video_id": "Candidate_2"
}"""
class FootballReplayMatchLive:
def __init__(self, base_url: str, model: str, temperature: float = 0.0, api_key: str = 'no_key', cache_dir:str=None, save_frames_enable:bool=False):
self.base_url = base_url
self.model = model
self.temperature = temperature
self.api_key = api_key
# self.model = ChatOllama(base_url="http://192.168.1.59:11434", model="qwen3.6:35b-a3b-q8_0", temperature=0.7,
# keep_alive=-1, reasoning=False)
# self.model = ChatOpenAI(base_url="http://192.168.1.59:11434/v1", model="qwen3.6:35b-a3b-q8_0", temperature=0.7,
# api_key='no_key')
self.model = ChatOpenAI(base_url=base_url, model=model, temperature=temperature, api_key=api_key,
extra_body={"chat_template_kwargs": {"enable_thinking": False}})
self.cache_dir = cache_dir
self.video2frame = Video2Frame(cache_dir=cache_dir, save_frames_enable=save_frames_enable)
def _match_once(self, replay_video_contents: list, live_videos: list[dict], record: list = None):
if len(live_videos) == 0:
return None
elif len(live_videos) == 1:
live_video_path = live_videos[0].get("url", None)
live_video_id = live_videos[0].get("video_id", os.path.basename(live_video_path))
asr_text = live_videos[0].get("asr_text", '')
live = {
"video_id": live_video_id,
"video_path": live_video_path,
"asr_text": asr_text,
}
if record is not None:
record.append({"live": live, "llm_result": None, 'live_list': [live]})
return live
user_contents = []
# replay_video_contents = video_contents(replay_video["video_path"], "\n【回放片段信息】\n",
# prompt_end=f"\n回放解说内容：{replay_video['asr_text']}\n",
# video_name=os.path.basename(replay_video["video_path"]),
# fps=2, max_frames=999, sampling_mode="head", max_short_edge=480)
live_videos_contents = []
live_map = {}
live_records = {}
for live_video in live_videos:
live_video_path = live_video.get("url", None)
live_video_id = live_video.get("video_id", os.path.basename(live_video_path))
event_utc = live_video.get("event_utc", None)
asr_text = live_video.get("asr_text", '')
live_video_contents = live_video.get("llm_contents", None)
if live_video_contents is None:
# event_start = event_utc - timedelta(seconds=10) if event_utc is not None else None
event_start = event_utc - 10000 if event_utc is not None else None
# event_end = event_utc + timedelta(seconds=5) if event_utc is not None else None
event_end = event_utc + 5000 if event_utc is not None else None
live_video_contents = self.video2frame.to_llm_contents( live_video_path,
cache=self.cache_dir,
fps=2,
start=event_start,
end=event_end,
roi=None,
max_px_area=400_000,
prompt_start=f"### 候选片段 video_id: {live_video_id} ###",
prompt_end=f"\n该片段解说内容: {asr_text}\n"
)
live_map[live_video_id] = {
"video_id": live_video_id,
"video_path": live_video_path,
"event_utc": event_utc,
"asr_text": asr_text,
"contents": live_video_contents,
}
live_records[live_video_id] = {
"video_id": live_video_id,
"video_path": live_video_path,
"event_utc": event_utc,
"asr_text": asr_text,
}
live_videos_contents.extend(live_video_contents)
user_contents.extend(replay_video_contents)
user_contents.append({'type': "text", "text": "\n【候选直播片段列表】\n"})
user_contents.extend(live_videos_contents)
user_contents.append({'type': "text", "text": "\n请根据上述片段进行匹配并按要求输出结果\n"})
system_message = SystemMessage(content=req_prompt)
user_message = HumanMessage(content=user_contents)
result = self.model.invoke([system_message, user_message]).content
try:
result_json = json.loads(result)
except json.JSONDecodeError:
try:
result_json = json.loads(result.replace("```json", "").replace("```", ""))
except Exception as e:
print("JSON解析失败:", result)
raise e
video_id = result_json.get("video_id", None)
result_live = live_map.get(video_id, None)
if record is not None:
record.append(
{"live": live_records.get(video_id,None), "llm_result": result_json, 'live_list': list(live_records.values())})
return result_live
def _match_batch(self, replay_video_contents: list, live_videos: list[dict], max_parallel: int = 3, cache_path=None,
record: list = None):
"""
Match a replay video with live videos to find the most likely match.
:param max_parallel:
:param replay_video: Path to the replay video.(video_path,asr_text)
:param live_videos: [(video_id,video_path,asr_text)]
:param cache_path: Path to cache the result.
:return: JSON object containing the match result.
"""
if cache_path is not None and os.path.exists(cache_path):
with open(cache_path, 'r', encoding='utf-8') as f:
return json.loads(f.read())
# 按照max_parallel对live_videos进行分组
live_videos_groups = [live_videos[i::max_parallel] for i in range(max_parallel)]
# 过滤掉空的分组
live_videos_groups = [g for g in live_videos_groups if g]
# 如果group 大于1 并且最后一个group 只有一个元素，将其唯一元素放入倒数第二个group，并删除最后一个group
if len(live_videos_groups) > 1 and len(live_videos_groups[-1]) == 1:
live_videos_groups[-2].append(live_videos_groups[-1][0])
live_videos_groups.pop()
if len(live_videos_groups) > 1:
match_result = []
for live_videos_group in live_videos_groups:
g_live = self._match_once(replay_video_contents, live_videos_group, record)
if g_live is None:
continue
match_result.append(g_live)
if len(match_result) == 0:
return None
else:
return self._match_batch(replay_video_contents, match_result, max_parallel, cache_path, record)
elif len(live_videos_groups) == 1:
return self._match_once(replay_video_contents, live_videos_groups[0], record)
else:
return None
def match_batch(self, replay_video: dict, live_videos: list[dict], max_parallel: int = 3, cache_dir=None):
self.cache_dir = cache_dir
cache_path = os.path.join(cache_dir, 'match_live.json')
if cache_path is not None and os.path.exists(cache_path):
try:
with open(cache_path, 'r', encoding='utf-8') as f:
return json.loads(f.read()).get("result", None)
except:
os.remove(cache_path)
replay_video_path = replay_video.get("url", None)
event_start = replay_video.get("start_utc", None)
event_end = replay_video.get("end_utc", None)
replay_video_contents = self.video2frame.to_llm_contents(replay_video_path,
cache=os.path.join(os.path.dirname(cache_dir), "replay"),
fps=2,
start=event_start,
end=event_end,
roi=None,
max_px_area=400_000,
prompt_start="\n【回放片段信息】\n",
prompt_end=f"\n回放解说内容：无\n"
)
live_record = []
result = self._match_batch(replay_video_contents, live_videos, max_parallel, cache_path, live_record)
if result is not None:
result_no_content = {
"video_id": result.get("video_id", None),
"video_path": result.get("video_path", None),
"event_utc": result.get("event_utc", None),
"asr_text": result.get("asr_text", None),
}
else:
result_no_content = None
record = {
"request": {
"replay_video": replay_video,
"live_videos": live_videos,
"max_parallel": max_parallel,
"cache_path": cache_path
},
"result": result_no_content,
"live_record": live_record
}
if cache_path is not None:
os.makedirs(Path(cache_path).parent, exist_ok=True)
with open(cache_path, 'w', encoding='utf-8') as f:
f.write(json.dumps(record, ensure_ascii=False, indent=4))
import json
import os.path
from pathlib import Path
from datetime import timedelta
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_openai import ChatOpenAI
try:
from .llm_image import Video2Frame
except:
from llm_image import Video2Frame
req_prompt = """
# Role
你是一名拥有20年经验的足球视频技术分析师，擅长结合**视觉画面**与**解说音频（ASR）**进行跨镜头的事件匹配。你能通过解说员的描述锁定时间背景，并通过视觉特征确认物理细节。
# Task
输入包含：
1. 【回放片段】（Replay）：包含视频帧 + **对应的解说文本**。
2. 【直播进球片段列表】（Live Candidates）：包含视频帧 + **对应的解说文本**。
目标：在【直播片段】中找到与【回放片段】属于**同一个进球事件**的片段。如果没有任何片段匹配，返回 null。
# Analysis Workflow (多模态分析流程)
请按照以下步骤进行推理：
### 第一步：解说词元数据提取 (听觉线索)
分析【回放片段】的解说文本，寻找以下关键信息：
- **时间指代**：解说员是否提到了具体时间？（如“上半场”、“第10分钟”、“开场不久”）。
- **事件描述**：解说员如何描述这个进球？（如“世界波”、“点球”、“补射”、“乌龙球”）。
- **球员提及**：解说员念出了谁的名字？
### 第二步：视觉物理指纹提取 (视觉线索)
忽略镜头语言（慢放、特写），提取核心物理特征：
- **进攻与射门**：进攻方式（传中/直塞）、射门部位（头/脚）、射门位置。
- **球路与防守**：球的轨迹（高/低/折射）、门将扑救动作（侧扑/倒地/未动）。
- **庆祝**：进球后的庆祝动作（仅作为辅助验证）。
### 第三步：跨模态匹配与验证
遍历【直播片段】，结合视觉和听觉进行判断：
- **视觉一致性**：直播片段的动作、轨迹、门将反应是否与回放完全吻合？
- **听觉一致性**：
- 如果回放解说提到“这是第5分钟的进球”，而直播片段的时间戳是90分钟，**不要直接排除**，而是检查直播片段的解说是否也提到了“回顾第5分钟”或者直播片段的视觉内容确实是第5分钟的动作。
- **核心原则**：视觉物理特征 > 解说词描述 > 时间戳元数据。
- **特殊情况**：如果视觉特征高度相似（如同一个球员、同一个角度），但解说词明确说“这是另一个进球（例如：这是他的第二个进球，而回放说是第一个）”，则判定为不匹配。
# Logic Constraints (逻辑约束 - 必须严格遵守)
- **时间单向性原则（铁律）**：
- **回看中的比赛时间一定在比赛片段画面时间之后**。
- 逻辑：回放是对过去发生事件的回顾。如果【回放片段】中解说提到的比赛时间（或画面显示的比赛时钟）**早于**【直播片段】中解说提到的比赛时间（或画面时钟），则**绝对不可能**是同一个事件。
- *示例*：回放解说在描述“第10分钟的进球”，而直播片段明确发生在“第5分钟”，则该直播片段**一定不是**目标。
- **时间陷阱**：回放可能是赛后集锦。如果解说员说“让我们看看**刚才**那个球”或者“**上半场**那个球”，即使当前比赛时间是90分钟，也要去匹配对应时间段的直播片段（或视觉特征）。
- **同名陷阱**：如果解说提到“又是**凯恩**进球了”，不能只看凯恩，必须看**怎么进的**（头球还是点球）。
- **无匹配处理**：如果所有候选片段在视觉动作（如射门方式、进球位置）或关键事件逻辑上与回放明显不符，必须判定为无匹配，将 `video_id` 设为 `null`。
### 输出要求
请仅输出一个JSON格式的结果，不要输出任何分析过程。不要包含 markdown 标记（如 ```json ... ```），不要包含任何解释或额外文本。
格式如下：
{
"replay_summary": {
"audio_cues": "解说提到的关键信息（如：'第15分钟', '远射', '德布劳内'）",
"visual_cues": "视觉关键特征（如：'禁区外右脚', '球挂死角', '门将飞身扑救'）"
},
"reasoning": "综合分析：回放解说提到是'上半场的远射'，视觉显示'17号球员禁区外起脚'。Candidate_1 视觉是'近距离推射'，排除。Candidate_2 视觉是'禁区外远射'，且门将动作一致，虽然直播时间显示是下半场（可能是集锦回顾），但解说也提到了'回顾上半场'，确认为同一事件。",
"video_id": "Candidate_2"
}"""
class FootballReplayMatchLive:
def __init__(self, base_url: str, model: str, temperature: float = 0.0, api_key: str = 'no_key', cache_dir:str=None, save_frames_enable:bool=False):
self.base_url = base_url
self.model = model
self.temperature = temperature
self.api_key = api_key
# self.model = ChatOllama(base_url="http://192.168.1.59:11434", model="qwen3.6:35b-a3b-q8_0", temperature=0.7,
# keep_alive=-1, reasoning=False)
# self.model = ChatOpenAI(base_url="http://192.168.1.59:11434/v1", model="qwen3.6:35b-a3b-q8_0", temperature=0.7,
# api_key='no_key')
self.model = ChatOpenAI(base_url=base_url, model=model, temperature=temperature, api_key=api_key,
extra_body={"chat_template_kwargs": {"enable_thinking": False}})
self.cache_dir = cache_dir
self.video2frame = Video2Frame(cache_dir=cache_dir, save_frames_enable=save_frames_enable)
def _match_once(self, replay_video_contents: list, live_videos: list[dict], record: list = None):
if len(live_videos) == 0:
return None
elif len(live_videos) == 1:
live_video_path = live_videos[0].get("url", None)
live_video_id = live_videos[0].get("video_id", os.path.basename(live_video_path))
asr_text = live_videos[0].get("asr_text", '')
live = {
"video_id": live_video_id,
"video_path": live_video_path,
"asr_text": asr_text,
}
if record is not None:
record.append({"live": live, "llm_result": None, 'live_list': [live]})
return live
user_contents = []
# replay_video_contents = video_contents(replay_video["video_path"], "\n【回放片段信息】\n",
# prompt_end=f"\n回放解说内容：{replay_video['asr_text']}\n",
# video_name=os.path.basename(replay_video["video_path"]),
# fps=2, max_frames=999, sampling_mode="head", max_short_edge=480)
live_videos_contents = []
live_map = {}
live_records = {}
for live_video in live_videos:
live_video_path = live_video.get("url", None)
live_video_id = live_video.get("video_id", os.path.basename(live_video_path))
event_utc = live_video.get("event_utc", None)
asr_text = live_video.get("asr_text", '')
live_video_contents = live_video.get("llm_contents", None)
if live_video_contents is None:
# event_start = event_utc - timedelta(seconds=10) if event_utc is not None else None
event_start = event_utc - 10000 if event_utc is not None else None
# event_end = event_utc + timedelta(seconds=5) if event_utc is not None else None
event_end = event_utc + 5000 if event_utc is not None else None
live_video_contents = self.video2frame.to_llm_contents( live_video_path,
cache=self.cache_dir,
fps=2,
start=event_start,
end=event_end,
roi=None,
max_px_area=400_000,
prompt_start=f"### 候选片段 video_id: {live_video_id} ###",
prompt_end=f"\n该片段解说内容: {asr_text}\n"
)
live_map[live_video_id] = {
"video_id": live_video_id,
"video_path": live_video_path,
"event_utc": event_utc,
"asr_text": asr_text,
"contents": live_video_contents,
}
live_records[live_video_id] = {
"video_id": live_video_id,
"video_path": live_video_path,
"event_utc": event_utc,
"asr_text": asr_text,
}
live_videos_contents.extend(live_video_contents)
user_contents.extend(replay_video_contents)
user_contents.append({'type': "text", "text": "\n【候选直播片段列表】\n"})
user_contents.extend(live_videos_contents)
user_contents.append({'type': "text", "text": "\n请根据上述片段进行匹配并按要求输出结果\n"})
system_message = SystemMessage(content=req_prompt)
user_message = HumanMessage(content=user_contents)
result = self.model.invoke([system_message, user_message]).content
try:
result_json = json.loads(result)
except json.JSONDecodeError:
try:
result_json = json.loads(result.replace("```json", "").replace("```", ""))
except Exception as e:
print("JSON解析失败:", result)
raise e
video_id = result_json.get("video_id", None)
result_live = live_map.get(video_id, None)
if record is not None:
record.append(
{"live": live_records.get(video_id,None), "llm_result": result_json, 'live_list': list(live_records.values())})
return result_live
def _match_batch(self, replay_video_contents: list, live_videos: list[dict], max_parallel: int = 3, cache_path=None,
record: list = None):
"""
Match a replay video with live videos to find the most likely match.
:param max_parallel:
:param replay_video: Path to the replay video.(video_path,asr_text)
:param live_videos: [(video_id,video_path,asr_text)]
:param cache_path: Path to cache the result.
:return: JSON object containing the match result.
"""
if cache_path is not None and os.path.exists(cache_path):
with open(cache_path, 'r', encoding='utf-8') as f:
return json.loads(f.read())
# 按照max_parallel对live_videos进行分组
live_videos_groups = [live_videos[i::max_parallel] for i in range(max_parallel)]
# 过滤掉空的分组
live_videos_groups = [g for g in live_videos_groups if g]
# 如果group 大于1 并且最后一个group 只有一个元素，将其唯一元素放入倒数第二个group，并删除最后一个group
if len(live_videos_groups) > 1 and len(live_videos_groups[-1]) == 1:
live_videos_groups[-2].append(live_videos_groups[-1][0])
live_videos_groups.pop()
if len(live_videos_groups) > 1:
match_result = []
for live_videos_group in live_videos_groups:
g_live = self._match_once(replay_video_contents, live_videos_group, record)
if g_live is None:
continue
match_result.append(g_live)
if len(match_result) == 0:
return None
else:
return self._match_batch(replay_video_contents, match_result, max_parallel, cache_path, record)
elif len(live_videos_groups) == 1:
return self._match_once(replay_video_contents, live_videos_groups[0], record)
else:
return None
def match_batch(self, replay_video: dict, live_videos: list[dict], max_parallel: int = 3, cache_dir=None):
self.cache_dir = cache_dir
cache_path = os.path.join(cache_dir, 'match_live.json')
if cache_path is not None and os.path.exists(cache_path):
try:
with open(cache_path, 'r', encoding='utf-8') as f:
return json.loads(f.read()).get("result", None)
except:
os.remove(cache_path)
replay_video_path = replay_video.get("url", None)
event_start = replay_video.get("start_utc", None)
event_end = replay_video.get("end_utc", None)
replay_video_contents = self.video2frame.to_llm_contents(replay_video_path,
cache=os.path.join(os.path.dirname(cache_dir), "replay"),
fps=2,
start=event_start,
end=event_end,
roi=None,
max_px_area=400_000,
prompt_start="\n【回放片段信息】\n",
prompt_end=f"\n回放解说内容：无\n"
)
live_record = []
result = self._match_batch(replay_video_contents, live_videos, max_parallel, cache_path, live_record)
if result is not None:
result_no_content = {
"video_id": result.get("video_id", None),
"video_path": result.get("video_path", None),
"event_utc": result.get("event_utc", None),
"asr_text": result.get("asr_text", None),
}
else:
result_no_content = None
record = {
"request": {
"replay_video": replay_video,
"live_videos": live_videos,
"max_parallel": max_parallel,
"cache_path": cache_path
},
"result": result_no_content,
"live_record": live_record
}
if cache_path is not None:
os.makedirs(Path(cache_path).parent, exist_ok=True)
with open(cache_path, 'w', encoding='utf-8') as f:
f.write(json.dumps(record, ensure_ascii=False, indent=4))
return result_no_content
\ No newline at end of file
... ...
import json
import os.path
from pathlib import Path
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_openai import ChatOpenAI
try:
from .llm_image import Video2Frame
except:
from llm_image import Video2Frame
req_prompt = """
### 角色设定
你是一位拥有20年经验的足球赛事视频分析专家。你的任务是结合**视频画面**和**解说音频文本**，精准判断视频片段中是否发生了"有效进球"。
### 输入内容
1. **视频片段**：可能包含赛场、观众、教练、回放、演播室、宣传片、广告等多种画面。
2. **解说文本**：该片段对应的实时解说内容（ASR）。
### 分析逻辑（思维链）
请综合以下三个维度的信息进行推理：
#### 1. 场景维度（非比赛画面过滤 - 最高优先级）
- **非比赛内容识别**：检查画面是否为**宣传片**、**商业广告**、**纯演播室解说**、**集锦混剪**（无连续比赛画面）或*
*静态图文**。
- **处理规则**：如果视频内容主要是上述非比赛画面，且没有包含明确的实时进球片段，**直接判定为“无进球”**
，无需进行后续的进球逻辑分析。
#### 2. 视觉维度（寻找关键证据）
- **核心画面**：足球入网、球在网内静止、裁判指中圈。
- **行为线索**：进攻方疯狂庆祝、防守方抱头懊恼、全场观众起立欢呼。
- **画面容忍度**：即使画面主要是观众或教练特写，只要上下文（如切入的进球回放）或行为暗示了进球，也应视为进球。
#### 3. 听觉维度（解说语义分析）
- **进球关键词**：寻找如“球进啦”、“Goal”、“得分”、“世界波”、“破门”、“1比0”等肯定性词汇。
- **情绪语调**：解说员音量突然升高、语速加快、情绪激动（通常伴随进球发生）。
- **否定排除**：如果解说提到“越位在先”、“进球无效”、“击中横梁”、“偏出”，则视为未进球。
### 判定规则
- **判定为“无进球”**：
- **画面为宣传片、广告、纯演播室或其他非比赛实时内容；**
- 视觉显示未进（偏出/被扑/中柱）；
- 解说明确表示未进或进球无效；
- 画面与解说均无进球迹象（如普通传球、界外球）。
- **判定为“进球”**：
- 画面为比赛内容，且视觉清晰显示进球；
- **或** 画面为比赛内容，视觉模糊但解说员明确喊出“球进了”且情绪激动；
- **或** 画面显示庆祝/回放，配合解说确认进球。
### 输出要求
请仅输出一个JSON格式的结果，不要输出任何分析过程。不要包含 markdown 标记（如 ```json ... ```），不要包含任何解释或额外文本。
格式如下：
{
"event_name": "进球" 或 "无进球",
"description": "判定理由，若是宣传片请直接注明"
}"""
class FootballReplayVideoEvent:
def __init__(self, base_url: str, model: str, temperature: float = 0.0, api_key: str = 'no_key', cache_dir:str=None, save_frames_enable:bool=False):
self.base_url = base_url
self.model = model
self.temperature = temperature
self.api_key = api_key
# self.model = ChatOllama(base_url="http://192.168.1.59:11434", model="qwen3.6:35b-a3b-q8_0", temperature=0.7,
# keep_alive=-1, reasoning=False)
# self.model = ChatOpenAI(base_url="http://192.168.1.59:11434/v1", model="qwen3.6:35b-a3b-q8_0", temperature=0.7,
# api_key='no_key')
self.model = ChatOpenAI(base_url=base_url, model=model, temperature=temperature, api_key=api_key,
extra_body={"chat_template_kwargs": {"enable_thinking": False}})
self.video2frame = Video2Frame(cache_dir=cache_dir, save_frames_enable=save_frames_enable)
def video_event(self, replay_pack: dict, asr_text: str = '无', cache_dir=None):
replay_video_path = replay_pack.get("url", None)
cache_path = os.path.join(cache_dir, 'video_event.json')
if cache_path is not None and os.path.exists(cache_path):
with open(cache_path, 'r', encoding='utf-8') as f:
return json.loads(f.read())
event_start = replay_pack.get("start_utc", None)
event_end = replay_pack.get("end_utc", None)
contents = self.video2frame.to_llm_contents(replay_video_path,
cache=cache_dir,
fps=2,
start=event_start,
end=event_end,
roi=None,
max_px_area=400_000,
prompt_start="\n【回放片段信息】\n",
prompt_end=f"\n回放解说内容：{asr_text}\n"
)
system_message = SystemMessage(content=req_prompt)
video_message = HumanMessage(content=contents)
asr_message = HumanMessage(content=f"解说内容：{asr_text}")
result = self.model.invoke([system_message, video_message, asr_message]).content
try:
result_json = json.loads(result)
except json.JSONDecodeError:
result_json = json.loads(result.replace("```json", "").replace("```", ""))
if cache_path is not None:
os.makedirs(Path(cache_path).parent, exist_ok=True)
with open(cache_path, 'w', encoding='utf-8') as f:
f.write(result)
return result_json
if __name__ == "__main__":
from aabd.base.patched_logging import init_logging
import os
os.environ["APP_LOG_TYPE"] = "console"
init_logging()
fbrv = FootballReplayVideoEvent(base_url="http://192.168.1.59:11434/v1",
model="Qwen3.6-35B-A3B-UD-Q8_K_XL.gguf",
temperature=0.7,
cache_dir="/root/lzw/tmp_0518_replay_cache",
save_frames_enable=True
)
replay_pack = {"url": "http://video.mam.miguvideo.com/mnt6/fastclip3/wsc/2026/04/29/df760b8d15394cbc9ed0d0a4a9c4b786_1080PS/29133605/vodtmp/3b3e99cf4ca84c3782503d8817242de2.m3u8",
"start_utc": 0,
"end_utc": 999999999,
"asr_text": '无'
}
replay_goals = fbrv.video_event(replay_pack, cache_path="/root/lzw/tmp_0518_replay_cache")
import json
import os.path
from pathlib import Path
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_openai import ChatOpenAI
try:
from .llm_image import Video2Frame
except:
from llm_image import Video2Frame
req_prompt = """
### 角色设定
你是一位拥有20年经验的足球赛事视频分析专家。你的任务是结合**视频画面**和**解说音频文本**，精准判断视频片段中是否发生了"有效进球"。
### 输入内容
1. **视频片段**：可能包含赛场、观众、教练、回放、演播室、宣传片、广告等多种画面。
2. **解说文本**：该片段对应的实时解说内容（ASR）。
### 分析逻辑（思维链）
请综合以下三个维度的信息进行推理：
#### 1. 场景维度（非比赛画面过滤 - 最高优先级）
- **非比赛内容识别**：检查画面是否为**宣传片**、**商业广告**、**纯演播室解说**、**集锦混剪**（无连续比赛画面）或*
*静态图文**。
- **处理规则**：如果视频内容主要是上述非比赛画面，且没有包含明确的实时进球片段，**直接判定为“无进球”**
，无需进行后续的进球逻辑分析。
#### 2. 视觉维度（寻找关键证据）
- **核心画面**：足球入网、球在网内静止、裁判指中圈。
- **行为线索**：进攻方疯狂庆祝、防守方抱头懊恼、全场观众起立欢呼。
- **画面容忍度**：即使画面主要是观众或教练特写，只要上下文（如切入的进球回放）或行为暗示了进球，也应视为进球。
#### 3. 听觉维度（解说语义分析）
- **进球关键词**：寻找如“球进啦”、“Goal”、“得分”、“世界波”、“破门”、“1比0”等肯定性词汇。
- **情绪语调**：解说员音量突然升高、语速加快、情绪激动（通常伴随进球发生）。
- **否定排除**：如果解说提到“越位在先”、“进球无效”、“击中横梁”、“偏出”，则视为未进球。
### 判定规则
- **判定为“无进球”**：
- **画面为宣传片、广告、纯演播室或其他非比赛实时内容；**
- 视觉显示未进（偏出/被扑/中柱）；
- 解说明确表示未进或进球无效；
- 画面与解说均无进球迹象（如普通传球、界外球）。
- **判定为“进球”**：
- 画面为比赛内容，且视觉清晰显示进球；
- **或** 画面为比赛内容，视觉模糊但解说员明确喊出“球进了”且情绪激动；
- **或** 画面显示庆祝/回放，配合解说确认进球。
### 输出要求
请仅输出一个JSON格式的结果，不要输出任何分析过程。不要包含 markdown 标记（如 ```json ... ```），不要包含任何解释或额外文本。
格式如下：
{
"event_name": "进球" 或 "无进球",
"description": "判定理由，若是宣传片请直接注明"
}"""
class FootballReplayVideoEvent:
def __init__(self, base_url: str, model: str, temperature: float = 0.0, api_key: str = 'no_key', cache_dir:str=None, save_frames_enable:bool=False):
self.base_url = base_url
self.model = model
self.temperature = temperature
self.api_key = api_key
# self.model = ChatOllama(base_url="http://192.168.1.59:11434", model="qwen3.6:35b-a3b-q8_0", temperature=0.7,
# keep_alive=-1, reasoning=False)
# self.model = ChatOpenAI(base_url="http://192.168.1.59:11434/v1", model="qwen3.6:35b-a3b-q8_0", temperature=0.7,
# api_key='no_key')
self.model = ChatOpenAI(base_url=base_url, model=model, temperature=temperature, api_key=api_key,
extra_body={"chat_template_kwargs": {"enable_thinking": False}})
self.video2frame = Video2Frame(cache_dir=cache_dir, save_frames_enable=save_frames_enable)
def video_event(self, replay_pack: dict, asr_text: str = '无', cache_dir=None):
replay_video_path = replay_pack.get("url", None)
cache_path = os.path.join(cache_dir, 'video_event.json')
if cache_path is not None and os.path.exists(cache_path):
with open(cache_path, 'r', encoding='utf-8') as f:
return json.loads(f.read())
event_start = replay_pack.get("start_utc", None)
event_end = replay_pack.get("end_utc", None)
contents = self.video2frame.to_llm_contents(replay_video_path,
cache=cache_dir,
fps=2,
start=event_start,
end=event_end,
roi=None,
max_px_area=400_000,
prompt_start="\n【回放片段信息】\n",
prompt_end=f"\n回放解说内容：{asr_text}\n"
)
system_message = SystemMessage(content=req_prompt)
video_message = HumanMessage(content=contents)
asr_message = HumanMessage(content=f"解说内容：{asr_text}")
result = self.model.invoke([system_message, video_message, asr_message]).content
try:
result_json = json.loads(result)
except json.JSONDecodeError:
result_json = json.loads(result.replace("```json", "").replace("```", ""))
if cache_path is not None:
os.makedirs(Path(cache_path).parent, exist_ok=True)
with open(cache_path, 'w', encoding='utf-8') as f:
f.write(result)
return result_json
if __name__ == "__main__":
from aabd.base.patched_logging import init_logging
import os
os.environ["APP_LOG_TYPE"] = "console"
init_logging()
fbrv = FootballReplayVideoEvent(base_url="http://192.168.1.59:11434/v1",
model="Qwen3.6-35B-A3B-UD-Q8_K_XL.gguf",
temperature=0.7,
cache_dir="/root/lzw/tmp_0518_replay_cache",
save_frames_enable=True
)
replay_pack = {"url": "http://video.mam.miguvideo.com/mnt6/fastclip3/wsc/2026/04/29/df760b8d15394cbc9ed0d0a4a9c4b786_1080PS/29133605/vodtmp/3b3e99cf4ca84c3782503d8817242de2.m3u8",
"start_utc": 0,
"end_utc": 999999999,
"asr_text": '无'
}
replay_goals = fbrv.video_event(replay_pack, cache_path="/root/lzw/tmp_0518_replay_cache")
print(replay_goals)
\ No newline at end of file
... ...
import base64
import shutil
import subprocess
import sys
from pathlib import Path
from typing import Optional
from urllib.parse import urlparse
import cv2
from aabd.stream_chain import video2frames
import json
import logging
import hashlib
logger = logging.getLogger(__name__)
_IS_ABSOLUTE_UTC_THRESHOLD = 1_000_000_000
def download_to_mp4(url: str, output_path: str, duration: Optional[int] = None) -> None:
cmd = ["ffmpeg", "-y", "-fflags", "+discardcorrupt", "-i", url]
if duration is not None:
cmd.extend(["-t", str(duration)])
cmd.extend(["-c", "copy", "-bsf:a", "aac_adtstoasc", "-movflags", "+faststart", output_path])
result = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace")
if result.returncode != 0:
raise RuntimeError(f"ffmpeg 下载失败: {result.stderr}")
def resize_frame(frame, max_px_area):
if max_px_area is None:
return frame
h, w = frame.shape[:2]
area = h * w
if area > max_px_area:
scale = (max_px_area / area) ** 0.5
new_w = int(w * scale)
new_h = int(h * scale)
frame = cv2.resize(frame, (new_w, new_h), interpolation=cv2.INTER_AREA)
return frame
class Video2Frame:
def __init__(self, cache_dir, save_frames_enable=False):
self.cache_dir = Path(cache_dir)
self.save_frames_enable = save_frames_enable
def get_root_path(self, url, cache=None):
if cache is None:
parsed_url = urlparse(url)
raw = f"{parsed_url.scheme}/{parsed_url.netloc}{parsed_url.path}"
cache = hashlib.md5(raw.encode('utf-8')).hexdigest()
return self.cache_dir / cache
def iter_mp4(self, file, start, end, fps):
if (start is not None and start > _IS_ABSOLUTE_UTC_THRESHOLD) or (
end is not None and end > _IS_ABSOLUTE_UTC_THRESHOLD):
video_start_time = 0
with video2frames.AVAnyKeyStreamDecoder(file, sei_enable=True, tqdm_enable=False) as (sd, _):
for frame in sd:
if start is not None:
video_start_time = start - frame.get('src_frame_time')
if end is not None:
video_end_time = end - frame.get('src_frame_time') or 0
break
else:
video_start_time = start if start is not None else 0
video_end_time = end if end is not None else sys.maxsize
with video2frames.AVAnyKeyStreamDecoder(file, start=video_start_time, end=video_end_time, control_type='time',
sei_enable=True, max_fps=fps, frame_type='numpy_bgr') as (sd, _):
yield from sd
def to_frames(self, url, cache=None, fps=None, start=None, end=None, roi=None, max_px_area=None) -> list:
video_root_path = self.get_root_path(url, cache)
download_video_path = video_root_path / "video.mp4"
if not download_video_path.exists():
logger.info(f"download_video: {url} -> {download_video_path}")
video_root_path.mkdir(parents=True, exist_ok=True)
download_to_mp4(url, str(download_video_path), duration=None)
logger.info(f"video_path: {download_video_path}")
cache_dir = video_root_path / 'caches' / f"{start}-{end}-{'_'.join(roi) if roi else 'None'}-{max_px_area}"
cache_frames = cache_dir / "frames"
cache_name = cache_dir / "data.json"
if cache_name.exists():
logger.info(f"use_cache: {cache_name}")
# return json.load(cache_name.open('r'))
with cache_name.open('r', encoding='utf-8') as f:
return json.load(f)
else:
cache_dir.mkdir(parents=True, exist_ok=True)
data = {
'fps': fps,
'frames': []
}
if self.save_frames_enable:
cache_frames.mkdir(parents=True, exist_ok=True)
for f in self.iter_mp4(str(download_video_path), start, end, fps):
frame = f['frame']
time = f['src_frame_time']
fps = f['fps']
data['fps'] = fps
if roi is not None:
x1, y1, x2, y2 = roi
frame = frame[y1:y2, x1:x2]
if max_px_area is not None:
frame = resize_frame(frame, max_px_area)
_, buffer = cv2.imencode('.jpg', frame)
b64_str = base64.b64encode(buffer).decode('utf-8')
data['frames'].append({'time': time, 'image': b64_str})
if self.save_frames_enable:
frame_path = cache_frames / f"{time:015d}.jpg"
cv2.imwrite(frame_path, frame)
cache_name.parent.mkdir(parents=True, exist_ok=True)
# json.dump(data, cache_name.open('w'), indent=4)
with cache_name.open('w', encoding='utf-8') as f:
json.dump(data, f, indent=4)
logger.info(f"save_cache: {cache_name}")
return data
def to_llm_contents(self, url, cache=None, fps=None, start=None, end=None, roi=None, max_px_area=None,
prompt_start=None,
prompt_end=None):
data = self.to_frames(url, cache, fps, start, end, roi, max_px_area)
fps = data['fps']
frames = data['frames']
contents = []
if prompt_start is not None:
contents.append({"type": "text", "text": prompt_start})
video_prompt = (
f"以下是从视频中按时间顺序提取的 {len(frames)} 帧画面，fps={fps}，请将它们视为一个连续的视频进行分析。"
)
contents.append({"type": "text", "text": video_prompt})
for frame in frames:
contents.append({
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{frame['image']}"
}
})
if prompt_end is not None:
contents.append({"type": "text", "text": prompt_end})
return contents
if __name__ == '__main__':
from aabd.base.patched_logging import init_logging
import os
os.environ["APP_LOG_TYPE"] = "console"
init_logging()
# url = "http://video.mam.miguvideo.com/mnt6/fastclip3/wsc/2026/04/29/df760b8d15394cbc9ed0d0a4a9c4b786_1080PS/29133605/vodtmp/3b3e99cf4ca84c3782503d8817242de2.m3u8"
url = rf"/root/lzw/finished/69dd5845dd0412067b8d5587-auto-1776074760653/live/videos/00-14-09-052.mp4"
caceh_dir = r"/root/lzw/aigc-embedding-service/src/football_replay_match/core"
vf = Video2Frame(caceh_dir, save_frames_enable=True)
# vf = Video2Frame(rf'D:\Code\migu\aigc-embedding-service\src\football_replay_match\core', save_frames_enable=True)
# a = vf.to_llm_contents(url, fps=2, start=1777444502543, end=1777444531063, max_px_area=1920 * 1080 // 6)
a = vf.to_llm_contents(url, fps=2, start=0, end=10_000, max_px_area=1920 * 1080 // 6)
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
model = ChatOpenAI(base_url="http://192.168.1.59:11434/v1", model="Qwen3.6-35B-A3B-UD-Q8_K_XL.gguf",
temperature=0.7, api_key='no_key',
extra_body={"chat_template_kwargs": {"enable_thinking": False}})
result = model.invoke([SystemMessage(content="描述一下视频的内容。"), HumanMessage(content=a)])
print(result)
import base64
import shutil
import subprocess
import sys
from pathlib import Path
from typing import Optional
from urllib.parse import urlparse
import cv2
from aabd.stream_chain import video2frames
import json
import logging
import hashlib
logger = logging.getLogger(__name__)
_IS_ABSOLUTE_UTC_THRESHOLD = 1_000_000_000
def download_to_mp4(url: str, output_path: str, duration: Optional[int] = None) -> None:
cmd = ["ffmpeg", "-y", "-fflags", "+discardcorrupt", "-i", url]
if duration is not None:
cmd.extend(["-t", str(duration)])
cmd.extend(["-c", "copy", "-bsf:a", "aac_adtstoasc", "-movflags", "+faststart", output_path])
result = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace")
if result.returncode != 0:
raise RuntimeError(f"ffmpeg 下载失败: {result.stderr}")
def resize_frame(frame, max_px_area):
if max_px_area is None:
return frame
h, w = frame.shape[:2]
area = h * w
if area > max_px_area:
scale = (max_px_area / area) ** 0.5
new_w = int(w * scale)
new_h = int(h * scale)
frame = cv2.resize(frame, (new_w, new_h), interpolation=cv2.INTER_AREA)
return frame
class Video2Frame:
def __init__(self, cache_dir, save_frames_enable=False):
self.cache_dir = Path(cache_dir)
self.save_frames_enable = save_frames_enable
def get_root_path(self, url, cache=None):
if cache is None:
parsed_url = urlparse(url)
raw = f"{parsed_url.scheme}/{parsed_url.netloc}{parsed_url.path}"
cache = hashlib.md5(raw.encode('utf-8')).hexdigest()
return self.cache_dir / cache
def iter_mp4(self, file, start, end, fps):
if (start is not None and start > _IS_ABSOLUTE_UTC_THRESHOLD) or (
end is not None and end > _IS_ABSOLUTE_UTC_THRESHOLD):
video_start_time = 0
with video2frames.AVAnyKeyStreamDecoder(file, sei_enable=True, tqdm_enable=False) as (sd, _):
for frame in sd:
if start is not None:
video_start_time = start - frame.get('src_frame_time')
if end is not None:
video_end_time = end - frame.get('src_frame_time') or 0
break
else:
video_start_time = start if start is not None else 0
video_end_time = end if end is not None else sys.maxsize
with video2frames.AVAnyKeyStreamDecoder(file, start=video_start_time, end=video_end_time, control_type='time',
sei_enable=True, max_fps=fps, frame_type='numpy_bgr') as (sd, _):
yield from sd
def to_frames(self, url, cache=None, fps=None, start=None, end=None, roi=None, max_px_area=None) -> list:
video_root_path = self.get_root_path(url, cache)
download_video_path = video_root_path / "video.mp4"
if not download_video_path.exists():
logger.info(f"download_video: {url} -> {download_video_path}")
video_root_path.mkdir(parents=True, exist_ok=True)
download_to_mp4(url, str(download_video_path), duration=None)
logger.info(f"video_path: {download_video_path}")
cache_dir = video_root_path / 'caches' / f"{start}-{end}-{'_'.join(roi) if roi else 'None'}-{max_px_area}"
cache_frames = cache_dir / "frames"
cache_name = cache_dir / "data.json"
if cache_name.exists():
logger.info(f"use_cache: {cache_name}")
# return json.load(cache_name.open('r'))
with cache_name.open('r', encoding='utf-8') as f:
return json.load(f)
else:
cache_dir.mkdir(parents=True, exist_ok=True)
data = {
'fps': fps,
'frames': []
}
if self.save_frames_enable:
cache_frames.mkdir(parents=True, exist_ok=True)
for f in self.iter_mp4(str(download_video_path), start, end, fps):
frame = f['frame']
time = f['src_frame_time']
fps = f['fps']
data['fps'] = fps
if roi is not None:
x1, y1, x2, y2 = roi
frame = frame[y1:y2, x1:x2]
if max_px_area is not None:
frame = resize_frame(frame, max_px_area)
_, buffer = cv2.imencode('.jpg', frame)
b64_str = base64.b64encode(buffer).decode('utf-8')
data['frames'].append({'time': time, 'image': b64_str})
if self.save_frames_enable:
frame_path = cache_frames / f"{time:015d}.jpg"
cv2.imwrite(frame_path, frame)
cache_name.parent.mkdir(parents=True, exist_ok=True)
# json.dump(data, cache_name.open('w'), indent=4)
with cache_name.open('w', encoding='utf-8') as f:
json.dump(data, f, indent=4)
logger.info(f"save_cache: {cache_name}")
return data
def to_llm_contents(self, url, cache=None, fps=None, start=None, end=None, roi=None, max_px_area=None,
prompt_start=None,
prompt_end=None):
data = self.to_frames(url, cache, fps, start, end, roi, max_px_area)
fps = data['fps']
frames = data['frames']
contents = []
if prompt_start is not None:
contents.append({"type": "text", "text": prompt_start})
video_prompt = (
f"以下是从视频中按时间顺序提取的 {len(frames)} 帧画面，fps={fps}，请将它们视为一个连续的视频进行分析。"
)
contents.append({"type": "text", "text": video_prompt})
for frame in frames:
contents.append({
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{frame['image']}"
}
})
if prompt_end is not None:
contents.append({"type": "text", "text": prompt_end})
return contents
if __name__ == '__main__':
from aabd.base.patched_logging import init_logging
import os
os.environ["APP_LOG_TYPE"] = "console"
init_logging()
# url = "http://video.mam.miguvideo.com/mnt6/fastclip3/wsc/2026/04/29/df760b8d15394cbc9ed0d0a4a9c4b786_1080PS/29133605/vodtmp/3b3e99cf4ca84c3782503d8817242de2.m3u8"
url = rf"/root/lzw/finished/69dd5845dd0412067b8d5587-auto-1776074760653/live/videos/00-14-09-052.mp4"
caceh_dir = r"/root/lzw/aigc-embedding-service/src/football_replay_match/core"
vf = Video2Frame(caceh_dir, save_frames_enable=True)
# vf = Video2Frame(rf'D:\Code\migu\aigc-embedding-service\src\football_replay_match\core', save_frames_enable=True)
# a = vf.to_llm_contents(url, fps=2, start=1777444502543, end=1777444531063, max_px_area=1920 * 1080 // 6)
a = vf.to_llm_contents(url, fps=2, start=0, end=10_000, max_px_area=1920 * 1080 // 6)
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
model = ChatOpenAI(base_url="http://192.168.1.59:11434/v1", model="Qwen3.6-35B-A3B-UD-Q8_K_XL.gguf",
temperature=0.7, api_key='no_key',
extra_body={"chat_template_kwargs": {"enable_thinking": False}})
result = model.invoke([SystemMessage(content="描述一下视频的内容。"), HumanMessage(content=a)])
print(result)
... ...
from football_replay_match_live import FootballReplayMatchLive
# from qwen_asr_util import QwenAsr
from football_replay_video_event_by_llm import FootballReplayVideoEvent
import os
import json
def batch_match():
cache_dir = "/root/lzw/tmp_0518_replay_cache03"
replay_match_live = FootballReplayMatchLive(base_url="http://192.168.1.59:11434/v1",
model="Qwen3.6-35B-A3B-UD-Q8_K_XL.gguf",
temperature=0.7,
cache_dir=cache_dir,
save_frames_enable=True
)
# qwen_asr = QwenAsr(base_url="http://192.168.1.59:8101/v1", model="Qwen/Qwen3-ASR-1.7B")
fbrv = FootballReplayVideoEvent(base_url="http://192.168.1.59:11434/v1",
model="Qwen3.6-35B-A3B-UD-Q8_K_XL.gguf",
temperature=0.7,
cache_dir=cache_dir,
save_frames_enable=True
)
videos_dir = rf"/root/lzw/finished"
for video_name in sorted(os.listdir(videos_dir)):
live_dir = os.path.join(videos_dir, video_name, 'live')
print(f"live_dir: {live_dir}")
live_video_dir = os.path.join(live_dir, 'videos')
live_asr_dir = os.path.join(live_dir, 'asr')
live_packs = []
for live_video_name in sorted(os.listdir(live_video_dir)):
live_video_path = os.path.join(live_video_dir, live_video_name)
live_asr_path = os.path.join(live_asr_dir, f"{live_video_name}.txt")
# live_asr_text = qwen_asr.asr(live_video_path, cache_path=live_asr_path)
live_asr_text = ''
live_packs.append({
"video_id": os.path.basename(live_video_path),
"url": live_video_path,
"asr_text": live_asr_text
})
replays_dir = os.path.join(videos_dir, video_name, 'replays')
replay_videos_dir = os.path.join(replays_dir, 'videos')
replay_goals_dir = os.path.join(replays_dir, 'goals')
replay_asr_dir = os.path.join(replays_dir, 'asr')
replay_matches_dir = os.path.join(replays_dir, 'matches')
for replay_video_name in sorted(os.listdir(replay_videos_dir)):
replay_video_path = os.path.join(replay_videos_dir, replay_video_name)
# replay_goals_path = os.path.join(replay_goals_dir, f"{replay_video_name}.json")
# replay_asr_path = os.path.join(replay_asr_dir, f"{replay_video_name}.txt")
# replay_match_path = os.path.join(replay_matches_dir, f"{replay_video_name}.json")
# print(100*"*")
# print(replay_match_path)
replay_match_path = None
# replay_asr_text = qwen_asr.asr(replay_video_path, cache_path=replay_asr_path)
replay_asr_text = ''
replay_pack = {"url": "http://video.mam.miguvideo.com/mnt6/fastclip3/wsc/2026/04/29/df760b8d15394cbc9ed0d0a4a9c4b786_1080PS/29133605/vodtmp/3b3e99cf4ca84c3782503d8817242de2.m3u8",
"start_utc": 1777444502543,
"end_utc": 1777444531063,
"asr_text": '无'
}
replay_goals = fbrv.video_event(replay_pack, cache_dir=cache_dir)
if replay_goals['event_name'] == '无进球':
continue
# replay_pack = {"video_path": replay_video_path, "asr_text": replay_asr_text}
try:
import time
start_time = time.time()
print("Start matching...")
print(f"len(live_packs): {len(live_packs)}")
result = replay_match_live.match_batch(replay_pack, live_packs, max_parallel=2, cache_dir=cache_dir)
end_time = time.time()
print(f"Time taken for matching: {end_time - start_time:.2f} seconds")
print(replay_video_path)
print(result)
break
except Exception as e:
print(f"Error processing {replay_video_path}: {e}")
print('-' * 20)
break
if __name__ == '__main__':
batch_match()
from football_replay_match_live import FootballReplayMatchLive
# from qwen_asr_util import QwenAsr
from football_replay_video_event_by_llm import FootballReplayVideoEvent
import os
import json
def batch_match():
cache_dir = "/root/lzw/tmp_0518_replay_cache03"
replay_match_live = FootballReplayMatchLive(base_url="http://192.168.1.59:11434/v1",
model="Qwen3.6-35B-A3B-UD-Q8_K_XL.gguf",
temperature=0.7,
cache_dir=cache_dir,
save_frames_enable=True
)
# qwen_asr = QwenAsr(base_url="http://192.168.1.59:8101/v1", model="Qwen/Qwen3-ASR-1.7B")
fbrv = FootballReplayVideoEvent(base_url="http://192.168.1.59:11434/v1",
model="Qwen3.6-35B-A3B-UD-Q8_K_XL.gguf",
temperature=0.7,
cache_dir=cache_dir,
save_frames_enable=True
)
videos_dir = rf"/root/lzw/finished"
for video_name in sorted(os.listdir(videos_dir)):
live_dir = os.path.join(videos_dir, video_name, 'live')
print(f"live_dir: {live_dir}")
live_video_dir = os.path.join(live_dir, 'videos')
live_asr_dir = os.path.join(live_dir, 'asr')
live_packs = []
for live_video_name in sorted(os.listdir(live_video_dir)):
live_video_path = os.path.join(live_video_dir, live_video_name)
live_asr_path = os.path.join(live_asr_dir, f"{live_video_name}.txt")
# live_asr_text = qwen_asr.asr(live_video_path, cache_path=live_asr_path)
live_asr_text = ''
live_packs.append({
"video_id": os.path.basename(live_video_path),
"url": live_video_path,
"asr_text": live_asr_text
})
replays_dir = os.path.join(videos_dir, video_name, 'replays')
replay_videos_dir = os.path.join(replays_dir, 'videos')
replay_goals_dir = os.path.join(replays_dir, 'goals')
replay_asr_dir = os.path.join(replays_dir, 'asr')
replay_matches_dir = os.path.join(replays_dir, 'matches')
for replay_video_name in sorted(os.listdir(replay_videos_dir)):
replay_video_path = os.path.join(replay_videos_dir, replay_video_name)
# replay_goals_path = os.path.join(replay_goals_dir, f"{replay_video_name}.json")
# replay_asr_path = os.path.join(replay_asr_dir, f"{replay_video_name}.txt")
# replay_match_path = os.path.join(replay_matches_dir, f"{replay_video_name}.json")
# print(100*"*")
# print(replay_match_path)
replay_match_path = None
# replay_asr_text = qwen_asr.asr(replay_video_path, cache_path=replay_asr_path)
replay_asr_text = ''
replay_pack = {"url": "http://video.mam.miguvideo.com/mnt6/fastclip3/wsc/2026/04/29/df760b8d15394cbc9ed0d0a4a9c4b786_1080PS/29133605/vodtmp/3b3e99cf4ca84c3782503d8817242de2.m3u8",
"start_utc": 1777444502543,
"end_utc": 1777444531063,
"asr_text": '无'
}
replay_goals = fbrv.video_event(replay_pack, cache_dir=cache_dir)
if replay_goals['event_name'] == '无进球':
continue
# replay_pack = {"video_path": replay_video_path, "asr_text": replay_asr_text}
try:
import time
start_time = time.time()
print("Start matching...")
print(f"len(live_packs): {len(live_packs)}")
result = replay_match_live.match_batch(replay_pack, live_packs, max_parallel=2, cache_dir=cache_dir)
end_time = time.time()
print(f"Time taken for matching: {end_time - start_time:.2f} seconds")
print(replay_video_path)
print(result)
break
except Exception as e:
print(f"Error processing {replay_video_path}: {e}")
print('-' * 20)
break
if __name__ == '__main__':
batch_match()
... ...
from aabd.base.patched_logging import init_logging, get_logger
init_logging()
logger = get_logger(__name__)
import json
import time
try:
from config import settings
except:
from .config import settings
pass
try:
from .core.api import FootballReplayMatch
except:
from core.api import FootballReplayMatch
pass
from threading import Thread
from aabd.mq.kafka_client import KafkaMessageIterator, KafkaProducer
def get_message_iterator(kafka_config):
return KafkaMessageIterator(bootstrap_servers=kafka_config.get('servers', None),
group_id=kafka_config.get('group_id', None),
topic=kafka_config.get('topic', None),
sasl_plain_username=kafka_config.get('username', None),
sasl_plain_password=kafka_config.get('password', None),
value_deserializer="str")
def get_message_producer(kafka_config):
return KafkaProducer(bootstrap_servers=kafka_config.get('servers', None),
sasl_plain_username=kafka_config.get('username', None),
sasl_plain_password=kafka_config.get('password', None))
def get_callback_func(producer, topic):
def callback_func(task_id, call_data):
try:
st = time.time()
producer.send_message_async(topic, call_data, key=task_id, timeout=5)
logger.info(f"sent_message[{task_id}][{time.time() - st:.2f}s]: {call_data}")
except:
logger.exception(f"Error sending message: {call_data}")
return callback_func
def kafka_message_iterator_thread(message_iterator, config, callback_func):
frm = FootballReplayMatch(config)
with message_iterator:
for message in message_iterator:
try:
json_message = json.loads(message)
result = frm.replay_match_event(json_message)
task_id = json_message.get("id")
callback_func(task_id, json.dumps(result, ensure_ascii=False))
logger.info(f"Processed task_id={task_id}, result={result}")
except Exception:
logger.exception(f"Error processing message: {message}")
def main():
message_iter = get_message_iterator(settings.input_kafka)
kafka_producer = get_message_producer(settings.output_kafka)
try:
Thread(target=kafka_message_iterator_thread,
args=(message_iter, settings,
get_callback_func(kafka_producer, settings.output_kafka.get('topic', None)))).start()
except KeyboardInterrupt:
message_iter.running = False
kafka_producer.close()
if __name__ == '__main__':
main()
from aabd.base.patched_logging import init_logging, get_logger
init_logging()
logger = get_logger(__name__)
import json
import time
try:
from config import settings
except:
from .config import settings
pass
try:
from .core.api import FootballReplayMatch
except:
from core.api import FootballReplayMatch
pass
from threading import Thread
from aabd.mq.kafka_client import KafkaMessageIterator, KafkaProducer
def get_message_iterator(kafka_config):
return KafkaMessageIterator(bootstrap_servers=kafka_config.get('servers', None),
group_id=kafka_config.get('group_id', None),
topic=kafka_config.get('topic', None),
sasl_plain_username=kafka_config.get('username', None),
sasl_plain_password=kafka_config.get('password', None),
value_deserializer="str")
def get_message_producer(kafka_config):
return KafkaProducer(bootstrap_servers=kafka_config.get('servers', None),
sasl_plain_username=kafka_config.get('username', None),
sasl_plain_password=kafka_config.get('password', None))
def get_callback_func(producer, topic):
def callback_func(task_id, call_data):
try:
st = time.time()
producer.send_message_async(topic, call_data, key=task_id, timeout=5)
logger.info(f"sent_message[{task_id}][{time.time() - st:.2f}s]: {call_data}")
except:
logger.exception(f"Error sending message: {call_data}")
return callback_func
def kafka_message_iterator_thread(message_iterator, config, callback_func):
frm = FootballReplayMatch(config)
with message_iterator:
for message in message_iterator:
try:
json_message = json.loads(message)
result = frm.replay_match_event(json_message)
task_id = json_message.get("id")
callback_func(task_id, json.dumps(result, ensure_ascii=False))
logger.info(f"Processed task_id={task_id}, result={result}")
except Exception:
logger.exception(f"Error processing message: {message}")
def main():
message_iter = get_message_iterator(settings.input_kafka)
kafka_producer = get_message_producer(settings.output_kafka)
try:
Thread(target=kafka_message_iterator_thread,
args=(message_iter, settings,
get_callback_func(kafka_producer, settings.output_kafka.get('topic', None)))).start()
except KeyboardInterrupt:
message_iter.running = False
kafka_producer.close()
if __name__ == '__main__':
main()
... ...
This diff could not be displayed because it is too large.