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~100分。
以上每项需求请分类。需求类型：比如获取服务类、获取信息类、交流互动类、内容分享类等，类型字数请控制在10个字以内。
针对你推理出的这些用户需求，请给出与每个需求相匹配且合理的产品解决方案，如有其它视频平台类似形式的案例也请描述出来。
其中产品解决方案中包括：
1、用户具体的需求描述并对需求进行分类。比如，需求描述为：“确认自身正在服用的药物的具体忌口信息，确保安全用药，避免潜在风险。”该需求可分类为：信息获取类、服务获取类。需求分类：比如，信息获取、服务获取、商品获取、互动交流、活动体验等
2、推测出该点需求的原因。
3、描述出与需求对应的产品钩子。需要包含钩子形式描述以及钩子形式分类。还需明确产品形式在视频播放过程中出现到消失的时间。
注意：产品钩子要求能和需求匹配又能合理地吸引用户。
钩子形式描述：比如，视频结束时，出现“看更多养老院视频”按键入口；钩子形式分类：比如内容观看入口。
钩子形式类型举例：比如tips提示、语音播报、贴片广告、购物车链接、内容观看入口、社群二维码图片、智能搜索按键等。类型字数请控制在10个字以内。
4、点击钩子后的产品落地形态描述以及产品落地形态分类。
注意：产品落地形态应与产品钩子合理承接。
点击钩子后的产品落地形态描述：比如，点击“搜索更多养老院视频”按键后，自动搜索“养老院”相关内容且页面跳转落地至搜索结果页，展示养老院视频列表供用户挑选；
产品落地形态分类：搜索结果列表。
产品落地形态分类举例：比如商品广告h5、搜索结果列表、内容列表、私域群二维码、AI机器人对话等。类型字数请控制在10个字以内。
通过以上这种结构化的分类和整理，最终想通过AI对内容进行理解，分析推测出内容不同时机的价值点与用户需求的对应关系，为后续AI在产品中的应用场景提供更多可能。
输出内容单引号换成双引号，每一个字段都要有值，输出内容格式：
output in JSON format with keys:
需求排序序号(str)
需求强烈程度分值(str)
用户具体的需求描述(str)
需求分类(str)
推测出该点需求的原因(str)
描述出与需求对应的产品钩子(str)
产品形式出现到消失的时间点(str 用分:秒-分:秒 的格式)
钩子形式描述(str)
钩子形式类型(str)
点击钩子后的产品落地形态描述(str)
产品落地形态分类(str)
其他平台案例(str)
'''
# 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-1.5-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())
            genai.delete_file(name=video.name)
            os.remove(video_path)
            return text
        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")