Files
Aislo/B01_Dashboard/B01_Dashboard_Repository.py
T
eomsangdonandClaude Opus 5 2c03dbee3e fix(B01): 값 변경 없이 참여자만 저장할 때 404 반환 문제 수정
- update_project 가 rowcount 0(변경된 컬럼 없음)을 프로젝트 없음으로 오판하던 자리에 존재 확인 추가
- 참여자 수정권한·시스템 회사 역할 규칙 검증 테스트 추가 (tmp/tests, git 추적 제외)

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-06 11:46:11 +09:00

741 lines
31 KiB
Python

"""B01_Dashboard aiomysql Raw SQL 저장소."""
from __future__ import annotations
import json
import shutil
from datetime import datetime, timedelta
from typing import Any
import aiomysql
import psutil
from config.config_db import get_db_pool
from config.config_system import ADMIN_EMAIL, EMAIL_REVERIFY_DAYS
def _role(value: str | None) -> str:
return {"MASTER": "ADMIN", "MEMBER": "USER"}.get(value or "USER", value or "USER")
async def get_system_company_id() -> int | None:
"""시스템 관리 회사 = `.env` 관리자 계정이 속한 회사 (2026-09-06 사용자 확정).
개발사 자기 회사 한 곳뿐이라 따로 표시 칸을 두지 않고 이 한 줄로 판정한다.
"""
if not ADMIN_EMAIL:
return None
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
await cursor.execute(
"SELECT company_id FROM users WHERE email = %s AND deleted_at IS NULL",
(ADMIN_EMAIL.lower(),),
)
row = await cursor.fetchone()
return int(row["company_id"]) if row and row["company_id"] else None
async def role_for_company(company_id: int | None) -> str:
"""회사에 들어갈 때 받는 역할 — 시스템 회사면 시스템관리자, 아니면 일반사용자."""
if company_id is not None and int(company_id) == (await get_system_company_id() or 0):
return "SYSTEM_ADMIN"
return "USER"
def _stage_from_status(status: str | None) -> tuple[int, int]:
value = status or "NEW"
if value in {"WF1_ANALYZING", "WF1_FAILED"}:
return 1, round(1 / 7 * 100)
order = [
("FILE_UPLOADED", 1),
("WF1_COMPLETE", 2),
("WF2_COMPLETE", 3),
("WF3_COMPLETE", 4),
("WF4_COMPLETE", 5),
("WF5_COMPLETE", 6),
("WF6_COMPLETE", 7),
("DONE", 7),
("CONFIRMED", 7),
]
stage = 0
for token, idx in order:
if token in value:
stage = max(stage, idx)
return stage, round(stage / 7 * 100)
def _project_row(row: dict[str, Any]) -> dict[str, Any]:
stage, progress = _stage_from_status(row.get("status"))
return {**row, "workflow_stage": stage, "progress_percent": progress}
async def _project_rows(cursor: Any, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
from .B01_Dashboard_Repository_Members import list_project_member_ids
states = await get_workflow_states_for_projects(cursor, [r["id"] for r in rows])
members = await list_project_member_ids(cursor, [r["id"] for r in rows])
result = []
for r in rows:
p_row = _project_row(r)
p_row["workflow_state"] = states.get(r["id"], {"current_stage": 0, "stages": []})
# 참여자는 일반 사용자여도 그 프로젝트를 수정할 수 있다 (2026-09-06 사용자 확정).
p_row["member_user_ids"] = members.get(str(r["id"]), [])
result.append(p_row)
return result
async def get_dashboard_me(user_id: int) -> dict[str, Any] | None:
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
await cursor.execute(
"""SELECT u.id, u.email, u.name, u.position, u.department, u.phone,
u.company_id, u.role, u.is_master, u.status, u.last_login,
DATE_ADD(u.last_email_verified_at, INTERVAL %s DAY) AS auth_expires_at,
c.name AS company_name
FROM users u LEFT JOIN companies c ON c.id = u.company_id
WHERE u.id = %s AND u.deleted_at IS NULL""",
(EMAIL_REVERIFY_DAYS, user_id),
)
row = await cursor.fetchone()
if row:
row["role"] = _role(row.get("role"))
row["is_master"] = bool(row.get("is_master"))
return row
async def update_user_profile(user_id: int, data: dict[str, Any]) -> dict[str, Any] | None:
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor() as cursor:
await cursor.execute(
"""UPDATE users SET name = %s, position = %s, department = %s, phone = %s
WHERE id = %s AND deleted_at IS NULL""",
(
data["name"],
data.get("position"),
data.get("department"),
data.get("phone"),
user_id,
),
)
await connection.commit()
return await get_dashboard_me(user_id)
async def get_workflow_states_for_projects(
cursor: aiomysql.DictCursor, project_ids: list[str]
) -> dict[str, dict[str, Any]]:
if not project_ids:
return {}
format_strings = ",".join(["%s"] * len(project_ids))
await cursor.execute(
f"""
SELECT project_id, stage_no, stage_key, state, progress_percent, params, message,
DATE_FORMAT(started_at, '%%Y-%%m-%%dT%%H:%%i:%%s') AS started_at,
DATE_FORMAT(completed_at, '%%Y-%%m-%%dT%%H:%%i:%%s') AS completed_at
FROM project_workflow_stages
WHERE project_id IN ({format_strings})
ORDER BY project_id, stage_no ASC
""",
tuple(project_ids),
)
rows = await cursor.fetchall()
project_stages = {}
for r in rows:
pid = r["project_id"]
if pid not in project_stages:
project_stages[pid] = []
params_val = None
if r.get("params"):
try:
params_val = (
json.loads(r["params"]) if isinstance(r["params"], str) else r["params"]
)
except Exception:
params_val = r["params"]
project_stages[pid].append(
{
"stage_no": r["stage_no"],
"stage_key": r["stage_key"],
"state": r["state"],
"progress_percent": r["progress_percent"],
"params": params_val,
"message": r["message"],
"started_at": r["started_at"],
"completed_at": r["completed_at"],
}
)
result = {}
for pid in project_ids:
stages = project_stages.get(pid, [])
current_stage = 0
for stage in stages:
if stage["stage_no"] == 0:
continue
prev_stage = stages[stage["stage_no"] - 1]
if prev_stage["state"] == "COMPLETE":
current_stage = stage["stage_no"]
else:
break
result[pid] = {"current_stage": current_stage, "stages": stages}
return result
async def list_user_projects(user_id: int) -> list[dict[str, Any]]:
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
await cursor.execute(
"""SELECT id, company_id, name, region, road_type, project_year,
estimated_length_m, route_start_m, route_end_m,
memo, status, updated_at, created_at,
client_org, project_number, work_amount, design_date,
pm_user_id, field_lead_user_id, designer_user_id,
logo_asset_id, signature_asset_id
FROM projects WHERE user_id = %s AND deleted_at IS NULL
ORDER BY updated_at DESC, created_at DESC""",
(user_id,),
)
return await _project_rows(cursor, await cursor.fetchall())
async def list_company_projects(company_id: int) -> list[dict[str, Any]]:
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
await cursor.execute(
"""SELECT p.id, p.company_id, p.name, p.region, p.road_type, p.project_year,
p.estimated_length_m, p.route_start_m, p.route_end_m,
p.memo, p.status, p.updated_at, p.created_at,
p.client_org, p.project_number, p.work_amount, p.design_date,
p.pm_user_id, p.field_lead_user_id, p.designer_user_id,
p.logo_asset_id, p.signature_asset_id,
u.name AS owner_name, u.email AS owner_email
FROM projects p LEFT JOIN users u ON u.id = p.user_id
WHERE p.company_id = %s AND p.deleted_at IS NULL
ORDER BY p.updated_at DESC, p.created_at DESC""",
(company_id,),
)
return await _project_rows(cursor, await cursor.fetchall())
async def list_all_projects() -> list[dict[str, Any]]:
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
await cursor.execute(
"""SELECT p.id, p.company_id, p.name, p.region, p.road_type, p.project_year,
p.estimated_length_m, p.route_start_m, p.route_end_m,
p.memo, p.status, p.updated_at, p.created_at,
p.client_org, p.project_number, p.work_amount, p.design_date,
p.pm_user_id, p.field_lead_user_id, p.designer_user_id,
p.logo_asset_id, p.signature_asset_id,
u.name AS owner_name, u.email AS owner_email
FROM projects p LEFT JOIN users u ON u.id = p.user_id
WHERE p.deleted_at IS NULL
ORDER BY p.updated_at DESC, p.created_at DESC"""
)
return await _project_rows(cursor, await cursor.fetchall())
async def get_project(project_id: str) -> dict[str, Any] | None:
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
await cursor.execute(
"""SELECT id, user_id, company_id, name, region, road_type, project_year,
estimated_length_m, route_start_m, route_end_m, memo, status,
client_org, project_number, work_amount, design_date,
pm_user_id, field_lead_user_id, designer_user_id,
logo_asset_id, signature_asset_id
FROM projects WHERE id = %s AND deleted_at IS NULL""",
(project_id,),
)
return await cursor.fetchone()
async def update_project(project_id: str, data: dict[str, Any], actor_id: int) -> bool:
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor() as cursor:
await connection.begin()
await cursor.execute(
"""UPDATE projects
SET name = %s, region = %s, road_type = %s, project_year = %s,
estimated_length_m = %s, route_start_m = %s, route_end_m = %s,
memo = %s, status = COALESCE(%s, status),
client_org = %s, project_number = %s, work_amount = %s,
design_date = %s, pm_user_id = %s, field_lead_user_id = %s,
designer_user_id = %s, logo_asset_id = %s, signature_asset_id = %s
WHERE id = %s AND deleted_at IS NULL""",
(
data["name"],
data.get("region"),
data.get("road_type"),
data.get("project_year"),
data.get("estimated_length_m"),
data.get("route_start_m"),
data.get("route_end_m"),
data.get("memo"),
data.get("status"),
data.get("client_org"),
data.get("project_number"),
data.get("work_amount"),
data.get("design_date"),
data.get("pm_user_id"),
data.get("field_lead_user_id"),
data.get("designer_user_id"),
data.get("logo_asset_id"),
data.get("signature_asset_id"),
project_id,
),
)
changed = cursor.rowcount > 0
if not changed:
# 값이 하나도 안 바뀌면 rowcount 가 0 이다 — 프로젝트가 없는 것과는 다르다
# (참여자만 바꿀 때 「찾을 수 없습니다」로 튕기던 자리, 2026-09-06).
await cursor.execute(
"SELECT 1 FROM projects WHERE id = %s AND deleted_at IS NULL", (project_id,)
)
changed = await cursor.fetchone() is not None
else:
await cursor.execute(
"""INSERT INTO system_audit_logs (user_id, action, resource_type, resource_id)
VALUES (%s, 'PROJECT_UPDATE', 'project', NULL)""",
(actor_id,),
)
await connection.commit()
return changed
async def soft_delete_project(project_id: str, actor_id: int) -> bool:
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor() as cursor:
await connection.begin()
await cursor.execute(
"""UPDATE projects SET deleted_at = CURRENT_TIMESTAMP
WHERE id = %s AND deleted_at IS NULL""",
(project_id,),
)
changed = cursor.rowcount > 0
if changed:
await cursor.execute(
"""INSERT INTO system_audit_logs (user_id, action, resource_type, resource_id)
VALUES (%s, 'PROJECT_DELETE', 'project', NULL)""",
(actor_id,),
)
await connection.commit()
return changed
async def get_user_company(company_id: int | None) -> dict[str, Any] | None:
if company_id is None:
return None
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
await cursor.execute(
"""SELECT c.id, c.name, c.business_registration_number, c.business_address,
c.business_owner, c.business_status, c.logo_asset_id,
COUNT(DISTINCT u.id) AS user_count,
COUNT(DISTINCT p.id) AS project_count
FROM companies c
LEFT JOIN users u ON u.company_id = c.id AND u.deleted_at IS NULL
LEFT JOIN projects p ON p.company_id = c.id AND p.deleted_at IS NULL
WHERE c.id = %s AND c.deleted_at IS NULL
GROUP BY c.id""",
(company_id,),
)
return await cursor.fetchone()
async def update_company(company_id: int, data: dict[str, Any]) -> bool:
"""회사 정보 수정 (2026-09-06 사용자 지시) — 회사 관리자는 자기 회사, 시스템관리자는 전체."""
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor() as cursor:
await cursor.execute(
"""UPDATE companies
SET name = %s, business_registration_number = %s,
business_address = %s, business_owner = %s
WHERE id = %s AND deleted_at IS NULL""",
(
data["name"],
data["business_registration_number"],
data.get("business_address"),
data.get("business_owner"),
company_id,
),
)
changed = cursor.rowcount > 0
await connection.commit()
return changed
async def search_companies(query: str) -> list[dict[str, Any]]:
pool = get_db_pool()
pattern = f"%{query}%"
async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
await cursor.execute(
"""SELECT id, name, business_registration_number, business_status
FROM companies
WHERE deleted_at IS NULL AND (name LIKE %s OR business_registration_number LIKE %s)
ORDER BY name LIMIT 20""",
(pattern, pattern),
)
return list(await cursor.fetchall())
async def create_company(user_id: int, data: dict[str, Any]) -> dict[str, Any]:
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor() as cursor:
try:
await connection.begin()
await cursor.execute(
"""INSERT INTO companies
(name, business_registration_number, business_address, business_owner,
business_status, master_user_id, created_by, status)
VALUES (%s, %s, %s, %s, '활동중', %s, %s, 'ACTIVE')""",
(
data["name"],
data["business_registration_number"],
data.get("business_address"),
data.get("business_owner"),
user_id,
user_id,
),
)
company_id = cursor.lastrowid
await cursor.execute(
"""UPDATE users
SET company_id = %s, role = 'ADMIN', is_master = TRUE,
status = 'ACTIVE'
WHERE id = %s""",
(company_id, user_id),
)
await cursor.execute(
"""INSERT INTO system_audit_logs (user_id, action, resource_type, resource_id)
VALUES (%s, 'COMPANY_CREATE', 'company', %s)""",
(user_id, company_id),
)
await connection.commit()
return {"company_id": company_id, "status": "ACTIVE"}
except Exception:
await connection.rollback()
raise
async def create_system_company(actor_id: int, data: dict[str, Any]) -> dict[str, Any]:
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor() as cursor:
try:
await connection.begin()
await cursor.execute(
"""INSERT INTO companies
(name, business_registration_number, business_address, business_owner,
business_status, created_by, status)
VALUES (%s, %s, %s, %s, '활동중', %s, 'ACTIVE')""",
(
data["name"],
data["business_registration_number"],
data.get("business_address"),
data.get("business_owner"),
actor_id,
),
)
company_id = cursor.lastrowid
await cursor.execute(
"""INSERT INTO system_audit_logs (user_id, action, resource_type, resource_id)
VALUES (%s, 'COMPANY_CREATE', 'company', %s)""",
(actor_id, company_id),
)
await connection.commit()
return {"company_id": company_id, "status": "ACTIVE"}
except Exception:
await connection.rollback()
raise
async def join_company(user_id: int, company_id: int) -> dict[str, Any] | None:
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor() as cursor:
try:
await connection.begin()
# 이미 소속이 있는 사람의 재신청은 막는다 (2026-09-06 사용자 지시) —
# 예전에는 신청 즉시 소속이 풀리고 승인대기로 떨어져 회사를 잃었다.
await cursor.execute(
"SELECT company_id FROM users WHERE id = %s AND deleted_at IS NULL FOR UPDATE",
(user_id,),
)
current = await cursor.fetchone()
if current and current[0]:
await connection.rollback()
return None
await cursor.execute(
"""INSERT INTO join_requests (user_id, company_id, status)
VALUES (%s, %s, 'PENDING')
ON DUPLICATE KEY UPDATE status = 'PENDING', requested_at = CURRENT_TIMESTAMP,
reviewed_by = NULL, reviewed_at = NULL""",
(user_id, company_id),
)
request_id = cursor.lastrowid
await cursor.execute(
"UPDATE users SET status = 'PENDING', company_id = NULL WHERE id = %s", (user_id,)
)
await connection.commit()
return {"join_request_id": request_id, "status": "PENDING"}
except Exception:
await connection.rollback()
raise
async def list_join_requests(company_id: int | None = None) -> list[dict[str, Any]]:
# 처리 끝난 신청은 목록에 남기지 않는다 (2026-09-06 사용자 지시) — 승인된 사람은
# 사용자 관리 목록에 이미 있어 같은 사람이 두 번 보였다.
where = "WHERE jr.status = 'PENDING'" + (" AND jr.company_id = %s" if company_id else "")
params = (company_id,) if company_id else ()
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
await cursor.execute(
f"""SELECT jr.id, jr.user_id, jr.company_id, jr.requested_at, jr.status,
u.email AS user_email, u.name AS user_name, c.name AS company_name
FROM join_requests jr
JOIN users u ON u.id = jr.user_id
JOIN companies c ON c.id = jr.company_id
{where}
ORDER BY jr.requested_at DESC LIMIT 100""",
params,
)
return list(await cursor.fetchall())
async def process_join_request(
request_id: int, reviewer_id: int, approved: bool, company_id: int | None = None
) -> bool:
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
await connection.begin()
where = "id = %s AND status = 'PENDING'"
params: tuple[Any, ...] = (request_id,)
if company_id is not None:
where += " AND company_id = %s"
params = (request_id, company_id)
await cursor.execute(f"SELECT * FROM join_requests WHERE {where} FOR UPDATE", params)
row = await cursor.fetchone()
if not row:
await connection.rollback()
return False
status = "APPROVED" if approved else "REJECTED"
await cursor.execute(
"""UPDATE join_requests SET status = %s, reviewed_by = %s,
reviewed_at = CURRENT_TIMESTAMP WHERE id = %s""",
(status, reviewer_id, request_id),
)
if approved:
# 시스템 회사로 들어오면 역할도 함께 올린다 (2026-09-06 사용자 확정).
role = await role_for_company(int(row["company_id"]))
await cursor.execute(
"""UPDATE users SET company_id = %s, status = 'ACTIVE', role = %s
WHERE id = %s""",
(row["company_id"], role, row["user_id"]),
)
else:
await cursor.execute(
"UPDATE users SET status = 'REJECTED' WHERE id = %s", (row["user_id"],)
)
await connection.commit()
return True
async def list_all_companies() -> list[dict[str, Any]]:
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
await cursor.execute(
"""SELECT c.id, c.name, c.business_registration_number, c.business_status,
c.logo_asset_id, c.created_at, COUNT(DISTINCT u.id) AS user_count,
COUNT(DISTINCT p.id) AS project_count
FROM companies c
LEFT JOIN users u ON u.company_id = c.id AND u.deleted_at IS NULL
LEFT JOIN projects p ON p.company_id = c.id AND p.deleted_at IS NULL
WHERE c.deleted_at IS NULL GROUP BY c.id ORDER BY c.created_at DESC"""
)
return list(await cursor.fetchall())
async def list_all_users() -> list[dict[str, Any]]:
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
await cursor.execute(
"""SELECT u.id, u.email, u.name, u.position, u.department, u.phone,
u.role, u.status, u.company_id,
c.name AS company_name, u.last_login
FROM users u LEFT JOIN companies c ON c.id = u.company_id
WHERE u.deleted_at IS NULL ORDER BY u.created_at DESC LIMIT 200"""
)
rows = list(await cursor.fetchall())
for row in rows:
row["role"] = _role(row.get("role"))
return rows
async def change_user_role(user_id: int, role: str) -> bool:
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor() as cursor:
# is_master 도 권한 판정에 쓰이므로 역할과 어긋나지 않게 함께 맞춘다.
await cursor.execute(
"UPDATE users SET role = %s, is_master = %s WHERE id = %s",
(role, role in ("ADMIN", "SYSTEM_ADMIN"), user_id),
)
changed = cursor.rowcount > 0
await connection.commit()
return changed
async def get_user_admin_target(user_id: int) -> dict[str, Any] | None:
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
await cursor.execute(
"""SELECT id, company_id, role, is_master, status
FROM users WHERE id = %s AND deleted_at IS NULL""",
(user_id,),
)
row = await cursor.fetchone()
if row:
row["role"] = _role(row.get("role"))
row["is_master"] = bool(row.get("is_master"))
return row
async def update_admin_user(user_id: int, data: dict[str, Any]) -> bool:
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor() as cursor:
await cursor.execute(
"""UPDATE users
SET name = %s, position = %s, department = %s, phone = %s,
status = COALESCE(%s, status)
WHERE id = %s AND deleted_at IS NULL""",
(
data["name"],
data.get("position"),
data.get("department"),
data.get("phone"),
data.get("status"),
user_id,
),
)
changed = cursor.rowcount > 0
await connection.commit()
return changed
async def count_company_admins(company_id: int) -> int:
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
await cursor.execute(
"""SELECT COUNT(*) AS cnt FROM users
WHERE company_id = %s AND role = 'ADMIN' AND deleted_at IS NULL""",
(company_id,),
)
row = await cursor.fetchone()
return int(row["cnt"] if row else 0)
async def soft_delete_user(user_id: int) -> bool:
"""사용자 계정을 지운다 (2026-09-06 사용자 확정 — 회사 관리자도 자기 회사 사람은 삭제).
지운 표시만 남기고 소속을 푼다. 상태가 INACTIVE 라 다음 요청에서 세션이 끊긴다.
이메일은 그대로 둔다 — 같은 주소로 다시 가입하려면 시스템 관리자가 되살려야 한다.
"""
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor() as cursor:
await cursor.execute(
"""UPDATE users
SET deleted_at = CURRENT_TIMESTAMP, status = 'INACTIVE',
company_id = NULL, is_master = FALSE
WHERE id = %s AND deleted_at IS NULL""",
(user_id,),
)
changed = cursor.rowcount > 0
await connection.commit()
return changed
async def assign_user_company(user_id: int, company_id: int | None) -> bool:
status = "ACTIVE" if company_id else "NO_COMPANY"
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor() as cursor:
await cursor.execute(
"UPDATE users SET company_id = %s, status = %s WHERE id = %s",
(company_id, status, user_id),
)
changed = cursor.rowcount > 0
await connection.commit()
return changed
async def list_audit_logs(limit: int, offset: int) -> dict[str, Any]:
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
await cursor.execute("SELECT COUNT(*) AS total FROM system_audit_logs")
total = (await cursor.fetchone())["total"]
await cursor.execute(
"""SELECT l.id, l.user_id, u.email, l.action, l.resource_type,
l.resource_id, l.timestamp
FROM system_audit_logs l LEFT JOIN users u ON u.id = l.user_id
ORDER BY l.timestamp DESC LIMIT %s OFFSET %s""",
(limit, offset),
)
return {"total": total, "items": list(await cursor.fetchall())}
async def get_system_resources(days: int = 30) -> dict[str, Any]:
total, used, _free = shutil.disk_usage(".")
disk_percent = used / total * 100 if total else 0
cpu_percent = psutil.cpu_percent(interval=0.1)
memory_percent = psutil.virtual_memory().percent
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
await cursor.execute(
"""SELECT
(SELECT COUNT(*) FROM sessions
WHERE expires_at > CURRENT_TIMESTAMP) AS active_user_count,
(SELECT COUNT(*) FROM projects WHERE deleted_at IS NULL) AS active_project_count"""
)
stats = await cursor.fetchone()
current = {
"cpu_usage_percent": round(cpu_percent, 2),
"memory_usage_percent": round(memory_percent, 2),
"disk_usage_percent": round(disk_percent, 2),
"active_user_count": stats["active_user_count"],
"active_project_count": stats["active_project_count"],
"total_storage_mb": round(used / 1024 / 1024, 2),
}
since = datetime.utcnow() - timedelta(days=days)
# 다운샘플링: 조회 구간을 최대 TARGET_POINTS개 시간버킷으로 나눠 평균.
# 계측 간격(2분)보다 버킷이 작으면 원본 그대로(버킷=간격) 반환.
target_points = 300
await cursor.execute(
"SELECT COUNT(*) AS cnt FROM system_resources WHERE timestamp >= %s",
(since,),
)
row_cnt_res = await cursor.fetchone()
row_count = row_cnt_res["cnt"] if row_cnt_res else 0
if row_count <= target_points:
bucket_seconds = 120
else:
bucket_seconds = max(120, (days * 86400) // target_points)
date_fmt = "%%Y-%%m-%%dT%%H:%%i:%%s"
await cursor.execute(
f"""SELECT
DATE_FORMAT(
FROM_UNIXTIME(FLOOR(UNIX_TIMESTAMP(timestamp) / %s) * %s),
'{date_fmt}'
) AS timestamp,
ROUND(AVG(cpu_usage_percent), 2) AS cpu_usage_percent,
ROUND(AVG(memory_usage_percent), 2) AS memory_usage_percent,
ROUND(AVG(disk_usage_percent), 2) AS disk_usage_percent,
MAX(active_user_count) AS active_user_count,
MAX(active_project_count) AS active_project_count,
ROUND(AVG(total_storage_mb), 2) AS total_storage_mb
FROM system_resources
WHERE timestamp >= %s
GROUP BY FLOOR(UNIX_TIMESTAMP(timestamp) / %s)
ORDER BY timestamp""",
(bucket_seconds, bucket_seconds, since, bucket_seconds),
)
return {"current": current, "history": list(await cursor.fetchall()), "stats": current}