diff --git a/src/BowongModalFunctions/models/web_model.py b/src/BowongModalFunctions/models/web_model.py
index f6aa0b7..4b59385 100644
--- a/src/BowongModalFunctions/models/web_model.py
+++ b/src/BowongModalFunctions/models/web_model.py
@@ -4,8 +4,7 @@ from enum import Enum
from typing import List, Union, Optional, Dict, Any
import pydantic
-from pydantic import BaseModel, Field, field_validator, ConfigDict, HttpUrl, computed_field
-from pydantic.json_schema import JsonSchemaValue
+from pydantic import BaseModel, Field, field_validator, ConfigDict, HttpUrl, computed_field, RootModel
from .ffmpeg_worker_model import FFMpegSliceSegment
from .media_model import MediaSource, MediaSources, MediaProtocol
@@ -561,6 +560,64 @@ class GoogleUploadResponse(BaseModel):
@computed_field(description="适用于Google内部服务的URN")
@property
def urn(self) -> str:
- return self.id.replace(self.bucket, "gs:/")
+ return ("gs://" + self.id).replace(f"/{self.generation}","")
model_config = ConfigDict(populate_by_name=True)
+
+class VisualFeatures(BaseModel):
+ """
+ 代表产品的视觉特征
+ """
+ color: str = Field(
+ description="商品详细颜色配色等色彩特征"
+ )
+ pattern: str = Field(
+ description="商品详细图案纹理布料等材质特征, 材料可以根据产品名称和图像确定"
+ )
+ style: str = Field(
+ default=None,
+ description="商品详细款式风格设计等有辨识度的款式特征"
+ )
+
+class ProductInfo(BaseModel):
+ """
+ 表示从图像中提取单个产品的信息
+ """
+ image_order: int = Field(
+ description="图像出现的顺序"
+ )
+ image_name: str = Field(
+ description="图像上显示的原始文本"
+ )
+ matched_product: str = Field(
+ description="匹配的标准产品名称（必须与产品列表中的标准产品名称保持一致），或者如果没有匹配，则空"
+ )
+ match_confidence: int = Field(
+ ge=0,
+ le=100,
+ description="产品匹配的置信度范围从0到100"
+ )
+ visual_features: VisualFeatures = Field(
+ description="产品的详细视觉特征"
+ )
+
+class GeminiFirstStageResponseModel(RootModel):
+ """
+ 从图像中提取的产品信息列表
+ """
+ root: List[ProductInfo]
+
+class ProductTimeline(BaseModel):
+ """
+ 表示单个商品及其在视频时间线中的出现信息。
+ """
+ product: str = Field(description="标准商品名称")
+ timeline: List[str] = Field(
+ description="该商品在视频中出现的时间段列表。每个字符串表示一个时间段，格式为 '开始时间 - 结束时间 (描述)'"
+ )
+
+class GeminiSecondStageResponseModel(RootModel):
+ """
+ 表示一个包含多个商品及其时间线信息的列表。
+ """
+ root: List[ProductTimeline]
\ No newline at end of file
diff --git a/src/BowongModalFunctions/utils/HTTPUtils.py b/src/BowongModalFunctions/utils/HTTPUtils.py
index ee779a5..c01d2a6 100644
--- a/src/BowongModalFunctions/utils/HTTPUtils.py
+++ b/src/BowongModalFunctions/utils/HTTPUtils.py
@@ -1,11 +1,13 @@
import random
-from typing import Union, Dict, BinaryIO, List, Optional
+from typing import Union, Dict, BinaryIO, List, Optional, Any
from urllib.parse import quote
import backoff
import httpx
import asyncio
+
+from google.genai.types import GenerateContentResponse
from loguru import logger
import aiofiles
from pathlib import Path
@@ -126,6 +128,8 @@ class GoogleAuthUtils:
if not v:
return None
elif isinstance(v, types.GenerateContentConfig):
+ if not v.response_schema:
+ return v
if issubclass(v.response_schema, BaseModel):
flat_schema = v.response_schema.model_json_schema(schema_generator=FlatJsonSchemaGenerator)
v.response_schema = flat_schema
@@ -153,7 +157,7 @@ class GoogleAuthUtils:
def generate_content(self, model_id: str, contents: Union[types.ContentListUnion, types.ContentListUnionDict],
config: Optional[types.GenerateContentConfig] = None,
- timeout: int = 30) -> types.GenerateContentResponse:
+ timeout: int = 30) -> tuple[dict[Any, Any], int] | tuple[GenerateContentResponse, int]:
parameter_model = GoogleAuthUtils.VertexAIRequestModel(contents=contents, generationConfig=config)
json_body = parameter_model.model_dump_json(indent=2, exclude_none=True, by_alias=True,
exclude={"generation_config": {"safety_settings"}})
@@ -165,9 +169,11 @@ class GoogleAuthUtils:
headers={"Content-Type": "application/json",
"Authorization": f"Bearer {self.access_token}"},
content=json_body)
- logger.info(response.text)
- response.raise_for_status()
- return types.GenerateContentResponse.model_validate_json(response.text)
+ # response.raise_for_status()
+ # logger.info(response.text)
+ if response.status_code != 200:
+ return {}, response.status_code
+ return types.GenerateContentResponse.model_validate_json(response.text), response.status_code
@staticmethod
async def get_google_auth_jwt(service_account_info: Dict[str, str], scopes: list[str]) -> GoogleAuthResponse:
@@ -186,7 +192,7 @@ class GoogleAuthUtils:
return GoogleAuthUtils.GoogleAuthResponse.model_validate_json(response.text)
@staticmethod
- async def google_upload_file(file_stream: BinaryIO, content_type: str, google_api_key: str, bucket_name: str,
+ async def google_upload_file(file_stream: Union[BinaryIO ,bytes], content_type: str, google_api_key: str, bucket_name: str,
filename: str) -> GoogleUploadResponse:
safe_filename = quote(filename)
@@ -196,7 +202,7 @@ class GoogleAuthUtils:
headers={
"Authorization": f"Bearer {google_api_key}",
"Content-Type": content_type
- })
+ }, timeout=900)
upload_response.raise_for_status()
# upload_url = f"gs://dy-media-storage/video/{filename}"
diff --git a/src/Douyin_TikTok_Download_API/app/api/router.py b/src/Douyin_TikTok_Download_API/app/api/router.py
index e9d593e..e6c5d50 100644
--- a/src/Douyin_TikTok_Download_API/app/api/router.py
+++ b/src/Douyin_TikTok_Download_API/app/api/router.py
@@ -1,10 +1,10 @@
from fastapi import APIRouter
from Douyin_TikTok_Download_API.app.api.endpoints import (
- tiktok_web,
- tiktok_app,
+ # tiktok_web,
+ # tiktok_app,
douyin_web,
- bilibili_web,
- hybrid_parsing, ios_shortcut, download,
+ # bilibili_web,
+ # hybrid_parsing, ios_shortcut, download,
)
router = APIRouter()
diff --git a/src/Douyin_TikTok_Download_API/crawlers/base_crawler.py b/src/Douyin_TikTok_Download_API/crawlers/base_crawler.py
index 06cbeb1..5402ec3 100644
--- a/src/Douyin_TikTok_Download_API/crawlers/base_crawler.py
+++ b/src/Douyin_TikTok_Download_API/crawlers/base_crawler.py
@@ -95,7 +95,7 @@ class BaseCrawler:
# 异步客户端 / Asynchronous client
self.aclient = httpx.AsyncClient(
headers=self.crawler_headers,
- proxies=self.proxies,
+ mounts=self.proxies,
timeout=self.timeout,
limits=self.limits,
transport=self.atransport,
diff --git a/src/Douyin_TikTok_Download_API/crawlers/douyin/web/utils.py b/src/Douyin_TikTok_Download_API/crawlers/douyin/web/utils.py
index 9fa684b..ae7d1ea 100644
--- a/src/Douyin_TikTok_Download_API/crawlers/douyin/web/utils.py
+++ b/src/Douyin_TikTok_Download_API/crawlers/douyin/web/utils.py
@@ -107,7 +107,7 @@ class TokenManager:
}
transport = httpx.HTTPTransport(retries=5)
- with httpx.Client(transport=transport, proxies=cls.proxies) as client:
+ with httpx.Client(transport=transport, mounts=cls.proxies) as client:
try:
response = client.post(
cls.token_conf["url"], content=payload, headers=headers
diff --git a/src/cluster/video_apps/hls_slice_inference.py b/src/cluster/video_apps/hls_slice_inference.py
index a5e2d1c..1606f1e 100644
--- a/src/cluster/video_apps/hls_slice_inference.py
+++ b/src/cluster/video_apps/hls_slice_inference.py
@@ -2,24 +2,22 @@ import re
import uuid
import modal
+from google.genai import types
+from BowongModalFunctions.utils.HTTPUtils import GoogleAuthUtils, FlatJsonSchemaGenerator
from ..video import downloader_image, app, config
with downloader_image.imports():
- import os, httpx, json, time, requests
+ import os, json
from typing import List
from loguru import logger
from modal import current_function_call_id
- from ffmpy import FFmpeg, FFprobe
from fastapi import HTTPException
- from httpx import Timeout
from starlette import status
- import asyncio
- import random
- import subprocess
from BowongModalFunctions.utils.SentryUtils import SentryUtils
from BowongModalFunctions.models.media_model import MediaSource
- from BowongModalFunctions.models.web_model import SentryTransactionInfo, WebhookNotify
+ from BowongModalFunctions.models.web_model import SentryTransactionInfo, WebhookNotify, \
+ GeminiFirstStageResponseModel, GeminiSecondStageResponseModel
from BowongModalFunctions.models.ffmpeg_worker_model import FFMpegSliceSegment
from BowongModalFunctions.utils.TimeUtils import TimeDelta, merge_product_data
from BowongModalFunctions.utils.VideoUtils import FFMPEGSliceOptions, VideoUtils
@@ -75,6 +73,13 @@ with downloader_image.imports():
f"\noptions || {options.model_dump_json()}, "
f"\nscale || {scale}")
+ client = GoogleAuthUtils.GoogleGenaiClient(
+ cloudflare_project_id="67720b647ff2b55cf37ba3ef9e677083",
+ cloudflare_gateway_id="bowong-dev",
+ google_project_id="gen-lang-client-0413414134",
+ regions=gemini_region, access_token=google_api_key,
+ )
+
# 动态Prompt
if "IMAGE_PRODUCT_IDENTIFICATION_PROMPT" in os.environ:
IMAGE_PRODUCT_IDENTIFICATION_PROMPT = os.environ["IMAGE_PRODUCT_IDENTIFICATION_PROMPT"]
@@ -86,12 +91,15 @@ with downloader_image.imports():
**输入材料**：
- 🖼️ **商品图片网格**：包含多个黑色边框区域，每个区域内有商品图片+商品名称文字
- 📋 **商品列表**：标准商品名称参考清单
+
+ **Token自动管理**
+ 适时记录思考token使用量，根据任务复杂程度自动调整思考token的使用量(不允许超过上限65535)，任务复杂时使用摘要思考模式，仅记录决策结果不记录决策详细过程。
**核心任务**：
1. **扫描黑色边框区域**：从左上角开始，按行扫描每个黑色边框区域
2. **提取文字信息**：精确提取每个区域内的所有文字信息
3. **与商品列表匹配**：将图片文字与商品列表进行高相似度匹配
- 4. **提取商品图片特征**：从商品图片提取详细可识别特征，包括颜色、图案、纹理、材质、版型、款式等
+ 4. **提取商品图片特征**：从商品图片提取详细有辨识度可以用于商品辨认任务的特征，包括颜色、图案、纹理、材质、版型、款式等详细特征
**严格约束**：
- 🚫 只识别有黑色边框包围的商品区域
@@ -99,22 +107,24 @@ with downloader_image.imports():
- 🚫 不得推测或添加图片中不存在的商品
- ✅ 输出商品数量不得超过图片中的黑色边框区域数量
- **商品列表**：
+ **商品列表-标准商品名称**：
+
{PRODUCT_LIST}
+
-
+
请按以下JSON格式输出：
[
{{
"image_order": 1,
"image_name": "图片上显示的原始文字",
- "matched_product": "匹配到的标准商品名称或null",
+ "matched_product": "匹配到的标准商品名称(必须与商品列表-标准商品名称一致)或null",
"match_confidence": 95,
"visual_features": {{
- "color": "商品详细颜色配色等有辨识度的色彩特征",
- "pattern": "商品详细图案纹理等有辨识度的材质特征",
- "style": "商品详细款式版型等有辨识度的风格特征"
+ "color": "商品详细配色等有辨识度的色彩特征",
+ "pattern": "商品详细图案纹理布料等有辨识度的材质特征，材质可根据商品名以及图片综合进行判断",
+ "style": "商品详细款式风格等有辨识度的款式特征"
}}
}}
]
@@ -130,36 +140,55 @@ with downloader_image.imports():
**输入材料**：
- 📹 **视频**：直播带货片段
- - 📋 **已识别商品清单**：第一阶段确认的商品列表
+ - 📋 **已识别商品清单**：第一阶段确认的商品列表及商品特征
+
+ **Token自动管理**
+ 适时记录思考token使用量，根据任务复杂程度自动调整思考token的使用量(不允许超过上限65535)，任务复杂时使用摘要思考模式，仅记录决策结果不记录决策详细过程。
+
+ **商品匹配逻辑示例**
+ - 🔄 示例：
+ * 📋 材质一致性: 商品名称中"100纯棉" ↔ 视频质感"棉质" ✅一致
+ * 🖼️ 风格一致性: 商品特征中"紧身包臀" ↔ 视频"紧身收腰，凸显曲线，包臀" ✅一致
+ * 📹 视频一致性: 视频"修身版型，天使图案" ↔ 商品特征"修身显瘦，天使印花" ✅一致
+ * 🎤 语音一致性: 语音"纯棉材质，天使图案" ↔ 商品特征"纯棉" ✅一致
+ * 综合判断: 四维一致性95% → 保留
+ - 重点关注：
+ * **如果图案的形状、设计、构图完全相同，但颜色搭配不同，这属于相同商品的不同配色**
+ * 例如：同样的鱼图案，一个是"紫色鱼+黄色底"，另一个是"红色鱼+蓝色底"，这是**相同商品的不同配色**
+ * 例如：同样的条纹图案，一个是"黑白条纹"，另一个是"蓝白条纹"，这是**相同商品的不同配色**
+ * 例如：同样为衬衫，一个是"棉质黑色"，"一个为丝质米色"，这**不是同款，是不同的商品**
**分析任务**：
- 1. **时间段识别**：找出每个商品在视频中的展示时间
- 2. **内容类型分类**：判断每个时间段的具体内容类型
- 3. **时间段合并**：合并连续或相近的时间段
+ 1. **时间段识别**：根据匹配逻辑对视频画面逐一匹配找出每个商品在视频中的展示时间，不要遗漏相同商品的不同配色
+ 2. **去重原则**：视频画面、语音等都要综合考虑得出置信度值， 如果同一时间段内展示的同一商品可能对应商品列表中的多个商品，只选取置信度值最高的一个
+ 2. **内容类型分类**：判断每个时间段的具体内容类型(必须为以下三种内容类型之一：穿着本品+介绍本品、穿着本品+他品、穿着本品+无关)
+ 3. **时间段合并**：合并连续或相近的时间段，去除合并后仍小于5秒的时间段
**内容类型**：
- - (穿着本品+介绍本品) - 主播穿着该商品且正在介绍该商品本身
- - (穿着本品+他品) - 主播穿着该商品但正在介绍其他商品
- - (穿着本品+无关) - 主播穿着该商品但在做无关商品的事情
+ - (穿着本品+介绍本品) - 主播穿着该商品或页面主体为该商品且正在介绍该商品本身，主播话语里正在出现该商品相关描述语句或手指向或眼睛看向该商品
+ - (穿着本品+他品) - 主播穿着该商品或页面主体为该商品但正在介绍其他商品，主播话语里正在出现其他商品相关描述语句或手指向或眼睛看向其他商品
+ - (穿着本品+无关) - 主播穿着该商品或页面主体为该商品但在做无关商品的事情，主播话语里正在出现与商品无关的语句或者做换衣服等无关商品介绍的动作
**时间格式标准**：
- - 必须严格使用格式：HH:MM:SS.mmm - HH:MM:SS.mmm
- - HH是小时(00-23)，MM是分钟(00-59)，SS是秒(00-59)，mmm是毫秒(000-999)
- - 示例：00:05:23.500 表示0小时5分钟23.5秒，不是5小时23分钟
- - 常见错误：不要将05:23:500解释为5小时23分钟，应该是0小时5分钟23.5秒
+ - 必须严格使用格式：MM:SS.mmm - MM:SS.mmm
+ - MM是分钟(00-59)，SS是秒(00-59)，mmm是毫秒(000-999)
+ - 示例：05:23.500 表示5分钟23.5秒500毫秒
+ - 常见错误：将05:23.500解释为5分钟23.5秒
- **已识别商品清单(product-商品名 feature-商品特征 feature.color-商品详细颜色配色等色彩特征 feature.pattern-商品详细图案纹理等材质特征 feature.style-商品详细款式版型等风格特征)**：
+ **已识别商品清单(product-商品名 feature-商品特征 feature.color-商品详细颜色配色等色彩特征 feature.pattern-商品详细图案纹理布料等材质特征 feature.style-商品详细款式风格等有辨识度的款式特征)**：
+
{IDENTIFIED_PRODUCTS}
+
-
+
- 请按以下JSON格式输出：
+ 请按以下JSON格式输出，确保输出格式复核JSON格式要求：
[
{{
- "product": "商品名称",
+ "product": "标准商品名称",
"timeline": [
- "00:01:15.200 - 00:02:30.800 (穿着本品+介绍本品)",
- "00:05:10.100 - 00:06:25.600 (穿着本品+他品)"
+ "01:15.200 - 02:30.800 (穿着本品+介绍本品)",
+ "05:10.100 - 06:25.600 (穿着本品+他品)"
]
}}
]
@@ -167,7 +196,7 @@ with downloader_image.imports():
"""
- def upload(file_path):
+ async def upload(file_path):
with open(file_path, "rb") as video:
video = video.read()
content_length = len(video)
@@ -178,19 +207,82 @@ with downloader_image.imports():
raise HTTPException(status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE)
logger.info(
f"Uploading name = {filename}, size = {content_length}, type = {content_type} to google file")
- with httpx.Client(timeout=1800) as client:
- upload_response = client.post(
- url=f"https://storage.googleapis.com/upload/storage/v1/b/dy-media-storage/o?uploadType=media&name=videos%2F{filename}",
- content=video,
- headers={
- "Authorization": f"Bearer {google_api_key}",
- "Content-Type": content_type
- })
- upload_response.raise_for_status()
+ upload_response = await GoogleAuthUtils.google_upload_file(file_stream=video,
+ content_type=content_type,
+ google_api_key=google_api_key,
+ bucket_name="dy-media-storage",
+ filename=f"videos/{filename}",)
+
+ return upload_response.urn
+
+ def convert_json(json_like_str:str, json_model):
+ try:
+ resp, resp_code = client.generate_content(model_id="gemini-2.5-flash",
+ contents=[types.Content(role='user',
+ parts=[
+ types.Part.from_text(
+ text=""
+ ""
+ "请格式化以下一段非标准json格式字符串为json标准格式 \n{0}"
+ ""
+ "".format(
+ json_like_str
+ )
+ )
+ ]
+ )
+ ],
+ config=types.GenerateContentConfig(
+ temperature=0.01,
+ top_p=0.7,
+ safety_settings=[
+ types.SafetySetting(
+ category=types.HarmCategory.HARM_CATEGORY_HARASSMENT,
+ threshold=types.HarmBlockThreshold.BLOCK_NONE,
+ ),
+ types.SafetySetting(
+ category=types.HarmCategory.HARM_CATEGORY_CIVIC_INTEGRITY,
+ threshold=types.HarmBlockThreshold.BLOCK_NONE,
+ ),
+ types.SafetySetting(
+ category=types.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT,
+ threshold=types.HarmBlockThreshold.BLOCK_NONE,
+ ),
+ types.SafetySetting(
+ category=types.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT,
+ threshold=types.HarmBlockThreshold.BLOCK_NONE,
+ ),
+ types.SafetySetting(
+ category=types.HarmCategory.HARM_CATEGORY_HATE_SPEECH,
+ threshold=types.HarmBlockThreshold.BLOCK_NONE,
+ )
+ ],
+ response_mime_type="application/json",
+ response_schema=json_model.model_json_schema(
+ schema_generator=FlatJsonSchemaGenerator)
+ ), timeout=900)
+ except Exception as e:
+ logger.exception(f"😭格式化json请求失败 {e}")
+ return None
+
+ if resp_code == 200:
+ reason = resp.candidates[0].finish_reason
+ logger.info("😊格式化json用量信息" + resp.usage_metadata.model_dump_json(indent=2))
+ if reason == "STOP":
+ result_text: str = resp.candidates[0].content.parts[0].text.replace("\n","").replace("\\n", "").replace("\\", "")
+ # 解析识别结果
+ result_json = json.loads(result_text)
+ return result_json
+ else:
+ logger.error(f"😭格式化json推理失败, Reason {reason}")
+ return None
+ else:
+ logger.error(f"😭格式化json推理失败, 状态码{resp_code}")
+ if resp_code == 429:
+ logger.warning("🥵请求负载过高, 随机更换地区重试")
+ return None
- upload_url = f"gs://dy-media-storage/videos/{filename}"
- return upload_url, upload_response.status_code
def parse_stage1_result(result_text):
"""解析第一阶段结果，提取已识别的商品列表"""
@@ -203,8 +295,12 @@ with downloader_image.imports():
clean_result = clean_result[4:].strip()
# 解析JSON
- stage1_data = json.loads(clean_result)
-
+ try:
+ stage1_data = json.loads(clean_result)
+ except json.decoder.JSONDecodeError:
+ stage1_data = convert_json(clean_result, GeminiFirstStageResponseModel)
+ if not stage1_data:
+ raise Exception("json为空")
identified_products = []
if isinstance(stage1_data, list):
for item in stage1_data:
@@ -235,8 +331,6 @@ with downloader_image.imports():
retry_time: int = 3,
scale:float = 0.9):
- video_gemini_uri = None
-
try:
# 1、首先获取全量商品列表
logger.info("1、获取直播间商品列表")
@@ -273,71 +367,68 @@ with downloader_image.imports():
# 3、上传到gemini
logger.info("3、视频文件开始上传到Gemini")
- video_gemini, code = upload(video_path)
- if code == 200:
- video_gemini_uri = video_gemini
- else:
- logger.error("视频文件上传Gemini失败")
- raise Exception("视频文件上传Gemini失败")
+ video_gemini = await upload(video_path)
async def inference_api_first_stage():
try:
logger.info("🎈开始一阶段推理")
product_list_str = ""
for i, product in enumerate(product_title_list):
- product_list_str += f"{i}. {product}\n"
+ product_list_str += f"- {i}. {product}
\n"
image_parts = []
for i in product_grid_list:
image_parts.append(
- {
- "file_data": {
- "mime_type": "image/jpeg",
- "file_uri": f"{i}"
- }
- }
+ types.Part(
+ file_data=types.FileData(
+ mime_type="image/jpeg",
+ file_uri=f"{i}"
+ )
+ )
)
image_parts.append(
- {
- "text": IMAGE_PRODUCT_IDENTIFICATION_PROMPT.format(PRODUCT_LIST=product_list_str)
- }
- )
- json_data = {
- "safetySettings": [
- {"category": "HARM_CATEGORY_HARASSMENT", "threshold": "BLOCK_NONE"},
- {"category": "HARM_CATEGORY_HATE_SPEECH", "threshold": "BLOCK_NONE"},
- {"category": "HARM_CATEGORY_SEXUALLY_EXPLICIT", "threshold": "BLOCK_NONE"},
- {"category": "HARM_CATEGORY_DANGEROUS_CONTENT", "threshold": "BLOCK_NONE"},
- {"category": "HARM_CATEGORY_CIVIC_INTEGRITY", "threshold": "BLOCK_NONE"}
- ],
- "contents": [
- {
- "role": "user",
- "parts": image_parts
- }
- ],
- "generationConfig": {
- "temperature": 0.01,
- "top_p": 0.7
- }
- }
- resp = requests.post(
- f"https://gateway.ai.cloudflare.com/v1/67720b647ff2b55cf37ba3ef9e677083/bowong-dev/google-vertex-ai/v1/projects/gen-lang-client-0413414134/locations/{random.choice(gemini_region)}/publishers/google/models/gemini-2.5-flash:generateContent",
- headers={
- "Content-Type": "application/json",
- "Authorization": f"Bearer {google_api_key}"
- },
- json=json_data, timeout=900
+ types.Part.from_text(
+ text=IMAGE_PRODUCT_IDENTIFICATION_PROMPT.format(PRODUCT_LIST=product_list_str)
+ )
)
+ resp, resp_code = client.generate_content(model_id="gemini-2.5-flash",
+ contents=[types.Content(role='user',
+ parts=image_parts
+ )
+ ],
+ config=types.GenerateContentConfig(
+ temperature=0.01,
+ top_p=0.7,
+ safety_settings=[
+ types.SafetySetting(
+ category=types.HarmCategory.HARM_CATEGORY_HARASSMENT,
+ threshold=types.HarmBlockThreshold.BLOCK_NONE,
+ ),
+ types.SafetySetting(
+ category=types.HarmCategory.HARM_CATEGORY_CIVIC_INTEGRITY,
+ threshold=types.HarmBlockThreshold.BLOCK_NONE,
+ ),
+ types.SafetySetting(
+ category=types.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT,
+ threshold=types.HarmBlockThreshold.BLOCK_NONE,
+ ),
+ types.SafetySetting(
+ category=types.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT,
+ threshold=types.HarmBlockThreshold.BLOCK_NONE,
+ ),
+ types.SafetySetting(
+ category=types.HarmCategory.HARM_CATEGORY_HATE_SPEECH,
+ threshold=types.HarmBlockThreshold.BLOCK_NONE,
+ )
+ ],
+ ), timeout=900)
except Exception as e:
logger.exception(f"😭Gemini一阶段推理请求失败: {e}")
raise e
- if resp.status_code == 200:
- target_json = resp.json()
- reason = target_json["candidates"][0]["finishReason"]
- if "usageMetadata" in target_json:
- logger.info("😊一阶段用量信息"+json.dumps(target_json["usageMetadata"], indent=4))
+ if resp_code == 200:
+ reason = resp.candidates[0].finish_reason
+ logger.info("😊一阶段用量信息"+resp.usage_metadata.model_dump_json(indent=2))
if reason == "STOP":
- result_text: str = target_json["candidates"][0]["content"]["parts"][0]["text"]
+ result_text: str = resp.candidates[0].content.parts[0].text
# 解析识别结果
identified_products = parse_stage1_result(result_text)
return identified_products
@@ -345,8 +436,8 @@ with downloader_image.imports():
logger.error(f"😭Gemini一阶段推理失败, Reason {reason}")
return None
else:
- logger.error(f"😭Gemini一阶段推理失败, 状态码{resp.status_code}")
- if resp.status_code == 429:
+ logger.error(f"😭Gemini一阶段推理失败, 状态码{resp_code}")
+ if resp_code == 429:
logger.warning("🥵请求负载过高, 随机更换地区重试")
return None
@@ -356,61 +447,74 @@ with downloader_image.imports():
# 构建已识别商品清单字符串
identified_products_str = ""
for i, product in enumerate(identified_products):
- identified_products_str += f"{i}. {json.dumps(product,ensure_ascii=False)}\n"
- image_parts = []
- image_parts.append(
- {
- "file_data": {
- "mime_type": "video/mp4",
- "file_uri": f"{video_gemini_uri}"
- }
- })
- image_parts.append(
- {
- "text": VIDEO_TIMELINE_ANALYSIS_PROMPT.format(IDENTIFIED_PRODUCTS=identified_products_str)
- }
+ product_info_json = json.dumps(product,ensure_ascii=False)
+ identified_products_str += f"- {i}. {product_info_json}
- \n"
+ video_parts = []
+ video_parts.append(
+ types.Part(
+ file_data=types.FileData(
+ mime_type="video/mp4",
+ file_uri=f"{video_gemini}"
+ )
+ )
)
- json_data = {
- "safetySettings": [
- {"category": "HARM_CATEGORY_HARASSMENT", "threshold": "BLOCK_NONE"},
- {"category": "HARM_CATEGORY_HATE_SPEECH", "threshold": "BLOCK_NONE"},
- {"category": "HARM_CATEGORY_SEXUALLY_EXPLICIT", "threshold": "BLOCK_NONE"},
- {"category": "HARM_CATEGORY_DANGEROUS_CONTENT", "threshold": "BLOCK_NONE"},
- {"category": "HARM_CATEGORY_CIVIC_INTEGRITY", "threshold": "BLOCK_NONE"}
- ],
- "contents": [
- {
- "role": "user",
- "parts": image_parts
- }
- ],
- "generationConfig": {
- "temperature": 0.01,
- "top_p": 0.7
- }
- }
- resp = requests.post(
- f"https://gateway.ai.cloudflare.com/v1/67720b647ff2b55cf37ba3ef9e677083/bowong-dev/google-vertex-ai/v1/projects/gen-lang-client-0413414134/locations/{random.choice(gemini_region)}/publishers/google/models/gemini-2.5-flash:generateContent",
- headers={
- "Content-Type": "application/json",
- "Authorization": f"Bearer {google_api_key}"
- },
- json=json_data, timeout=900
+ video_parts.append(
+ types.Part.from_text(
+ text=VIDEO_TIMELINE_ANALYSIS_PROMPT.format(IDENTIFIED_PRODUCTS=identified_products_str)
+ )
)
+ resp, resp_code = client.generate_content(model_id="gemini-2.5-flash",
+ contents=[types.Content(role='user',
+ parts=video_parts
+ )
+ ],
+ config=types.GenerateContentConfig(
+ temperature=0.01,
+ top_p=0.7,
+ safety_settings=[
+ types.SafetySetting(
+ category=types.HarmCategory.HARM_CATEGORY_HARASSMENT,
+ threshold=types.HarmBlockThreshold.BLOCK_NONE,
+ ),
+ types.SafetySetting(
+ category=types.HarmCategory.HARM_CATEGORY_CIVIC_INTEGRITY,
+ threshold=types.HarmBlockThreshold.BLOCK_NONE,
+ ),
+ types.SafetySetting(
+ category=types.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT,
+ threshold=types.HarmBlockThreshold.BLOCK_NONE,
+ ),
+ types.SafetySetting(
+ category=types.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT,
+ threshold=types.HarmBlockThreshold.BLOCK_NONE,
+ ),
+ types.SafetySetting(
+ category=types.HarmCategory.HARM_CATEGORY_HATE_SPEECH,
+ threshold=types.HarmBlockThreshold.BLOCK_NONE,
+ )
+ ],
+ ), timeout=900)
except Exception as e:
logger.exception(f"😭Gemini二阶段推理请求失败: {e}")
raise e
- if resp.status_code == 200:
- target_json = resp.json()
- reason = target_json["candidates"][0]["finishReason"]
- if "usageMetadata" in target_json:
- logger.info("😊二阶段用量信息"+json.dumps(target_json["usageMetadata"], indent=4))
+ if resp_code == 200:
+ reason = resp.candidates[0].finish_reason
+ logger.info("😊二阶段用量信息"+resp.usage_metadata.model_dump_json(indent=2))
if reason == "STOP":
- result_text: str = target_json["candidates"][0]["content"]["parts"][0]["text"]
+ result_text: str = resp.candidates[0].content.parts[0].text
if len(result_text)>10:
- parts = result_text.split("```")[-2].replace("```", "").replace("json\n", "").replace("\n",
- "").replace("\\", "")
- parts = json.loads(parts)
+ if result_text.startswith("```") and result_text.endswith("```"):
+ parts_text = result_text.split("```")[-2].replace("json","").replace("\n","").replace("\\", "").replace("\\n", "")
+ else:
+ parts_text = result_text.replace("json", "").replace("\n",
+ "").replace(
+ "\\", "").replace("\\n", "")
+ try:
+ parts = json.loads(parts_text)
+ except json.decoder.JSONDecodeError:
+ parts = convert_json(parts_text, GeminiSecondStageResponseModel)
+ if not parts:
+ raise Exception("json为空")
logger.info(f"👌合并前 {json.dumps(parts, indent=4, ensure_ascii=False)}")
# 合并产品和时间线
parts = merge_product_data(parts, start_time, end_time, merge_diff=5)
@@ -423,8 +527,8 @@ with downloader_image.imports():
logger.error(f"😭Gemini二阶段推理失败, Reason {reason}")
return None
else:
- logger.error(f"😭Gemini二阶段推理失败, 状态码{resp.status_code}")
- if resp.status_code == 429:
+ logger.error(f"😭Gemini二阶段推理失败, 状态码{resp_code}")
+ if resp_code == 429:
logger.warning("🥵请求负载过高, 随机更换地区重试")
return None
@@ -437,9 +541,9 @@ with downloader_image.imports():
product_info = await inference_api_first_stage()
first_retry_time -= 1
if product_info is None:
- raise Exception("一阶段推理失败")
+ raise Exception("😭一阶段推理失败")
else:
- logger.info(f"一阶段推理完成JSON \n{json.dumps(product_info, indent=4, ensure_ascii=False)}")
+ logger.info(f"🥳一阶段推理完成JSON \n{json.dumps(product_info, indent=4, ensure_ascii=False)}")
# 二阶段推理
product_timeline_info = None
@@ -448,15 +552,15 @@ with downloader_image.imports():
product_timeline_info = await inference_api_second_stage(product_info)
second_retry_time -= 1
if product_timeline_info is None:
- raise Exception("二阶段推理失败")
+ raise Exception("😭二阶段推理失败")
else:
- logger.info(f"二阶段推理完成JSON \n{json.dumps(product_timeline_info, indent=4, ensure_ascii=False)}")
+ logger.info(f"🥳二阶段推理完成JSON \n{json.dumps(product_timeline_info, indent=4, ensure_ascii=False)}")
return product_timeline_info, sentry_trace
except Exception as e:
- logger.exception(f"推理失败, {e}")
- raise Exception(f"推理失败, {e}")
+ logger.exception(f"🥲推理失败, {e}")
+ raise Exception(f"🥲推理失败, {e}")
return await _handler(media, google_api_key, product_grid_list, product_list, start_time, end_time, options,
sentry_trace, retry_time, scale)
diff --git a/src/cluster/video_apps/make_grid_upload.py b/src/cluster/video_apps/make_grid_upload.py
index 22a44e1..6742f35 100644
--- a/src/cluster/video_apps/make_grid_upload.py
+++ b/src/cluster/video_apps/make_grid_upload.py
@@ -4,6 +4,7 @@ import urllib
import aiofiles
import modal
+from BowongModalFunctions.utils.HTTPUtils import GoogleAuthUtils
from ..video import downloader_image, app, config
with downloader_image.imports():
@@ -270,30 +271,18 @@ with downloader_image.imports():
print(f"上传文件: {filename}, 大小: {content_length} bytes, 类型: {content_type}")
- # 构建上传URL
+ # 构建上传文件名
object_name = f"images/{filename}"
- object_name_quote = object_name.replace("/", "%2F")
- upload_url = f"https://storage.googleapis.com/upload/storage/v1/b/{bucket_name}/o?uploadType=media&name={object_name_quote}"
# 读取文件内容
- async with aiofiles.open(file_path, 'rb') as f:
- file_content = await f.read()
+ with open(file_path, 'rb') as f:
+ response = await GoogleAuthUtils.google_upload_file(file_stream=f,
+ content_type=content_type,
+ google_api_key=google_api_key,
+ bucket_name=bucket_name,
+ filename=object_name)
- # 发送异步请求
- async with httpx.AsyncClient(timeout=1800) as client:
- response = await client.post(
- url=upload_url,
- content=file_content,
- headers={
- "Authorization": f"Bearer {google_api_key}",
- "Content-Type": content_type,
- "Content-Length": str(content_length)
- }
- )
- response.raise_for_status()
- # 处理响应
- upload_url = f"gs://{bucket_name}/{object_name}"
- return upload_url, response.status_code
+ return response.urn
@SentryUtils.sentry_tracker(sentry_trace.x_trace_id, sentry_trace.x_baggage, op="make_grid_gemini",
name="将输入图拼为网格上传到Gemini网盘", fn_id=current_function_call_id())
@@ -310,12 +299,7 @@ with downloader_image.imports():
if not image_grid_path:
raise Exception("创建图片网格失败")
- image_grid_gemini, code = await upload(image_grid_path, google_api_key)
- if code == 200:
- image_gemini_uri = image_grid_gemini
- else:
- logger.error("图片网格文件上传Gemini失败")
- raise Exception("图片网格文件上传Gemini失败")
- return image_gemini_uri
+ image_grid_gemini = await upload(image_grid_path, google_api_key)
+ return image_grid_gemini
return await _handler(google_api_key, pic_info_list, image_size, text_height, font_size, padding, separator)
\ No newline at end of file