orion/filecodebox
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: remove admin recommendation summaries

Browse Source
This commit is contained in:
Lan
2026-06-03 10:24:07 +08:00
parent 32af822e3b
commit 63754e806a
2 changed files with 6 additions and 669 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/admin/services.py
+4 -624
View File
@@ -447,15 +447,9 @@ class FileService:
page = max(page, 1)
size = min(max(size, 1), 100)
keyword = keyword.strip().lower()
status = self._normalize_file_view_preset_choice(
status, self.VIEW_PRESET_STATUS_VALUES
)
file_type = self._normalize_file_view_preset_choice(
file_type, self.VIEW_PRESET_TYPE_VALUES
)
health = self._normalize_file_view_preset_choice(
health, self.VIEW_PRESET_HEALTH_VALUES
)
status = status.strip().lower()
file_type = file_type.strip().lower()
health = health.strip().lower()
sort_by = self._normalize_sort_by(sort_by)
reverse = sort_order.strip().lower() != "asc"
@@ -496,27 +490,8 @@ class FileService:
key=lambda item: self._get_sort_value(item, sort_by),
reverse=reverse,
)
view_summary = self._build_file_view_summary(
files=enriched_files,
all_file_count=len(all_files),
filters={
"keyword": keyword,
"status": status or "all",
"type": file_type or "all",
"health": health or "all",
"sortBy": sort_by,
"sort_by": sort_by,
"sortOrder": "desc" if reverse else "asc",
"sort_order": "desc" if reverse else "asc",
},
)
offset = (page - 1) * size
return (
enriched_files[offset : offset + size],
len(enriched_files),
summary,
view_summary,
)
return enriched_files[offset : offset + size], len(enriched_files), summary
def _empty_health_summary(self) -> dict[str, int]:
return {
@@ -553,189 +528,6 @@ class FileService:
if "never_retrieved" in reasons:
summary["neverRetrievedCount"] += 1
def _build_file_view_summary(
self,
files: list[dict[str, Any]],
all_file_count: int,
filters: dict[str, Any],
) -> dict[str, Any]:
view_counts = {
"totalFiles": len(files),
"activeCount": 0,
"expiredCount": 0,
"textCount": 0,
"fileCount": 0,
"chunkedCount": 0,
**self._empty_health_summary(),
"storageUsed": sum(item.get("size") or 0 for item in files),
"usedCount": sum(item.get("usedCount") or 0 for item in files),
}
for item in files:
if item.get("isExpired"):
view_counts["expiredCount"] += 1
else:
view_counts["activeCount"] += 1
if item.get("isText"):
view_counts["textCount"] += 1
else:
view_counts["fileCount"] += 1
if item.get("isChunked"):
view_counts["chunkedCount"] += 1
self._accumulate_health_summary(view_counts, item)
cards = self._build_file_view_summary_cards(view_counts)
actions = self._build_file_view_summary_actions(cards)
strongest_severity = self._get_strongest_summary_severity(cards)
active_filter_count = sum(
[
1 if str(filters.get("keyword") or "").strip() else 0,
1 if str(filters.get("status") or "all").strip() != "all" else 0,
1 if str(filters.get("type") or "all").strip() != "all" else 0,
1 if str(filters.get("health") or "all").strip() != "all" else 0,
]
)
return {
"total": len(files),
"filteredTotal": len(files),
"filtered_total": len(files),
"allTotal": max(int(all_file_count or 0), 0),
"all_total": max(int(all_file_count or 0), 0),
"activeFilterCount": active_filter_count,
"active_filter_count": active_filter_count,
"hasFilters": active_filter_count > 0,
"has_filters": active_filter_count > 0,
"strongestSeverity": strongest_severity,
"strongest_severity": strongest_severity,
"filters": filters,
"summary": view_counts,
"healthSummary": view_counts,
"health_summary": view_counts,
"cards": cards,
"items": cards,
"actions": actions,
}
def _build_file_view_summary_cards(
self, summary: dict[str, Any]
) -> list[dict[str, Any]]:
candidates = [
{
"key": "storage_issue",
"severity": "danger",
"priority": 100,
"count": summary["storageIssueCount"],
"health": "storage_issue",
},
{
"key": "expired",
"severity": "danger" if summary["expiredCount"] >= 10 else "warning",
"priority": 90,
"count": summary["expiredCount"],
"health": "expired",
},
{
"key": "expiring_soon",
"severity": "warning",
"priority": 80,
"count": summary["expiringSoonCount"],
"health": "expiring_soon",
},
{
"key": "never_retrieved",
"severity": "neutral",
"priority": 60,
"count": summary["neverRetrievedCount"],
"health": "never_retrieved",
},
{
"key": "permanent",
"severity": "neutral",
"priority": 40,
"count": summary["permanentCount"],
"health": "permanent",
},
]
cards = [
self._build_file_view_summary_item(
key=item["key"],
severity=item["severity"],
priority=item["priority"],
count=item["count"],
health=item["health"],
)
for item in candidates
if item["count"] > 0
]
if not cards:
key = "empty" if summary["totalFiles"] == 0 else "healthy"
cards.append(
self._build_file_view_summary_item(
key=key,
severity="success" if summary["totalFiles"] > 0 else "neutral",
priority=10,
count=summary["totalFiles"],
health="healthy" if summary["totalFiles"] > 0 else "all",
)
)
cards.sort(key=lambda item: (-item["priority"], item["key"]))
return cards[:4]
def _build_file_view_summary_actions(
self, cards: list[dict[str, Any]]
) -> list[dict[str, Any]]:
action_key_map = {
"storage_issue": "inspect_storage",
"expired": "cleanup_expired",
"expiring_soon": "extend_expiring",
"never_retrieved": "review_never_retrieved",
"permanent": "review_permanent",
"healthy": "monitor",
"empty": "clear_filters",
}
return [
{
**item,
"sourceKey": item["key"],
"source_key": item["key"],
"suggestedAction": action_key_map.get(item["key"], "review"),
"suggested_action": action_key_map.get(item["key"], "review"),
}
for item in cards
if item["severity"] != "success"
]
def _build_file_view_summary_item(
self,
key: str,
severity: str,
priority: int,
count: int,
health: str,
) -> dict[str, Any]:
return {
"key": key,
"severity": severity,
"priority": priority,
"count": max(int(count or 0), 0),
"actionType": "filter",
"action_type": "filter",
"health": health,
"targetHealth": health,
"target_health": health,
}
def _get_strongest_summary_severity(self, items: list[dict[str, Any]]) -> str:
severity_order = {"danger": 3, "warning": 2, "neutral": 1, "success": 0}
return max(
(item["severity"] for item in items),
key=lambda severity: severity_order.get(severity, 0),
default="success",
)
async def build_file_health_summary(
self, file_codes: list[FileCodes], now: Optional[datetime] = None
) -> dict[str, int]:
@@ -747,418 +539,6 @@ class FileService:
self._accumulate_health_summary(summary, item)
return summary
async def get_dashboard_maintenance_queue(self) -> dict[str, Any]:
file_codes = await FileCodes.all()
now = await get_now()
today_start = now.replace(hour=0, minute=0, second=0, microsecond=0)
today_size = sum(
file_code.size
for file_code in file_codes
if file_code.created_at and file_code.created_at >= today_start
)
expired_count = 0
for file_code in file_codes:
if await file_code.is_expired():
expired_count += 1
health_summary = await self.build_file_health_summary(file_codes, now=now)
return self.build_dashboard_maintenance_queue(
health_summary=health_summary,
total_files=len(file_codes),
expired_count=expired_count,
today_size=today_size,
upload_size_limit=settings.uploadSize,
open_upload=settings.openUpload,
enable_chunk=settings.enableChunk,
max_save_seconds=settings.max_save_seconds,
)
def build_dashboard_operational_insights(
self,
health_summary: dict[str, int],
total_files: int,
expired_count: int,
today_size: int,
upload_size_limit: int,
open_upload: int,
enable_chunk: int,
max_save_seconds: int,
) -> list[dict[str, Any]]:
def to_int(value: Any) -> int:
try:
return int(value or 0)
except (TypeError, ValueError):
return 0
total_files = to_int(total_files)
expired_count = to_int(expired_count)
today_size = to_int(today_size)
upload_size_limit = to_int(upload_size_limit)
open_upload = to_int(open_upload)
enable_chunk = to_int(enable_chunk)
max_save_seconds = to_int(max_save_seconds)
insights = []
if health_summary.get("storageIssueCount", 0) > 0:
insights.append(
self._build_dashboard_operational_insight(
key="storage_issue",
severity="danger",
priority=100,
count=health_summary["storageIssueCount"],
action_type="file_queue",
health="storage_issue",
)
)
if expired_count > 0:
insights.append(
self._build_dashboard_operational_insight(
key="expired_cleanup",
severity="danger" if expired_count >= 10 else "warning",
priority=90,
count=expired_count,
action_type="file_queue",
health="expired",
)
)
if health_summary.get("expiringSoonCount", 0) > 0:
insights.append(
self._build_dashboard_operational_insight(
key="expiring_soon",
severity="warning",
priority=80,
count=health_summary["expiringSoonCount"],
action_type="file_queue",
health="expiring_soon",
)
)
never_retrieved_count = health_summary.get("neverRetrievedCount", 0)
if total_files > 0 and never_retrieved_count >= max(3, total_files // 5):
insights.append(
self._build_dashboard_operational_insight(
key="never_retrieved",
severity="neutral",
priority=60,
count=never_retrieved_count,
action_type="file_queue",
health="never_retrieved",
)
)
if open_upload and max_save_seconds <= 0:
insights.append(
self._build_dashboard_operational_insight(
key="guest_upload_retention",
severity="warning",
priority=50,
count=1,
action_type="settings",
)
)
if (
upload_size_limit > 0
and today_size >= upload_size_limit
and not enable_chunk
):
insights.append(
self._build_dashboard_operational_insight(
key="chunking_disabled",
severity="neutral",
priority=40,
count=1,
action_type="settings",
)
)
if not insights:
insights.append(
self._build_dashboard_operational_insight(
key="healthy",
severity="success",
priority=10,
count=total_files,
action_type="file_queue",
health="healthy",
)
)
insights.sort(key=lambda item: (-item["priority"], item["key"]))
return insights[:4]
def build_dashboard_maintenance_queue(
self,
health_summary: dict[str, int],
total_files: int,
expired_count: int,
today_size: int,
upload_size_limit: int,
open_upload: int,
enable_chunk: int,
max_save_seconds: int,
) -> dict[str, Any]:
def to_int(value: Any) -> int:
try:
return int(value or 0)
except (TypeError, ValueError):
return 0
total_files = to_int(total_files)
expired_count = to_int(expired_count)
today_size = to_int(today_size)
upload_size_limit = to_int(upload_size_limit)
open_upload = to_int(open_upload)
enable_chunk = to_int(enable_chunk)
max_save_seconds = to_int(max_save_seconds)
items = []
if health_summary.get("storageIssueCount", 0) > 0:
items.append(
self._build_dashboard_maintenance_item(
key="storage_issue",
severity="danger",
category="storage",
priority=100,
count=health_summary["storageIssueCount"],
action_type="file_queue",
health="storage_issue",
suggested_action="inspect_storage",
)
)
if expired_count > 0:
items.append(
self._build_dashboard_maintenance_item(
key="expired_cleanup",
severity="danger" if expired_count >= 10 else "warning",
category="retention",
priority=90,
count=expired_count,
action_type="file_queue",
health="expired",
suggested_action="cleanup_or_extend",
)
)
if health_summary.get("expiringSoonCount", 0) > 0:
items.append(
self._build_dashboard_maintenance_item(
key="expiring_soon",
severity="warning",
category="retention",
priority=80,
count=health_summary["expiringSoonCount"],
action_type="file_queue",
health="expiring_soon",
suggested_action="extend_expiration",
)
)
never_retrieved_count = health_summary.get("neverRetrievedCount", 0)
if never_retrieved_count > 0:
items.append(
self._build_dashboard_maintenance_item(
key="never_retrieved",
severity="neutral",
category="adoption",
priority=60,
count=never_retrieved_count,
action_type="file_queue",
health="never_retrieved",
suggested_action="review_usage",
)
)
permanent_count = health_summary.get("permanentCount", 0)
if permanent_count > 0:
items.append(
self._build_dashboard_maintenance_item(
key="permanent_review",
severity="neutral",
category="retention",
priority=45,
count=permanent_count,
action_type="file_queue",
health="permanent",
suggested_action="review_retention",
)
)
if open_upload and max_save_seconds <= 0:
items.append(
self._build_dashboard_maintenance_item(
key="guest_upload_retention",
severity="warning",
category="settings",
priority=70,
count=1,
action_type="settings",
suggested_action="set_retention_limit",
)
)
if (
upload_size_limit > 0
and today_size >= upload_size_limit
and not enable_chunk
):
items.append(
self._build_dashboard_maintenance_item(
key="chunking_disabled",
severity="neutral",
category="settings",
priority=40,
count=1,
action_type="settings",
suggested_action="enable_chunking",
)
)
if not items:
items.append(
self._build_dashboard_maintenance_item(
key="healthy",
severity="success",
category="system",
priority=10,
count=total_files,
action_type="file_queue",
health="healthy",
suggested_action="monitor",
)
)
items.sort(key=lambda item: (-item["priority"], item["key"]))
summary = self._build_dashboard_maintenance_summary(items)
return {
"items": items,
"maintenanceItems": items,
"maintenance_items": items,
"summary": summary,
"maintenanceSummary": summary,
"maintenance_summary": summary,
}
def _build_dashboard_maintenance_item(
self,
key: str,
severity: str,
category: str,
priority: int,
count: int,
action_type: str,
suggested_action: str,
health: Optional[str] = None,
) -> dict[str, Any]:
action = {
"type": action_type,
"actionType": action_type,
"action_type": action_type,
"suggestedAction": suggested_action,
"suggested_action": suggested_action,
}
if health:
action.update(
{
"health": health,
"targetHealth": health,
"target_health": health,
}
)
return {
"key": key,
"severity": severity,
"category": category,
"priority": priority,
"count": max(int(count or 0), 0),
"action": action,
"actionType": action_type,
"action_type": action_type,
"suggestedAction": suggested_action,
"suggested_action": suggested_action,
"targetHealth": health,
"target_health": health,
}
def _build_dashboard_maintenance_summary(
self, items: list[dict[str, Any]]
) -> dict[str, Any]:
severity_order = {"danger": 3, "warning": 2, "neutral": 1, "success": 0}
strongest_severity = max(
(item["severity"] for item in items),
key=lambda severity: severity_order.get(severity, 0),
default="success",
)
actionable_items = [item for item in items if item["severity"] != "success"]
file_queue_count = sum(
item["count"]
for item in actionable_items
if item.get("actionType") == "file_queue"
)
settings_count = sum(
item["count"]
for item in actionable_items
if item.get("actionType") == "settings"
)
return {
"total": len(items),
"actionableCount": len(actionable_items),
"actionable_count": len(actionable_items),
"dangerCount": sum(1 for item in items if item["severity"] == "danger"),
"danger_count": sum(1 for item in items if item["severity"] == "danger"),
"warningCount": sum(1 for item in items if item["severity"] == "warning"),
"warning_count": sum(1 for item in items if item["severity"] == "warning"),
"successCount": sum(1 for item in items if item["severity"] == "success"),
"success_count": sum(1 for item in items if item["severity"] == "success"),
"neutralCount": sum(1 for item in items if item["severity"] == "neutral"),
"neutral_count": sum(1 for item in items if item["severity"] == "neutral"),
"fileQueueCount": file_queue_count,
"file_queue_count": file_queue_count,
"settingsCount": settings_count,
"settings_count": settings_count,
"strongestSeverity": strongest_severity,
"strongest_severity": strongest_severity,
}
def _build_dashboard_operational_insight(
self,
key: str,
severity: str,
priority: int,
count: int,
action_type: str,
health: Optional[str] = None,
) -> dict[str, Any]:
action = {
"type": action_type,
"actionType": action_type,
"action_type": action_type,
}
if health:
action.update(
{
"health": health,
"targetHealth": health,
"target_health": health,
}
)
return {
"key": key,
"severity": severity,
"priority": priority,
"count": max(int(count or 0), 0),
"action": action,
"actionType": action_type,
"action_type": action_type,
"targetHealth": health,
"target_health": health,
}
async def _build_admin_file_item(
self, file_code: FileCodes, now: Optional[datetime] = None
) -> dict[str, Any]: