外部文章质量判断

src/domains/content_quality/__init__.py

@@ -1,3 +1,15 @@
-"""
-文章质量指标
-"""
+"""内容质量评估域 —— 量化评估 + 标题质量评分 + 品类识别"""
+
+from src.domains.content_quality._const import ContentQualityConst
+from src.domains.content_quality._mapper import QualityMapper
+from src.domains.content_quality.article_quality_quantify import ArticleQualityQuantify
+from src.domains.content_quality.article_quality_title import ArticleQualityTitle
+from src.domains.content_quality.article_quality_category import ArticleQualityCategory
+
+__all__ = [
+    "ContentQualityConst",
+    "QualityMapper",
+    "ArticleQualityQuantify",
+    "ArticleQualityTitle",
+    "ArticleQualityCategory",
+]

src/domains/content_quality/_const.py

@@ -0,0 +1,193 @@
+"""内容质量评估域常量 —— 量化评估 + 标题质量 + 品类识别"""
+
+
+class ContentQualityConst:
+    """内容质量评估域所有常量的聚合根"""
+
+    # ═══════════════════════════════════════════════════════════════
+    # 表名
+    # ═══════════════════════════════════════════════════════════════
+
+    QUANTILE_TABLE = "demand_article_quality_quantile"
+    TITLE_TABLE = "demand_article_quality_title"
+    CATEGORY_TABLE = "demand_article_quality_category"
+    DETAIL_TABLE = "demand_search_article_detail"
+    ARTICLES_TABLE = "demand_account_articles"
+    ACCOUNT_TABLE = "demand_search_accounts"
+    RELATION_TABLE = "demand_search_article_relation"
+
+    # ═══════════════════════════════════════════════════════════════
+    # 通用状态（三表共用）
+    # ═══════════════════════════════════════════════════════════════
+
+    class Status:
+        INIT = 0
+        PROCESSING = 1
+        DONE = 2
+        FAIL = 99
+
+    # ═══════════════════════════════════════════════════════════════
+    # 筛选条件
+    # ═══════════════════════════════════════════════════════════════
+
+    MASS_SEND_TYPE = 9
+    VALID_POSITIONS = (1, 2)
+
+    # ═══════════════════════════════════════════════════════════════
+    # 量化参数
+    # ═══════════════════════════════════════════════════════════════
+
+    AVG_WINDOW_DAYS = 60
+    MIN_SAMPLE_COUNT = 3
+
+    # ═══════════════════════════════════════════════════════════════
+    # LLM 参数
+    # ═══════════════════════════════════════════════════════════════
+
+    LLM_BATCH_SIZE = 20
+    LLM_INTERVAL_SEC = 2
+
+    # ═══════════════════════════════════════════════════════════════
+    # XXL-JOB 参数
+    # ═══════════════════════════════════════════════════════════════
+
+    QUANTIFY_DEAL_LIMIT = 500
+    TITLE_DEAL_LIMIT = 100
+    CATEGORY_DEAL_LIMIT = 100
+
+    # ═══════════════════════════════════════════════════════════════
+    # Prompt 模板
+    # ═══════════════════════════════════════════════════════════════
+
+    PROMPT_TITLE_QUALITY = """爆款标题特征与权重分如下，特征权重分的总分为100分
+1.情绪极端化：25分
+2.名人/大国冲突：20分
+3.悬念制造：15分
+4.数字具体化：10分
+5.时间紧迫感：10分
+6.对比/意外转折：8分
+7.阵营对立：7分
+8.军事危机暗示：5分
+
+基于以上的爆款标题特征分评分标准，如果下列标题符合某个特征的、那么就加上这个特征对应的权重，标题最终的得分是各个特征分的总和。请直接输出最终的分数，不用给任何理由和描述、解释。
+待评分的标题为：
+{title_list}"""
+
+    PROMPT_TITLE_CATEGORY = """
+    请帮我完成以下任务：输入为文章的标题，根据标题判断其内容所属的类目，输出为文章标题及其对应的类目。
+类目需从以下15个品类内选择：
+1. 知识科普
+定义：以通俗易懂的方式普及科学、技术、健康、安全、生活常识、财产保护、医保政策、为人处事方式等内容，旨在提高公众的知识水平和认知能力。内容通常具有教育性和实用性，涵盖自然、社会、文化等多个领域。
+标题示例：
+我国存款最安全的五大银行，永远都不会倒闭，你知道是哪五家吗？
+借条上不要写“这3个字”，不然变成一张废纸，否则用法律也没用
+不能二次加热的3种食物！再次提醒：这3种食物吃不完最好扔掉
+
+2. 军事历史
+定义：聚焦于历史上的军事事件、战争故事、军事策略、英雄人物等内容，旨在还原战争场景、探讨军事决策、揭示历史真相，并展现战争中的人物命运与历史影响。内容通常以叙事、分析或回忆的形式呈现，兼具历史深度和故事性。
+标题示例：
+对越作战永远失踪的332人，陵园没有墓碑，没有名字，只有烈士证
+淮海大战丢失阵地，师长带头冲锋！最后出一口恶气：活捉敌最高指挥官
+抗战时，一村民被敌拉去带路，半道回头忽发现:后面跟个游击队员
+
+3. 家长里短
+定义：围绕家庭成员之间的关系、矛盾、情感、道德、等展开的故事或讨论，内容常涉及婚姻、亲子、婆媳、兄弟姐妹等关系，或是人情往来、金钱纠纷、情感变化等内容，反映家庭生活中的温情、冲突与人性。
+标题示例：
+父母越老越能暴露家庭最真实的一面：当父母70岁，子女不该抱有这三种期待
+老母亲分家产，给亲闺女30万，给养女一筐青菜，养女意外摔倒，看到筐子里的东西，瞬间愣住了
+我花150一天雇了阿姨，两天后上班回来给她300，阿姨说我账算错了
+
+4. 社会法治
+定义：聚焦社会事件、法律纠纷、法院判决、社会现象等内容，通常涉及道德、法律、公平正义等议题，旨在揭示社会问题、探讨法律规则或反映人性与社会现实。
+标题示例：
+山东，女子在小区捡到16万天价项链，业主悬赏3万找回，女子归还后，失主拒绝支付报酬，还说：要有格局，女子认为被骗，将失主告上法庭
+陕西，女子22万买26层房，2年后，楼盘24层就已经封顶！开发商：你闹事造成100万损失，道歉才给赔偿！
+上海：男子超市连续购买46枚过期咸鸭蛋，2天分46次交易，向厂家索赔金14万，法院判了！
+
+5. 奇闻趣事
+定义：以猎奇、娱乐为主，涵盖罕见、奇特、有趣的事件、发现或故事，内容通常具有趣味性和话题性，能够引发读者的好奇心和讨论。
+标题示例：
+狗屎运？江西男子钓鱼时发现青鱼尸骸，扒开后捡到鸡蛋大小的青鱼石，网友：起码值几千！
+内蒙古小伙河边捡到金牌，拒绝上交将其熔成金手镯，专家气愤不已
+男子买了一辆废弃坦克，拆油箱时，他发现了一根又一根的金条……
+
+6. 名人八卦
+定义：围绕名人的生活、言论、事件、八卦等内容展开，通常涉及娱乐圈、政界、历史人物等，旨在满足公众对名人隐私和动态的好奇心。
+标题示例：
+难怪王扶林说陈晓旭不够漂亮，看看他选的原黛玉候选人，那才叫美
+心狠手辣的容嬷嬷年轻时是校花？看了照片后，网友直接闭嘴了！
+李玉成终于说出实话，公开吐槽马玉琴年纪太大，结婚28年疑似后悔
+
+7. 健康养生
+定义：关注健康、养生、疾病预防、生活习惯等方面的知识和建议，内容通常具有实用性和指导性，旨在帮助读者改善生活质量、提升健康水平。
+标题示例：
+72岁老人每天一个蒸苹果，半年后体检，看到指标变化让他乐开了花
+40岁女子每天吃水煮蛋，一年后去体检，检查报告令医生都羡慕不已
+2024年血糖新标准已公布，不再是3.9~6.1，你的血糖还不算高吗？
+
+8. 情感故事
+定义：以人与人之间的情感交流、感人故事、情感经历为主题，内容通常充满温情、感动或反思，旨在引发读者的情感共鸣和思考。
+标题示例：
+男孩饭店吃饭，发现陌生女子和去世母亲很像，走过去说：我妈妈去世了，能抱一下我吗？
+河南一女子直播时，被失散 32 年的父亲认出：闺女等着爸爸接你回家
+1987年，江苏男子借好友一千元，25年后朋友成富豪还他1000万报恩
+流浪狗跟着骑行夫妻跑了一百多公里，一直守护在女主身边，赶都赶不走，当男主得知原因后竟抱着狗狗大哭起来
+
+9. 国家大事
+定义：涉及国家实力、科技发展、资源发现、国际合作等内容，通常以宏观视角展现国家的综合实力、科技成就或国际影响力，体现国家的崛起与发展。
+标题示例：
+我国在南极发现“海上粮仓”，储量高达10亿吨，世界各国眼红不已
+我国贵州发现7000万吨宝藏，价值高达上万亿，多国求合作被拒绝
+距我国3000公里，塞班岛明明归美国管辖，为何岛上大多是中国人？
+
+10. 现代人物
+定义：聚焦活跃在21世纪后具有传奇色彩或巨大贡献的人物、事迹、成就等，内容通常充满戏剧性和启发性，旨在展现人物的非凡经历或历史贡献。
+标题示例：
+她曾狂贪国家上百亿，被发现时已经移居美国，最终还风光一时得善终
+山东女子因坐月子无聊，破译美国2套绝密系统的密码，国家：奖励711万！
+牺牲太大了！航天女英雄刘洋：结婚8年未生子，回地面后“消失”的她怎样了？
+
+11. 怀旧时光
+定义：以回忆和怀旧为主题，涉及过去的历史、文化、生活、照片等内容，旨在唤起读者对过去时光的情感共鸣和怀念。
+标题示例：
+1975年“下馆子”的老照片，2元能吃些什么，勾起那段最难忘的时光
+82年，北京老人捡回两张“破椅子”，遭家人数落，29年后拍出2300万
+这张老照片第一次看到，邓颖超和李讷的罕见合影！
+
+12. 政治新闻
+定义：聚焦政治事件、领导人动态、国际关系等内容，通常以新闻或分析的形式呈现，旨在揭示政治局势、政策变化或国际关系的动态。
+标题示例：
+中方外长行程有变，提前结束访欧匆匆回国，带回来一个好消息
+宋庆龄在北京逝世后，远在美国的宋美龄只说了7个字，字字揪心！
+庐山会议后，叶帅去劝彭德怀认个错，哭着说了一句心里话
+
+13. 历史人物
+定义：聚焦于21世纪前具有重要影响的人物，包括他们的生平、事迹、成就、性格、趣事及其对历史进程的贡献。内容通常以传记、回忆录或历史分析的形式呈现，旨在还原人物的真实面貌并探讨其历史意义。
+标题示例：
+林彪去世后，蒋介石收到林彪与戴笠的一份密谈文件，看后拍桌大骂
+张学良软禁时的一张实拍照片，头发秃顶，两眼无光，像个中年老头
+1912年，孙中山和两个女儿罕见留影，面对镜头父女三人看起来很幸福
+
+14. 社会现象
+定义：关注社会中出现的普遍现象、趋势或问题，通常涉及文化、经济、教育、民生等领域。内容以观察、分析或评论为主，旨在揭示现象背后的原因、影响及社会意义，引发公众的思考和讨论。
+标题示例：
+22年河南男子跳河救人，体力耗尽留遗言，被救女子猛然抓住他：一起走
+浙江一老人刑满释放，靠蹬三轮为生，6年后，政府领导登门拜访：我们帮您分配工作
+儿子收到清华通知书，父亲花5万请全村吃席，镇长看一眼竟说：这是假的
+
+15.财经科技
+定义：聚焦于经济、金融、投资及行业发展的分析与预测，涵盖未来经济趋势、资产价值变化、行业变革及个人理财策略等内容。可以提供前瞻性的财经视角和实用的理财建议，帮助其把握经济动态、优化财务规划并应对行业变化。
+标题示例：
+未来10年，现金和房子都将贬值，只有2样东西最值钱
+外卖时代将被终结？一个全新行业正悄悄取代外卖，你准备好了吗？
+准备存款的一定要知道，今明两年，定期存款要记住“4不存”
+
+输入是一个 LIST， LIST 中的每个元素是一个元组，元组的第一个元素是文章的 ID，第二个元素是文章的标题。
+最后输出结果请用JSON格式输出，key为ID，value为品类，仅输出JSON，不要markdown格式，不要任何其他内容
+如果标题中包含半角双引号，则进行转义
+输入的 LIST 是 {title_list}
+检查你的输出，输出的品类一定需要是输入的 15个品类，只需要输出品类的中文汉字
+"""
+
+
+__all__ = ["ContentQualityConst"]

