[Python] NCP 월별 사용 요금 api로 가져오기

개요

프로젝트에 사용되는 계정 요금 확인이 주기적으로 필요했습니다.

지원팀은 일반 사무직군으로 하나씩 접속해서 확인할 계획을 가지고 있었습니다.

바로 옆자리에서 지켜보니 사람 손으로 확인하기에는 계정이 많아(대략 100개..?) 달에 한번만 확인하려 했습니다.

옆에서 스트레스 받고 있는 걸 보니, 도움을 주고 싶어 자동화를 해보기로 결정했습니다.

 

동작 방식

Naver Cloud에서 제공하는 API를 통해 비용을 가져와야 하기 때문에 각 계정에 API Key가 필요했습니다.

먼저 엑셀에 아래와 같은 유형으로 계정명, accesskey, secretkey가 입력되어야 합니다.

 

api key는 클라우드 운영에 있어서 매우 민감 정보로 문서에 암호를 지정하지 않을 수가 없었고,

암호를 코드 내에서 해제하기로 하였습니다.

 

패스워드 입력 방법은 두가지였습니다.

 

1. 사용자가 input 한다.

2. env에 넣는다.

 

실제 사용할 담당자들의 의견은 입력을 안했으면 더 좋겠다~ 가 있어서 env 파일을 사용하기로 했습니다.

또 이 프로그램을 사용하는 대상이 일반 사무 직군이기 때문에 python 코드를 실행하기에는 어려움이 있다고 판단하여, flask를 통해 웹에 표시해주기로 했습니다.

 

flask에서 표시하는 방법은 api 처리가 완료된 csv 파일의 데이터를 pandas를 통해 df에 넣는 방식을 사용했습니다.

단, flask 앱 종료 시 보안을 위해 후 처리된 csv 파일은 삭제하도록 정했습니다.

 

전체적인 프로세스는 아래와 같습니다.

 

• bill_api.py

import sys
import os
import hashlib
import hmac
import base64
import requests
import time
import json
from datetime import datetime, timedelta
import csv
import msoffcrypto
import io
import pandas as pd
import openpyxl
from dotenv import load_dotenv

# api 시그니처 생성
def get_api_signature(method, url, timestamp, access_key, secret_key):
    message = method + " " + url + "\n" + timestamp + "\n" + access_key
    message = bytes(message, 'UTF-8')
    secret_key = bytes(secret_key, 'UTF-8')
    signingKey = base64.b64encode(hmac.new(secret_key, message, digestmod=hashlib.sha256).digest())
    return signingKey

# api 호출
def process_api_key(access_key, secret_key, start_month, end_month):
    timestamp = str(int(time.time() * 1000))
    apicall_method = "GET"
    api_server = "https://billingapi.apigw.ntruss.com"
    api_url = "/billing/v1/cost/getDemandCostList"
    api_url = api_url + f"?startMonth={start_month}&endMonth={end_month}&responseFormatType=json"

    signingKey = get_api_signature(apicall_method, api_url, timestamp, access_key, secret_key)

    http_header = {
        'x-ncp-apigw-timestamp': timestamp,
        'x-ncp-iam-access-key': access_key,
        'x-ncp-apigw-signature-v2': signingKey
    }

    response = requests.get(api_server + api_url, headers=http_header)

    try:
        amount = response.json()
        demand_cost_list = amount["getDemandCostListResponse"]["demandCostList"]
        
        total_use_amount = 0
        monthly_amounts = {}
        for item in demand_cost_list:
            demand_month = item["demandMonth"]
            use_amount = item["useAmount"]
            monthly_amounts[demand_month] = use_amount
            total_use_amount += use_amount
            residuum_credit = 3000000 - total_use_amount
    
        return monthly_amounts, total_use_amount, residuum_credit

    except (json.JSONDecodeError, KeyError, IndexError) as e:
        print(f"Error processing API response: {str(e)}")
        print("API response:", response.text)
        return {}, 0

# api 저장된 excel을 data frame으로 저장
def read_api_keys_from_excel(file_path, password):
    try:
        temp = io.BytesIO()
        with open(file_path, 'rb') as f:
            excel = msoffcrypto.OfficeFile(f)
            excel.load_key(password)
            excel.decrypt(temp)
        
        df = pd.read_excel(temp)
        del temp

        api_keys = []
        for _, row in df.iterrows():
            if len(row) >= 4:
                account = row[1]
                access_key = row[2]
                secret_key = row[3]
                api_keys.append((account, access_key, secret_key))
        
        return api_keys
    except Exception as e:
        print(f"Excel 파일을 읽는 중 오류가 발생했습니다: {e}")
        return []

