Merge branch 'feature/20250126-audit-title-by-llm'

applications/llm_sensitivity.py

@@ -0,0 +1,135 @@
+#! /usr/bin/env python
+# -*- coding: utf-8 -*-
+# vim:fenc=utf-8
+
+import json
+from openai import OpenAI
+import pandas as pd
+
+def request_llm_api(prompt, text):
+    client = OpenAI(
+        api_key='sk-30d00642c8b643cab3f54e5672f651c9',
+        base_url="https://api.deepseek.com"
+    )
+    chat_completion = client.chat.completions.create(
+        messages=[
+            {
+                "role": "user",
+                "content": prompt + text,
+            }
+        ],
+        model="deepseek-chat",
+        temperature=0.2,
+        response_format={"type": "json_object"}
+    )
+    response = chat_completion.choices[0].message.content
+    return response
+
+
+def do_check_titles(text):
+    """
+    :param text:
+    :return:
+    """
+    text_prompt = """
+你是一位专业的微信公众号内容编辑，公众号的主要读者是中老年群体，主要内容来源为其它公众号发布文章。你需要对公众号待发布的文章标题进行可用性审核，判断其是否适合发布。不可用的情形包括以下7大类，每个大类情形中包括多个小类，以多级序号表示：
+1.违规诱导、对抗平台审查规则
+1.1 使用不完全或擦边的标题诱导用户点击
+1.2 使用了生僻字/绕过字体等形式规避微信审核
+2.违反国家法律法规
+2.1 涉及国内政治敏感人物、敏感政治事件
+3.有煽动、夸大、误导嫌疑
+3.1 涉及国内外时局政治内容，尤其是与中国关系密切的国家的
+3.2 在标题中使用内部、内参、解封等词语，将内容伪装成官方内部流出的
+3.3 非官方通知或者公告，但标题假借官方名义或者暗喻为官方发布的
+3.4 标题滥用“领导”、“主席”等头衔，容易造成曲解的
+3.5 在标题中使用重大消息、通知、紧急提醒等，或以信息来源机密、看完即删来诱导用户的
+3.6 在标题中使用过度夸张的词语，比如含有危害人身安全、恐吓侮辱、惊悚、极端内容，或者以命令式语气强迫用户阅读，煽动诱导的意味较强的
+4.色情低俗内容
+4.1 文字描述男女性隐私部位，或误导用户误以为是隐私部位
+4.2 文字明示或暗示，容易让人联想到性、性行为、不正当性关系的
+4.3 文字出现不正当婚恋观，如出轨、找小三、约炮等
+5.宣扬封建迷信
+5.1 标题捏造神迹或捏造所谓的特殊日子，用于接福、求福的
+5.2 标题涉及民间迷信谚语、民俗，容易宣扬迷信思想的
+6.不适合在非特定时间、向不特定群体发布的
+6.1 具有较强新闻时效性的事件，尤其是包含“突发”、“刚刚”或特定日期的
+6.2 面向特定范围群体（某个省市的居民、企事业单位员工等）的通知、宣传稿、新闻稿等
+7.与中老年人阅读偏好不相关的
+7.1 学习、影视等资源下载
+7.2 其它与一般中老年人偏好不相关的情形
+审核时，请特别注意以下事项：
+ - 标题是否涉及政治敏感、色情低俗、封建迷信等内容。
+ - 标题是否适合中老年群体的阅读偏好。
+ - 标题是否具有时效性或面向特定群体，不适合广泛发布。例如：
+  - 标题中包含“突破”、“刚刚”等词语时，需注意是否具有时效性（如年初、特定节日等）。
+  - 标题中包含特定地名、群体名称（如省市区县旗盟乡镇）时，需注意是否面向特定群体，如果包含特定地域或人群但具有传播性和普适性则也可以发布。
+审核结果分为2种：
+ - 通过：标题未违反规则，适合发布。
+ - 不通过：标题明确违反规则，不适合发布。
+对于“不通过”的情形，请同时标注命中的规则大类序号（如 1、2、3 等）。
+输出格式必须为 JSON 格式，示例如下：
+[
+{
+"title": "标题",
+"status": "不通过",
+"hit_rule": 5
+}
+]
+现在，请对以下文章标题的可用性进行判断：
+"""
+    return request_llm_api(text_prompt, text)
+
+def check_titles(titles, retun_map=False):
+    n_titles = len(titles)
+
+    batch_size = 20
+    n_batches = (n_titles + batch_size - 1) // batch_size
+    json_data = []
+    for i in range(0, n_batches):
+        offset_begin = i * batch_size
+        offset_end = min(offset_begin + batch_size, n_titles)
+        current_batch = titles[offset_begin:offset_end]
+        try:
+            res = do_check_titles('\n'.join(current_batch))
+            json_data = json.loads(res)
+        except Exception as e:
+            print(e)
+        if isinstance(json_data, dict):
+            json_data = [json_data]
+        if not json_data:
+            for title in current_batch:
+                try:
+                    res = do_check_titles(title)
+                    json_res = json.loads(res)
+                    if isinstance(res, list):
+                        json_data.append(json_res[0])
+                    elif isinstance(res, dict):
+                        json_data.append(json_res)
+                    else:
+                        raise Exception('Invalid Response')
+                except Exception as e:
+                    print(e)
+                    json_data.append({'title': title, 'status': '通过'})
+    for item in json_data:
+        if item['status'] == '不通过':
+            try:
+                item['hit_rule'] = int(float(item['hit_rule']))
+            except Exception as e:
+                item['hit_rule'] = 99
+            # 保证不为0 避免LLM返回异常
+            if item['hit_rule'] == 0:
+                item['hit_rule'] = 99
+        else:
+            item['hit_rule'] = 0
+    if retun_map:
+        result_map = {}
+        for item in json_data:
+            if item['title'] not in result_map:
+                result_map[item['title']] = item
+            # 安全起见，保留最后一次不通过的记录
+            elif item['status'] == '不通过':
+                result_map[item['title']] = item
+        return result_map
+    else:
+        return json_data

