zioinfo-mail/workspace/guardia-itsm/routers/asset_qr.py

"""
자산 QR 태그 — 서버·장비에 QR 부착 → 스캔 → CMDB 조회

엔드포인트:
  POST /api/asset-qr/generate/{server_id} — QR 코드 생성
  GET  /api/asset-qr/scan/{qr_token}      — QR 스캔 → 자산 정보
  POST /api/asset-qr/checkin/{qr_token}   — 실사 체크인
  GET  /api/asset-qr/print/{server_id}    — 인쇄용 QR 라벨 HTML
  GET  /api/asset-qr/batch-print          — 다수 자산 일괄 인쇄
  GET  /api/asset-qr/list                 — QR 발행 목록
  GET  /api/asset-qr/audit-log/{server_id}— 스캔 이력
"""
from __future__ import annotations

import base64
import io
import uuid
from datetime import datetime
from typing import Optional

from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi.responses import HTMLResponse
from pydantic import BaseModel
from sqlalchemy import select, desc
from sqlalchemy.ext.asyncio import AsyncSession

from core.auth import get_current_user, require_admin_role
from database import get_db
from models import User, Server, AssetQRToken, AssetQRScanLog

router = APIRouter(prefix="/api/asset-qr", tags=["자산 QR 태그"])

BASE_URL = "https://zioinfo.co.kr:8443"


def _gen_qr(url: str) -> str:
    """QR base64 PNG 생성."""
    try:
        import qrcode
        qr = qrcode.QRCode(version=1, box_size=8, border=3)
        qr.add_data(url)
        qr.make(fit=True)
        img = qr.make_image(fill_color="#003366", back_color="white")
        buf = io.BytesIO()
        img.save(buf, format='PNG')
        return base64.b64encode(buf.getvalue()).decode()
    except ImportError:
        return ""


class CheckinRequest(BaseModel):
    note: Optional[str] = None
    location: Optional[str] = None


@router.post("/generate/{server_id}")
async def generate_qr(
    server_id: int,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(require_admin_role),
):
    srv_row = await db.execute(select(Server).where(Server.id == server_id))
    server = srv_row.scalar_one_or_none()
    if not server:
        raise HTTPException(404, "서버 없음")

    # 기존 QR 확인
    existing = await db.execute(select(AssetQRToken).where(AssetQRToken.server_id == server_id))
    token_obj = existing.scalar_one_or_none()

    if not token_obj:
        token = uuid.uuid4().hex
        scan_url = f"{BASE_URL}/api/asset-qr/scan/{token}"
        qr_b64 = _gen_qr(scan_url)
        token_obj = AssetQRToken(
            qr_token=token,
            server_id=server_id,
            qr_data=qr_b64,
            scan_count=0,
            created_at=datetime.utcnow(),
        )
        db.add(token_obj)
        await db.commit()
        await db.refresh(token_obj)

    return {
        "ok": True,
        "qr_token": token_obj.qr_token,
        "scan_url": f"{BASE_URL}/api/asset-qr/scan/{token_obj.qr_token}",
        "print_url": f"{BASE_URL}/api/asset-qr/print/{server_id}",
        "qr_image_b64": token_obj.qr_data,
    }


