LongArticlesJob/applications/utils/item.py

"""
@author: luojunhui
"""

import time

default_single_video_table_fields = {
    "platform": "gzh",
    "article_title": None,
    "content_trace_id": None,
    "read_cnt": 0,
    "article_index": None,
    "out_account_name": None,
    "article_url": None,
    "url_unique_md5": None,
    "category": None,
    "publish_timestamp": None,
    "out_account_id": None,
    "cover_url": None,
    "crawler_timestamp": int(time.time()),
    "source_account": 1,
    "article_publish_type": None,
    "like_cnt": 0,
    "bad_status": 0,
    "tags": None,
    "video_oss_path": None,
    "audit_status": 0,
    "category_status": 0,
    "audit_video_id": None,
    "mini_program_title": None
}


class Item(object):
    """
    format save to article meta table or single video source table
    """

    def __init__(self):
        self.item = {}

    def add(self, key, value):
        """
        add key value to item
        """
        self.item[key] = value

    def check_video_item(self):
        """
        check video item
        """
        fields = list(default_single_video_table_fields.keys())
        for field in fields:
            if self.item.get(field, None) is not None:
                continue
            else:
                self.item[field] = default_single_video_table_fields[field]

    def check_article_item(self):
        """
        check article item
        """
        return

    def check(self, source):
        """
        check item
        """
        match source:
            case "video":
                self.check_video_item()
            case "article":
                self.check_article_item()