228 lines
10 KiB
Python
228 lines
10 KiB
Python
# [DEF:backend.src.services.reports.report_service:Module]
|
|
# @TIER: CRITICAL
|
|
# @SEMANTICS: reports, service, aggregation, filtering, pagination, detail
|
|
# @PURPOSE: Aggregate, normalize, filter, and paginate task reports for unified list/detail API use cases.
|
|
# @LAYER: Domain
|
|
# @RELATION: DEPENDS_ON -> backend.src.core.task_manager.manager.TaskManager
|
|
# @RELATION: DEPENDS_ON -> backend.src.models.report
|
|
# @RELATION: DEPENDS_ON -> backend.src.services.reports.normalizer
|
|
# @INVARIANT: List responses are deterministic and include applied filter echo metadata.
|
|
|
|
# [SECTION: IMPORTS]
|
|
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
|
|
# [/SECTION]
|
|
|
|
|
|
# [DEF:ReportsService:Class]
|
|
# @PURPOSE: Service layer for list/detail report retrieval and normalization.
|
|
# @TIER: CRITICAL
|
|
# @PRE: TaskManager dependency is initialized.
|
|
# @POST: Provides deterministic list/detail report responses.
|
|
# @INVARIANT: Service methods are read-only over task history source.
|
|
class ReportsService:
|
|
# [DEF:__init__:Function]
|
|
# @TIER: CRITICAL
|
|
# @PURPOSE: Initialize service with TaskManager dependency.
|
|
# @PRE: task_manager is a live TaskManager instance.
|
|
# @POST: self.task_manager is assigned and ready for read operations.
|
|
# @INVARIANT: Constructor performs no task mutations.
|
|
# @PARAM: task_manager (TaskManager) - Task manager providing source task history.
|
|
def __init__(self, task_manager: TaskManager):
|
|
with belief_scope("__init__"):
|
|
self.task_manager = task_manager
|
|
# [/DEF:__init__:Function]
|
|
|
|
# [DEF:_load_normalized_reports:Function]
|
|
# @PURPOSE: Build normalized reports from all available tasks.
|
|
# @PRE: Task manager returns iterable task history records.
|
|
# @POST: Returns normalized report list preserving source cardinality.
|
|
# @INVARIANT: Every returned item is a TaskReport.
|
|
# @RETURN: List[TaskReport] - Reports sorted later by list logic.
|
|
def _load_normalized_reports(self) -> List[TaskReport]:
|
|
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]
|
|
# @PURPOSE: Normalize naive/aware datetime values to UTC-aware datetime for safe comparisons.
|
|
# @PRE: value is either datetime or None.
|
|
# @POST: Returns UTC-aware datetime or None.
|
|
# @INVARIANT: Naive datetimes are interpreted as UTC to preserve deterministic ordering/filtering.
|
|
# @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]:
|
|
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]
|
|
# @PURPOSE: Produce stable numeric sort key for report timestamps.
|
|
# @PRE: report contains updated_at datetime.
|
|
# @POST: Returns float timestamp suitable for deterministic sorting.
|
|
# @INVARIANT: Mixed naive/aware datetimes never raise TypeError.
|
|
# @PARAM: report (TaskReport) - Report item.
|
|
# @RETURN: float - UTC timestamp key.
|
|
def _datetime_sort_key(self, report: TaskReport) -> float:
|
|
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]
|
|
# @PURPOSE: Apply query filtering to a report.
|
|
# @PRE: report and query are normalized schema instances.
|
|
# @POST: Returns True iff report satisfies all active query filters.
|
|
# @INVARIANT: Filter evaluation is side-effect free.
|
|
# @PARAM: report (TaskReport) - Candidate report.
|
|
# @PARAM: query (ReportQuery) - Applied query.
|
|
# @RETURN: bool - True if report matches all filters.
|
|
def _matches_query(self, report: TaskReport, query: ReportQuery) -> bool:
|
|
with belief_scope("_matches_query"):
|
|
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:
|
|
return False
|
|
return True
|
|
# [/DEF:_matches_query:Function]
|
|
|
|
# [DEF:_sort_reports:Function]
|
|
# @PURPOSE: Sort reports deterministically according to query settings.
|
|
# @PRE: reports contains only TaskReport items.
|
|
# @POST: Returns reports ordered by selected sort field and order.
|
|
# @INVARIANT: Sorting criteria are deterministic for equal input.
|
|
# @PARAM: reports (List[TaskReport]) - Filtered reports.
|
|
# @PARAM: query (ReportQuery) - Sort config.
|
|
# @RETURN: List[TaskReport] - Sorted reports.
|
|
def _sort_reports(self, reports: List[TaskReport], query: ReportQuery) -> List[TaskReport]:
|
|
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)
|
|
|
|
return reports
|
|
# [/DEF:_sort_reports:Function]
|
|
|
|
# [DEF:list_reports:Function]
|
|
# @PURPOSE: Return filtered, sorted, paginated report collection.
|
|
# @PRE: query has passed schema validation.
|
|
# @POST: Returns {items,total,page,page_size,has_next,applied_filters}.
|
|
# @PARAM: query (ReportQuery) - List filters and pagination.
|
|
# @RETURN: ReportCollection - Paginated unified reports payload.
|
|
def list_reports(self, query: ReportQuery) -> ReportCollection:
|
|
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
|
|
|
|
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]
|
|
# @PURPOSE: Return one normalized report with timeline/diagnostics/next actions.
|
|
# @PRE: report_id exists in normalized report set.
|
|
# @POST: Returns normalized detail envelope with diagnostics and next actions where applicable.
|
|
# @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]:
|
|
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()})
|
|
|
|
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"]
|
|
|
|
return ReportDetailView(
|
|
report=target,
|
|
timeline=timeline,
|
|
diagnostics=diagnostics,
|
|
next_actions=next_actions,
|
|
)
|
|
# [/DEF:get_report_detail:Function]
|
|
# [/DEF:ReportsService:Class]
|
|
|
|
import sys
|
|
from generate_semantic_map import parse_file
|
|
|
|
file_path = "backend/src/core/task_manager/task_logger.py"
|
|
entities, issues = parse_file(file_path, file_path, "python")
|
|
|
|
for e in entities:
|
|
e.validate()
|
|
|
|
def print_entity(ent, indent=0):
|
|
print(" " * indent + f"{ent.type} {ent.name} Tags: {list(ent.tags.keys())} Belief: {ent.has_belief_scope}")
|
|
for i in ent.compliance_issues:
|
|
print(" " * (indent + 1) + f"ISSUE: {i.message}")
|
|
for c in ent.children:
|
|
print_entity(c, indent + 1)
|
|
|
|
for e in entities:
|
|
print_entity(e)
|
|
|
|
for i in issues:
|
|
print(f"GLOBAL ISSUE: {i.message} at line {i.line_number}")
|
|
|
|
# [/DEF:backend.src.services.reports.report_service:Module]
|