LongArticlesJob/coldStartTasks/ai_pipeline/basic.py

import os
import re
import json
import time
import datetime
import requests
from typing import Optional, Dict

from requests.exceptions import RequestException
from tenacity import retry

from applications.db import DatabaseConnector
from applications.utils import request_retry

retry_desc = request_retry(retry_times=3, min_retry_delay=2, max_retry_delay=30)


def get_status_field_by_task(task: str) -> tuple[str, str]:
    match task:
        case "upload":
            status = "upload_status"
            update_timestamp = "upload_status_ts"
        case "extract":
            status = "extract_status"
            update_timestamp = "extract_status_ts"
        case "get_cover":
            status = "get_cover_status"
            update_timestamp = "get_cover_status_ts"
        case _:
            raise ValueError(f"Unexpected task: {task}")
    return status, update_timestamp


def roll_back_lock_tasks(
    db_client: DatabaseConnector,
    task: str,
    max_process_time: int,
    init_status: int,
    processing_status: int,
) -> int:
    """
    rollback tasks which have been locked for a long time
    """
    status, update_timestamp = get_status_field_by_task(task)
    now_timestamp = int(time.time())
    timestamp_threshold = datetime.datetime.fromtimestamp(
        now_timestamp - max_process_time
    )
    update_query = f"""
        update long_articles_new_video_cover
        set {status} = %s
        where {status} = %s and {update_timestamp} < %s;
    """
    rollback_rows = db_client.save(
        query=update_query, params=(init_status, processing_status, timestamp_threshold)
    )
    return rollback_rows


def download_file(task_id, oss_path):
    """
    下载视频文件
    """
    video_url = "https://rescdn.yishihui.com/" + oss_path
    file_name = "static/{}.mp4".format(task_id)
    if os.path.exists(file_name):
        return file_name

    proxies = {"http": None, "https": None}
    with open(file_name, "wb") as f:
        response = requests.get(video_url, proxies=proxies)
        f.write(response.content)
    return file_name


def update_task_queue_status(
    db_client: DatabaseConnector,
    task_id: int,
    task: str,
    ori_status: int,
    new_status: int,
) -> int:
    # update task queue status
    status, update_timestamp = get_status_field_by_task(task)
    update_query = f"""
        update long_articles_new_video_cover
        set {status} = %s, {update_timestamp} = %s
        where {status} = %s and id = %s;
    """
    update_rows = db_client.save(
        query=update_query,
        params=(
            new_status,
            datetime.datetime.now(),
            ori_status,
            task_id,
        ),
    )
    return update_rows


