Merge branch '2025-02-25-article-association-publish' of luojunhui/LongArticlesJob into master

applications/const/__init__.py

@@ -4,7 +4,7 @@
 """
 
 
-class coldStartTaskConst:
+class ColdStartTaskConst:
     """
     冷启动任务常量配置
     """
@@ -12,6 +12,44 @@ class coldStartTaskConst:
     INIT_STATUS = 1  # 文章初始状态
     BAD_STATUS = 0  # 低质量文章状态
 
+    # 常量
+    ACCOUNT_GOOD_STATUS = 1
+
+    # 账号是否每日抓取
+    ACCOUNT_DAILY_SCRAPE = 1
+    ACCOUNT_NOT_DAILY_SCRAPE = 0
+
+    # 默认值
+    DEFAULT_VIEW_COUNT = 0
+    DEFAULT_LIKE_COUNT = 0
+    DEFAULT_ARTICLE_STATUS = 1
+    DEFAULT_TIMESTAMP = 1717171200
+
+    # 标题sensitivity
+    TITLE_SENSITIVE = 1
+    TITLE_NOT_SENSITIVE = 0
+
+    # 文章联想深度
+    ARTICLE_ASSOCIATION_MAX_DEPTH = 4
+
+    # 相关分百分位阈值
+    PERCENT_THRESHOLD = 95
+
+    # 相关性分阈值
+    CORRELATION_THRESHOLD = 0.5
+
+    # 阅读量阈值
+    READ_COUNT_THRESHOLD = 1000
+
+    # 阅读均值倍数阈值
+    READ_AVG_THRESHOLD = 1.3
+
+    # 群发类型
+    BULK_PUBLISH_TYPE = 9
+
+    # 种子文章数量
+    SEED_ARTICLE_LIMIT_NUM = 30
+
 
 class updatePublishedMsgTaskConst:
     """

applications/utils/__init__.py

@@ -0,0 +1,4 @@
+"""
+utils
+"""
+from .cold_start import *

applications/utils/cold_start.py

@@ -0,0 +1,30 @@
+"""
+@author: luojunhui
+"""
+import json
+
+from applications import aiditApi
+from config import apolloConfig
+
+config = apolloConfig()
+sensitive_word_list = json.loads(config.getConfigValue("sensitive_word_list"))
+
+
+def whether_title_sensitive(title: str) -> bool:
+    """
+    : param title:
+    判断视频是否的标题是否包含敏感词
+    """
+    for word in sensitive_word_list:
+        if word in title:
+            return True
+    return False
+
+
+def get_inner_account_set() -> set:
+    """
+    get inner account set
+    """
+    accounts = aiditApi.get_publish_account_from_aigc()
+    gh_id_list = [i['ghId'] for i in accounts]
+    return set(gh_id_list)

applications/wxSpiderApi.py

@@ -1,9 +1,12 @@
 """
 @author: luojunhui
 """
+
 import json
+import time
 import requests
 
+from applications.aliyunLogApi import log
 from applications.decoratorApi import retryOnNone
 
 
@@ -11,13 +14,12 @@ class WeixinSpider(object):
     """
     Update account articles
     """
+
     # ip = "8.217.190.241"
     # ip = "47.98.154.124"
     # port = "8888"
     base_url = "http://crawler-cn.aiddit.com/crawler/wei_xin"
-    headers = {
-        "Content-Type": "application/json"
-    }
+    headers = {"Content-Type": "application/json"}
 
     @classmethod
     @retryOnNone()
@@ -27,11 +29,10 @@ class WeixinSpider(object):
         :return:
         """
         url = "{}/keyword".format(cls.base_url)
-        payload = json.dumps({
-            "keyword": title,
-            "cursor": page
-        })
-        response = requests.request("POST", url, headers=cls.headers, data=payload, timeout=120)
+        payload = json.dumps({"keyword": title, "cursor": page})
+        response = requests.request(
+            "POST", url, headers=cls.headers, data=payload, timeout=120
+        )
         return response.json()
 
     @classmethod
@@ -45,13 +46,17 @@ class WeixinSpider(object):
         :return:
         """
         url = "{}/detail".format(cls.base_url)