# 3개월 동안의 데이터 추출
def get_all_months(start_month, end_month):
    start = datetime.strptime(start_month, "%Y%m")
    end = datetime.strptime(end_month, "%Y%m")
    months = []
    current = start
    while current <= end:
        months.append(current.strftime("%Y%m"))
        current += timedelta(days=32)
        current = current.replace(day=1)
    return months

# API 결과값 csv 파일에 저장, 계정별 row로 저장
def save_results_to_csv(results, start_month, end_month, output_path):
    try:
        all_months = get_all_months(start_month, end_month)
        fieldnames = ['계정'] + all_months + ['3개월 합산 요금', '잔여 크레딧']

        with open(output_path, 'w', newline='', encoding='utf-8-sig') as csvfile:
            writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
            writer.writeheader()

            for account, data in results.items():
                row = {'계정': account}
                for month in all_months:
                    amount = data['monthly'].get(month, 0)
                    row[month] = f"{amount:,}원"
                row['3개월 합산 요금'] = f"{data['total']:,}원"
                row['잔여 크레딧'] = f"{data['credit']:,}원"
                writer.writerow(row)

        print(f"결과가 {output_path}에 저장되었습니다.")
    except Exception as e:
        print(f"결과 저장 중 오류 발생: {e}")

def main():
    print("최대 3개월만 조회 가능")
    
    current_date = datetime.now()
    end_month = current_date.strftime("%Y%m")
    start_month = str(int(current_date.strftime("%Y%m")) - 2)
    
    print(f"기본 조회 기간: {start_month} ~ {end_month}")
    
    #excel_file_path = "./api_e.xlsx"
    excel_file_path = "/usr/src/app/api_e.xlsx"
    load_dotenv()
    password = os.getenv("EXCEL_PW")
    
    api_keys = read_api_keys_from_excel(excel_file_path, password)
    
    if not api_keys:
        print("API 키를 읽어올 수 없습니다. 프로그램을 종료합니다.")
        return
    global output_file
    #output_file = f"./billing_result_{current_date.strftime('%Y%m%d')}.csv"
    output_file = f"/usr/src/app/billing_result_{current_date.strftime('%Y%m%d')}.csv"
    print(f"결과는 {output_file}에 저장됩니다.")

    results = {}
    grand_total = 0
    all_months = get_all_months(start_month, end_month)

    for account, access_key, secret_key in api_keys:
        print(f"\n처리 중인 계정: {account}")
        monthly_amounts, total, residuum_credit = process_api_key(access_key, secret_key, start_month, end_month)
        
        # 모든 월에 대해 데이터 확인, 없으면 0으로 설정
        for month in all_months:
            if month not in monthly_amounts:
                monthly_amounts[month] = 0
        
        results[account] = {
            'monthly': monthly_amounts,
            'total': total,
            'credit': residuum_credit
        }
        
        for month in all_months:
            amount = monthly_amounts[month]
            formatted_amount = "{:,}".format(amount)
            print(f"{month} 사용량: {formatted_amount}원")
        
        formatted_total = "{:,}".format(total)
        print(f"계정 총 사용량: {formatted_total}원")
        print(f"잔여 크레딧: {residuum_credit:,}원")
        grand_total += total

        save_results_to_csv(results, start_month, end_month, output_file)
        
        
if __name__ == "__main__":
    main()

 

• flask.py

import threading
import time
from flask import Flask, render_template
import pandas as pd
import bill_test as bill
import atexit
import os
import signal

bill.main()

# Flask가 종료되면 csv 파일 삭제되도록 함수 선언
def delete_file(file_path):
  if os.path.exists(file_path):
      os.remove(file_path)

app = Flask(__name__)

file_to_delete = bill.output_file
atexit.register(delete_file, file_to_delete)

# bill_test의 output csv 파일 호출해서 data frame에 import
df = pd.read_csv(bill.output_file)

@app.route('/')
@app.route('/table')

def table():
  data = pd.read_csv(bill.output_file)
  return render_template('table.html', tables=[data.to_html()], titles=[''])

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=int("5000"))

 

• Dockerfile

FROM python:3.8.10-slim
WORKDIR /usr/src/app

COPY . .

RUN apt-get update && apt-get install -y build-essential
RUN pip install --upgrade pip && pip install -r requirements.txt

EXPOSE 5000

CMD ["sh", "entrypoint.sh"]

 

 

결과

아래는 단순히 flask 출력 예제입니다.

번호	계정명	06월	07월	08월	3개월 총액
1	ssungz_test	100,000원	150,000원	240,000원	490,000원