""" @author: luojunhui """ import time from pydantic import BaseModel 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 MetaArticleItem(BaseModel): platform: str | None = None mode: str | None = None category: str | None = None out_account_id: str | None = None article_index: int | None = None title: str | None = None link: str | None = None read_cnt: int = 0 like_cnt: int = 0 description: str | None = None publish_time: int | None = None crawler_time: int | None = None status: str | None = None channel_content_id: str | None = None unique_index: str | None = None source_article_title: str | None = None source_account: str | None = None llm_sensitivity: float | None = None title_sensitivity: float | None = 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 """ article_item = MetaArticleItem(**self.item) return article_item def check(self, source): """ check item """ match source: case "video": self.check_video_item() case "article": return self.check_article_item() return None