获取凭证的摘要信息，支持分页和状态筛选
(
self,
offset: int = 0,
limit: Optional[int] = None,
status_filter: str = "all",
mode: str = "geminicli",
error_code_filter: Optional[str] = None,
cooldown_filter: Optional[str] = None,
preview_filter: Optional[str] = None,
tier_filter: Optional[str] = None
)
| 610 | return {} |
| 611 | |
| 612 | async def get_credentials_summary( |
| 613 | self, |
| 614 | offset: int = 0, |
| 615 | limit: Optional[int] = None, |
| 616 | status_filter: str = "all", |
| 617 | mode: str = "geminicli", |
| 618 | error_code_filter: Optional[str] = None, |
| 619 | cooldown_filter: Optional[str] = None, |
| 620 | preview_filter: Optional[str] = None, |
| 621 | tier_filter: Optional[str] = None |
| 622 | ) -> Dict[str, Any]: |
| 623 | """获取凭证的摘要信息，支持分页和状态筛选""" |
| 624 | self._ensure_initialized() |
| 625 | |
| 626 | try: |
| 627 | table_name = self._get_table_name(mode) |
| 628 | current_time = time.time() |
| 629 | |
| 630 | async with self._pool.acquire() as conn: |
| 631 | # 全局统计 |
| 632 | stats_rows = await conn.fetch( |
| 633 | f"SELECT disabled, COUNT(*) AS cnt FROM {table_name} GROUP BY disabled" |
| 634 | ) |
| 635 | global_stats = {"total": 0, "normal": 0, "disabled": 0} |
| 636 | for r in stats_rows: |
| 637 | global_stats["total"] += r["cnt"] |
| 638 | if r["disabled"]: |
| 639 | global_stats["disabled"] = r["cnt"] |
| 640 | else: |
| 641 | global_stats["normal"] = r["cnt"] |
| 642 | |
| 643 | # WHERE 子句 |
| 644 | where_clauses = [] |
| 645 | if status_filter == "enabled": |
| 646 | where_clauses.append("disabled = 0") |
| 647 | elif status_filter == "disabled": |
| 648 | where_clauses.append("disabled = 1") |
| 649 | |
| 650 | where_clause = ("WHERE " + " AND ".join(where_clauses)) if where_clauses else "" |
| 651 | |
| 652 | # 查询 |
| 653 | if mode == "geminicli": |
| 654 | all_rows = await conn.fetch(f""" |
| 655 | SELECT filename, disabled, error_codes, last_success, |
| 656 | user_email, rotation_order, model_cooldowns, preview, tier |
| 657 | FROM {table_name} |
| 658 | {where_clause} |
| 659 | ORDER BY rotation_order |
| 660 | """) |
| 661 | else: |
| 662 | all_rows = await conn.fetch(f""" |
| 663 | SELECT filename, disabled, error_codes, last_success, |
| 664 | user_email, rotation_order, model_cooldowns, tier, enable_credit |
| 665 | FROM {table_name} |
| 666 | {where_clause} |
| 667 | ORDER BY rotation_order |
| 668 | """) |
| 669 |
no test coverage detected