@router.get("/scan/{qr_token}")
async def scan_qr(
    qr_token: str,
    request: Request,
    db: AsyncSession = Depends(get_db),
):
    """QR 스캔 → 자산 정보 반환. 인증 불필요 (공개 기본 정보)."""
    row = await db.execute(select(AssetQRToken).where(AssetQRToken.qr_token == qr_token))
    token_obj = row.scalar_one_or_none()
    if not token_obj:
        raise HTTPException(404, "유효하지 않은 QR 코드")

    srv_row = await db.execute(select(Server).where(Server.id == token_obj.server_id))
    server = srv_row.scalar_one_or_none()
    if not server:
        raise HTTPException(404, "서버 정보 없음")

    # 스캔 횟수 증가
    token_obj.scan_count = (token_obj.scan_count or 0) + 1
    token_obj.last_scan_at = datetime.utcnow()

    # 스캔 로그
    log = AssetQRScanLog(
        qr_token=qr_token,
        scan_type="VIEW",
        user_agent=request.headers.get("User-Agent", "")[:200],
        scanned_at=datetime.utcnow(),
    )
    db.add(log)
    await db.commit()

    return {
        "server_id": server.id,
        "hostname": server.hostname or "미설정",
        "ip_addr": "***.***.***.**",   # 공개 응답에서 IP 마스킹
        "os_type": server.os_type or "미상",
        "cpu_cores": server.cpu_cores,
        "memory_gb": round((server.memory_mb or 0) / 1024, 1),
        "scan_count": token_obj.scan_count,
        "last_scan_at": token_obj.last_scan_at,
        "checkin_url": f"{BASE_URL}/api/asset-qr/checkin/{qr_token}",
    }


@router.post("/checkin/{qr_token}")
async def checkin(
    qr_token: str,
    req: CheckinRequest,
    request: Request,
    db: AsyncSession = Depends(get_db),
):
    """실사 체크인 — 장비 위치·상태 기록."""
    row = await db.execute(select(AssetQRToken).where(AssetQRToken.qr_token == qr_token))
    token_obj = row.scalar_one_or_none()
    if not token_obj:
        raise HTTPException(404)

    log = AssetQRScanLog(
        qr_token=qr_token,
        scan_type="CHECKIN",
        user_agent=request.headers.get("User-Agent", "")[:200],
        location=req.location,
        note=req.note,
        scanned_at=datetime.utcnow(),
    )
    db.add(log)
    await db.commit()
    return {"ok": True, "message": "실사 체크인 완료", "checked_at": log.scanned_at}


@router.get("/print/{server_id}", response_class=HTMLResponse)
async def print_label(
    server_id: int,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(get_current_user),
):
    """인쇄용 QR 라벨 HTML (50×30mm 라벨 크기)."""
    srv_row = await db.execute(select(Server).where(Server.id == server_id))
    server = srv_row.scalar_one_or_none()
    if not server:
        raise HTTPException(404)

    token_row = await db.execute(select(AssetQRToken).where(AssetQRToken.server_id == server_id))
    token_obj = token_row.scalar_one_or_none()
    if not token_obj:
        raise HTTPException(400, "먼저 QR을 생성하세요. POST /api/asset-qr/generate/{server_id}")

    qr_img = f'<img src="data:image/png;base64,{token_obj.qr_data}" width="80" height="80">' if token_obj.qr_data else ''

    html = f"""<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<title>QR 라벨 — {server.hostname}</title>
<style>
  @page {{ size: 50mm 30mm; margin: 1mm }}
  body {{ font-family: sans-serif; margin: 0; padding: 2mm }}
  .label {{ display: flex; align-items: center; gap: 2mm; width: 46mm }}
  .qr {{ flex-shrink: 0 }}
  .info {{ font-size: 7pt; line-height: 1.4 }}
  .host {{ font-weight: bold; font-size: 8pt; color: #003366 }}
  .btn {{ display: inline-block; margin: 4mm 0; padding: 4px 12px; background: #003366; color: white; border: none; border-radius: 4px; cursor: pointer; font-size: 12px }}
  @media print {{ .btn {{ display: none }} }}
</style>
</head>
<body>
<button class="btn" onclick="window.print()">🖨️ 인쇄</button>
<div class="label">
  <div class="qr">{qr_img}</div>
  <div class="info">
    <div class="host">{server.hostname or '미설정'}</div>
    <div>ID: {server.id}</div>
    <div>{server.os_type or ''}</div>
    <div>GUARDiA ITSM</div>
  </div>
</div>
</body>
</html>"""
    return HTMLResponse(html)