-        payload = json.dumps({
-            "content_link": content_link,
-            "is_count": is_count,
-            "is_ad": False,
-            "is_cache": is_cache
-        })
-        response = requests.request("POST", url, headers=cls.headers, data=payload, timeout=120)
+        payload = json.dumps(
+            {
+                "content_link": content_link,
+                "is_count": is_count,
+                "is_ad": False,
+                "is_cache": is_cache,
+            }
+        )
+        response = requests.request(
+            "POST", url, headers=cls.headers, data=payload, timeout=120
+        )
         return response.json()
 
     @classmethod
@@ -60,12 +65,14 @@ class WeixinSpider(object):
         """
         :return:
         """
-        url = '{}/blogger'.format(cls.base_url)
+        url = "{}/blogger".format(cls.base_url)
         payload = {
-            'account_id': ghId,
-            'cursor': index,
+            "account_id": ghId,
+            "cursor": index,
         }
-        response = requests.post(url=url, headers=cls.headers, data=json.dumps(payload), timeout=120)
+        response = requests.post(
+            url=url, headers=cls.headers, data=json.dumps(payload), timeout=120
+        )
         return response.json()
 
     @classmethod
@@ -76,9 +83,11 @@ class WeixinSpider(object):
         :param content_url:
         :return:
         """
-        url = '{}/account_info'.format(cls.base_url)
+        url = "{}/account_info".format(cls.base_url)
         data = {"content_link": content_url}
-        response = requests.request("POST", url=url, headers=cls.headers, json=data, timeout=120)
+        response = requests.request(
+            "POST", url=url, headers=cls.headers, json=data, timeout=120
+        )
         return response.json()
 
     @classmethod
@@ -89,8 +98,35 @@ class WeixinSpider(object):
         :return:
         """
         url = "{}/recommend".format(cls.base_url)
+        payload = json.dumps({"content_link": content_link})
+        response = requests.request(
+            "POST", url=url, headers=cls.headers, data=payload, timeout=120
+        )
+        response_json = response.json()
+        if response_json["code"] != 0:
+            return cls.get_recommend_articles(content_link)
+        time.sleep(3)
+        return response.json()
+
+    @classmethod
+    def get_recommend_articles_v2(cls, content_link) -> dict:
+        """
+        use content link to get recommend articles
+        :param content_link:
+        :return:
+        """
+        url = "http://datapi.top/wxapi/relatedarticle"
         payload = json.dumps(
-            {"content_link": content_link}
+            {"content_link": content_link, "token": "401e4d3c85068bb5"}
+        )
+        response = requests.request(
+            "POST", url=url, headers=cls.headers, data=payload, timeout=120
+        )
+        log(
+            task="article_association_crawler",
+            function="get_recommend_articles_v2",
+            message="获取推荐链接，付费接口",
+            data={"content_link": content_link, "response": response.json()},
         )
-        response = requests.request("POST", url=url, headers=cls.headers, data=payload, timeout=120)
+        time.sleep(3)
         return response.json()

article_association_task.py

@@ -0,0 +1,27 @@
+"""
+@author: luojunhui
+"""
+from argparse import ArgumentParser
+
+from coldStartTasks.crawler.wechat import ArticleAssociationCrawler
+
+
+def main():
+    """
+    main function
+    """
+    parser = ArgumentParser()
+    parser.add_argument("--biz_date", type=str, help="format 2025-01-01")
+    args = parser.parse_args()
+
+    if args.biz_date:
+        biz_date = args.biz_date
+    else:
+        biz_date = None
+
+    article_association_crawler = ArticleAssociationCrawler()
+    article_association_crawler.deal(biz_date=biz_date)
+
+
+if __name__ == "__main__":
+    main()

coldStartTasks/crawler/wechat/__init__.py

@@ -0,0 +1,4 @@
+"""
+@author: luojunhui
+"""
+from .article_association import ArticleAssociationCrawler

