video-insight/utils/google_ai_studio.py

import os
import time
import uuid
from typing import  Optional

import google.generativeai as genai
import orjson
import requests
from google.generativeai.types import (HarmBlockThreshold, HarmCategory)
from loguru import logger

from utils.feishu_data import Material

CACHE_DIR = '/app/cache/'
prompt = '''
请帮助我做以下分析：
我将提供给你视频，需要你模拟成中老年用户在视频消费平台观看这些视频。
需要你充分发挥想象力，推测出中老年用户在观看这些视频过程中，可能产生什么具体的需求。并推测出该点需求的原因。
-每项需求请根据观看者的需求强烈程度打分排序，越强烈分值越高，分值越高排序越前。分值范围：0-10分。
-每项需求请分类。需求类型：文字问答类、视频推荐类、
-针对这些需求，平台有两部分功能：钩子及落地方案，钩子指在视频需求产生时，平台会有相关提示，提示用户进入落地方案；落地方案指针对用户需求，平台提供的相关信息、服务能力，解决用户的需求
-针对这些需求，平台有几类解决方案，1. 文字问答；可提供问答信息获取服务；2. 内容推荐；推荐平台内相关视频内容；
-需求是用户在观看过程中有欲望产生除观看外的动作，需求不要泛泛而谈，而是针对这个视频的强烈需求欲望。第一要与视频紧密相关，第二要包含用户真正的好奇心、交流欲等。第三需求范围不要大而空，要具体详细真实，如“了解适合老年人的旅游景点和路线”是不好的需求提取与描述，要根据视频说明具体的景点及具体出行行为
-由于用户观看行为与注意力，需求最好与视频初始部分或整体内容相关；
-合理为需求打分，如判断需求不强烈，请打低分；一个视频可都为低分需求；打分考虑以下维度：问题与视频关联程度；问题精准度；问题可获得信息的丰富度；
-需求详细query是将用户需求及视频背景信息总结为一段给AI模型的需求问题或指令；query格式为：我看了一段内容为“（视频50字简介）”的视频，我想了解“（具体需求）”
-需求详细query描述时，不要用“它””他们””视频中的***”指代视频内容，描述清楚其名称及场景；视频信息完全由query内容传递。
-总结11字内钩子话术，引导用户产生兴趣点击落地方案；产品落地形态应与产品钩子合理承接。

输出内容单引号换成双引号，每一个字段都要有值

输出内容格式：
output in JSON format with keys:
需求排序序号(str)
需求强烈程度分值(str)
用户具体的需求描述(str)
需求详细query(str)
需求分类(str)
推测出该点需求的原因(str)
需求钩子话术(str)
落地方案类型(str)
落地方案形态描述(str)
'''

prompt1 = '''
请帮助我做以下分析：
我将提供给你视频，需要你模拟成中老年用户在视频消费平台观看这些视频。
需要你充分发挥想象力，推测出中老年用户在观看这些视频过程中，可能产生的具体的需求。
根据视频内容，分析用户是否会因为内容产生强烈的衍生需求，需求可能包含：了解获取更多相关信息、进行相关社交交流、获取相关商业服务等衍生需求
请根据视频内容，为用户观看视频时产生需求的强烈程度打分，1-10分，10分为非常强烈，1分为基本没有衍生需求。换言之，评估此视频是否适合为老年人提供衍生信息与服务
评判标准可从以下方面考虑：视频信息量，是否有较多信息量可引出相关信息需求；是否有特定事件、人物、现象等可能引起知识需求的内容；视频是否与商业化服务较直接相关等
综合用户需求为视频整体打一个需求强烈程度分值，打分区分度要大
我们会根据打分对视频需求排序，请客观评价；直接输出JSON结果，不要有其他内容;
输出内容单引号换成双引号，每一个字段都要有值;只输出一个结果
输出内容格式：
output in JSON format with keys:
需求强烈程度(str)
理由(str)

示例:
{
  "需求强烈程度":4,
  "理由":"XXXX"
}
'''
# CACHE_DIR = '/Users/z/Downloads/'
# PROXY_ADDR = 'http://localhost:1081'
# os.environ['http_proxy'] = PROXY_ADDR
# os.environ['https_proxy'] = PROXY_ADDR

class GoogleAI(object):

    @classmethod
    def download_video(cls, video_link: str) -> Optional[str]:
        file_path = os.path.join(CACHE_DIR, f'{str(uuid.uuid4())}.mp4')
        for _ in range(3):
            try:
                response = requests.get(url=video_link, timeout=60)
                if response.status_code == 200:
                    with open(file_path, 'wb') as f:
                        f.write(response.content)
                    logger.info(f'[内容分析] 视频链接: {video_link}, 存储地址: {file_path}')
                    return file_path
            except Exception:
                time.sleep(1)
                continue
        return

    @classmethod
    def run(cls, api_key, video_url):
        video_path = None
        try:
            genai.configure(api_key=api_key)
            video_path = cls.download_video(video_link=video_url)
            if not video_path:
                logger.error(f'[内容分析] 视频下载失败, 跳过任务')
                os.remove(video_path)
                logger.info(f"[内容分析] 文件已删除: {video_path}")
                return "[异常] 视频下载失败",""

            video = genai.upload_file(path=video_path, mime_type='video/mp4')
            while video.state.name == 'PROCESSING':
                time.sleep(1)
                video = genai.get_file(name=video.name)
            if video.state.name != 'ACTIVE':
                genai.delete_file(name=video.name)
                os.remove(video_path)
                return "[异常] 上传视频失败", ""
            model = genai.GenerativeModel(
                model_name='gemini-2.0-flash',
                generation_config=genai.GenerationConfig(response_mime_type='application/json'),
                safety_settings={
                    HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: HarmBlockThreshold.BLOCK_NONE,
                },
            )
            # mark, prompt = Material.feishu_list()
            response = model.generate_content(
                contents=[
                    video,
                    f'{prompt}',
                ],
                stream=False,
                request_options={
                    'timeout': 600,
                },
            )
            text = orjson.loads(response.text.strip())

            response2 = model.generate_content(
                contents=[
                    video,
                    f'{prompt1}',
                ],
                stream=False,
                request_options={
                    'timeout': 600,
                },
            )
            text2 = orjson.loads(response2.text.strip())

            genai.delete_file(name=video.name)
            os.remove(video_path)
            return text,text2
        except Exception as e:
            logger.error(f"[内容分析] 处理异常,异常信息{e}")
            os.remove(video_path)
            return f"[异常] {e}",""


if __name__ == '__main__':
    ai = GoogleAI()
    ai.run("AIzaSyA_dXKbcW8s0fFIoo89rtPRWU34pJry7mU",
                 "http://rescdn.yishihui.com/jq_oss/video/2025012215472528213")