공공 api 읽어온 후 json 파일 csv로 전환한 후 저장하기 - python

https://do-one-more.tistory.com/7

공공 api로 국회의원 코드 읽어오기 - python

 

 

저번에는 공공 api로 코드를 읽어 json 파일로 확인까지 했다

이번에는 json파일을 pandas를 이용해 df 형식으로 바꾼 뒤 CSV로 저장해줬다.

 

[STEP : 1] JSON파일을 df로 전환한 뒤, csv파일에 합치고 csv 파일로 저장하기

import requests
from dotenv import load_dotenv
import json
import os
import pandas as pd
        
def read_api_to_df(url,params):
    response = requests.get(url, params=params)
    
    if response.status_code == 200:
        print(f"API READ CORRECTLY {response.status_code}")
    else:
        print(f"Error Code :{response.status_code}")
    
    response_json = json.loads(response.content)
    json_to_df = pd.json_normalize(response_json)
    
    return json_to_df

def api_to_csv(url):
    
    df = pd.DataFrame()
    file_path = 'public_vote_code'
    
    for i in range(1,1001):
        params = {'serviceKey' : serviceKey, 'resultType' : 'json', 'pageNo' : f'{i}', 'numOfRows' : '100'}
        new_df = read_api_to_df(url,params=params)
        df = pd.concat([df, new_df])
        
    df.to_csv("public_vote_code.csv")
        
if __name__ == "__main__":    
    # api key
    load_dotenv()
    serviceKey = os.getenv("GONGGONG_API_KEY")
    
    # 공공데이터 국회의원 선거공약정보확인을 위한 코드 정보
    govcode_url = "http://apis.data.go.kr/9760000/CommonCodeService/getCommonSgCodeList"
    
    # api code csv로 저장
    api_to_csv(govcode_url)

완전히 이상한 결과가 나왔다.

json파일 안에 json 파일이 겹겹이 쌓인 형식이어서 값이 원하는 것과 다르게 나왔다.
그리고 데이터가 그렇게 많지도 않았다. response.body.items.item에 있는 값만 가져오도록 다시 만들어야 할 거 같다.

 

[STEP 2] json안에 있는 response.body.items.item만으로 csv파일 만들도록 하기

import requests
from dotenv import load_dotenv
import json
import os
import pandas as pd
        
def read_api_to_df(url,params):
    try:
        response = requests.get(url, params=params)
        response.raise_for_status()  # HTTP 에러가 있으면 예외 발생
    except requests.exceptions.RequestException as e:
        print(f"Error during API request: {e}")
        return pd.DataFrame()

	# 이 부분에서 items까지 가져오도록 만듦
    try: 
        response_json = response.json()
        items = response_json.get('response', {}).get('body', {}).get('items', {}).get('item', [])
        print(items)

        if items:
            json_to_df = pd.json_normalize(items)
        else:
            print("No items found in response")
            return pd.DataFrame()
            
    except (KeyError, ValueError, TypeError) as e:
        print(f"Error during JSON processing: {e}")
        return pd.DataFrame()
    
    return json_to_df

def api_to_csv(url):
    
    df = pd.DataFrame()
    file_path = 'public_vote_code'
    
    for i in range(1,10):
        params = {'serviceKey' : serviceKey, 'resultType' : 'json', 'pageNo' : f'{i}', 'numOfRows' : '100'}
        new_df = read_api_to_df(url,params=params)
        df = pd.concat([df, new_df])
        
    df.to_csv("public_vote_code.csv")
        
if __name__ == "__main__":    
    # api key
    load_dotenv()
    serviceKey = os.getenv("GONGGONG_API_KEY")
    
    # 공공데이터 국회의원 선거공약정보확인을 위한 코드 정보
    govcode_url = "http://apis.data.go.kr/9760000/CommonCodeService/getCommonSgCodeList"
    
    # api code csv로 저장
    api_to_csv(govcode_url)

잘 들어온 모습을 확인할 수 있다.

[STEP 3] CSV에 있는 국회의원 선거 코드를 가져와서  국회의원 코드 가져오기

https://www.data.go.kr/tcs/dss/selectApiDataDetailView.do?publicDataPk=15000864

중앙선거관리위원회 당선인정보

 

 

위에서 만든 csv에서 sgId를 list로 가져온 뒤, 당선인 정보를 가져온다.

import requests
from dotenv import load_dotenv
import json
import os
import pandas as pd

def get_votecode():
    try:
        csv_file = pd.read_csv('public_vote_code.csv')
        congress_vote_id_list = csv_file[csv_file['sgTypecode'] == 2]['sgId'].to_list()
        return congress_vote_id_list
    except FileNotFoundError:
        print("Error: 'public_vote_code.csv' not found.")
        return []