coldStartTasks/crawler/wechat/article_association.py

@@ -0,0 +1,210 @@
+"""
+@author: luojunhui
+"""
+
+import time
+import traceback
+from datetime import datetime
+
+import numpy as np
+
+from pymysql.cursors import DictCursor
+from tqdm import tqdm
+
+
+from applications import WeixinSpider, log
+from applications.api import similarity_between_title_list
+from applications.const import ColdStartTaskConst
+from applications.db import DatabaseConnector
+from applications.functions import Functions
+from applications.utils import get_inner_account_set
+from applications.utils import whether_title_sensitive
+from config import long_articles_config
+
+spider = WeixinSpider()
+functions = Functions()
+const = ColdStartTaskConst()
+
+
+class ArticleAssociationCrawler(object):
+    """
+    article association crawler task
+    """
+
+    def __init__(self):
+        self.db_client = DatabaseConnector(db_config=long_articles_config)
+        self.db_client.connect()
+        self.inner_account_set = get_inner_account_set()
+
+    def get_seed_url_list(self, biz_date):
+        """
+        获取种子url列表
+        """
+        sql = f"""
+            select gh_id, title, link
+            from datastat_sort_strategy
+            where date_str > DATE_FORMAT(DATE_SUB('{biz_date}', INTERVAL 2 DAY), '%Y%m%d') 
+                and view_count > {const.READ_COUNT_THRESHOLD} 
+                and read_rate > {const.READ_AVG_THRESHOLD} 
+                and type = {const.BULK_PUBLISH_TYPE}
+            order by read_rate desc 
+            limit {const.SEED_ARTICLE_LIMIT_NUM};
+        """
+        seed_article_list = self.db_client.fetch(query=sql, cursor_type=DictCursor)
+        return seed_article_list
+
+    def get_level_up_title_list(self):
+        """
+        获取晋级文章标题列表
+        status: 1 表示文章已经溯源完成
+        deleted: 0 表示文章正常
+        level = 'autoArticlePoolLevel1' 表示头条
+        """
+        sql = f"""
+            select distinct title 
+            from article_pool_promotion_source 
+            where level = 'autoArticlePoolLevel1' and status = 1 and deleted = 0;
+        """
+        mysql_response = self.db_client.fetch(query=sql)
+        title_list = [i[0] for i in mysql_response]
+        return title_list
+
+    def get_recommend_url_list_with_depth(
+        self, seed_url, source_title, source_account, base_title_list, depth=1
+    ):
+        """
+        @param seed_url: good url from data_sort_strategy
+        @param depth: association depth
+        @param source_title: article title
+        @param source_account: article account
+        """
+        if depth > const.ARTICLE_ASSOCIATION_MAX_DEPTH:
+            return
+
+        res = spider.get_recommend_articles(content_link=seed_url)
+        related_articles = res["data"]["data"]["list"]
+        if related_articles:
+            title_list = [i["title"] for i in related_articles]
+            similarity_array = similarity_between_title_list(
+                title_list, base_title_list
+            )
+
+            recommend_articles = []
+            for index, score_list in enumerate(similarity_array):
+                sorted_score_list = sorted(score_list)
+                percent_threshold_score = np.percentile(
+                    sorted_score_list, const.PERCENT_THRESHOLD
+                )
+                if percent_threshold_score < const.CORRELATION_THRESHOLD:
+                    continue
+
+                else:
+                    article_obj = related_articles[index]
+                    article_obj["score"] = percent_threshold_score
+                    recommend_articles.append(article_obj)
+
+            recommend_process_bar = tqdm(
+                recommend_articles, desc="save recommend articles"
+            )
+            for article in recommend_process_bar:
+                obj = {
+                    "title": article["title"],
+                    "url": article["url"],
+                    "gh_id": article["username"],
+                    "index": article["idx"],
+                    "send_time": article["send_time"],
+                    "read_cnt": article["read_num"],
+                    "depth": depth,
+                    "source_article_title": source_title,
+                    "source_account": source_account,
+                }
+                self.insert_recommend_article(obj)
+                recommend_process_bar.set_postfix(
+                    {"title": article["title"], "depth": depth}
+                )
+                self.get_recommend_url_list_with_depth(
+                    seed_url=obj["url"],
+                    source_title=obj["title"],
+                    source_account=obj["gh_id"],
+                    base_title_list=base_title_list,
+                    depth=depth + 1,
+                )
+        else:
+            return
+
+    def insert_recommend_article(self, obj):
+        """
+        insert recommend article
+        """
+        # whether account inside
+        if obj["gh_id"] in self.inner_account_set:
+            return
+
+        # whether article title exists
+        title = obj["title"]
+        select_sql = "select article_id from crawler_meta_article where title = %s;"
+        res = self.db_client.fetch(query=select_sql, params=(title,))
+        if res:
+            return
+
+        # whether title sensitive
+        title_sensitivity = (
+            const.TITLE_SENSITIVE
+            if whether_title_sensitive(title)
+            else const.TITLE_NOT_SENSITIVE
+        )
+
+        # insert this article
+        insert_sql = f"""
+            insert into crawler_meta_article 
+            (platform, mode, category, out_account_id, article_index, title, link, read_cnt, publish_time, crawler_time, status, unique_index, source_article_title, source_account, title_sensitivity)
+            values (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s);
+        """
+        self.db_client.save(
+            query=insert_sql,
+            params=(
+                "weixin",
+                "recommend",
+                "article_association",
+                obj["gh_id"],
+                obj["index"],
+                obj["title"],
+                obj["url"],
+                obj["read_cnt"],
+                obj["send_time"],
+                int(time.time()),
+                const.DEFAULT_ARTICLE_STATUS,
+                functions.generateGzhId(obj["url"]),
+                obj["source_article_title"],
+                obj["source_account"],
+                title_sensitivity,
+            ),
+        )
+
+    def deal(self, biz_date=None):
+        """
+        class entrance
+        :param biz_date:
+        """
+        if biz_date is None:
+            biz_date = datetime.today().strftime("%Y-%m-%d")
+
+        seed_article_list = self.get_seed_url_list(biz_date)
+        deal_bar = tqdm(seed_article_list, desc="article association crawler")
+        base_title_list = self.get_level_up_title_list()
+        for article in deal_bar:
+            try:
+                self.get_recommend_url_list_with_depth(
+                    seed_url=article["link"],
+                    source_title=article["title"],
+                    source_account=article["gh_id"],
+                    base_title_list=base_title_list,
+                )
+                deal_bar.set_postfix({"article_title": article["title"]})
+            except Exception as e:
+                log(
+                    task="article_association_crawler",
+                    function="deal",
+                    message=f"article association crawler error, article title: {article['title']}, error: {e}",
+                    data={"article": article, "traceback": traceback.format_exc()},
+                )

