diff --git a/app/agents/safety/berth_alerts.py b/app/agents/safety/berth_alerts.py index ee468b2..4ae870e 100644 --- a/app/agents/safety/berth_alerts.py +++ b/app/agents/safety/berth_alerts.py @@ -24,15 +24,8 @@ from sqlalchemy.ext.asyncio import AsyncSession from .bulk_compatibility import build_bulk_conflicts -from .graph_queries import ( - find_adjacent_berth_conflicts, - find_bulk_exceptions, - find_bulk_group_conflicts, - find_bulk_groups, - find_imdg_no_segregation_required, - find_imdg_segregation_conflicts, - find_incompatible_conflicts, -) +from .graph_queries import find_adjacent_berth_conflicts +from .graph_snapshot import get_snapshot from .rule_engine import ( compute_bulk_compatibility_floor, compute_imdg_berth_adjacency_floor, @@ -86,26 +79,22 @@ async def _segregation_facts( driver: AsyncDriver, sem: asyncio.Semaphore, target_id: str, other_ids: list[str] ) -> tuple: - """대상 화물 하나의 혼재 판정에 필요한 그래프 조회 결과(판정은 하지 않는다).""" - async with sem: - return ( - await find_incompatible_conflicts( - driver, target_chem_id=target_id, adjacent_chem_ids=other_ids - ), - await find_imdg_segregation_conflicts( - driver, target_chem_id=target_id, adjacent_chem_ids=other_ids - ), - await find_imdg_no_segregation_required( - driver, target_chem_id=target_id, adjacent_chem_ids=other_ids - ), - await find_bulk_group_conflicts( - driver, target_chem_id=target_id, adjacent_chem_ids=other_ids - ), - await find_bulk_groups(driver, chem_ids=[target_id, *other_ids]), - await find_bulk_exceptions( - driver, target_chem_id=target_id, adjacent_chem_ids=other_ids - ), - ) + """대상 화물 하나의 혼재 판정에 필요한 그래프 조회 결과(판정은 하지 않는다). + + [2026-09-30] Neo4j 조회 6개 → 메모리 사본(graph_snapshot). service.assess_safety 가 9/29 에 + 옮겨 간 것과 같은 사본·같은 답이다(tests/test_graph_snapshot_20260929.py). 운영(Aura)에서 + 대상 수십 개 × 조회 6개가 캐시 빈 첫 화면을 30초 붙잡았다. sem 은 테스트가 바꿔 끼우는 + 시그니처를 지키려고 남겨 둔다(쓰지 않음). + """ + graph = await get_snapshot(driver) + return ( + graph.incompatible_conflicts(target_id, other_ids), + graph.imdg_segregation_conflicts(target_id, other_ids), + graph.imdg_no_segregation_required(target_id, other_ids), + graph.bulk_group_conflicts(target_id, other_ids), + graph.bulk_groups([target_id, *other_ids]), + graph.bulk_exceptions(target_id, other_ids), + ) async def _segregation_alerts(rows: list[dict], driver: AsyncDriver) -> list[dict]: @@ -349,21 +338,15 @@ async def _adjacent_berth_alerts(rows: list[dict], driver: AsyncDriver) -> list[ seen: set[frozenset] = set() alerts: list[dict] = [] - # 조회는 동시에, 판정은 원래 순서대로(_segregation_alerts 와 같은 이유) + # 조회는 한 번에(대상 전부), 판정은 원래 순서대로(_segregation_alerts 와 같은 이유) targets = [ (wharf, target) for wharf, cargos in by_wharf.items() for target in {c["chem_id"]: c for c in cargos}.values() ] - sem = asyncio.Semaphore(_NEO4J_CONCURRENCY) - - async def _hits(wharf: str, target_id: str) -> list[dict]: - async with sem: - return await find_adjacent_berth_conflicts( - driver, wharf_name=wharf, target_chem_id=target_id, neighbor_cargo=cargo - ) - - all_hits = await asyncio.gather(*(_hits(w, t["chem_id"]) for w, t in targets)) + all_hits = await find_adjacent_berth_conflicts( + driver, targets=[(w, t["chem_id"]) for w, t in targets], neighbor_cargo=cargo + ) for (wharf, target), hits in zip(targets, all_hits): for hit in hits: diff --git a/app/agents/safety/graph_queries.py b/app/agents/safety/graph_queries.py index 56bcdf8..3417556 100644 --- a/app/agents/safety/graph_queries.py +++ b/app/agents/safety/graph_queries.py @@ -424,14 +424,20 @@ async def _tx(tx): # 가장 가까운 거리만 남긴다(min) — 같은 부두쌍이 여러 줄로 불어나지 않게. # ───────────────────────────────────────────────────────────────────────────── +# [2026-09-30] 대상(부두·화물)마다 한 번씩 부르던 것을 대상 목록 한 번으로 묶었다 — 운영(Aura)은 +# 왕복이 약 300ms 라 대상 74개가 캐시 빈 관제 경고를 7.6초 붙잡았다. 대상 하나의 탐색·반환 행· +# 정렬 키는 예전 그대로이고, 결과를 대상 순번(i)으로 나눠 돌려준다. _CYPHER_ADJACENT_BERTH_CONFLICTS = """ -MATCH (here:Berth {wharf_name: $wharf_name})-[adj:ADJACENT_TO]->(nb:Berth) -WHERE nb.wharf_name <> $wharf_name -WITH nb.wharf_name AS neighbor_wharf, min(adj.distance_m) AS distance_m +UNWIND range(0, size($targets) - 1) AS i +CALL (i) { +WITH $targets[i].wharf_name AS wharf_name, $targets[i].target AS target +MATCH (here:Berth {wharf_name: wharf_name})-[adj:ADJACENT_TO]->(nb:Berth) +WHERE nb.wharf_name <> wharf_name +WITH wharf_name, target, nb.wharf_name AS neighbor_wharf, min(adj.distance_m) AS distance_m UNWIND $neighbor_cargo AS nc -WITH neighbor_wharf, distance_m, nc +WITH wharf_name, target, neighbor_wharf, distance_m, nc WHERE nc.wharf_name = neighbor_wharf -MATCH (a:Chemical {id: $target}) +MATCH (a:Chemical {id: target}) MATCH (b:Chemical {id: nc.chem_id}) // 충돌 근거 세 갈래를 한 서브쿼리로 합친다. 근거가 다른 신호라 어느 하나로 // 대체할 수 없다 — MSDS 텍스트 마이닝 2방향(비대칭이라 양쪽을 다 봐야 한다. @@ -462,7 +468,7 @@ async def _tx(tx): via AS via, code AS segregation_code, direction AS direction, - $wharf_name + ' —인접(' + + wharf_name + ' —인접(' + CASE WHEN distance_m IS NULL THEN '거리 미상' ELSE toString(toInteger(round(distance_m))) + 'm' END + ')→ ' + neighbor_wharf + @@ -471,21 +477,25 @@ async def _tx(tx): WHEN 'MSDS_INCOMPATIBLE' THEN '혼재금지(' + via + ')' ELSE 'IMDG격리(' + via + ' → ' + coalesce(code, '?') + ')' END + - '→ ' + coalesce(a.name_ko, $target) AS path_text + '→ ' + coalesce(a.name_ko, target) AS path_text +} +RETURN i, neighbor_wharf, distance_m, chem_id, name_ko, callsgn, basis, via, + segregation_code, direction, path_text // 거리 미상(SK5~8 처럼 좌표 없이 부두번호로 이은 쌍)은 맨 뒤로. Cypher 는 // ORDER BY ... NULLS LAST 를 안 받아서 정렬 키를 직접 만든다. -ORDER BY coalesce(distance_m, 1000000.0), neighbor_wharf, chem_id +ORDER BY i, coalesce(distance_m, 1000000.0), neighbor_wharf, chem_id """ async def find_adjacent_berth_conflicts( driver: AsyncDriver, *, - wharf_name: str, - target_chem_id: str, + targets: list[tuple[str, str]], neighbor_cargo: list[dict], -) -> list[dict]: - """대상 화물 하나에 대해 **인접 부두** 재항 화물과의 충돌을 한 번에 찾는다. +) -> list[list[dict]]: + """대상 (부두, 화물) 각각에 대해 **인접 부두** 재항 화물과의 충돌을 한 번에 찾는다. + + 반환은 targets 와 같은 순서·같은 길이의 목록이고, 각 원소가 그 대상의 충돌 행이다. neighbor_cargo 는 호출부가 이미 읽어 둔 재항 현황 그대로 넘긴다 — 각 항목은 최소한 {"wharf_name", "chem_id"} 를 갖고, "callsgn"·"cargo_name" @@ -494,8 +504,15 @@ async def find_adjacent_berth_conflicts( 반환 행에는 `path_text`(그래프가 만든 경로 문장)가 들어 있다. 화면·프롬프트는 이 문장을 그대로 쓰면 되고, 근거를 다시 조립하지 않는다. """ - if not wharf_name or not target_chem_id or not neighbor_cargo: - return [] + out: list[list[dict]] = [[] for _ in targets] + # 부두나 화물이 빈 대상은 예전처럼 조회 없이 빈 결과 + query_targets = [ + {"wharf_name": w, "target": t, "idx": n} + for n, (w, t) in enumerate(targets) + if w and t + ] + if not query_targets or not neighbor_cargo: + return out payload = [ { @@ -508,17 +525,19 @@ async def find_adjacent_berth_conflicts( if c.get("wharf_name") and c.get("chem_id") ] if not payload: - return [] + return out async with driver.session() as session: async def _tx(tx): result = await tx.run( _CYPHER_ADJACENT_BERTH_CONFLICTS, - wharf_name=wharf_name, - target=target_chem_id, + targets=query_targets, neighbor_cargo=payload, ) return [record.data() async for record in result] - return await session.execute_read(_tx) + rows = await session.execute_read(_tx) + for row in rows: + out[query_targets[row.pop("i")]["idx"]].append(row) + return out diff --git a/app/api/v1/arrivals.py b/app/api/v1/arrivals.py index 5ce5bc0..3df6eaf 100644 --- a/app/api/v1/arrivals.py +++ b/app/api/v1/arrivals.py @@ -32,12 +32,29 @@ pv.agency_name, pv.prev_port_nm, pv.entry_purpose_nm, pv.arrival_at_utc, pv.departure_sched_utc, pv.arrival_report_type, pv.arrival_facility_cd, pv.arrival_facility_nm, pv.collected_at_utc, - pv.entry_year, pv.entry_count + pv.entry_year, pv.entry_count, + CASE WHEN pv.entry_year IS NOT NULL AND pv.entry_count IS NOT NULL + THEN upper(btrim(pv.callsgn)) || '_' || pv.entry_year::text || '_' || lpad(pv.entry_count::text, 3, '0') + END AS port_call_key FROM portmis_vessel pv WHERE pv.is_liquid_cargo_vessel AND pv.arrival_at_utc BETWEEN now() - make_interval(hours => :past_hours) AND now() + make_interval(hours => :ahead_hours) ), + -- [2026-09-25] 이 입항 건(콜사인·입항연도·입항횟수)에 단 화물 전부. 입항 후 + -- 위치 화면(mart.vessel_current_call 경유)과 같은 키라 같은 화물이 나온다. + -- [2026-09-30] 행마다 LATERAL 로 cargo_msds 를 다시 훑던 것(100건 × 23ms)을 한 번 집계로 바꿈. + cg AS ( + SELECT cm.port_call_key, + json_agg(json_build_object( + 'name', coalesce(cm.msds_name_ko, cm.cargo_name_raw), + 'un_no', cm.dg_un_no, 'chem_id', cm.chem_id, + 'cas_no', cm.cas_no, 'is_synthetic', cm.is_synthetic + ) ORDER BY (cm.chem_id IS NULL), cm.bl_no) AS cargos + FROM mart.cargo_msds cm + WHERE cm.port_call_key IN (SELECT port_call_key FROM arr) + GROUP BY cm.port_call_key + ), fac AS ( SELECT DISTINCT ON (source_name) source_name, wharf_name, facility_type FROM mart.facility_alias @@ -85,30 +102,18 @@ a.collected_at_utc, cg.cargos, -- [2026-09-29] 이 행의 입항 건 키와 판정 잡이 지금 보는 입항 건 키. 판정 기록은 같은 입항 건 것만 붙인다. - CASE WHEN a.entry_year IS NOT NULL AND a.entry_count IS NOT NULL - THEN upper(btrim(a.callsgn)) || '_' || a.entry_year::text || '_' || lpad(a.entry_count::text, 3, '0') - END AS port_call_key, - (SELECT vc.port_call_key FROM mart.vessel_current_call vc - WHERE vc.callsgn = upper(btrim(a.callsgn))) AS current_call_key + a.port_call_key, + vc.port_call_key AS current_call_key FROM arr a LEFT JOIN fac f ON f.source_name = a.arrival_facility_nm LEFT JOIN berth b ON b.wharf_name = f.wharf_name LEFT JOIN pos p ON p.cs = upper(btrim(a.callsgn)) LEFT JOIN draught d ON d.cs = upper(btrim(a.callsgn)) LEFT JOIN spec s ON s.cs = upper(btrim(a.callsgn)) - -- [2026-09-25] 이 입항 건(콜사인·입항연도·입항횟수)에 단 화물 전부. 입항 후 - -- 위치 화면(mart.vessel_current_call 경유)과 같은 키라 같은 화물이 나온다. - LEFT JOIN LATERAL ( - SELECT json_agg(json_build_object( - 'name', coalesce(cm.msds_name_ko, cm.cargo_name_raw), - 'un_no', cm.dg_un_no, 'chem_id', cm.chem_id, - 'cas_no', cm.cas_no, 'is_synthetic', cm.is_synthetic - ) ORDER BY (cm.chem_id IS NULL), cm.bl_no) AS cargos - FROM mart.cargo_msds cm - WHERE a.entry_year IS NOT NULL AND a.entry_count IS NOT NULL - AND cm.port_call_key = upper(btrim(a.callsgn)) || '_' || a.entry_year::text - || '_' || lpad(a.entry_count::text, 3, '0') - ) cg ON true + LEFT JOIN cg ON cg.port_call_key = a.port_call_key + -- [2026-09-30] 상관 서브쿼리로 행마다 뷰를 다시 계산하던 것(100건 × 20ms)을 조인으로 바꿈. + -- vessel_current_call 은 콜사인당 1행(스칼라 서브쿼리였으므로)이라 행 수가 늘지 않는다. + LEFT JOIN mart.vessel_current_call vc ON vc.callsgn = upper(btrim(a.callsgn)) ORDER BY a.arrival_at_utc """) diff --git a/app/database.py b/app/database.py index 20456ef..2c25a5b 100644 --- a/app/database.py +++ b/app/database.py @@ -6,7 +6,14 @@ settings = get_settings() -engine = create_async_engine(settings.database_url, echo=False, future=True) +# JIT 끔: mart 뷰 조인의 행 수 과대추정(실제 1.7천 → 추정 477만)으로 비용이 JIT 기준을 +# 넘어, 0.2초 쿼리에 LLVM 컴파일 3초가 붙었다(2026-09-30 로컬 PG16 실측). RDS 는 기본이 off. +engine = create_async_engine( + settings.database_url, + echo=False, + future=True, + connect_args={"server_settings": {"jit": "off"}}, +) AsyncSessionFactory = async_sessionmaker(bind=engine, expire_on_commit=False, class_=AsyncSession)