task.py
3.81 KB
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
"""
数据模型 - 补货任务
"""
from dataclasses import dataclass, field
from decimal import Decimal
from datetime import datetime
from typing import Optional
from enum import IntEnum
class TaskStatus(IntEnum):
"""任务状态"""
RUNNING = 0
SUCCESS = 1
FAILED = 2
@dataclass
class ReplenishmentTask:
"""AI补货任务"""
task_no: str
group_id: int
dealer_grouping_id: int
id: Optional[int] = None
dealer_grouping_name: Optional[str] = None
brand_grouping_id: Optional[int] = None
plan_amount: Decimal = Decimal("0")
actual_amount: Decimal = Decimal("0")
part_count: int = 0
base_ratio: Optional[Decimal] = None
status: TaskStatus = TaskStatus.RUNNING
error_message: str = ""
llm_provider: str = ""
llm_model: str = ""
llm_total_tokens: int = 0
statistics_date: str = ""
start_time: Optional[datetime] = None
end_time: Optional[datetime] = None
create_time: Optional[datetime] = None
def to_dict(self) -> dict:
"""转换为字典"""
return {
"task_no": self.task_no,
"group_id": self.group_id,
"dealer_grouping_id": self.dealer_grouping_id,
"dealer_grouping_name": self.dealer_grouping_name,
"brand_grouping_id": self.brand_grouping_id,
"plan_amount": float(self.plan_amount),
"actual_amount": float(self.actual_amount),
"part_count": self.part_count,
"base_ratio": float(self.base_ratio) if self.base_ratio else None,
"status": int(self.status),
"error_message": self.error_message,
"llm_provider": self.llm_provider,
"llm_model": self.llm_model,
"llm_total_tokens": self.llm_total_tokens,
"statistics_date": self.statistics_date,
}
@dataclass
class ReplenishmentDetail:
"""AI补货建议明细"""
task_no: str
group_id: int
dealer_grouping_id: int
shop_id: int
part_code: str
id: Optional[int] = None
brand_grouping_id: Optional[int] = None
shop_name: Optional[str] = None
part_name: Optional[str] = None
unit: Optional[str] = None
cost_price: Decimal = Decimal("0")
current_ratio: Optional[Decimal] = None
base_ratio: Optional[Decimal] = None
post_plan_ratio: Optional[Decimal] = None
valid_storage_cnt: Decimal = Decimal("0")
avg_sales_cnt: Decimal = Decimal("0")
suggest_cnt: int = 0
suggest_amount: Decimal = Decimal("0")
suggestion_reason: str = ""
priority: int = 2
llm_confidence: float = 0.8
statistics_date: str = ""
create_time: Optional[datetime] = None
def to_dict(self) -> dict:
"""转换为字典"""
return {
"task_no": self.task_no,
"group_id": self.group_id,
"dealer_grouping_id": self.dealer_grouping_id,
"brand_grouping_id": self.brand_grouping_id,
"shop_id": self.shop_id,
"shop_name": self.shop_name,
"part_code": self.part_code,
"part_name": self.part_name,
"unit": self.unit,
"cost_price": float(self.cost_price),
"current_ratio": float(self.current_ratio) if self.current_ratio else None,
"base_ratio": float(self.base_ratio) if self.base_ratio else None,
"post_plan_ratio": float(self.post_plan_ratio) if self.post_plan_ratio else None,
"valid_storage_cnt": float(self.valid_storage_cnt),
"avg_sales_cnt": float(self.avg_sales_cnt),
"suggest_cnt": self.suggest_cnt,
"suggest_amount": float(self.suggest_amount),
"suggestion_reason": self.suggestion_reason,
"priority": self.priority,
"llm_confidence": self.llm_confidence,
"statistics_date": self.statistics_date,
}