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.config import moon_shot
from applications.config import deep_seek
from applications.log import logging


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

    @classmethod
    async def search_kimi_schedule(cls, params, safe_score):
        """
        搜索阶段 kimi 操作
        :param params:
        :param safe_score:
        :return:
        """
        title = params['article_title'].split("@@")[-1]
        contents = params['article_text']
        trace_id = params['content_id']
        try:
            kimi_title = await cls.kimi_title(title, ai_model=moon_shot)
            # 判断kimi 标题是否安全
            title_score = await cls.get_kimi_title_safe_score(kimi_title, ai_model=moon_shot)
            kimi_green_title = None
            if int(title_score) > safe_score:
                kimi_green_title = await cls.make_kimi_title_safer(title, ai_model=moon_shot)
                
            logging(
                code="1005",
                info="generate_title",
                data={
                    "ori_title": title,
                    "kimi_title": kimi_title,
                    "title_score": title_score,
                    "new_kimi_title": kimi_green_title
                },
                trace_id=trace_id
            )
            kimi_title = kimi_green_title if kimi_green_title else kimi_title

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

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

    @classmethod
    async def get_kimi_title_safe_score(cls, kimi_title, ai_model):
        """
        获取kimi_title的安全分数
        """
        prompt = f"""
        请你学习一下内容规范，以下标题可能会违反了某条内容规范。请你对标题做一个内容风险评级，1-10分，等级越高内容违规风险越大。 
        请直接输出内容风险评级的分数，不要输出你的理由、分析等内容。 
        输出:
            只需要输出危险分级分数，不要输出任何其他内容。
        内容规范为： 
        4.2 色俗内容 
            4.2.1 散布淫秽、色情内容，包括但不限于招嫖、寻找一夜情、性伴侣等。 
            4.2.2 发布有色情意味的情色文字、情色视频、情色漫画等内容。 
            4.2.3 以投稿/爆料等形式描述约炮经历、性交体验、偷情、涉隐私部位偷拍等伤风败俗的话题内容。 
            4.2.4 以低俗的配图引诱用户阅读文章、关注微信公众号。包含性撩拨、性挑逗画面；疑似女性性高潮/性虐场面；偷拍的沐浴/更衣/如厕/亲热等私密画面；女性故意露出敏感部位 (纯裸露的胸、生殖器官)以及敏感部位未打码的真人写真/艺术摄影等。 
            4.2.5 文内以低俗类的动图或引导图，诱导用户点击进而跳转至另一篇图文页或关注某个公众号。 
            4.2.6 文章主要描述PUA撩妹、撩汉等相关话题，且引导用户关注公众号/加个人微信号/加群。 
        4.11 煽动、夸大、误导类内容 平台鼓励创作者提供准确、清晰、能体现文章内容主旨的标题，不允许通过标题噱头诱导用户点击或误导用户。
            包括但不限于以下情况： 
            4.11.1 标题含有危害人身安全、恐吓侮辱、惊悚、极端内容，或者以命令式语气强迫用户阅读。 
            4.11.2 标题无依据夸大事件严重程度、紧急程度、受影响面以及事件引发的情绪。 
            4.11.3 标题以浮夸的描述，反常识强调某种食物/行为对人体健康的影响，煽动人群要/不要去做某行为。 
            4.11.4 非官方通知或者公告，但标题假借官方名义煽动获取流量，或以信息来源机密、看完即删来诱导用户。 
            4.11.5 标题故意隐藏关键信息，或无中生有部分信息，给用户造成误导。 
            4.12 违反国家法律法规禁止的内容 
                （1）违反宪法确定的基本原则的； 
                （2）危害国家安全，泄露国家秘密，颠覆国家政权，破坏国家统一的； 
                （3）损害国家荣誉和利益的； 
                （4）煽动民族仇恨、民族歧视，破坏民族团结的； 
                （5）破坏国家宗教政策，宣扬邪教和封建迷信的； 
                （6）散布不实信息，扰乱社会秩序，破坏社会稳定的； 
                （7）散布淫秽、色情、赌博、暴力、恐怖或者教唆犯罪的； 
                （8）侮辱或者诽谤他人，侵害他人合法权益的； 
                （9）煽动非法集会、结社、游行、示威、聚众扰乱社会秩序； 
                （10）以非法民间组织名义活动的； 
                （11）不符合《即时通信工具公众信息服务发展管理暂行规定》及遵守法律法规、社会主义制度、国家利益、公民合法利益、公共秩序、社会道德风尚和信息真实性等“七条底线”要求的； 
                （12）含有法律、行政法规禁止的其他内容的。
        输入的标题是： {kimi_title}
        """
        client = OpenAI(
            api_key=ai_model['api_key'],
            base_url=ai_model['base_url']
        )
        chat_completion = client.chat.completions.create(
            messages=[
                {
                    "role": "user",
                    "content": prompt
                }
            ],
            model=ai_model['model']
        )
        response = chat_completion.choices[0].message.content
        return response

    @classmethod
    async def make_kimi_title_safer(cls, ori_title, ai_model):
        """
        安全化 kimi title
        """
        prompt = """
            以下每行为一个文章的标题，请用尽量平实的语言对以上标题进行改写，保持在10～15字左右，请注意：
            1. 不要虚构或改变标题的含义。
            2. 不要用笃定的语气描述存疑的可能性，不要将表述可能性的问句改为肯定句。
            直接输出改写后的标题列表。
            在改写完成后，再输出一次，在改写的标题前增加和标题情感、语气匹配的特殊符号，如：“🔴”、“😄”、“🔥”、“😨”等等
            输出：
                输出结果是Dict, 格式为: 
                {
                "title_v1": 请填写第一次输出的标题,
                "title_v2": 请填写第二次输出的标题
                }
            输入的标题是: 
        """
        client = OpenAI(
            api_key=ai_model['api_key'],
            base_url=ai_model['base_url']
        )
        chat_completion = client.chat.completions.create(
            messages=[
                {
                    "role": "user",
                    "content": prompt + ori_title,
                }
            ],
            model=ai_model['model'],
            response_format={"type": "json_object"}
        )
        response = chat_completion.choices[0].message.content
        response_json = json.loads(response)
        safer_title = response_json["title_v2"]
        return safer_title