Issue/23/특정 역 지하철정보

environment_setup_full.bat

@@ -28,6 +28,7 @@ set /p pepper=AIT_PW_PEPPER:
 set /p deepl_key=DEEPL_API_KEY:
 set /p nlp_key=NLP_API_KEY:
 set /p ipinfo_key=IPINFO_TOKEN:
+set /p subway_key=SUBWAY_API_KEY:
 
 echo setting environments...
 :: »ç¿ëÀÚ ÀÔ·ÂÀ» ȯ°æ º¯¼ö¿¡ ÀúÀå
@@ -39,5 +40,6 @@ setx AIT_PW_PEPPER %pepper%
 setx DEEPL_API_KEY %deepl_key%
 setx NLP_API_KEY %nlp_key%
 setx IPINFO_TOKEN %ipinfo_key%
+setx SUBWAY_API_KEY %subway_key%
 echo setup complete
 pause

main.py

@@ -6,6 +6,7 @@
 
 from core.sql_util import check_and_create_table
 from oauth import auth
+from service.subway import subway
 from service.summary import summary
 from service.weather import weather
 
@@ -14,6 +15,7 @@
 app.include_router(auth.router, prefix="/auth", tags=["auth"])
 app.include_router(weather.router, prefix="/service/weather", tags=["weather"])
 app.include_router(summary.router, prefix="/service/summary", tags=["summary"])
+app.include_router(subway.router, prefix="/service/subway", tags=["subway"])
 
 check_and_create_table()
 
@@ -27,8 +29,12 @@ def root():
     uvicorn.run("main:app", port=1777, reload=True)
 
     if not os.path.exists(".env"):
-        print("ENV >> Cannot found .env file, use pre-configured environment variables.")
-        print("ENV >> If not prepared environment variables, server will be throw exception.")
+        print(
+            "ENV >> Cannot found .env file, use pre-configured environment variables."
+        )
+        print(
+            "ENV >> If not prepared environment variables, server will be throw exception."
+        )
     else:
         print("ENV >> Found .env file! load environment variables from .env file.")
         load_dotenv()

service/subway/dto/subway_arrival_dto.py

@@ -0,0 +1,28 @@
+from typing import List
+
+from service.subway.dto.subway_arrival_item_dto import SubwayArrivalItemDto
+from service.subway.dto.subway_error_msg_dto import SubwayErrorMsgDto
+
+
+class SubwayArrivalDto:
+    def __init__(
+        self,
+        error_msg: SubwayErrorMsgDto,
+        items: List[SubwayArrivalItemDto],
+    ):
+        self.error_msg = error_msg
+        self.items = items
+
+    @classmethod
+    def from_json(cls, data: dict):
+        error_msg = data.get["errorMessage"]
+        return cls(
+            error_msg=SubwayErrorMsgDto.from_json(
+                error_msg if error_msg is not None else data
+            ),
+            items=[
+                SubwayArrivalItemDto.from_json(e) for e in data["realtimeArrivalList"]
+            ]
+            if error_msg is not None
+            else [],
+        )

service/subway/dto/subway_arrival_item_dto.py