coldStartTasks/publish/basic.py

@@ -0,0 +1,276 @@
+"""
+@author: luojunhui
+"""
+
+import json
+import time
+import datetime
+import pandas as pd
+import traceback
+
+from pandas import DataFrame
+from tqdm import tqdm
+
+from applications import log, aiditApi, bot
+from applications.const import ColdStartTaskConst
+from config import apolloConfig
+
+const = ColdStartTaskConst()
+config = apolloConfig()
+
+category_cold_start_threshold = json.loads(
+    config.getConfigValue("category_cold_start_threshold")
+)
+READ_TIMES_THRESHOLD = category_cold_start_threshold.get("READ_TIMES_THRESHOLD", 1.3)
+READ_THRESHOLD = category_cold_start_threshold.get("READ_THRESHOLD", 5000)
+LIMIT_TITLE_LENGTH = category_cold_start_threshold.get("LIMIT_TITLE_LENGTH", 15)
+TITLE_LENGTH_MAX = category_cold_start_threshold.get("TITLE_LENGTH_MAX", 50)
+
+
+def get_article_from_meta_table(db_client, category: str, platform: str) -> DataFrame:
+    """
+    get article from meta data
+    :param db_client: database connector
+    :param category: article category
+    :param platform: article platform
+    :return: article dataframe
+    """
+    sql = f"""
+        select 
+            article_id, out_account_id, article_index, title, link, read_cnt, status, llm_sensitivity, score
+        from crawler_meta_article
+        where category = "{category}" and platform = "{platform}" and title_sensitivity = {const.TITLE_NOT_SENSITIVE}
+        order by score desc;
+    """
+    article_list = db_client.fetch(sql)
+    log(
+        task="category_publish_task",
+        function="get_articles_from_meta_table",
+        message="获取品类文章总数",
+        data={"total_articles": len(article_list), "category": category},
+    )
+    article_df = pd.DataFrame(
+        article_list,
+        columns=[
+            "article_id",
+            "gh_id",
+            "position",
+            "title",
+            "link",
+            "read_cnt",
+            "status",
+            "llm_sensitivity",
+            "score",
+        ],
+    )
+    return article_df
+
+
+def update_published_articles_status(db_client) -> None:
+    """
+    filter published articles
+    """
+    category_map = json.loads(config.getConfigValue("category_cold_start_map"))
+    category_list = list(category_map.keys())
+    processing_bar = tqdm(category_list, desc="update_published_articles")
+    for category in processing_bar:
+        plan_id = category_map.get(category)
+        if plan_id:
+            article_list = aiditApi.get_generated_article_list(plan_id)
+            title_list = [i[1] for i in article_list]
+            if title_list:
+                update_sql = f"""
+                        update crawler_meta_article
+                        set status = %s 
+                        where title in %s and status = %s;
+                """
+                affected_rows = db_client.save(
+                    query=update_sql,
+                    params=(
+                        const.PUBLISHED_STATUS,
+                        tuple(title_list),
+                        const.INIT_STATUS,
+                    ),
+                )
+                processing_bar.set_postfix(
+                    {"category": category, "affected_rows": affected_rows}
+                )
+        else:
+            return
+
+
+def filter_by_read_times(article_df: DataFrame) -> DataFrame:
+    """
+    filter by read times
+    """
+    article_df["average_read"] = article_df.groupby(["gh_id", "position"])[
+        "read_cnt"
+    ].transform("mean")
+    article_df["read_times"] = article_df["read_cnt"] / article_df["average_read"]
+    filter_df = article_df[article_df["read_times"] >= READ_TIMES_THRESHOLD]
+    return filter_df
+
+
+def filter_by_status(article_df: DataFrame) -> DataFrame:
+    """
+    filter by status
+    """
+    filter_df = article_df[article_df["status"] == const.INIT_STATUS]
+    return filter_df
+
+
+def filter_by_read_cnt(article_df: DataFrame) -> DataFrame:
+    """
+    filter by read cnt
+    """
+    filter_df = article_df[article_df["read_cnt"] >= READ_THRESHOLD]
+    return filter_df
+
+
+def filter_by_title_length(article_df: DataFrame) -> DataFrame:
+    """
+    filter by title length
+    """
+    filter_df = article_df[
+        (article_df["title"].str.len() >= LIMIT_TITLE_LENGTH)
+        & (article_df["title"].str.len() <= TITLE_LENGTH_MAX)
+    ]
+    return filter_df
+
+
+def filter_by_sensitive_words(article_df: DataFrame) -> DataFrame:
+    """
+    filter by sensitive words
+    """
+    filter_df = article_df[
+        (~article_df["title"].str.contains("农历"))
+        & (~article_df["title"].str.contains("太极"))
+        & (~article_df["title"].str.contains("节"))
+        & (~article_df["title"].str.contains("早上好"))
+        & (~article_df["title"].str.contains("赖清德"))
+        & (~article_df["title"].str.contains("普京"))
+        & (~article_df["title"].str.contains("俄"))
+        & (~article_df["title"].str.contains("南海"))
+        & (~article_df["title"].str.contains("台海"))
+        & (~article_df["title"].str.contains("解放军"))
+        & (~article_df["title"].str.contains("蔡英文"))
+        & (~article_df["title"].str.contains("中国"))
+    ]
+    return filter_df
+
+
+def filter_by_similarity_score(article_df: DataFrame, score) -> DataFrame:
+    """
+    filter by similarity score
+    """
+    filter_df = article_df[article_df["score"] >= score]
+    return filter_df
+
+
+def insert_into_article_crawler_plan(
+    db_client, crawler_plan_id, crawler_plan_name, create_timestamp
+):
+    """
+    insert into article crawler plan
+    """
+    insert_sql = f"""
+        insert into article_crawler_plan (crawler_plan_id, name, create_timestamp)
+        values (%s, %s, %s);
+    """
+    try:
+        db_client.save(
+            query=insert_sql,
+            params=(crawler_plan_id, crawler_plan_name, create_timestamp),
+        )
+    except Exception as e:
+        bot(
+            title="品类冷启任务，记录抓取计划id失败",
+            detail={
+                "error": str(e),
+                "error_msg": traceback.format_exc(),
+                "crawler_plan_id": crawler_plan_id,
+                "crawler_plan_name": crawler_plan_name,
+            },
+        )
+
+
+def create_crawler_plan(url_list, plan_tag, platform) -> tuple:
+    """
+    create crawler plan
+    """
+    crawler_plan_response = aiditApi.auto_create_crawler_task(
+        plan_id=None,
+        plan_name="自动绑定-{}--{}--{}".format(
+            plan_tag, datetime.date.today().__str__(), len(url_list)
+        ),
+        plan_tag=plan_tag,
+        article_source=platform,
+        url_list=url_list,
+    )
+    log(
+        task="category_publish_task",
+        function="publish_filter_articles",
+        message="成功创建抓取计划",
+        data=crawler_plan_response,
+    )
+    # save to db
+    create_timestamp = int(time.time()) * 1000
+    crawler_plan_id = crawler_plan_response["data"]["id"]
+    crawler_plan_name = crawler_plan_response["data"]["name"]
+    return crawler_plan_id, crawler_plan_name, create_timestamp
+
+
+def bind_to_generate_plan(category, crawler_plan_id, crawler_plan_name, platform):
+    """
+    auto bind to generate plan
+    """
+    match platform:
+        case "weixin":
+            input_source_channel = 5
+        case "toutiao":
+            input_source_channel = 6
+        case _:
+            input_source_channel = 5
+
+    new_crawler_task_list = [
+        {
+            "contentType": 1,
+            "inputSourceType": 2,
+            "inputSourceSubType": None,
+            "fieldName": None,
+            "inputSourceValue": crawler_plan_id,
+            "inputSourceLabel": crawler_plan_name,
+            "inputSourceModal": 3,
+            "inputSourceChannel": input_source_channel,
+        }
+    ]
+    category_map = json.loads(config.getConfigValue("category_cold_start_map"))
+    generate_plan_response = aiditApi.bind_crawler_task_to_generate_task(
+        crawler_task_list=new_crawler_task_list, generate_task_id=category_map[category]
+    )
+    log(
+        task="category_publish_task",
+        function="publish_filter_articles",
+        message="成功绑定到生成计划",
+        data=generate_plan_response,
+    )
+
+
+def update_article_status_after_publishing(db_client, article_id_list):
+    """
+    update article status after publishing
+    """
+    update_sql = f"""
+        update crawler_meta_article
+        set status = %s
+        where article_id in %s and status = %s;
+    """
+    affect_rows = db_client.save(
+        query=update_sql,
+        params=(const.PUBLISHED_STATUS, tuple(article_id_list), const.INIT_STATUS),
+    )
+    if affect_rows != len(article_id_list):
+        bot(
+            title="品类冷启任务中，出现更新状文章状态失败异常",
+            detail={"affected_rows": affect_rows, "task_rows": len(article_id_list)},
+        )

