File size: 4,085 Bytes
5378afe | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 | """
任务领域模型
定义任务实体及其业务逻辑
"""
from pydantic import BaseModel, Field, validator
from typing import Optional
from enum import Enum
class TaskStatus(str, Enum):
"""任务状态枚举"""
STOPPED = "stopped"
RUNNING = "running"
SCHEDULED = "scheduled"
class Task(BaseModel):
"""任务实体"""
id: Optional[int] = None
task_name: str
enabled: bool
keyword: str
description: Optional[str] = ""
max_pages: int
personal_only: bool
min_price: Optional[str] = None
max_price: Optional[str] = None
cron: Optional[str] = None
ai_prompt_base_file: str
ai_prompt_criteria_file: str
account_state_file: Optional[str] = None
free_shipping: bool = True
new_publish_option: Optional[str] = None
region: Optional[str] = "江苏/南京/全南京"
is_running: bool = False
class Config:
use_enum_values = True
def can_start(self) -> bool:
"""检查任务是否可以启动"""
return self.enabled and not self.is_running
def can_stop(self) -> bool:
"""检查任务是否可以停止"""
return self.is_running
def apply_update(self, update: 'TaskUpdate') -> 'Task':
"""应用更新并返回新的任务实例"""
update_data = update.dict(exclude_unset=True)
return self.copy(update=update_data)
class TaskCreate(BaseModel):
"""创建任务的DTO"""
task_name: str
enabled: bool = True
keyword: str
description: Optional[str] = ""
max_pages: int = 3
personal_only: bool = True
min_price: Optional[str] = None
max_price: Optional[str] = None
cron: Optional[str] = None
ai_prompt_base_file: str = "prompts/base_prompt.txt"
ai_prompt_criteria_file: str
account_state_file: Optional[str] = None
free_shipping: bool = True
new_publish_option: Optional[str] = None
region: Optional[str] = "江苏/南京/全南京"
class TaskUpdate(BaseModel):
"""更新任务的DTO"""
task_name: Optional[str] = None
enabled: Optional[bool] = None
keyword: Optional[str] = None
description: Optional[str] = None
max_pages: Optional[int] = None
personal_only: Optional[bool] = None
min_price: Optional[str] = None
max_price: Optional[str] = None
cron: Optional[str] = None
ai_prompt_base_file: Optional[str] = None
ai_prompt_criteria_file: Optional[str] = None
account_state_file: Optional[str] = None
free_shipping: Optional[bool] = None
new_publish_option: Optional[str] = None
region: Optional[str] = None
is_running: Optional[bool] = None
class TaskGenerateRequest(BaseModel):
"""AI生成任务的请求DTO"""
task_name: str
keyword: str
description: str
personal_only: bool = True
min_price: Optional[str] = None
max_price: Optional[str] = None
max_pages: int = 3
cron: Optional[str] = None
account_state_file: Optional[str] = None
free_shipping: bool = True
new_publish_option: Optional[str] = None
region: Optional[str] = "江苏/南京/全南京"
@validator('min_price', 'max_price', pre=True)
def convert_price_to_str(cls, v):
"""将价格转换为字符串，处理空字符串和数字"""
if v == "" or v == "null" or v == "undefined" or v is None:
return None
# 如果是数字，转换为字符串
if isinstance(v, (int, float)):
return str(v)
return v
@validator('cron', pre=True)
def empty_str_to_none(cls, v):
"""将空字符串转换为 None"""
if v == "" or v == "null" or v == "undefined":
return None
return v
@validator('account_state_file', pre=True)
def empty_account_to_none(cls, v):
if v == "" or v == "null" or v == "undefined":
return None
return v
@validator('new_publish_option', 'region', pre=True)
def empty_str_to_none_for_strings(cls, v):
if v == "" or v == "null" or v == "undefined":
return None
return v
|