src/domains/content_quality/_mapper.py

@@ -0,0 +1,388 @@
+"""内容质量评估域 DB 读写 —— quantile / title / category 三表 + detail / articles 查询"""
+
+from typing import Dict, List, Optional
+
+from src.infra.database.mysql.manager import MysqlManager
+
+from ._const import ContentQualityConst
+
+
+class QualityMapper(ContentQualityConst):
+    """三张质量评估表 + detail.stat_status 的读写"""
+
+    def __init__(self, pool: MysqlManager):
+        self.pool = pool
+
+    # ═══════════════════════════════════════════════════════════════
+    # detail 表 — 前置校验
+    # ═══════════════════════════════════════════════════════════════
+
+    async def fetch_crawled_account_ids(self) -> List[str]:
+        """取已完成深翻的账号 channel_account_id 列表"""
+        query = f"""
+            SELECT channel_account_id
+            FROM {self.ACCOUNT_TABLE}
+            WHERE crawl_deep_done = 1 AND channel_account_id != ''
+        """
+        rows = await self.pool.fetch(query=query)
+        return [r["channel_account_id"] for r in rows]
+
+    async def fetch_pending_detail_by_account(
+        self, account_id: str, *, limit: int = 500
+    ) -> List[Dict]:
+        """取 detail 中 stat_status=INIT 且属于指定账号的文章"""
+        query = f"""
+            SELECT id, wx_sn, title, view_count, like_count
+            FROM {self.DETAIL_TABLE}
+            WHERE stat_status = %s AND channel_account_id = %s
+            ORDER BY id ASC
+            LIMIT %s
+        """
+        return await self.pool.fetch(
+            query=query,
+            params=(self.Status.INIT, account_id, limit),
+        )
+
+    async def mark_detail_processing(self, ids: List[int]) -> int:
+        if not ids:
+            return 0
+        placeholders = ",".join(["%s"] * len(ids))
+        query = f"""
+            UPDATE {self.DETAIL_TABLE}
+            SET stat_status = %s
+            WHERE id IN ({placeholders}) AND stat_status = %s
+        """
+        return await self.pool.save(
+            query=query,
+            params=(self.Status.PROCESSING, *ids, self.Status.INIT),
+        )
+
+    async def mark_detail_done(self, item_id: int) -> int:
+        query = f"""
+            UPDATE {self.DETAIL_TABLE}
+            SET stat_status = %s
+            WHERE id = %s AND stat_status = %s
+        """
+        return await self.pool.save(
+            query=query,
+            params=(self.Status.DONE, item_id, self.Status.PROCESSING),
+        )
+
+    async def mark_detail_failed(self, item_id: int, remark: str = "") -> int:
+        query = f"""
+            UPDATE {self.DETAIL_TABLE}
+            SET stat_status = %s, remark = %s
+            WHERE id = %s AND stat_status = %s
+        """
+        return await self.pool.save(
+            query=query,
+            params=(self.Status.FAIL, remark[:512], item_id, self.Status.PROCESSING),
+        )
+
+    async def revert_detail_init(self, item_id: int) -> int:
+        """PROCESSING → INIT（数据未就绪，回退等下次重试）"""
+        query = f"""
+            UPDATE {self.DETAIL_TABLE}
+            SET stat_status = %s
+            WHERE id = %s AND stat_status = %s
+        """
+        return await self.pool.save(
+            query=query,
+            params=(self.Status.INIT, item_id, self.Status.PROCESSING),
+        )
+
+    async def fetch_relation_titles_not_in_table(
+        self, table: str, *, limit: int = 500
+    ) -> List[Dict]:
+        """取 relation 中 title 不在 title/category 表的去重标题"""
+        query = f"""
+            SELECT DISTINCT r.title_md5, r.title
+            FROM {self.RELATION_TABLE} r
+            WHERE r.title != '' AND r.title_md5 != ''
+              AND NOT EXISTS (
+                  SELECT 1 FROM {table} t WHERE t.title_md5 = r.title_md5
+              )
+            LIMIT %s
+        """
+        return await self.pool.fetch(query=query, params=(limit,))
+
+    async def backfill_relation_quality_score(self) -> int:
+        """将 title 表中已评分的 score 回写到 relation.quality_score"""
+        query = f"""
+            UPDATE {self.RELATION_TABLE} r
+            JOIN {self.TITLE_TABLE} t ON r.title_md5 = t.title_md5
+            SET r.quality_score = t.score
+            WHERE t.status = %s AND r.quality_score != t.score
+        """
+        return await self.pool.save(
+            query=query,
+            params=(self.Status.DONE,),
+        )
+
+    # ═══════════════════════════════════════════════════════════════
+    # articles 表 — 查群发位置
+    # ═══════════════════════════════════════════════════════════════
+
+    async def get_article_position(self, wx_sn: str) -> Optional[Dict]:
+        query = f"""
+            SELECT gh_id, item_index, type
+            FROM {self.ARTICLES_TABLE}
+            WHERE wx_sn = %s
+            LIMIT 1
+        """
+        return await self.pool.fetch_one(query=query, params=(wx_sn,))
+
+    # ═══════════════════════════════════════════════════════════════
+    # quantile 表
+    # ═══════════════════════════════════════════════════════════════
+
+    async def insert_quantile(self, row: Dict) -> int:
+        query = f"""
+            INSERT IGNORE INTO {self.QUANTILE_TABLE}
+                (wx_sn, view_count, like_count)
+            VALUES (%s, %s, %s)
+        """
+        return await self.pool.save(
+            query=query,
+            params=(row["wx_sn"], row.get("view_count", 0), row.get("like_count", 0)),
+        )
+
+    async def fetch_pending_quantile(self, *, limit: int = 500) -> List[Dict]:
+        query = f"""
+            SELECT q.id, q.wx_sn, q.view_count, q.like_count, d.publish_at
+            FROM {self.QUANTILE_TABLE} q
+            JOIN {self.DETAIL_TABLE} d ON q.wx_sn = d.wx_sn
+            WHERE q.status = %s
+            ORDER BY q.id ASC
+            LIMIT %s
+        """
+        return await self.pool.fetch(
+            query=query,
+            params=(self.Status.INIT, limit),
+        )
+
+    async def mark_quantile_processing(self, ids: List[int]) -> int:
+        if not ids:
+            return 0
+        placeholders = ",".join(["%s"] * len(ids))
+        query = f"""
+            UPDATE {self.QUANTILE_TABLE}
+            SET status = %s
+            WHERE id IN ({placeholders}) AND status = %s
+        """
+        return await self.pool.save(
+            query=query,
+            params=(self.Status.PROCESSING, *ids, self.Status.INIT),
+        )
+
+    async def revert_quantile_init(self, item_id: int) -> int:
+        """PROCESSING → INIT（数据未就绪，回退等下次重试）"""
+        query = f"""
+            UPDATE {self.QUANTILE_TABLE}
+            SET status = %s
+            WHERE id = %s AND status = %s
+        """
+        return await self.pool.save(
+            query=query,
+            params=(self.Status.INIT, item_id, self.Status.PROCESSING),
+        )
+
+    async def get_avg_view_count_60d(
+        self, gh_id: str, item_index: int, publish_time,
+    ) -> Optional[Dict]:
+        query = f"""
+            SELECT AVG(show_view_count) AS avg_count, COUNT(*) AS sample_count
+            FROM {self.ARTICLES_TABLE}
+            WHERE gh_id = %s
+              AND item_index = %s
+              AND send_time >= UNIX_TIMESTAMP(DATE_SUB(%s, INTERVAL %s DAY))
+              AND send_time < UNIX_TIMESTAMP(%s)
+              AND type = %s
+              AND show_view_count > 0
+        """
+        return await self.pool.fetch_one(
+            query=query,
+            params=(
+                gh_id, item_index, publish_time,
+                self.AVG_WINDOW_DAYS, publish_time, self.MASS_SEND_TYPE,
+            ),
+        )
+
+    async def update_quantile_result(
+        self,
+        item_id: int,
+        *,
+        avg_view_count_60d: float | None = None,
+        view_ratio: float | None = None,
+        like_ratio: float | None = None,
+        sample_count_60d: int = 0,
+        status: int,
+    ) -> int:
+        query = f"""
+            UPDATE {self.QUANTILE_TABLE}
+            SET avg_view_count_60d = %s,
+                view_ratio = %s,
+                like_ratio = %s,
+                sample_count_60d = %s,
+                status = %s
+            WHERE id = %s AND status = %s
+        """
+        return await self.pool.save(
+            query=query,
+            params=(
+                avg_view_count_60d, view_ratio, like_ratio,
+                sample_count_60d, status, item_id, self.Status.PROCESSING,
+            ),
+        )
+
+    # ═══════════════════════════════════════════════════════════════
+    # title 表（标题粒度，按 title_md5 去重）
+    # ═══════════════════════════════════════════════════════════════
+
+    async def seed_titles(self, *, limit: int = 500) -> int:
+        """relation → 新标题 INSERT title 表"""
+        rows = await self.fetch_relation_titles_not_in_table(self.TITLE_TABLE, limit=limit)
+        if not rows:
+            return 0
+        query = f"""
+            INSERT IGNORE INTO {self.TITLE_TABLE} (title_md5, title)
+            VALUES (%s, %s)
+        """
+        params = [(r["title_md5"], r["title"]) for r in rows]
+        return await self.pool.save(query=query, params=params, batch=True)
+
+    async def fetch_pending_titles(self, *, limit: int = 100) -> List[Dict]:
+        query = f"""
+            SELECT id, title_md5, title
+            FROM {self.TITLE_TABLE}
+            WHERE status = %s
+            ORDER BY id ASC
+            LIMIT %s
+        """
+        return await self.pool.fetch(
+            query=query,
+            params=(self.Status.INIT, limit),
+        )
+
+    async def mark_title_processing(self, ids: List[int]) -> int:
+        if not ids:
+            return 0
+        placeholders = ",".join(["%s"] * len(ids))
+        query = f"""
+            UPDATE {self.TITLE_TABLE}
+            SET status = %s
+            WHERE id IN ({placeholders}) AND status = %s
+        """
+        return await self.pool.save(
+            query=query,
+            params=(self.Status.PROCESSING, *ids, self.Status.INIT),
+        )
+
+    async def batch_update_title_score(self, updates: List[Dict]) -> int:
+        if not updates:
+            return 0
+        total = 0
+        for u in updates:
+            query = f"""
+                UPDATE {self.TITLE_TABLE}
+                SET score = %s, status = %s
+                WHERE id = %s AND status = %s
+            """
+            n = await self.pool.save(
+                query=query,
+                params=(u["score"], self.Status.DONE, u["id"], self.Status.PROCESSING),
+            )
+            total += n
+        return total
+
+    async def mark_title_batch_failed(self, ids: List[int]) -> int:
+        if not ids:
+            return 0
+        placeholders = ",".join(["%s"] * len(ids))
+        query = f"""
+            UPDATE {self.TITLE_TABLE}
+            SET status = %s
+            WHERE id IN ({placeholders}) AND status = %s
+        """
+        return await self.pool.save(
+            query=query,
+            params=(self.Status.FAIL, *ids, self.Status.PROCESSING),
+        )
+
+    # ═══════════════════════════════════════════════════════════════
+    # category 表（标题粒度，按 title_md5 去重）
+    # ═══════════════════════════════════════════════════════════════
+
+    async def seed_categories(self, *, limit: int = 500) -> int:
+        rows = await self.fetch_relation_titles_not_in_table(
+            self.CATEGORY_TABLE, limit=limit,
+        )
+        if not rows:
+            return 0
+        query = f"""
+            INSERT IGNORE INTO {self.CATEGORY_TABLE} (title_md5, title)
+            VALUES (%s, %s)
+        """
+        params = [(r["title_md5"], r["title"]) for r in rows]
+        return await self.pool.save(query=query, params=params, batch=True)
+
+    async def fetch_pending_categories(self, *, limit: int = 100) -> List[Dict]:
+        query = f"""
+            SELECT id, title_md5, title
+            FROM {self.CATEGORY_TABLE}
+            WHERE status = %s
+            ORDER BY id ASC
+            LIMIT %s
+        """
+        return await self.pool.fetch(
+            query=query,
+            params=(self.Status.INIT, limit),
+        )
+
+    async def mark_category_processing(self, ids: List[int]) -> int:
+        if not ids:
+            return 0
+        placeholders = ",".join(["%s"] * len(ids))
+        query = f"""
+            UPDATE {self.CATEGORY_TABLE}
+            SET status = %s
+            WHERE id IN ({placeholders}) AND status = %s
+        """
+        return await self.pool.save(
+            query=query,
+            params=(self.Status.PROCESSING, *ids, self.Status.INIT),
+        )
+
+    async def batch_update_category(self, updates: List[Dict]) -> int:
+        if not updates:
+            return 0
+        total = 0
+        for u in updates:
+            query = f"""
+                UPDATE {self.CATEGORY_TABLE}
+                SET category = %s, status = %s
+                WHERE id = %s AND status = %s
+            """
+            n = await self.pool.save(
+                query=query,
+                params=(u["category"], self.Status.DONE, u["id"], self.Status.PROCESSING),
+            )
+            total += n
+        return total
+
+    async def mark_category_batch_failed(self, ids: List[int]) -> int:
+        if not ids:
+            return 0
+        placeholders = ",".join(["%s"] * len(ids))
+        query = f"""
+            UPDATE {self.CATEGORY_TABLE}
+            SET status = %s
+            WHERE id IN ({placeholders}) AND status = %s
+        """
+        return await self.pool.save(
+            query=query,
+            params=(self.Status.FAIL, *ids, self.Status.PROCESSING),
+        )
+
+
+__all__ = ["QualityMapper"]

