"""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 common_util.common_util_audit import record_audit 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) -> int: """프로젝트 상태 문자열에서 워크플로 단계만 뽑는다. 진행도(%)는 내지 않는다 (2026-09-06 사용자 지시) — 화면은 워크플로 배지로 보여 주고, 배지는 `project_workflow_stages` 표를 근거로 삼는다. 상태 문자열로 따로 세면 근거가 둘이 되어 배지와 숫자가 어긋났다. """ value = status or "NEW" if value in {"WF1_ANALYZING", "WF1_FAILED"}: return 1 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 def _project_row(row: dict[str, Any]) -> dict[str, Any]: return {**row, "workflow_stage": _stage_from_status(row.get("status"))} 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, request: Any | None = None ) -> 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 record_audit( cursor, actor_id=actor_id, action="PROJECT_UPDATE", resource_type="project", resource_ref=project_id, request=request, ) await connection.commit() return changed async def soft_delete_project(project_id: str, actor_id: int, request: Any | None = None) -> 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 record_audit( cursor, actor_id=actor_id, action="PROJECT_DELETE", resource_type="project", resource_ref=project_id, request=request, ) await connection.commit() return changed 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.resource_ref, l.ip_address, 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}