coldStartTasks/crawler/weixinCategoryCrawler.py

@@ -6,8 +6,7 @@
 import time
 
 from tqdm import tqdm
-
-from applications import WeixinSpider, Functions
+from applications import WeixinSpider, Functions, llm_sensitivity
 
 # 常量
 ACCOUNT_GOOD_STATUS = 1
@@ -56,6 +55,7 @@ class weixinCategory(object):
         将数据更新到数据库
         :return:
         """
+        success_records = []
         for article_obj in article_list:
             detail_article_list = article_obj["AppMsg"]["DetailInfo"]
             for obj in detail_article_list:
@@ -63,11 +63,15 @@ class weixinCategory(object):
                     show_stat = self.function.show_desc_to_sta(obj["ShowDesc"])
                     show_view_count = show_stat.get("show_view_count", DEFAULT_VIEW_COUNT)
                     show_like_count = show_stat.get("show_like_count", DEFAULT_LIKE_COUNT)
+                    unique_idx = self.function.generateGzhId(obj["ContentUrl"])
                     insert_sql = f"""
                         insert into crawler_meta_article
-                        (platform, mode, category, out_account_id, article_index, title, link, read_cnt, like_cnt, description, publish_time, crawler_time, status, unique_index)
+                        (
+                         platform, mode, category, out_account_id, article_index, title, link, read_cnt, like_cnt,
+                         description, publish_time, crawler_time, status, unique_index, llm_sensitivity
+                        )
                         VALUES 
-                        (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s);
+                        (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s);
                     """
                     self.db_client_lam.update(
                         sql=insert_sql,
@@ -85,11 +89,28 @@ class weixinCategory(object):
                             obj["send_time"],
                             int(time.time()),
                             DEFAULT_ARTICLE_STATUS,
-                            self.function.generateGzhId(obj["ContentUrl"]),
+                            unique_idx,
+                            obj.get("llm_sensitivity", -1)
                         ),
                     )
+                    success_records.append({
+                        'unique_index': unique_idx, 'title': obj['Title']
+                    })
                 except Exception as e:
                     print(e)
+        return success_records
+
+    def update_article_sensitive_status(self, category, unique_index, status):
+        """
+        更新文章敏感状态
+        :return:
+        """
+        update_sql = f"""
+            update crawler_meta_article
+            set llm_sensitivity = %s
+            where category = %s and unique_index = %s;
+        """
+        self.db_client_lam.update(sql=update_sql, params=(status, category, unique_index))
 
     def update_latest_account_timestamp(self, gh_id):
         """
@@ -121,13 +142,13 @@ class weixinCategory(object):
         msg_list = response.get("data", {}).get("data")
         if msg_list:
             last_article_in_this_msg = msg_list[-1]
-            self.insert_data_into_db(
+            success_records = self.insert_data_into_db(
                 gh_id=gh_id, category=category, article_list=msg_list
             )
             last_time_stamp_in_this_msg = last_article_in_this_msg["AppMsg"]["BaseInfo"]["UpdateTime"]
             if latest_time_stamp < last_time_stamp_in_this_msg:
                 next_cursor = response["data"]["next_cursor"]
-                return self.update_each_account(
+                return success_records + self.update_each_account(
                     gh_id=gh_id,
                     latest_time_stamp=latest_time_stamp,
                     category=category,
@@ -137,8 +158,10 @@ class weixinCategory(object):
                 # 更新最近抓取时间
                 self.update_latest_account_timestamp(gh_id=gh_id)
                 print("账号时间更新成功")
+                return success_records
         else:
             print("No more data")
+            return []
 
     def deal(self, category_list):
         """
@@ -147,6 +170,7 @@ class weixinCategory(object):
         :return:
         """
         for category in category_list:
+            success_records = []
             account_list = self.get_account_list(category)
             for account in tqdm(account_list):
                 try:
@@ -156,7 +180,7 @@ class weixinCategory(object):
                         timestamp = int(account['latest_timestamp'].timestamp())
                     except Exception as e:
                         timestamp = DEFAULT_TIMESTAMP
-                    self.update_each_account(
+                    success_records += self.update_each_account(
                         gh_id=gh_id,
                         category=category,
                         latest_time_stamp=timestamp
@@ -164,6 +188,18 @@ class weixinCategory(object):
                     print("success")
                 except Exception as e:
                     print("fail because of {}".format(e))
+            success_titles = [x['title'] for x in success_records]
+            if success_titles:
+                try:
+                    sensitive_results = llm_sensitivity.check_titles(success_titles)
+                    for record, sensitive_result in zip(success_records, sensitive_results):
+                        self.update_article_sensitive_status(
+                            category=category,
+                            unique_index=record['unique_index'],
+                            status=sensitive_result['hit_rule']
+                        )
+                except Exception as e:
+                    print("failed to update sensitive status: {}".format(e))
 
     def deal_accounts(self, account_list):
         """