wisheasy project (4) - 그래프 데이터는 어떻게 관리할까

19 Oct 2025 • yeonliyou

서비스 주요 알고리즘 단계

[출발역] → [도착역] 의 최소환승 경로 찾기
(1)에서 찾은 최단경로로, 거처가야 하는 모든 역의 역사내 에스컬레이터 우선 경로 구축하기

문제 상황

우리 팀의 프로젝트는 알고리즘 1번 단계에서 외부 교통 API를 활용하는 것을 전제로 설계되었다.

그러나 데이터센터 화재 사고로 API가 장기간 사용이 불가능하게 되었다.

따라서, 직접 호선 그래프를 구축해서 다익스트라 알고리즘으로 최소환승 찾기를 구현했다.

서버를 실행 시 migrate -> load_csv -> build_graph 순으로 작동하게 했다.

서버가 실행되면 csv가 DB에 적재되고, lines 테이블을 통해서 전역 캐시로 그래프를 구축하여 이를 계속 서비스에서 사용할 수 있게 한다.

전제 상황

DB와 데이터는 구축되어 있음

방법 프로세스

1. 함수 생성

apps/
└── journeys/
    ├── management/
    │    └── commands/
    │         └── load_csv.py
    │         └── bulid_graph.py <--- 추가!
    ├── models.py
    ├── views.py
    ├── ...

build_graph.py 함수

  from django.core.management.base import BaseCommand
  import networkx as nx
  from apps.journeys.models import Lines

  GRAPH_CACHE = None  # 서버 실행 후 유지할 전역 그래프

  def build_graph():
      """DB 데이터를 기반으로 최소환승 그래프 구성"""
      G = nx.Graph()

      # 1. 호선별 역 목록 불러오기
      line_names = Lines.objects.values_list('line', flat=True).distinct()

      # 2. 호선별 인접역 연결
      for line in line_names:
          stations = (
              Lines.objects.filter(line=line)
              .order_by('order_in_line')
              .values_list('station', flat=True)
          )
          for i in range(len(stations)):
              node = f"{stations[i]}-{line}"
              G.add_node(node)
              if i > 0:
                  prev_node = f"{stations[i-1]}-{line}"
                  G.add_edge(prev_node, node, weight=0)  # 인접역

      # 3. 환승 연결 (역 이름 동일, 호선 다름)
      all_nodes = list(G.nodes)
      for n1 in all_nodes:
          name1, line1 = n1.split("-")
          for n2 in all_nodes:
              name2, line2 = n2.split("-")
              if name1 == name2 and line1 != line2:
                  G.add_edge(n1, n2, weight=1)

      print("✅ Subway graph built successfully.")
      return G


  def get_graph():
      """전역 그래프 캐시 반환"""
      global GRAPH_CACHE
      if GRAPH_CACHE is None:
          GRAPH_CACHE = build_graph()
      return GRAPH_CACHE


  class Command(BaseCommand):
      help = "서버 실행 시 그래프를 메모리에 빌드합니다."

      def handle(self, *args, **options):
          global GRAPH_CACHE
          self.stdout.write("🚆 Building subway graph from DB...")
          GRAPH_CACHE = build_graph()
          self.stdout.write(self.style.SUCCESS(f"Graph built successfully ✅"))
          self.stdout.write(f"노드 수: {len(GRAPH_CACHE.nodes)} / 엣지 수: {len(GRAPH_CACHE.edges)}")

2. `python manage.py build_graph --settings=config.settings.local` 실행

실제 서버에서는 .local를 .prod로 변경해주면 됨.

3. 그래프가 잘 구축되었는지 확인해보기

python manage.py shell --settings=config.settings.local 실행

확인해보기

  from apps.journeys.management.commands.build_graph import get_graph
  G = get_graph()

  print(G.nodes)
  print(G.edges)

위의 코드를 실행해보면 그래프가 잘 구축된 것을 아래와 같이 확인할 수 있다.

graph

DataBase