Aislo/B01_Dashboard/B01_Dashboard_Repository_Assets.py

"""B01_Dashboard 회사 공유 도면 자산(로고·서명) 저장소 — `company_assets` (013).

그림은 `storage/{회사}/assets/` 에 파일로 두고 표에는 경로만 담는다. 자산은 사용자 계정에
물릴 수도(개인 서명) 아닐 수도(회사 공용 직인) 있다 — `user_id` NULL 이 공용.
"""

from __future__ import annotations

import os
import uuid
from typing import Any

import aiomysql

from config.config_db import get_db_pool
from config.config_system import STORAGE_BASE_DIR

ASSET_KINDS = ("LOGO", "SIGNATURE")
ASSET_EXTENSIONS = {".png", ".jpg", ".jpeg", ".webp", ".svg"}
ASSET_MAX_BYTES = 2 * 1024 * 1024


async def list_company_assets(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 a.id, a.company_id, a.kind, a.label, a.file_path, a.user_id,
                      u.name AS user_name, a.created_at
               FROM company_assets a LEFT JOIN users u ON u.id = a.user_id
               WHERE a.company_id = %s AND a.deleted_at IS NULL
               ORDER BY a.kind, a.label, a.id""",
            (company_id,),
        )
        return list(await cursor.fetchall())


async def get_company_asset(asset_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, kind, label, file_path, user_id
               FROM company_assets WHERE id = %s AND deleted_at IS NULL""",
            (asset_id,),
        )
        return await cursor.fetchone()


def write_company_asset_file(company_id: int, kind: str, suffix: str, blob: bytes) -> str:
    """그림을 `storage/{회사}/assets/` 에 쓰고 storage 기준 상대 경로를 돌려준다."""
    folder = os.path.join(STORAGE_BASE_DIR, str(company_id), "assets")
    os.makedirs(folder, exist_ok=True)
    filename = f"{kind.lower()}_{uuid.uuid4().hex}{suffix}"
    with open(os.path.join(folder, filename), "wb") as file:
        file.write(blob)
    return f"storage/{company_id}/assets/{filename}"


async def create_company_asset(
    company_id: int, kind: str, label: str, file_path: str, user_id: int | None, actor_id: int
) -> int:
    pool = get_db_pool()
    async with pool.acquire() as connection, connection.cursor() as cursor:
        await cursor.execute(
            """INSERT INTO company_assets (company_id, kind, label, file_path, user_id, created_by)
               VALUES (%s, %s, %s, %s, %s, %s)""",
            (company_id, kind, label, file_path, user_id, actor_id),
        )
        await connection.commit()
        return int(cursor.lastrowid)


async def update_company_asset(asset_id: int, label: str, user_id: int | None) -> bool:
    pool = get_db_pool()
    async with pool.acquire() as connection, connection.cursor() as cursor:
        await cursor.execute(
            """UPDATE company_assets SET label = %s, user_id = %s
               WHERE id = %s AND deleted_at IS NULL""",
            (label, user_id, asset_id),
        )
        await connection.commit()
        return cursor.rowcount > 0


async def delete_company_asset(asset_id: int) -> bool:
    """소프트 삭제 — 프로젝트가 물고 있던 참조는 도면 읽을 때 LEFT JOIN 으로 빈칸이 된다."""
    pool = get_db_pool()
    async with pool.acquire() as connection, connection.cursor() as cursor:
        await cursor.execute(
            """UPDATE company_assets SET deleted_at = CURRENT_TIMESTAMP
               WHERE id = %s AND deleted_at IS NULL""",
            (asset_id,),
        )
        await connection.commit()
        return cursor.rowcount > 0