busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

semantic update

Browse Source
This commit is contained in:
busya
2026-02-24 21:08:12 +03:00
parent 7a12ed0931
commit 95ae9c6af1
32 changed files with 60376 additions and 59911 deletions
Download Patch File Download Diff File Expand all files Collapse all files

166
backend/src/services/reports/report_service.py
View File

@@ -12,6 +12,8 @@
from datetime import datetime, timezone
from typing import List, Optional
from ...core.logger import belief_scope
from ...core.task_manager import TaskManager
from ...models.report import ReportCollection, ReportDetailView, ReportQuery, ReportStatus, TaskReport, TaskType
from .normalizer import normalize_task_report
@@ -33,7 +35,8 @@ class ReportsService:
# @INVARIANT: Constructor performs no task mutations.
# @PARAM: task_manager (TaskManager) - Task manager providing source task history.
def __init__(self, task_manager: TaskManager):
self.task_manager = task_manager
with belief_scope("__init__"):
self.task_manager = task_manager
# [/DEF:__init__:Function]
# [DEF:_load_normalized_reports:Function]
@@ -43,9 +46,10 @@ class ReportsService:
# @INVARIANT: Every returned item is a TaskReport.
# @RETURN: List[TaskReport] - Reports sorted later by list logic.
def _load_normalized_reports(self) -> List[TaskReport]:
tasks = self.task_manager.get_all_tasks()
reports = [normalize_task_report(task) for task in tasks]
return reports
with belief_scope("_load_normalized_reports"):
tasks = self.task_manager.get_all_tasks()
reports = [normalize_task_report(task) for task in tasks]
return reports
# [/DEF:_load_normalized_reports:Function]
# [DEF:_to_utc_datetime:Function]
@@ -56,11 +60,12 @@ class ReportsService:
# @PARAM: value (Optional[datetime]) - Source datetime value.
# @RETURN: Optional[datetime] - UTC-aware datetime or None.
def _to_utc_datetime(self, value: Optional[datetime]) -> Optional[datetime]:
if value is None:
return None
if value.tzinfo is None:
return value.replace(tzinfo=timezone.utc)
return value.astimezone(timezone.utc)
with belief_scope("_to_utc_datetime"):
if value is None:
return None
if value.tzinfo is None:
return value.replace(tzinfo=timezone.utc)
return value.astimezone(timezone.utc)
# [/DEF:_to_utc_datetime:Function]
# [DEF:_datetime_sort_key:Function]
@@ -71,10 +76,11 @@ class ReportsService:
# @PARAM: report (TaskReport) - Report item.
# @RETURN: float - UTC timestamp key.
def _datetime_sort_key(self, report: TaskReport) -> float:
updated = self._to_utc_datetime(report.updated_at)
if updated is None:
return 0.0
return updated.timestamp()
with belief_scope("_datetime_sort_key"):
updated = self._to_utc_datetime(report.updated_at)
if updated is None:
return 0.0
return updated.timestamp()
# [/DEF:_datetime_sort_key:Function]
# [DEF:_matches_query:Function]
@@ -86,24 +92,25 @@ class ReportsService:
# @PARAM: query (ReportQuery) - Applied query.
# @RETURN: bool - True if report matches all filters.
def _matches_query(self, report: TaskReport, query: ReportQuery) -> bool:
if query.task_types and report.task_type not in query.task_types:
return False
if query.statuses and report.status not in query.statuses:
return False
report_updated_at = self._to_utc_datetime(report.updated_at)
query_time_from = self._to_utc_datetime(query.time_from)
query_time_to = self._to_utc_datetime(query.time_to)
if query_time_from and report_updated_at and report_updated_at < query_time_from:
return False
if query_time_to and report_updated_at and report_updated_at > query_time_to:
return False
if query.search:
needle = query.search.lower()
haystack = f"{report.summary} {report.task_type.value} {report.status.value}".lower()
if needle not in haystack:
with belief_scope("_matches_query"):
if query.task_types and report.task_type not in query.task_types:
return False
return True
if query.statuses and report.status not in query.statuses:
return False
report_updated_at = self._to_utc_datetime(report.updated_at)
query_time_from = self._to_utc_datetime(query.time_from)
query_time_to = self._to_utc_datetime(query.time_to)
if query_time_from and report_updated_at and report_updated_at < query_time_from:
return False
if query_time_to and report_updated_at and report_updated_at > query_time_to:
return False
if query.search:
needle = query.search.lower()
haystack = f"{report.summary} {report.task_type.value} {report.status.value}".lower()
if needle not in haystack:
return False
return True
# [/DEF:_matches_query:Function]
# [DEF:_sort_reports:Function]
@@ -115,16 +122,17 @@ class ReportsService:
# @PARAM: query (ReportQuery) - Sort config.
# @RETURN: List[TaskReport] - Sorted reports.
def _sort_reports(self, reports: List[TaskReport], query: ReportQuery) -> List[TaskReport]:
reverse = query.sort_order == "desc"
with belief_scope("_sort_reports"):
reverse = query.sort_order == "desc"
if query.sort_by == "status":
reports.sort(key=lambda item: item.status.value, reverse=reverse)
elif query.sort_by == "task_type":
reports.sort(key=lambda item: item.task_type.value, reverse=reverse)
else:
reports.sort(key=self._datetime_sort_key, reverse=reverse)
if query.sort_by == "status":
reports.sort(key=lambda item: item.status.value, reverse=reverse)
elif query.sort_by == "task_type":
reports.sort(key=lambda item: item.task_type.value, reverse=reverse)
else:
reports.sort(key=self._datetime_sort_key, reverse=reverse)
return reports
return reports
# [/DEF:_sort_reports:Function]
# [DEF:list_reports:Function]
@@ -134,24 +142,25 @@ class ReportsService:
# @PARAM: query (ReportQuery) - List filters and pagination.
# @RETURN: ReportCollection - Paginated unified reports payload.
def list_reports(self, query: ReportQuery) -> ReportCollection:
reports = self._load_normalized_reports()
filtered = [report for report in reports if self._matches_query(report, query)]
sorted_reports = self._sort_reports(filtered, query)
with belief_scope("list_reports"):
reports = self._load_normalized_reports()
filtered = [report for report in reports if self._matches_query(report, query)]
sorted_reports = self._sort_reports(filtered, query)
total = len(sorted_reports)
start = (query.page - 1) * query.page_size
end = start + query.page_size
items = sorted_reports[start:end]
has_next = end < total
total = len(sorted_reports)
start = (query.page - 1) * query.page_size
end = start + query.page_size
items = sorted_reports[start:end]
has_next = end < total
return ReportCollection(
items=items,
total=total,
page=query.page,
page_size=query.page_size,
has_next=has_next,
applied_filters=query,
)
return ReportCollection(
items=items,
total=total,
page=query.page,
page_size=query.page_size,
has_next=has_next,
applied_filters=query,
)
# [/DEF:list_reports:Function]
# [DEF:get_report_detail:Function]
@@ -161,34 +170,35 @@ class ReportsService:
# @PARAM: report_id (str) - Stable report identifier.
# @RETURN: Optional[ReportDetailView] - Detailed report or None if not found.
def get_report_detail(self, report_id: str) -> Optional[ReportDetailView]:
reports = self._load_normalized_reports()
target = next((report for report in reports if report.report_id == report_id), None)
if not target:
return None
with belief_scope("get_report_detail"):
reports = self._load_normalized_reports()
target = next((report for report in reports if report.report_id == report_id), None)
if not target:
return None
timeline = []
if target.started_at:
timeline.append({"event": "started", "at": target.started_at.isoformat()})
timeline.append({"event": "updated", "at": target.updated_at.isoformat()})
timeline = []
if target.started_at:
timeline.append({"event": "started", "at": target.started_at.isoformat()})
timeline.append({"event": "updated", "at": target.updated_at.isoformat()})
diagnostics = target.details or {}
if not diagnostics:
diagnostics = {"note": "Not provided"}
if target.error_context:
diagnostics["error_context"] = target.error_context.model_dump()
diagnostics = target.details or {}
if not diagnostics:
diagnostics = {"note": "Not provided"}
if target.error_context:
diagnostics["error_context"] = target.error_context.model_dump()
next_actions = []
if target.error_context and target.error_context.next_actions:
next_actions = target.error_context.next_actions
elif target.status in {ReportStatus.FAILED, ReportStatus.PARTIAL}:
next_actions = ["Review diagnostics", "Retry task if applicable"]
next_actions = []
if target.error_context and target.error_context.next_actions:
next_actions = target.error_context.next_actions
elif target.status in {ReportStatus.FAILED, ReportStatus.PARTIAL}:
next_actions = ["Review diagnostics", "Retry task if applicable"]
return ReportDetailView(
report=target,
timeline=timeline,
diagnostics=diagnostics,
next_actions=next_actions,
)
return ReportDetailView(
report=target,
timeline=timeline,
diagnostics=diagnostics,
next_actions=next_actions,
)
# [/DEF:get_report_detail:Function]
# [/DEF:ReportsService:Class]