title_with_video/applications/functions/kimi.py

"""
@author: luojunhui
"""
import os
import json
from openai import OpenAI

from applications.functions.chatgpt import OpenAIServer
from applications.functions.log import logging


class KimiServer(object):
    """
    Kimi Server
    """

    @classmethod
    async def search_kimi_schedule(cls, params):
        """
        搜索阶段 kimi 操作
        :param params:
        :return:
        """
        title = params['title'].split("@@")[-1]
        contents = params['content']
        trace_id = params['trace_id']
        try:
            kimi_title = await cls.kimi_title(title)
        except Exception as e:
            logging(
                code="9000",
                info="kimi 挖掘失败--{}, 采用 gpt".format(e),
                trace_id=trace_id
            )
            gpt = OpenAIServer()
            kimi_title = gpt.gpt_mining(
                text=title,
                prompt=gpt.prompt_generate(
                    prompt_type="title"
                )
            )

        kimi_title = kimi_title.replace("'", "").replace('"', "").replace("\\", "")
        try:
            kimi_info = await cls.kimi_mining(contents)
        except Exception as e:
            logging(
                code="9000",
                info="kimi 挖掘失败--{}, 采用 gpt".format(e),
                trace_id=trace_id
            )
            gpt = OpenAIServer()
            kimi_info = gpt.gpt_mining(
                text=contents,
                prompt=gpt.prompt_generate(
                    prompt_type="text"
                )
            )
        kimi_info['k_title'] = kimi_title
        kimi_info['ori_title'] = title
        logging(
            code="8000",
            info="kimi_mining",
            data=kimi_info,
            trace_id=trace_id
        )
        return kimi_info

    @classmethod
    async def ask_kimi(cls, question):
        """
        Ask Kimi for information
        :param question: tiny text
        :return: "{}"
        """
        single_title_prompt = """
            我会给你一个视频标题，需要你帮我用你所学的知识来帮我分析出以下信息，信息我都写到 json 里面了
            {
                "key_words": [],  # 返回三个关键词
                "search_keys": [], # 标题可能的搜索关键词，返回 3 个
                "extra_keys": [], # 关心这个视频的用户还会关心哪些关键词， 返回 3 个
                "theme": 标题的主题， 用一个词概括
            }
            只需要返回一个 json，key 和上面的一样，
            我给你的标题是: 
            """
        client = OpenAI(
            api_key='sk-5DqYCa88kche6nwIWjLE1p4oMm8nXrR9kQMKbBolNAWERu7q',
            base_url="https://api.moonshot.cn/v1"
        )
        chat_completion = client.chat.completions.create(
            messages=[
                {
                    "role": "user",
                    "content": single_title_prompt + question,
                }
            ],
            model="moonshot-v1-8k",
        )
        response = chat_completion.choices[0].message.content.replace('```json', '').replace('```', '')
        try:
            response = json.loads(response)
            return response
        except:
            return {}

    @classmethod
    async def ask_kimi_and_save_to_local(cls, info_tuple):
        """
        save file to local
        :return:
        """
        title, trace_id, save_path = info_tuple[0], info_tuple[1], info_tuple[2]
        if os.path.exists(save_path):
            logging(
                code="2001",
                info="该 video 信息已经挖掘完成---{}".format(title),
                function="ask_kimi_and_save_to_local",
                trace_id=trace_id,
            )
        else:
            os.makedirs(os.path.dirname(save_path), exist_ok=True)
            if not title:
                result = {}
            else:
                result = await cls.ask_kimi(title)
            logging(
                code="2001",
                info="kimi-result",
                data=result,
                trace_id=trace_id,
                function="ask_kimi_and_save_to_local"
            )
            with open(save_path, "w", encoding="utf-8") as f:
                f.write(json.dumps(result, ensure_ascii=False))

    @classmethod
    async def kimi_title(cls, ori_title):
        """
        prompt + kimi + ori_title generate new title
        :param ori_title:
        :return:
        """
        single_title_prompt = """
        请将以上标题改写成适合小程序点击和传播的小程序标题，小程序标题的写作规范如下，请学习后进行小程序标题的编写。直接输出最终的小程序标题
        小程序标题写作规范：
        1.要点前置：将最重要的信息放在标题的最前面，以快速吸引读者的注意力。例如，“5月一辈子同学，三辈子亲，送给我的老同学，听哭无数人！”中的“5月”和“一辈子同学，三辈子亲”都是重要的信息点。
        2.激发情绪：使用能够触动人心的语言，激发读者的情感共鸣。如“只剩两人同学聚会，看后感动落泪。”使用“感动落泪”激发读者的同情和怀旧情绪。
        3.使用数字和特殊符号：数字可以提供具体性，而特殊符号如“🔴”、“😄”、“🔥”等可以吸引视觉注意力，增加点击率。
        4.悬念和好奇心：创建悬念或提出问题，激发读者的好奇心。例如，“太神奇了!长江水位下降，重庆出现惊奇一幕!”中的“惊奇一幕”就是一个悬念。
        5.名人效应：如果内容与知名人士相关，提及他们的名字可以增加标题的吸引力。
        6.社会价值观：触及读者的文化和社会价值观，如家庭、友情、国家荣誉等。
        7.标点符号的运用：使用感叹号、问号等标点来增强语气和情感表达。
        8.直接的语言：使用直白、口语化的语言，易于理解，如“狗屁股，笑死我了!”。
        9.热点人物或事件：提及当前的热点人物或事件，利用热点效应吸引读者。
        10.字数适中：保持标题在10-20个字之间，既不过长也不过短，确保信息的完整性和吸引力。
        11.适当的紧迫感：使用“最新”、“首次”、“紧急”等词汇，创造一种紧迫感，促使读者立即行动。
        12.情感或价值诉求：使用如“感动”、“泪目”、“经典”等词汇，直接与读者的情感或价值观产生共鸣。
        避免误导：确保标题准确反映内容，避免夸大或误导读者。
        """
        client = OpenAI(
            api_key='sk-5DqYCa88kche6nwIWjLE1p4oMm8nXrR9kQMKbBolNAWERu7q',
            base_url="https://api.moonshot.cn/v1"
        )
        chat_completion = client.chat.completions.create(
            messages=[
                {
                    "role": "user",
                    "content": ori_title + "\n" + single_title_prompt,
                }
            ],
            model="moonshot-v1-32k",
        )
        response = chat_completion.choices[0].message.content
        return response.split("\n")[0]

    @classmethod
    async def kimi_mining(cls, text):
        """
        通过文章来挖掘出有效的信息
        :param text:
        :return:
        """
        text_prompt = """
        请从我给你的文章中挖掘出以下信息并且返回如下结果。
        你返回的结果是一个 json, 格式如下:
        {
            "content_keys": [] # 同时提供三个与文章内容高度相关的关键词，这些关键词将用于网络上搜索相关视频内容,
            "content_title": 一个总结性的标题，该标题应简洁并能够反映文章的主要内容
        }
        你需要处理的文本是：
        """
        client = OpenAI(
            api_key='sk-5DqYCa88kche6nwIWjLE1p4oMm8nXrR9kQMKbBolNAWERu7q',
            base_url="https://api.moonshot.cn/v1"
        )
        chat_completion = client.chat.completions.create(
            messages=[
                {
                    "role": "user",
                    "content": text_prompt + text,
                }
            ],
            model="moonshot-v1-32k",
        )
        response = chat_completion.choices[0].message.content.replace('```json', '').replace('```', '')
        try:
            response = json.loads(response)
            return response
        except:
            return {}