src/domains/content_quality/_utils.py

@@ -0,0 +1,106 @@
+"""内容质量评估工具函数 —— Prompt 构建 + LLM 响应解析"""
+
+import hashlib
+import re
+import logging
+from typing import Dict, List
+
+from src.infra.shared.tools import safe_json_parse
+
+from ._const import ContentQualityConst
+
+logger = logging.getLogger(__name__)
+
+
+def md5(text: str) -> str:
+    return hashlib.md5(text.encode("utf-8")).hexdigest()
+
+
+# ═══════════════════════════════════════════════════════════════
+# Prompt 构建
+# ═══════════════════════════════════════════════════════════════
+
+
+def build_title_quality_prompt(titles: List[Dict]) -> str:
+    """构造标题质量评分 prompt，每行一个标题"""
+    lines = [t["title"] for t in titles]
+    return ContentQualityConst.PROMPT_TITLE_QUALITY.format(
+        title_list="\n".join(lines),
+    )
+
+
+def build_category_prompt(titles: List[Dict]) -> str:
+    """构造品类识别 prompt，格式: [(id, title), ...] 元组列表"""
+    tuples = [(t["id"], t["title"]) for t in titles]
+    return ContentQualityConst.PROMPT_TITLE_CATEGORY.format(
+        title_list=str(tuples),
+    )
+
+
+# ═══════════════════════════════════════════════════════════════
+# LLM 响应解析
+# ═══════════════════════════════════════════════════════════════
+
+
+def parse_title_quality_response(
+    raw: str | None,
+    batch_titles: List[Dict],
+) -> List[Dict]:
+    """解析 prompt1 返回值：LLM 按行输出纯数字分数，按顺序对应输入标题
+
+    返回: [{"id": 1, "score": 85}, ...]
+    """
+    result: List[Dict] = []
+    if not raw:
+        logger.warning("LLM 标题质量返回为空")
+        return result
+
+    # 从文本中提取所有数字
+    scores = re.findall(r"\d+", raw)
+    if not scores:
+        logger.warning("LLM 标题质量未提取到分数: %s", raw[:200])
+        return result
+
+    for i, s in enumerate(scores):
+        if i >= len(batch_titles):
+            break
+        score = int(s)
+        if 0 <= score <= 100:
+            result.append({"id": batch_titles[i]["id"], "score": score})
+
+    return result
+
+
+def parse_category_response(
+    raw: str | None,
+    batch_titles: List[Dict],
+) -> List[Dict]:
+    """解析 prompt2 返回值：{id: category} JSON 对象
+
+    返回: [{"id": 1, "category": "知识科普"}, ...]
+    """
+    result: List[Dict] = []
+    if not raw:
+        logger.warning("LLM 品类识别返回为空")
+        return result
+
+    parsed = safe_json_parse(raw)
+    if not isinstance(parsed, dict):
+        logger.warning("LLM 品类识别返回非对象: %s", raw[:200])
+        return result
+
+    for t in batch_titles:
+        category = parsed.get(str(t["id"]))
+        if category:
+            result.append({"id": t["id"], "category": str(category)[:32]})
+
+    return result
+
+
+__all__ = [
+    "md5",
+    "build_title_quality_prompt",
+    "build_category_prompt",
+    "parse_title_quality_response",
+    "parse_category_response",
+]

