AutoScraperX/core/models/message_parser.py

from pydantic import BaseModel, Field
from typing import Optional, Union, List
import json


class RuleField(BaseModel):
    min: Union[int, float] = 0
    max: Union[int, float] = 0


class RuleModel(BaseModel):
    period: Optional[RuleField] = None
    duration: Optional[RuleField] = None
    play_cnt: Optional[RuleField] = None
    like_cnt: Optional[RuleField] = None
    comment_cnt: Optional[RuleField] = None
    share_cnt: Optional[RuleField] = None
    videos_cnt: Optional[RuleField] = None
    width: Optional[RuleField] = None
    height: Optional[RuleField] = None


class MessageModel(BaseModel):
    id: int
    rule: str
    parsed_rule: List[RuleModel] = Field(default_factory=list)

    def model_post_init(self, __context) -> None:
        # 解析 rule 字符串
        try:
            if self.rule:
                rule_list = json.loads(self.rule)
                if not isinstance(rule_list, list):
                    rule_list = [rule_list]
                self.parsed_rule = []
                for item in rule_list:
                    try:
                        self.parsed_rule.append(RuleModel(**item))
                    except Exception as e:
                        print(f"解析规则项失败: {item}, 错误: {e}")
        except json.JSONDecodeError as e:
            print(f"JSON 解析错误: {e}")
        except Exception as e:
            print(f"其他错误: {e}")

        super().model_post_init(__context)


if __name__ == "__main__":
    message_data = {
        "createTime": 1747042241127,
        "id": 147,
        "interval": 86400,
        "machine": "",
        "mode": "related",
        "operator": "张良",
        "rule": '[{"videos_cnt":{"min":200,"max":0}},{"duration":{"min":30,"max":1200}}]',
        "source": "zhongqingkandianrelated",
        "spiderName": "run_zqkd_related",
        "startTime": 1758211500000,
        "status": 0,
        "taskName": "中青看点相关推荐",
        "updateTime": 1749178503914
    }

    try:
        message = MessageModel(**message_data)
        print(f"ID: {message.id}")
        print(f"原始规则字符串: {message.rule}")
        print("解析后的规则:")


        for idx, rule in enumerate(message.parsed_rule, 1):
            print(f"规则 {idx}:")
            rule_dict = rule.model_dump()
            for field_name, field_value in rule_dict.items():
                if field_value is not None:
                    print(f"  {field_name}: min={field_value['min']}, max={field_value['max']}")

        if not message.parsed_rule:
            print("  没有有效的规则")

    except Exception as e:
        print(f"解析失败: {e}")
        import traceback
        traceback.print_exc()