@router.get("/batch-print", response_class=HTMLResponse)
async def batch_print(
    server_ids: str,  # 쉼표 구분 "1,2,3"
    db: AsyncSession = Depends(get_db),
    user: User = Depends(get_current_user),
):
    """다수 서버 QR 라벨 일괄 인쇄."""
    ids = [int(x) for x in server_ids.split(",") if x.strip().isdigit()]
    labels = []
    for sid in ids[:50]:
        srv_row = await db.execute(select(Server).where(Server.id == sid))
        server = srv_row.scalar_one_or_none()
        if not server:
            continue
        token_row = await db.execute(select(AssetQRToken).where(AssetQRToken.server_id == sid))
        token_obj = token_row.scalar_one_or_none()
        if not token_obj:
            continue
        qr_img = f'<img src="data:image/png;base64,{token_obj.qr_data}" width="70" height="70">' if token_obj.qr_data else ''
        labels.append(f"""
        <div class="label">
          <div class="qr">{qr_img}</div>
          <div class="info">
            <div class="host">{server.hostname or '미설정'}</div>
            <div>ID: {server.id}</div>
            <div>{server.os_type or ''}</div>
          </div>
        </div>""")

    html = f"""<!DOCTYPE html>
<html><head><meta charset="UTF-8">
<style>
  @page {{ margin: 5mm }}
  body {{ font-family: sans-serif }}
  .grid {{ display: flex; flex-wrap: wrap; gap: 2mm }}
  .label {{ display: flex; align-items: center; gap: 2mm; width: 50mm; height: 30mm; border: 0.3mm solid #ccc; padding: 1mm; page-break-inside: avoid }}
  .info {{ font-size: 7pt; line-height: 1.4 }}
  .host {{ font-weight: bold; font-size: 8pt; color: #003366 }}
  .btn {{ padding: 4px 12px; background: #003366; color: white; border: none; border-radius: 4px; cursor: pointer; margin-bottom: 4mm }}
  @media print {{ .btn {{ display: none }} }}
</style>
</head>
<body>
<button class="btn" onclick="window.print()">🖨️ 전체 인쇄 ({len(labels)}개)</button>
<div class="grid">{''.join(labels)}</div>
</body></html>"""
    return HTMLResponse(html)


@router.get("/list")
async def list_qr_tokens(
    db: AsyncSession = Depends(get_db),
    user: User = Depends(get_current_user),
):
    rows = await db.execute(
        select(AssetQRToken, Server.hostname, Server.ip_addr).join(
            Server, AssetQRToken.server_id == Server.id
        ).order_by(desc(AssetQRToken.created_at)).limit(100)
    )
    return [
        {
            "server_id": r.AssetQRToken.server_id,
            "hostname": r.hostname or "미설정",
            "ip": r.ip_addr,
            "scan_count": r.AssetQRToken.scan_count,
            "last_scan": r.AssetQRToken.last_scan_at,
            "qr_url": f"{BASE_URL}/api/asset-qr/scan/{r.AssetQRToken.qr_token}",
            "print_url": f"{BASE_URL}/api/asset-qr/print/{r.AssetQRToken.server_id}",
        }
        for r in rows.all()
    ]


@router.get("/audit-log/{server_id}")
async def get_scan_log(
    server_id: int,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(get_current_user),
):
    token_row = await db.execute(select(AssetQRToken).where(AssetQRToken.server_id == server_id))
    token_obj = token_row.scalar_one_or_none()
    if not token_obj:
        return {"logs": []}
    rows = await db.execute(
        select(AssetQRScanLog).where(AssetQRScanLog.qr_token == token_obj.qr_token)
        .order_by(desc(AssetQRScanLog.scanned_at)).limit(50)
    )
    logs = rows.scalars().all()
    return {"logs": [
        {"type": l.scan_type, "location": l.location, "note": l.note,
         "scanned_at": l.scanned_at}
        for l in logs
    ]}