src/domains/content_quality/article_quality_category.py

@@ -0,0 +1,80 @@
+"""品类识别 Service —— 种子同步 + 批量 LLM 识别"""
+
+import asyncio
+import logging
+
+from tqdm import tqdm
+
+from src.infra.database.mysql.manager import MysqlManager
+from src.infra.external.volcengine import fetch_deepseek_completion
+
+from ._const import ContentQualityConst
+from ._mapper import QualityMapper
+from ._utils import build_category_prompt, parse_category_response
+
+logger = logging.getLogger(__name__)
+
+
+class ArticleQualityCategory(ContentQualityConst):
+
+    def __init__(self, pool: MysqlManager):
+        self.mapper = QualityMapper(pool)
+
+    async def deal(self, *, limit: int | None = None) -> dict:
+        _limit = limit if limit is not None else self.CATEGORY_DEAL_LIMIT
+
+        # Step 0: 种子同步
+        seeded = await self.mapper.seed_categories(limit=_limit)
+        logger.info("品类种子同步: seeded=%d", seeded)
+
+        # Step 1: LLM 识别
+        rows = await self.mapper.fetch_pending_categories(limit=_limit)
+        if not rows:
+            logger.info("无待识别品类")
+            return {"seeded": seeded, "evaluated": 0}
+
+        evaluated = 0
+        with tqdm(total=len(rows), desc="品类识别", unit="条") as pbar:
+            for start in range(0, len(rows), self.LLM_BATCH_SIZE):
+                batch = rows[start : start + self.LLM_BATCH_SIZE]
+                batch_ids = [r["id"] for r in batch]
+
+                affected = await self.mapper.mark_category_processing(batch_ids)
+                if affected == 0:
+                    pbar.update(len(batch))
+                    continue
+
+                titles = [{"id": r["id"], "title": r["title"]} for r in batch]
+                prompt = build_category_prompt(titles)
+
+                try:
+                    raw = await self._call_llm(prompt)
+                    if raw is None:
+                        raise RuntimeError("LLM 返回 None")
+
+                    parsed = parse_category_response(raw, titles)
+                    if not parsed:
+                        raise RuntimeError("LLM 响应解析为空")
+
+                    await self.mapper.batch_update_category(parsed)
+                    evaluated += len(parsed)
+
+                except Exception:
+                    logger.exception("品类识别失败 batch_start=%d", start)
+                    await self.mapper.mark_category_batch_failed(batch_ids)
+
+                pbar.update(len(batch))
+                pbar.set_postfix(evaluated=evaluated)
+
+                if start + self.LLM_BATCH_SIZE < len(rows):
+                    await asyncio.sleep(self.LLM_INTERVAL_SEC)
+
+        logger.info("品类识别完成: seeded=%d, evaluated=%d", seeded, evaluated)
+        return {"seeded": seeded, "evaluated": evaluated}
+
+    async def _call_llm(self, prompt: str) -> str | None:
+        loop = asyncio.get_running_loop()
+        return await loop.run_in_executor(None, fetch_deepseek_completion, prompt, "text")
+
+
+__all__ = ["ArticleQualityCategory"]

