1from typing import Dict, List, Union
 2
 3from fastapi import APIRouter
 4
 5from app import __version__
 6# from .. import __version__
 7from app.db import crud
 8from app.dependencies import DBDep, AdminDep, SudoAdminDep
 9from app.models.admin import Admin
10from app.models.proxy import InboundHost, Inbound
11from app.models.system import SystemStats
12from app.models.user import UserStatus
13from app.utils.system import memory_usage, cpu_usage, realtime_bandwidth
14
15router = APIRouter(tags=["System"])
16
17
18@router.get("/system", response_model=SystemStats)
19def get_system_stats(db: DBDep, admin: AdminDep):
20    mem = memory_usage()
21    cpu = cpu_usage()
22    system = crud.get_system_usage(db)
23    dbadmin: Union[Admin, None] = crud.get_admin(db, admin.username)
24
25    total_user = crud.get_users_count(db, admin=dbadmin if not admin.is_sudo else None)
26    users_active = crud.get_users_count(db, status=UserStatus.active, admin=dbadmin if not admin.is_sudo else None)
27    realtime_bandwidth_stats = realtime_bandwidth()
28
29    return SystemStats(
30        version=__version__,
31        mem_total=mem.total,
32        mem_used=mem.used,
33        cpu_cores=cpu.cores,
34        cpu_usage=cpu.percent,
35        total_user=total_user,
36        users_active=users_active,
37        incoming_bandwidth=system.uplink,
38        outgoing_bandwidth=system.downlink,
39        incoming_bandwidth_speed=realtime_bandwidth_stats.incoming_bytes,
40        outgoing_bandwidth_speed=realtime_bandwidth_stats.outgoing_bytes,
41    )
42
43
44@router.get("/hosts", response_model=Dict[int, List[InboundHost]])
45def get_hosts(db: DBDep, admin: SudoAdminDep):
46    hosts = crud.get_all_hosts(db)
47
48    return hosts
49
50
51@router.put('/api/hosts', response_model=Dict[str, List[InboundHost]])
52def modify_hosts(modified_hosts: Dict[str, List[InboundHost]],
53                 db: DBDep,
54                 admin: SudoAdminDep):
55    # validate
56    # for inbound_id, hosts in modified_hosts.items():
57    #     if not xray.config.inbounds_by_id.get(inbound_id):
58    #         raise HTTPException(status_code=400, detail=f"Inbound {inbound_id} doesn't exist")
59
60    for inbound_id, hosts in modified_hosts.items():
61        crud.update_hosts(db, inbound_id, hosts)
62
63    # xray.hosts.update()
64
65    hosts = {}
66    # for inbound_tag in xray.config.inbounds_by_tag:
67    #     hosts[inbound_tag] = crud.get_hosts(db, inbound_tag)
68
69    return hosts