def extract_best_frame_prompt():
    extract_prompt = """
    以下为视频爆款封面的评分体系，请根据视频内容，逐帧分析视频画面，根据以下标准进行打分，最终先把分数最高的画面出现的时间输出给我，并精确到几时几分几秒几毫秒，格式：hh:mm:ss.xxx，要求是确切时间而不是时间段。
    评分维度：
    一、现实关切度 (满分15分):
    高度相关（13-15分）： 封面主题直接涉及老年人的生活、健康、财产等切身利益，例如养老、退休、健康、食品安全、子女教育、财产安全等。
    中度相关（8-12分）： 封面主题与老年人的生活有一定的关联，例如家长里短、邻里关系、社会热点、生活窍门等。
    低度相关（4-7分）： 封面主题与老年人的生活关系较弱，例如娱乐八卦、时尚潮流等。
    无关（0-3分）： 封面主题与老年人的生活基本无关。
    二、社会认同感与亲切感 (满分15分):
    高度认同（13-15分）： 封面人物形象亲切、接地气，场景贴近老年人的生活，有强烈的代入感和归属感。
    中度认同（8-12分）： 封面人物形象尚可接受，场景有一定的熟悉感。
    低度认同（4-7分）： 封面人物形象陌生，场景较为遥远。
    无认同感（0-3分）： 封面人物形象令人反感，场景与老年人生活无关。
    三、信息传达效率 (满分15分):
    高效传达（13-15分）： 封面文字简洁直白、字体较大、重点突出，视觉元素聚焦，能让老年人快速理解视频内容，色彩明快，对比强烈，视觉冲击力强。
    中效传达（8-12分）： 封面文字尚可，视觉元素有一定吸引力，但略显复杂。
    低效传达（4-7分）： 封面文字晦涩难懂，视觉元素杂乱无章。色彩暗淡，对比度弱。
    无效传达（0-3分）： 封面信息表达不清，无法理解视频内容。
    四、情感共鸣与回忆杀 (满分20分):
    高度共鸣（17-20分）： 封面主题能够引发老年人对过去岁月的回忆，勾起他们对老朋友、老时光的思念，产生强烈的情感共鸣，引发对晚年生活的思考。例如同学情、怀旧主题等。
    中度共鸣（11-16分）： 封面主题有一定怀旧元素，能引发部分老年人的回忆，并进行一定程度的思考。
    低度共鸣（6-10分）： 封面主题怀旧元素较少，共鸣感不强。
    无共鸣（0-5分）： 封面主题与回忆无关。
    五、正能量与精神寄托 (满分15分):
    高度正能量（13-15分）： 封面内容积极向上，能够给予老年人希望和力量，或者包含宗教、祈福等元素，满足他们的精神寄托。
    中度正能量（8-12分）： 封面内容有一定的积极意义，但不够突出。
    低度正能量（4-7分）： 封面内容较为平淡，缺乏精神寄托。
    负能量 (0-3分): 封面内容消极悲观，或者与老年人的信仰不符。
    六、节日/时事关联性 (满分10分):
    高度关联（8-10分）： 封面与节日、时事热点紧密相关，能激发老年人的分享欲和参与感。
    中度关联（5-7分）： 封面与节日或时事有一定关联，但并非核心。
    低度关联（2-4分）： 封面与节日或时事关联较弱。
    无关联（0-1分）： 封面与节日、时事无关。
    七、传播动机 (满分10分):
    强传播动机（8-10分）： 封面内容能激发老年人强烈的情感，例如激动、感动、惊讶等，让他们想要分享给家人朋友，或者认为视频内容对他人有帮助，有分享的价值。
    中等传播动机（5-7分）： 封面内容有一定分享价值，但分享意愿不强烈。
    低传播动机（2-4分）： 封面内容平淡，缺乏分享动力。
    无传播动机（0-1分）： 封面内容无分享价值，或者会引起不适，降低传播意愿。
    八、附加分（满分60分）
    1.包含老年人为画面主体（0-5分）
    2.有超过3人为画面主体（0-5分）
    3.充斥画面的密集人群为画面主体（0-5分）
    4.存在知名历史、近代人物（0-5分）
    5.存在人物脸部、头部未完整出现在画面的情况（0-5分）
    6.是不以人为主体的鲜花、美景、知名建筑或风景（0-5分）
    7.是老照片、怀旧风格（0-5分）
    8.是农村、军事、综艺演出、历史画面（0-5分）
    9.有趣味、惊奇的形象或画面为主体（0-5分）
    10.以大号文字或密集文字为主体并且不包含人物（0-5分）
    11.是不包含人物的纯色画面（0-5分）
    12.是模糊的或清晰度、像素较低的（0-5分）
    
    总分评估：110-160分： 高度吸引老年人群体，有成为爆款的潜力。
    80-109分： 具有一定吸引力，值得尝试。
    65-79分： 吸引力一般，需要优化。
    65分以下： 吸引力不足，不建议使用。
    
    注意：输出只需要返回 'hh:mm:ss.xxx' 格式的时间, 无需返回任何其他东西
    """
    return extract_prompt


@retry(**retry_desc)
def get_video_cover(video_oss_path: str, time_millisecond_str: str) -> Optional[Dict]:
    """
    input video oss path and time millisecond
    output video cover image oss path
    """
    video_url = "https://rescdn.yishihui.com/" + video_oss_path
    url = "http://192.168.205.80:8080/ffmpeg/fetchKeyFrames"
    data = {"url": video_url, "timestamp": time_millisecond_str}
    headers = {
        "content-type": "application/json",
    }
    try:
        response = requests.post(url, headers=headers, json=data, timeout=60)
        response.raise_for_status()
        return response.json()
    except RequestException as e:
        print(f"API请求失败: {e}")
    except json.JSONDecodeError as e:
        print(f"响应解析失败: {e}")
    return None


def normalize_time_str(time_string: str) -> str | None:
    # 预处理：替换中文冒号、去除特殊标记、清理空格
    time_str = re.sub(r"：", ":", time_string)  # 中文冒号转英文
    time_str = re.sub(r"\s*(\d+\.\s*)?\*+\s*", "", time_str)  # 去除数字编号和**标记
    time_str = time_str.strip()

    # 组合式正则匹配（按优先级排序）
    patterns = [
        # hh:mm:ss.xxx
        (r"^(\d{2}):(\d{2}):(\d{2})\.(\d{3})$", None),
        # h:mm:ss.xxx
        (
            r"^(\d{1}):(\d{2}):(\d{2})\.(\d{3})$",
            lambda m: f"0{m[1]}:{m[2]}:{m[3]}.{m[4]}",
        ),
        # mm:ss.xxx
        (r"^(\d{2}):(\d{2})\.(\d{3})$", lambda m: f"00:{m[1]}:{m[2]}.{m[3]}"),
        # m:ss.xxx
        (r"^(\d{1}):(\d{2})\.(\d{3})$", lambda m: f"00:0{m[1]}:{m[2]}.{m[3]}"),
    ]

    for pattern, processor in patterns:
        if match := re.fullmatch(pattern, time_str):
            return processor(match) if processor else time_str

    # 特殊处理 dd:dd:ddd 格式（假设最后3位为毫秒）
    if m := re.fullmatch(r"(\d{2}:\d{2}):(\d{3})", time_str):
        return f"00:{m[1]}.{m[2]}"

    return None