src/domains/content_quality/article_quality_quantify.py

@@ -0,0 +1,190 @@
+"""量化评估 Service —— 前置校验 + 入库 + 量化计算"""
+
+import logging
+
+from tqdm import tqdm
+
+from src.infra.database.mysql.manager import MysqlManager
+
+from ._const import ContentQualityConst
+from ._mapper import QualityMapper
+
+logger = logging.getLogger(__name__)
+
+
+class ArticleQualityQuantify(ContentQualityConst):
+
+    def __init__(self, pool: MysqlManager):
+        self.mapper = QualityMapper(pool)
+
+    async def deal(self, *, limit: int | None = None) -> dict:
+        _limit = limit if limit is not None else self.QUANTIFY_DEAL_LIMIT
+
+        # 取已完成深翻的账号，挨个处理
+        account_ids = await self.mapper.fetch_crawled_account_ids()
+        if not account_ids:
+            logger.info("无已完成深翻的账号")
+            return {"validated": 0, "skipped": 0, "inserted": 0, "computed": 0, "errors": 0}
+        logger.info("已深翻账号数: %d", len(account_ids))
+
+        total_validated = 0
+        total_skipped = 0
+        total_inserted = 0
+        total_computed = 0
+        total_errors = 0
+
+        with tqdm(total=len(account_ids), desc="账号校验", unit="acc") as pbar:
+            for account_id in account_ids:
+                v = await self._validate_and_insert(account_id, limit=_limit)
+                total_validated += v["validated"]
+                total_skipped += v["skipped"]
+                total_inserted += v["inserted"]
+                total_errors += v["errors"]
+                pbar.update(1)
+                pbar.set_postfix(inserted=total_inserted, skipped=total_skipped)
+
+        c = await self._compute(limit=_limit)
+        total_computed = c["computed"]
+        total_errors += c["errors"]
+
+        logger.info(
+            "量化评估完成: validated=%d skipped=%d inserted=%d computed=%d errors=%d",
+            total_validated, total_skipped, total_inserted, total_computed, total_errors,
+        )
+        return {
+            "validated": total_validated,
+            "skipped": total_skipped,
+            "inserted": total_inserted,
+            "computed": total_computed,
+            "errors": total_errors,
+        }
+
+    async def _validate_and_insert(self, account_id: str, *, limit: int) -> dict:
+        """处理单个账号的待校验文章"""
+        rows = await self.mapper.fetch_pending_detail_by_account(account_id, limit=limit)
+        if not rows:
+            return {"validated": 0, "skipped": 0, "inserted": 0, "errors": 0}
+
+        skipped = 0
+        inserted = 0
+        errors = 0
+
+        for row in rows:
+            detail_id = row["id"]
+            wx_sn = row["wx_sn"]
+
+            affected = await self.mapper.mark_detail_processing([detail_id])
+            if affected == 0:
+                continue
+
+            try:
+                pos = await self.mapper.get_article_position(wx_sn)
+                if not pos:
+                    await self.mapper.revert_detail_init(detail_id)
+                    skipped += 1
+                    continue
+
+                article_type = pos.get("type", 0)
+                item_index = pos.get("item_index", 0)
+
+                if article_type != self.MASS_SEND_TYPE or item_index not in self.VALID_POSITIONS:
+                    await self.mapper.mark_detail_failed(
+                        detail_id,
+                        f"type={article_type}, pos={item_index}",
+                    )
+                    skipped += 1
+                    continue
+
+                await self.mapper.insert_quantile({
+                    "wx_sn": wx_sn,
+                    "view_count": row.get("view_count") or 0,
+                    "like_count": row.get("like_count") or 0,
+                })
+                await self.mapper.mark_detail_done(detail_id)
+                inserted += 1
+
+            except Exception:
+                logger.exception("校验失败 detail_id=%d wx_sn=%s", detail_id, wx_sn)
+                try:
+                    await self.mapper.mark_detail_failed(detail_id, "异常")
+                except Exception:
+                    pass
+                errors += 1
+
+        return {"validated": len(rows), "skipped": skipped, "inserted": inserted, "errors": errors}
+
+    async def _compute(self, *, limit: int) -> dict:
+        rows = await self.mapper.fetch_pending_quantile(limit=limit)
+        if not rows:
+            return {"computed": 0, "errors": 0}
+
+        computed = 0
+        errors = 0
+
+        with tqdm(total=len(rows), desc="量化计算", unit="条") as pbar:
+            for row in rows:
+                item_id = row["id"]
+                wx_sn = row["wx_sn"]
+                view_count = row.get("view_count") or 0
+                like_count = row.get("like_count") or 0
+
+                affected = await self.mapper.mark_quantile_processing([item_id])
+                if affected == 0:
+                    pbar.update(1)
+                    continue
+
+                try:
+                    pos = await self.mapper.get_article_position(wx_sn)
+                    if not pos:
+                        await self.mapper.revert_quantile_init(item_id)
+                        pbar.update(1)
+                        continue
+
+                    gh_id = (pos.get("gh_id") or "").strip()
+                    item_index = pos.get("item_index", 0)
+
+                    avg_row = await self.mapper.get_avg_view_count_60d(
+                        gh_id, item_index, row.get("publish_at"),
+                    )
+
+                    avg_count = None
+                    sample_count = 0
+                    view_ratio = None
+
+                    if avg_row and avg_row.get("avg_count") is not None:
+                        avg_count = float(avg_row["avg_count"])
+                        sample_count = int(avg_row.get("sample_count", 0))
+                        if avg_count > 0:
+                            view_ratio = round(view_count / avg_count, 4)
+
+                    like_ratio = None
+                    if view_count > 0:
+                        like_ratio = round(like_count / view_count, 4)
+
+                    await self.mapper.update_quantile_result(
+                        item_id,
+                        avg_view_count_60d=avg_count,
+                        view_ratio=view_ratio,
+                        like_ratio=like_ratio,
+                        sample_count_60d=sample_count,
+                        status=self.Status.DONE,
+                    )
+                    computed += 1
+
+                except Exception:
+                    logger.exception("量化计算失败 id=%d wx_sn=%s", item_id, wx_sn)
+                    try:
+                        await self.mapper.update_quantile_result(
+                            item_id, status=self.Status.FAIL,
+                        )
+                    except Exception:
+                        pass
+                    errors += 1
+
+                pbar.update(1)
+                pbar.set_postfix(computed=computed, errors=errors)
+
+        return {"computed": computed, "errors": errors}
+
+
+__all__ = ["ArticleQualityQuantify"]