@@ -0,0 +1,102 @@
+from enum import StrEnum
+from typing import List
+
+
+class SubwayId(StrEnum):
+    s1001 = "1호선"
+    s1002 = "2호선"
+    s1003 = "3호선"
+    s1004 = "4호선"
+    s1005 = "5호선"
+    s1006 = "6호선"
+    s1007 = "7호선"
+    s1008 = "8호선"
+    s1009 = "9호선"
+    s1061 = "중앙선"
+    s1063 = "경의중앙선"
+    s1065 = "공항철도"
+    s1067 = "경춘선"
+    s1075 = "수의분당선"
+    s1077 = "신분당선"
+    s1092 = "우이신설선"
+    s1093 = "서해선"
+    s1081 = "경강선"
+
+
+class ArrivalCode(StrEnum):
+    a0 = "진입"
+    a1 = "도착"
+    a2 = "출발"
+    a3 = "전역출발"
+    a4 = "전역진입"
+    a5 = "전역도착"
+    a99 = "운행중"
+
+
+class SubwayArrivalItemDto:
+    def __init__(
+        self,
+        subway_id: SubwayId,  # 지하철 호선 ID (1호선, 3호선, 경춘선 등)
+        up_down_line: str,  # (상행, 하행)
+        train_line_name: str,  #  종착지 - 다음역
+        prev_station_id: str,  # 이전역 id
+        next_station_id: str,  # 다음역 id
+        station_id: str,  # ì—­ id
+        station_name: str,  # 역 이름
+        trnsitCo: int,  # 환승 노선 수
+        # ord_key 는 필요없다 판단되어 추가 안함
+        subway_list: List[SubwayId],  # 연계 호선 Id
+        station_list: List[str],  # 연계 지하철역 Id
+        train_type: str,  # 급행, 일반 등의 열차 종류
+        pred_sec: int,  # 예상 도착 시간 (초)
+        train_no: int,  # 열차번호
+        station_end_id: int,  # 종착역 id
+        station_end_name: str,  # 종착역 이름
+        generated_date: str,  # 도착정보를 생성한 시각
+        arrival_msg_2: str,  # 첫 번째 도착 메세지 (도착, 출발, 진입 등)
+        arrival_msg_3: str,  # 두 번재 도착 메세지 (종합운동장 도착, 12분 후 등)
+        arrival_code: ArrivalCode,  # 도착 코드
+    ):
+        self.subway_id = subway_id
+        self.up_down_line = up_down_line
+        self.train_line_name = train_line_name
+        self.prev_station_id = prev_station_id
+        self.next_station_id = next_station_id
+        self.station_id = station_id
+        self.station_name = station_name
+        self.trnsitCo = trnsitCo
+        self.subway_list = subway_list
+        self.station_list = station_list
+        self.train_type = train_type
+        self.pred_sec = pred_sec
+        self.train_no = train_no
+        self.station_end_id = station_end_id
+        self.station_end_name = station_end_name
+        self.generated_date = generated_date
+        self.arrival_msg_2 = arrival_msg_2
+        self.arrival_msg_3 = arrival_msg_3
+        self.arrival_code = arrival_code
+
+    @classmethod
+    def from_json(cls, data: dict):
+        return cls(
+            subway_id=SubwayId["s" + data["subwayId"]],
+            up_down_line=data["updnLine"],
+            train_line_name=data["trainLineNm"],
+            prev_station_id=data["statnFid"],
+            next_station_id=data["statnTid"],
+            station_id=data["statnId"],
+            station_name=data["statnNm"],
+            trnsitCo=int(data["trnsitCo"]),
+            subway_list=[SubwayId["s" + e] for e in data["subwayList"].split(",")],
+            station_list=data["statnList"],
+            train_type=data["btrainSttus"],
+            pred_sec=int(data["barvlDt"]),
+            train_no=data["btrainNo"],
+            station_end_id=data["bstatnId"],
+            station_end_name=data["bstatnNm"],
+            generated_date=data["recptnDt"],
+            arrival_msg_2=data["arvlMsg2"],
+            arrival_msg_3=data["arvlMsg3"],
+            arrival_code=ArrivalCode["a" + data["arvlCd"]],
+        )

service/subway/dto/subway_error_msg_dto.py

@@ -0,0 +1,48 @@
+from enum import StrEnum
+
+
+class SubwayErrorCode(StrEnum):
+    INFO000 = "정상 처리되었습니다."
+    ERROR300 = "필수 값이 누락되어 있습니다."
+    INFO100 = "인증키가 유효하지 않습니다."
+    ERROR301 = "파일타입 값이 누락 혹은 유효하지 않습니다."
+    ERROR310 = "해당하는 서비스를 찾을 수 없습니다."
+    ERROR331 = "요청시작위치 값을 확인하십시오."
+    ERROR332 = "요청종료위치 값을 확인하십시오."
+    ERROR333 = "요청위치 값의 타입이 유효하지 않습니다."
+    ERROR334 = "요청종료위치 보다 요청시작위치가 더 큽니다."
+    ERROR335 = "샘플데이터(샘플키:sample) 는 한번에 최대 5건을 넘을 수 없습니다."
+    ERROR336 = "데이터요청은 한번에 최대 1000건을 넘을 수 없습니다."
+    ERROR500 = "서버 오류입니다."
+    ERROR600 = "데이터베이스 연결 오류입니다."
+    ERROR601 = "SQL 문장 오류 입니다."
+    INFO200 = "해당하는 데이터가 없습니다."
+
+
+class SubwayErrorMsgDto:
+    def __init__(
+        self,
+        status: int,
+        code: SubwayErrorCode,
+        message: str,
+        link: str,
+        dev_msg: str,
+        total: int,
+    ):
+        self.status = status
+        self.code = code
+        self.message = message
+        self.link = link
+        self.dev_msg = dev_msg
+        self.total = total
+
+    @classmethod
+    def from_json(cls, data: dict):
+        return cls(
+            status=int(data["status"]),
+            code=SubwayErrorCode[data["code"].replace("-", "")],
+            message=data["message"],
+            link=data["link"],
+            dev_msg=data["developerMessage"],
+            total=data["total"],
+        )

service/subway/subway.py

@@ -0,0 +1,35 @@
+import json
+import os
+
+import requests
+from fastapi import APIRouter
+
+from service.subway.dto.subway_arrival_dto import SubwayArrivalDto
+
+_URL = "http://swopenapi.seoul.go.kr/api/subway"
+_API_KEY = os.environ.get("SUBWAY_API_KEY")
+_DATA_TYPE = "json"
+_SERVICE = "realtimeStationArrival"
+_DATA_COUNT = "20"
+
+
+router = APIRouter()
+
+
+@router.get("/")
+def request_subway_data(station_name: str):
+    response = requests.get(
+        "{}/{}/{}/{}/{}/{}/{}".format(
+            _URL,
+            _API_KEY,
+            _DATA_TYPE,
+            _SERVICE,
+            0,
+            _DATA_COUNT,
+            station_name,
+        )
+    )
+
+    data = SubwayArrivalDto.from_json(json.loads(response.text))
+    data.items.sort(key=lambda x: x.pred_sec)
+    return data