coldStartTasks/publish/publishArticleAssociationArticles.py

@@ -1,4 +0,0 @@
-"""
-@author: luojunhui
-发布i2i文章
-"""

coldStartTasks/publish/publish_article_association_articles.py

@@ -0,0 +1,125 @@
+"""
+@author: luojunhui
+"""
+
+from pandas import DataFrame
+
+from applications import bot
+from applications.const import ColdStartTaskConst
+from applications.db import DatabaseConnector
+from config import long_articles_config
+
+from coldStartTasks.publish.basic import filter_by_status
+from coldStartTasks.publish.basic import filter_by_sensitive_words
+from coldStartTasks.publish.basic import filter_by_title_length
+from coldStartTasks.publish.basic import update_published_articles_status
+from coldStartTasks.publish.basic import get_article_from_meta_table
+from coldStartTasks.publish.basic import update_article_status_after_publishing
+from coldStartTasks.publish.basic import create_crawler_plan
+from coldStartTasks.publish.basic import insert_into_article_crawler_plan
+from coldStartTasks.publish.basic import bind_to_generate_plan
+
+const = ColdStartTaskConst()
+
+
+def filter_articles_before_create_plan(article_df: DataFrame) -> DataFrame:
+    """
+    filter articles before create plan
+    """
+    total_length = article_df.shape[0]
+
+    # filter by status
+    filter_df = filter_by_status(article_df)
+    filter_length0 = filter_df.shape[0]
+
+    # filter by sensitive words
+    filter_df = filter_by_sensitive_words(filter_df)
+    filter_length1 = filter_df.shape[0]
+
+    # filter by title length
+    filter_df = filter_by_title_length(filter_df)
+    filter_length2 = filter_df.shape[0]
+
+    bot(
+        title="文章联想任务，开始创建抓取计划",
+        detail={
+            "文章总数": total_length,
+            "发布状态过滤": "过滤： {}, 剩余： {}".format(
+                total_length - filter_length0, filter_length0
+            ),
+            "敏感词过滤": "过滤： {}, 剩余： {}".format(
+                filter_length0 - filter_length1, filter_length1
+            ),
+            "标题长度过滤": "过滤： {}, 剩余： {}".format(
+                filter_length1 - filter_length2, filter_length2
+            ),
+        },
+        mention=False,
+    )
+
+    return filter_df
+
+
+class ArticleAssociationPublish(object):
+    """
+    publish i2i articles
+    """
+
+    def __init__(self):
+        self.db_client = DatabaseConnector(db_config=long_articles_config)
+        self.db_client.connect()
+
+    def deal(self):
+        """
+        class entrance
+        """
+        # update published articles
+        update_published_articles_status(db_client=self.db_client)
+
+        # get data from meta table
+        article_dataframe = get_article_from_meta_table(
+            db_client=self.db_client, category="article_association", platform="weixin"
+        )
+
+        # fileter articles
+        filter_dataframe = filter_articles_before_create_plan(article_dataframe)
+
+        # create crawler plan
+        url_list = filter_dataframe["link"].values.tolist()
+        if url_list:
+            crawler_plan_id, crawler_plan_name, create_timestamp = create_crawler_plan(
+                url_list=url_list, plan_tag="article_association", platform="weixin"
+            )
+
+            # insert crawler plan
+            insert_into_article_crawler_plan(
+                db_client=self.db_client,
+                crawler_plan_id=crawler_plan_id,
+                crawler_plan_name=crawler_plan_name,
+                create_timestamp=create_timestamp,
+            )
+
+            # bind to generate plan
+            bind_to_generate_plan(
+                category="article_association",
+                crawler_plan_id=crawler_plan_id,
+                crawler_plan_name=crawler_plan_name,
+                platform="weixin",
+            )
+
+            # update status
+            article_id_list = filter_dataframe["article_id"].values.tolist()
+            update_article_status_after_publishing(
+                db_client=self.db_client, article_id_list=article_id_list
+            )
+
+            bot(
+                title="文章联想任务，创建抓取计划成功",
+                detail={
+                    "抓取计划id": crawler_plan_id,
+                    "抓取计划名称": crawler_plan_name,
+                    "抓取条数": len(url_list),
+                    "冷启动类型": "article_association",
+                },
+                mention=False,
+            )