src/domains/content_quality/article_quality_title.py

@@ -0,0 +1,86 @@
+"""标题质量评分 Service —— 种子同步 + 批量 LLM 评分"""
+
+import asyncio
+import logging
+
+from tqdm import tqdm
+
+from src.infra.database.mysql.manager import MysqlManager
+from src.infra.external.volcengine import fetch_deepseek_completion
+
+from ._const import ContentQualityConst
+from ._mapper import QualityMapper
+from ._utils import build_title_quality_prompt, parse_title_quality_response
+
+logger = logging.getLogger(__name__)
+
+
+class ArticleQualityTitle(ContentQualityConst):
+
+    def __init__(self, pool: MysqlManager):
+        self.mapper = QualityMapper(pool)
+
+    async def deal(self, *, limit: int | None = None) -> dict:
+        _limit = limit if limit is not None else self.TITLE_DEAL_LIMIT
+
+        # Step 0: 种子同步
+        seeded = await self.mapper.seed_titles(limit=_limit)
+        logger.info("标题种子同步: seeded=%d", seeded)
+
+        # Step 1: LLM 评分
+        rows = await self.mapper.fetch_pending_titles(limit=_limit)
+        if not rows:
+            logger.info("无待评分标题")
+            return {"seeded": seeded, "evaluated": 0}
+
+        evaluated = 0
+        batches = (len(rows) + self.LLM_BATCH_SIZE - 1) // self.LLM_BATCH_SIZE
+        with tqdm(total=len(rows), desc="标题评分", unit="条") as pbar:
+            for start in range(0, len(rows), self.LLM_BATCH_SIZE):
+                batch = rows[start : start + self.LLM_BATCH_SIZE]
+                batch_ids = [r["id"] for r in batch]
+
+                affected = await self.mapper.mark_title_processing(batch_ids)
+                if affected == 0:
+                    pbar.update(len(batch))
+                    continue
+
+                titles = [{"id": r["id"], "title": r["title"]} for r in batch]
+                prompt = build_title_quality_prompt(titles)
+
+                try:
+                    raw = await self._call_llm(prompt)
+                    if raw is None:
+                        raise RuntimeError("LLM 返回 None")
+
+                    parsed = parse_title_quality_response(raw, titles)
+                    if not parsed:
+                        raise RuntimeError("LLM 响应解析为空")
+
+                    await self.mapper.batch_update_title_score(parsed)
+                    evaluated += len(parsed)
+
+                except Exception:
+                    logger.exception("标题评分失败 batch_start=%d", start)
+                    await self.mapper.mark_title_batch_failed(batch_ids)
+
+                pbar.update(len(batch))
+                pbar.set_postfix(evaluated=evaluated)
+
+                if start + self.LLM_BATCH_SIZE < len(rows):
+                    await asyncio.sleep(self.LLM_INTERVAL_SEC)
+
+        # 回写 relation.quality_score，让详情/账号拉取按质量分排序
+        if evaluated > 0:
+            backfilled = await self.mapper.backfill_relation_quality_score()
+            logger.info("relation 质量分回写: %d 行", backfilled)
+
+        logger.info("标题评分完成: seeded=%d, evaluated=%d", seeded, evaluated)
+        return {"seeded": seeded, "evaluated": evaluated}
+
+    async def _call_llm(self, prompt: str) -> str | None:
+        loop = asyncio.get_running_loop()
+        return await loop.run_in_executor(None, fetch_deepseek_completion, prompt, "text")
+
+
+__all__ = ["ArticleQualityTitle"]

