26 lines
937 B
Python
26 lines
937 B
Python
from __future__ import annotations
|
|
|
|
from app.core.db import get_db
|
|
from app.core.files import load_import_records
|
|
|
|
|
|
def get_import_summary() -> dict:
|
|
with get_db() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute("select count(*) from domains")
|
|
domains_total = cur.fetchone()[0]
|
|
cur.execute("select count(*) from detect_tasks")
|
|
tasks_total = cur.fetchone()[0]
|
|
cur.execute("select max(create_time) from domains")
|
|
last_import_time = cur.fetchone()[0]
|
|
tasks = load_import_records()
|
|
running_tasks = sum(1 for item in tasks if item.get("status") in {"queued", "running"})
|
|
|
|
return {
|
|
"domains_total": domains_total,
|
|
"detect_tasks_total": tasks_total,
|
|
"last_import_time": last_import_time.isoformat() if last_import_time else None,
|
|
"import_task_total": len(tasks),
|
|
"running_import_tasks": running_tasks,
|
|
}
|