(
self,
page: int,
size: int,
keyword: str = "",
status: str = "",
file_type: str = "",
health: str = "",
sort_by: str = "created_at",
sort_order: str = "desc",
)
| 435 | } |
| 436 | |
| 437 | async def list_files( |
| 438 | self, |
| 439 | page: int, |
| 440 | size: int, |
| 441 | keyword: str = "", |
| 442 | status: str = "", |
| 443 | file_type: str = "", |
| 444 | health: str = "", |
| 445 | sort_by: str = "created_at", |
| 446 | sort_order: str = "desc", |
| 447 | ): |
| 448 | page = max(page, 1) |
| 449 | size = min(max(size, 1), 100) |
| 450 | keyword = keyword.strip().lower() |
| 451 | status = status.strip().lower() |
| 452 | file_type = file_type.strip().lower() |
| 453 | health = health.strip().lower() |
| 454 | sort_by = self._normalize_sort_by(sort_by) |
| 455 | reverse = sort_order.strip().lower() != "asc" |
| 456 | |
| 457 | all_files = await FileCodes.all() |
| 458 | now = await get_now() |
| 459 | enriched_files = [] |
| 460 | summary = { |
| 461 | "totalFiles": len(all_files), |
| 462 | "activeCount": 0, |
| 463 | "expiredCount": 0, |
| 464 | "textCount": 0, |
| 465 | "fileCount": 0, |
| 466 | "chunkedCount": 0, |
| 467 | **self._empty_health_summary(), |
| 468 | "storageUsed": sum(file_code.size for file_code in all_files), |
| 469 | "usedCount": sum(file_code.used_count for file_code in all_files), |
| 470 | } |
| 471 | |
| 472 | for file_code in all_files: |
| 473 | item = await self._build_admin_file_item(file_code, now=now) |
| 474 | if item["isExpired"]: |
| 475 | summary["expiredCount"] += 1 |
| 476 | else: |
| 477 | summary["activeCount"] += 1 |
| 478 | if item["isText"]: |
| 479 | summary["textCount"] += 1 |
| 480 | else: |
| 481 | summary["fileCount"] += 1 |
| 482 | if item["isChunked"]: |
| 483 | summary["chunkedCount"] += 1 |
| 484 | self._accumulate_health_summary(summary, item) |
| 485 | |
| 486 | if not self._match_admin_file(item, keyword, status, file_type, health): |
| 487 | continue |
| 488 | enriched_files.append(item) |
| 489 | |
| 490 | enriched_files.sort( |
| 491 | key=lambda item: self._get_sort_value(item, sort_by), |
| 492 | reverse=reverse, |
| 493 | ) |
| 494 | offset = (page - 1) * size |
no test coverage detected