def read_api_to_df(url, params):
    try:
        response = requests.get(url, params=params, timeout=10)
        response.raise_for_status()
        print(f"Response status code: {response.status_code}")
    except requests.exceptions.RequestException as e:
        print(f"Error during API request: {e}")
        return pd.DataFrame()

    try:
        response_json = response.json()
        items = response_json.get('response', {}).get('body', {}).get('items', {}).get('item', [])
        if items:
            return pd.json_normalize(items)
        else:
            print("No items found in response")
            return pd.DataFrame()
    except json.JSONDecodeError as e:
        print(f"Error decoding JSON: {e}")
        return pd.DataFrame()
    except (KeyError, ValueError, TypeError) as e:
        print(f"Error during JSON processing: {e}")
        return pd.DataFrame()

def api_to_csv(url):
    serviceKey = load_api_key()
    
    df = pd.DataFrame()
    vote_code = get_votecode()
    if not vote_code:
        print("Error: No vote codes found.")
        return
    
    file_path = "vote_erection.csv"
    
    for id in vote_code:
        for i in range(1, 10):
            params = {
                'serviceKey': f'{serviceKey}',
                'pageNo': f'{i}',
                'resultType': 'json',
                'numOfRows': '100',
                'sgId': f'{id}',
                'sgTypecode': '2',
                'sdName': '',
                'sggName': ''
            }
            new_df = read_api_to_df(url, params=params)
            df = pd.concat([df, new_df], ignore_index=True)
    
    df.drop_duplicates(inplace=True)
    df.to_csv(file_path, index=False, encoding='utf-8-sig')
    print(f"Data saved to {file_path}")

def load_api_key():
    load_dotenv()
    serviceKey = os.getenv("GONGGONG_API_KEY")
    if not serviceKey:
        raise ValueError("API key not found in environment variables.")
    return serviceKey

if __name__ == "__main__":
    erection_url = 'http://apis.data.go.kr/9760000/WinnerInfoInqireService2/getWinnerInfoInqire'
    api_to_csv(erection_url)

 

[step 5] 국회의원 당선인 정보를 가지고 국회의원의 공약을 확인한다.

2개의 컬럼값을 가져오려면 DataFrame 형태로 가져오기에 to_list()가 되지 않고 to_dict로 가져와야한다.

그리고 각 레코드를 읽어서 저장하기 위해 to_dict('records')로 추가로 선정한다.

import requests
from dotenv import load_dotenv
import json
import os
import pandas as pd

def get_votecode():
    try:
        csv_file = pd.read_csv('vote_erection.csv')
        pledge_code_list = [(row['huboid'], row['sgId']) for row in csv_file[['huboid', 'sgId']].to_dict('records')]
        return pledge_code_list
    except FileNotFoundError:
        print("Error: 'vote_erection.csv' not found.")
        return []

def read_api_to_df(url, params):
    try:
        response = requests.get(url, params=params)
        response.raise_for_status()
        print(f"Response status code: {response.status_code}")
    except requests.exceptions.RequestException as e:
        print(f"Error during API request: {e}")
        return pd.DataFrame()

    try:
        response_json = response.json()
        items = response_json.get('response', {}).get('body', {}).get('items', {}).get('item', [])
        if items:
            return pd.json_normalize(items)
        else:
            print("No items found in response")
            return pd.DataFrame()
    except json.JSONDecodeError as e:
        print(f"Error decoding JSON: {e}")
        return pd.DataFrame()
    except (KeyError, ValueError, TypeError) as e:
        print(f"Error during JSON processing: {e}")
        return pd.DataFrame()

def api_to_csv(url):
    serviceKey = load_api_key()
    
    df = pd.DataFrame()
    pledge_code_list = get_votecode()
    if not pledge_code_list:
        print("Error: No vote codes found.")
        return
    
    file_path = "vote_pledge.csv"
    
    for huboid,sgId in pledge_code_list:
        for i in range(1, 10):
            params = {
                'serviceKey': f'{serviceKey}',
                'pageNo': f'{i}',
                'resultType': 'json',
                'numOfRows': '100',
                'cnddtId' : f'{huboid}',
                'sgId': f'{sgId}',
                'sgTypecode': '2',
                'sdName': '',
                'sggName': ''
            }
            new_df = read_api_to_df(url, params=params)
            df = pd.concat([df, new_df], ignore_index=True)
    
    df.drop_duplicates(inplace=True)
    df.to_csv(file_path, index=False, encoding='utf-8-sig')
    print(f"Data saved to {file_path}")

def load_api_key():
    load_dotenv()
    serviceKey = os.getenv("GONGGONG_API_KEY")
    if not serviceKey:
        raise ValueError("API key not found in environment variables.")
    return serviceKey

if __name__ == "__main__":
    pledge_url = "http://apis.data.go.kr/9760000/ElecPrmsInfoInqireService/getCnddtElecPrmsInfoInqire"    
    api_to_csv(pledge_url)

트래픽 1000개가 생각보다 적기 때문에 날짜를 확인해서 특정날짜로 제한해야할 수도 있겠다