106 lines
2.8 KiB
Python
106 lines
2.8 KiB
Python
import datetime
|
|
import platform
|
|
from typing import Optional
|
|
|
|
|
|
class AnalyzeTask:
|
|
def __init__(self):
|
|
self.created_at = 0
|
|
self.url = ""
|
|
self.level = -1
|
|
self.task_id = -1
|
|
|
|
@classmethod
|
|
def from_data(cls, data):
|
|
res = cls()
|
|
res.created_at = data["created_at"]
|
|
res.url = data["url"]
|
|
res.level = data["level"]
|
|
res.task_id = data["task_id"]
|
|
return res
|
|
|
|
@property
|
|
def data(self):
|
|
return {
|
|
"created_at": self.created_at,
|
|
"url": self.url,
|
|
"level": self.level,
|
|
"task_id": self.task_id,
|
|
}
|
|
|
|
|
|
class AnalyzeResponse:
|
|
def __init__(self):
|
|
self.created_at = 0
|
|
self.url = ""
|
|
self.level = -1
|
|
self.task_id = -1
|
|
self.clone_duration = 0.0
|
|
self.duration = 0.0
|
|
self.improvement_absolute = 0
|
|
self.improvement_relative = 0
|
|
self.worker = ""
|
|
|
|
@classmethod
|
|
def from_data(cls, data):
|
|
res = cls()
|
|
res.created_at = data["created_at"]
|
|
res.url = data["url"]
|
|
res.level = data["level"]
|
|
res.task_id = data["task_id"]
|
|
res.duration = data["duration"]
|
|
res.clone_duration = data["clone_duration"]
|
|
res.worker = data["worker"]
|
|
res.improvement_absolute = data["improvement_absolute"]
|
|
res.improvement_relative = data["improvement_relative"]
|
|
return res
|
|
|
|
@property
|
|
def data(self):
|
|
if self.task_id < 0:
|
|
raise RuntimeError("No task_id")
|
|
return {
|
|
"created_at": self.created_at,
|
|
"url": self.url,
|
|
"level": self.level,
|
|
"task_id": self.task_id,
|
|
"duration": self.duration,
|
|
"clone_duration": self.clone_duration,
|
|
"worker": self.worker,
|
|
"improvement_absolute": self.improvement_absolute,
|
|
"improvement_relative": self.improvement_relative,
|
|
}
|
|
|
|
|
|
def create_analyze_task(url: str, level: int, task_id: Optional[int] = None) -> AnalyzeTask:
|
|
res = AnalyzeTask()
|
|
res.created_at = int(datetime.datetime.now().timestamp())
|
|
res.url = url
|
|
res.level = level
|
|
res.task_id = task_id
|
|
return res
|
|
|
|
|
|
def create_analyze_response(
|
|
task: AnalyzeTask,
|
|
initial_size: int,
|
|
final_size: int,
|
|
duration: float,
|
|
worker: str,
|
|
clone_duration: float = 0,
|
|
) -> AnalyzeResponse:
|
|
res = AnalyzeResponse()
|
|
res.created_at = task.created_at
|
|
res.url = task.url
|
|
res.level = task.level
|
|
res.task_id = task.task_id
|
|
res.clone_duration = clone_duration
|
|
res.duration = duration
|
|
res.worker = worker
|
|
if initial_size < 100:
|
|
res.improvement_absolute = res.improvement_relative = 0
|
|
else:
|
|
res.improvement_absolute = initial_size - final_size
|
|
res.improvement_relative = 1 - final_size / initial_size
|
|
return res
|