src/domains/content_supply/_mapper.py

@@ -103,16 +103,16 @@ class ArticleSearchRelationMapper(ContentSupplyConst):
 
         items 每条须包含:
           search_key, key_type, experiment_id, search_cursor,
-          url, wx_sn, biz, title, cover_url, publish_time
+          url, wx_sn, biz, title, title_md5, cover_url, publish_time
         """
         if not items:
             return 0
         query = f"""
             INSERT IGNORE INTO {self.RELATION_TABLE}
                 (search_key, key_type, experiment_id, search_cursor,
-                 url, wx_sn, biz, title, cover_url, publish_time)
+                 url, wx_sn, biz, title, title_md5, cover_url, publish_time)
             VALUES
-                (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
+                (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
         """
         params = [
             (
@@ -124,6 +124,7 @@ class ArticleSearchRelationMapper(ContentSupplyConst):
                 it["wx_sn"],
                 it["biz"],
                 it["title"],
+                it.get("title_md5", ""),
                 it["cover_url"],
                 it["publish_time"],
             )
@@ -140,12 +141,12 @@ class ArticleSearchRelationMapper(ContentSupplyConst):
         *,
         limit: int = 500,
     ) -> List[Dict]:
-        """查询待拉详情的行 (status=0)"""
+        """查询待拉详情的行 (status=0)，按标题质量分逆序"""
         query = f"""
             SELECT *
             FROM {self.RELATION_TABLE}
             WHERE status = %s
-            ORDER BY id ASC
+            ORDER BY quality_score DESC, id ASC
             LIMIT %s
         """
         params = (self.SearchStatus.INIT, limit)
@@ -332,13 +333,13 @@ class AccountMapper(ContentSupplyConst):
     # ═══════════════════════════════════════════════════════════════
 
     async def fetch_pending_relations(self, *, limit: int = 100) -> List[Dict]:
-        """取 save_account_status=0 的行（DISTINCT biz，按 id ASC）"""
+        """取 save_account_status=0 的行（DISTINCT biz，按最高质量分逆序）"""
         query = f"""
-            SELECT MIN(id) AS id, biz, MIN(url) AS url
+            SELECT MIN(id) AS id, biz, MIN(url) AS url, MAX(quality_score) AS max_score
             FROM {self.RELATION_TABLE}
             WHERE save_account_status = %s AND biz IS NOT NULL AND biz != ''
             GROUP BY biz
-            ORDER BY id ASC
+            ORDER BY max_score DESC, id ASC
             LIMIT %s
         """
         return await self.pool.fetch(

src/domains/content_supply/_utils.py

@@ -1,5 +1,6 @@
 """供给域工具函数 —— API 响应解析、数据转换、搜索/账号/文章通用工具"""
 
+import hashlib
 import logging
 import re
 from datetime import datetime
@@ -10,6 +11,15 @@ from src.infra.spider.wechat.gzh import get_article_list_from_account
 
 logger = logging.getLogger(__name__)
 
+_EM_PATTERN = re.compile(r"</?em[^>]*>")
+
+
+def _clean_title(title: str) -> str:
+    """去掉微信搜索返回标题中的 <em class="highlight"> 和 </em> 标签"""
+    if not title:
+        return ""
+    return _EM_PATTERN.sub("", title)
+
 
 # ═══════════════════════════════════════════════════════════════
 # URL 参数提取（通用）
@@ -74,6 +84,7 @@ def parse_search_result(
     ts = article.get("time")
     publish_time = datetime.fromtimestamp(ts) if ts and ts > 0 else None
     article_url = article.get("url", "")
+    title = _clean_title(article.get("title", ""))
     return {
         "search_key": keyword,
         "key_type": key_type,
@@ -82,7 +93,8 @@ def parse_search_result(
         "url": article_url,
         "wx_sn": extract_wx_sn(article_url) or "",
         "biz": extract_biz(article_url) or "",
-        "title": article.get("title", ""),
+        "title": title,
+        "title_md5": hashlib.md5(title.encode("utf-8")).hexdigest() if title else "",
         "cover_url": article.get("cover_url", ""),
         "publish_time": publish_time,
     }

src/handlers/article_quality_category.py

@@ -0,0 +1,39 @@
+"""品类识别 XXL-JOB Handler
+
+XXL-JOB Admin:
+    执行器: long-articles-agentic-src
+    处理器: articleQualityCategory
+    参数:   limit=100
+    超时:   建议 ≥600s
+"""
+
+import logging
+
+from src.infra.xxl_jobs import xxl_job
+from src.domains.content_quality import ArticleQualityCategory, ContentQualityConst
+from src.handlers._utils import get_db, parse_limit
+
+logger = logging.getLogger(__name__)
+
+
+@xxl_job("articleQualityCategory")
+async def article_quality_category(param: str) -> dict:
+    limit = parse_limit(param, ContentQualityConst.CATEGORY_DEAL_LIMIT)
+    logger.info("articleQualityCategory started, limit=%d", limit)
+
+    try:
+        mysql = get_db().mysql_manager("long_articles")
+        service = ArticleQualityCategory(mysql)
+        result = await service.deal(limit=limit)
+    except Exception:
+        logger.exception("品类识别失败")
+        return {"code": 500, "msg": "异常"}
+
+    logger.info(
+        "done: seeded=%d evaluated=%d",
+        result["seeded"], result["evaluated"],
+    )
+    return {
+        "code": 200,
+        "msg": f"seeded={result['seeded']} evaluated={result['evaluated']}",
+    }

src/handlers/article_quality_quantify.py

@@ -0,0 +1,43 @@
+"""量化评估 XXL-JOB Handler
+
+XXL-JOB Admin:
+    执行器: long-articles-agentic-src
+    处理器: articleQualityQuantify
+    参数:   limit=500
+    超时:   建议 ≥300s
+"""
+
+import logging
+
+from src.infra.xxl_jobs import xxl_job
+from src.domains.content_quality import ArticleQualityQuantify, ContentQualityConst
+from src.handlers._utils import get_db, parse_limit
+
+logger = logging.getLogger(__name__)
+
+
+@xxl_job("articleQualityQuantify")
+async def article_quality_quantify(param: str) -> dict:
+    limit = parse_limit(param, ContentQualityConst.QUANTIFY_DEAL_LIMIT)
+    logger.info("articleQualityQuantify started, limit=%d", limit)
+
+    try:
+        mysql = get_db().mysql_manager("long_articles")
+        service = ArticleQualityQuantify(mysql)
+        result = await service.deal(limit=limit)
+    except Exception:
+        logger.exception("量化评估失败")
+        return {"code": 500, "msg": "异常"}
+
+    logger.info(
+        "done: validated=%d skipped=%d inserted=%d computed=%d errors=%d",
+        result["validated"], result["skipped"],
+        result["inserted"], result["computed"], result["errors"],
+    )
+    return {
+        "code": 200,
+        "msg": (
+            f"validated={result['validated']} skipped={result['skipped']} "
+            f"inserted={result['inserted']} computed={result['computed']} errors={result['errors']}"
+        ),
+    }

src/handlers/article_quality_title.py

@@ -0,0 +1,39 @@
+"""标题质量评分 XXL-JOB Handler
+
+XXL-JOB Admin:
+    执行器: long-articles-agentic-src
+    处理器: articleQualityTitle
+    参数:   limit=100
+    超时:   建议 ≥600s（100条≈5批×30s+延迟）
+"""
+
+import logging
+
+from src.infra.xxl_jobs import xxl_job
+from src.domains.content_quality import ArticleQualityTitle, ContentQualityConst
+from src.handlers._utils import get_db, parse_limit
+
+logger = logging.getLogger(__name__)
+
+
+@xxl_job("articleQualityTitle")
+async def article_quality_title(param: str) -> dict:
+    limit = parse_limit(param, ContentQualityConst.TITLE_DEAL_LIMIT)
+    logger.info("articleQualityTitle started, limit=%d", limit)
+
+    try:
+        mysql = get_db().mysql_manager("long_articles")
+        service = ArticleQualityTitle(mysql)
+        result = await service.deal(limit=limit)
+    except Exception:
+        logger.exception("标题评分失败")
+        return {"code": 500, "msg": "异常"}
+
+    logger.info(
+        "done: seeded=%d evaluated=%d",
+        result["seeded"], result["evaluated"],
+    )
+    return {
+        "code": 200,
+        "msg": f"seeded={result['seeded']} evaluated={result['evaluated']}",
+    }