LongArticlesJob/update_history_kimi_title.py

"""
@author: luojunhui
"""
from tqdm import tqdm
from pymysql.cursors import DictCursor

from applications.api import fetch_deepseek_response
from applications import longArticlesMySQL

lam = longArticlesMySQL()


def generate_prompt(article_title):
    """
    generate prompt
    """
    prompt = f"""
        输入标题： {article_title}
        请将以上标题改写成适合小程序点击和传播的小程序标题，小程序标题的写作规范如下，请学习后进行小程序标题的编写。直接输出最终的小程序标题
        小程序标题写作规范：
        1.要点前置：将最重要的信息放在标题的最前面，以快速吸引读者的注意力。例如，“5月一辈子同学，三辈子亲，送给我的老同学，听哭无数人！”中的“5月”和“一辈子同学，三辈子亲”都是重要的信息点。
        2.激发情绪：使用能够触动人心的语言，激发读者的情感共鸣。如“只剩两人同学聚会，看后感动落泪。”使用“感动落泪”激发读者的同情和怀旧情绪。
        3.使用数字和特殊符号：数字可以提供具体性，而特殊符号如“🔴”、“😄”、“🔥”等可以吸引视觉注意力，增加点击率。
        4.悬念和好奇心：创建悬念或提出问题，激发读者的好奇心。
        5.名人效应：如果内容与知名人士相关，提及他们的名字可以增加标题的吸引力。
        6.社会价值观：触及读者的文化和社会价值观，如家庭、友情、国家荣誉等。
        7.标点符号的运用：使用感叹号、问号等标点来增强语气和情感表达。
        8.直接的语言：使用直白、口语化的语言，易于理解，如“狗屁股，笑死我了!”。
        9.热点人物或事件：提及当前的热点人物或事件，利用热点效应吸引读者。
        10.字数适中：保持标题在10-20个字之间，既不过长也不过短，确保信息的完整性和吸引力。
        11.情感或价值诉求：使用如“感动”、“泪目”、“经典”等词汇，直接与读者的情感或价值观产生共鸣。
        避免误导：确保标题准确反映内容，避免夸大或误导读者。
    """
    return prompt


def update_status(content_id, ori, new_):
    """
    update status
    """
    sql = """
        update long_articles_temp
        set status = %s
        where content_id = %s and status = %s;
        """
    rows = lam.update(sql=sql, params=(new_, content_id, ori))
    return rows


def get_task_list():
    """
    get task list
    """
    sql = """
        select t1.content_id, t2.article_title from long_articles_temp t1
        join long_articles_text t2 on t1.content_id = t2.content_id
        where t1.status = 0;
        """
    resp = lam.select(sql, cursor_type=DictCursor)
    return resp


def update_temp(content_id, new_title):
    """
    update temp
    """
    sql = """
        update long_articles_temp 
        set new_kimi_title = %s, status = %s
        where content_id = %s and status = %s;
        """
    lam.update(sql=sql, params=(new_title, 1, content_id, 101))


def process_task(task):
    """
    print
    """
    content_id = task['content_id']
    article_title = task['article_title']
    prompt = generate_prompt(article_title)
    # lock
    r = update_status(content_id=content_id, ori=0, new_=101)
    if not r:
        return
    try:
        title = fetch_deepseek_response(model="DeepSeek-V3", prompt=prompt)
        update_temp(content_id, title)
    except Exception as e:
        # roll back
        update_status(content_id=content_id, ori=101, new_=0)
        print(e)


def main():
    """
    main function
    """
    task_list = get_task_list()
    for task in tqdm(task_list):
        process_task(task)


if __name__ == '__main__':
    main()