Unify graph routing analyzer
All checks were successful
Build Docker Image / build (push) Successful in 1m53s

This commit is contained in:
Ruslan Bakiev
2026-02-07 11:28:16 +07:00
parent e99cbf4882
commit 4ec6506633

View File

@@ -933,7 +933,7 @@ class Query(graphene.ObjectType):
return [] return []
def resolve_hubs_near_offer(self, info, offer_uuid, limit=12): def resolve_hubs_near_offer(self, info, offer_uuid, limit=12):
"""Get nearest hubs to an offer location.""" """Get nearest hubs to an offer location (graph-based)."""
db = get_db() db = get_db()
nodes_col = db.collection('nodes') nodes_col = db.collection('nodes')
offer = nodes_col.get(offer_uuid) offer = nodes_col.get(offer_uuid)
@@ -947,36 +947,13 @@ class Query(graphene.ObjectType):
logger.info("Offer %s has no coordinates", offer_uuid) logger.info("Offer %s has no coordinates", offer_uuid)
return [] return []
aql = """ return self.resolve_nearest_hubs(
FOR node IN nodes info,
FILTER node.node_type == 'logistics' OR node.node_type == null lat=lat,
FILTER node.product_uuid == null lon=lon,
FILTER node.latitude != null AND node.longitude != null source_uuid=offer_uuid,
LET dist = DISTANCE(node.latitude, node.longitude, @lat, @lon) / 1000 limit=limit,
SORT dist ASC )
LIMIT @limit
RETURN MERGE(node, {distance_km: dist})
"""
try:
cursor = db.aql.execute(aql, bind_vars={'lat': lat, 'lon': lon, 'limit': limit})
hubs = []
for node in cursor:
hubs.append(NodeType(
uuid=node['_key'],
name=node.get('name'),
latitude=node.get('latitude'),
longitude=node.get('longitude'),
country=node.get('country'),
country_code=node.get('country_code'),
synced_at=node.get('synced_at'),
transport_types=node.get('transport_types') or [],
edges=[],
))
logger.info("Found %d hubs near offer %s", len(hubs), offer_uuid)
return hubs
except Exception as e:
logger.error("Error getting hubs near offer: %s", e)
return []
def resolve_suppliers(self, info): def resolve_suppliers(self, info):
"""Get unique suppliers from all offers.""" """Get unique suppliers from all offers."""
@@ -1059,7 +1036,7 @@ class Query(graphene.ObjectType):
return [] return []
def resolve_products_near_hub(self, info, hub_uuid, radius_km=500): def resolve_products_near_hub(self, info, hub_uuid, radius_km=500):
"""Get products available near a hub (within radius).""" """Get products available via graph routes from a hub."""
db = get_db() db = get_db()
nodes_col = db.collection('nodes') nodes_col = db.collection('nodes')
hub = nodes_col.get(hub_uuid) hub = nodes_col.get(hub_uuid)
@@ -1067,34 +1044,26 @@ class Query(graphene.ObjectType):
logger.info("Hub %s not found", hub_uuid) logger.info("Hub %s not found", hub_uuid)
return [] return []
lat = hub.get('latitude') matches = _graph_find_targets(
lon = hub.get('longitude') db,
if lat is None or lon is None: start_uuid=hub_uuid,
logger.info("Hub %s has no coordinates", hub_uuid) target_predicate=lambda doc: doc.get('node_type') == 'offer',
return [] limit=1000,
max_expansions=Query.MAX_EXPANSIONS,
)
aql = """ products = {}
FOR node IN nodes for match in matches:
FILTER node.node_type == 'offer' offer = match.get('node') or {}
FILTER node.product_uuid != null product_uuid = offer.get('product_uuid')
FILTER node.latitude != null AND node.longitude != null if not product_uuid:
LET dist = DISTANCE(node.latitude, node.longitude, @lat, @lon) / 1000 continue
FILTER dist <= @radius_km if product_uuid not in products:
COLLECT product_uuid = node.product_uuid INTO offers products[product_uuid] = offer.get('product_name')
LET first_offer = FIRST(offers).node
RETURN { result = [ProductType(uuid=uuid, name=name) for uuid, name in products.items()]
uuid: product_uuid, logger.info("Found %d products via graph for hub %s", len(result), hub_uuid)
name: first_offer.product_name return result
}
"""
try:
cursor = db.aql.execute(aql, bind_vars={'lat': lat, 'lon': lon, 'radius_km': radius_km})
products = [ProductType(uuid=p['uuid'], name=p.get('name')) for p in cursor]
logger.info("Found %d products near hub %s", len(products), hub_uuid)
return products
except Exception as e:
logger.error("Error getting products near hub: %s", e)
return []
def resolve_suppliers_for_product(self, info, product_uuid): def resolve_suppliers_for_product(self, info, product_uuid):
"""Get unique suppliers that have offers for this product.""" """Get unique suppliers that have offers for this product."""
@@ -1179,14 +1148,11 @@ class Query(graphene.ObjectType):
def resolve_offers_by_hub(self, info, hub_uuid, product_uuid=None, limit=10): def resolve_offers_by_hub(self, info, hub_uuid, product_uuid=None, limit=10):
""" """
Get offers for a product with routes to hub. Get offers for a product with routes to hub.
Uses phase-based routing: auto → rail* → auto Uses unified graph routing: auto → rail* → auto
Search goes from hub outward to find offers.
""" """
db = get_db() db = get_db()
ensure_graph()
nodes_col = db.collection('nodes') nodes_col = db.collection('nodes')
# Get hub
hub = nodes_col.get(hub_uuid) hub = nodes_col.get(hub_uuid)
if not hub: if not hub:
logger.info("Hub %s not found", hub_uuid) logger.info("Hub %s not found", hub_uuid)
@@ -1198,158 +1164,35 @@ class Query(graphene.ObjectType):
logger.info("Hub %s missing coordinates", hub_uuid) logger.info("Hub %s missing coordinates", hub_uuid)
return [] return []
# Phase-based routing: auto → rail* → auto matches = _graph_find_targets(
def allowed_next_phase(current_phase, transport_type): db,
""" start_uuid=hub_uuid,
Phases — расширение радиуса поиска, ЖД не обязателен: target_predicate=lambda doc: doc.get('node_type') == 'offer' and (
- end_auto: можно 1 авто, rail, или сразу offer product_uuid is None or doc.get('product_uuid') == product_uuid
- end_auto_done: авто использовано — rail или offer ),
- rail: любое кол-во rail, потом 1 авто или offer limit=limit,
- start_auto_done: авто использовано — только offer max_expansions=Query.MAX_EXPANSIONS,
""" )
if current_phase == 'end_auto':
if transport_type == 'offer':
return 'offer'
if transport_type == 'auto':
return 'end_auto_done'
if transport_type == 'rail':
return 'rail'
return None
if current_phase == 'end_auto_done':
if transport_type == 'offer':
return 'offer'
if transport_type == 'rail':
return 'rail'
return None
if current_phase == 'rail':
if transport_type == 'offer':
return 'offer'
if transport_type == 'rail':
return 'rail'
if transport_type == 'auto':
return 'start_auto_done'
return None
if current_phase == 'start_auto_done':
if transport_type == 'offer':
return 'offer'
return None
return None
def fetch_neighbors(node_key, phase):
"""Get neighbors based on allowed transport types for phase."""
if phase == 'end_auto':
types = ['auto', 'rail', 'offer']
elif phase == 'end_auto_done':
types = ['rail', 'offer']
elif phase == 'rail':
types = ['rail', 'auto', 'offer']
elif phase == 'start_auto_done':
types = ['offer']
else:
types = ['offer']
aql = """
FOR edge IN edges
FILTER edge.transport_type IN @types
FILTER edge._from == @node_id OR edge._to == @node_id
LET neighbor_id = edge._from == @node_id ? edge._to : edge._from
LET neighbor = DOCUMENT(neighbor_id)
FILTER neighbor != null
RETURN {
neighbor_key: neighbor._key,
neighbor_doc: neighbor,
from_id: edge._from,
to_id: edge._to,
transport_type: edge.transport_type,
distance_km: edge.distance_km,
travel_time_seconds: edge.travel_time_seconds
}
"""
cursor = db.aql.execute(
aql,
bind_vars={
'node_id': f"nodes/{node_key}",
'types': types,
},
)
return list(cursor)
# Priority queue: (cost, seq, node_key, phase)
queue = []
counter = 0
heapq.heappush(queue, (0, counter, hub_uuid, 'end_auto'))
visited = {}
predecessors = {}
node_docs = {hub_uuid: hub}
found_routes = [] found_routes = []
expansions = 0 for match in matches:
node_doc = match.get('node') or {}
while queue and len(found_routes) < limit and expansions < Query.MAX_EXPANSIONS: route = match.get('route')
cost, _, node_key, phase = heapq.heappop(queue) distance_km = match.get('distance_km')
if distance_km is None:
if (node_key, phase) in visited and cost > visited[(node_key, phase)]:
continue
# Found an offer (optionally filtered by product)
node_doc = node_docs.get(node_key)
if node_doc and node_doc.get('node_type') == 'offer' and (
product_uuid is None or node_doc.get('product_uuid') == product_uuid
):
path_edges = []
state = (node_key, phase)
current_key = node_key
while state in predecessors:
prev_state, edge_info = predecessors[state]
prev_key = prev_state[0]
path_edges.append((current_key, prev_key, edge_info))
state = prev_state
current_key = prev_key
route = _build_route_from_edges(path_edges, node_docs)
distance_km = None
src_lat = node_doc.get('latitude') src_lat = node_doc.get('latitude')
src_lon = node_doc.get('longitude') src_lon = node_doc.get('longitude')
if src_lat is not None and src_lon is not None: if src_lat is not None and src_lon is not None:
distance_km = _distance_km(src_lat, src_lon, hub_lat, hub_lon) distance_km = _distance_km(src_lat, src_lon, hub_lat, hub_lon)
found_routes.append(ProductRouteOptionType( found_routes.append(ProductRouteOptionType(
source_uuid=node_key, source_uuid=node_doc.get('_key'),
source_name=node_doc.get('name') or node_doc.get('product_name'), source_name=node_doc.get('name') or node_doc.get('product_name'),
source_lat=node_doc.get('latitude'), source_lat=node_doc.get('latitude'),
source_lon=node_doc.get('longitude'), source_lon=node_doc.get('longitude'),
distance_km=distance_km, distance_km=distance_km,
routes=[route] if route else [], routes=[route] if route else [],
)) ))
continue
neighbors = fetch_neighbors(node_key, phase)
expansions += 1
for neighbor in neighbors:
transport_type = neighbor.get('transport_type')
next_phase = allowed_next_phase(phase, transport_type)
if next_phase is None:
continue
travel_time = neighbor.get('travel_time_seconds')
distance_km = neighbor.get('distance_km')
neighbor_key = neighbor.get('neighbor_key')
node_docs[neighbor_key] = neighbor.get('neighbor_doc')
step_cost = travel_time if travel_time is not None else (distance_km or 0)
new_cost = cost + step_cost
state_key = (neighbor_key, next_phase)
if state_key in visited and new_cost >= visited[state_key]:
continue
visited[state_key] = new_cost
counter += 1
heapq.heappush(queue, (new_cost, counter, neighbor_key, next_phase))
predecessors[state_key] = ((node_key, phase), neighbor)
if not found_routes: if not found_routes:
logger.info("No offers found near hub %s", hub_uuid) logger.info("No offers found near hub %s", hub_uuid)
@@ -1367,78 +1210,31 @@ class Query(graphene.ObjectType):
# Graph-based nearest hubs when source_uuid provided # Graph-based nearest hubs when source_uuid provided
if source_uuid: if source_uuid:
nodes_col = db.collection('nodes') start_hub = _resolve_start_hub(db, source_uuid=source_uuid)
start = nodes_col.get(source_uuid) if not start_hub:
if not start: logger.warning("Source node %s not found for nearest hubs, falling back to coordinate search", source_uuid)
logger.warning("Source node %s not found for nearest hubs, falling back to radius search", source_uuid)
source_uuid = None
else: else:
start_uuid = start_hub.get('_key')
def is_target_hub(doc): def is_target_hub(doc):
if doc.get('_key') == source_uuid: if doc.get('_key') == start_uuid:
return False return False
if doc.get('node_type') not in ('logistics', None): if doc.get('node_type') not in ('logistics', None):
return False return False
types = doc.get('transport_types') or [] types = doc.get('transport_types') or []
return ('rail' in types) or ('sea' in types) return ('rail' in types) or ('sea' in types)
def fetch_neighbors(node_key): matches = _graph_find_targets(
aql = """ db,
FOR edge IN edges start_uuid=start_uuid,
FILTER edge.transport_type IN ['auto', 'rail', 'offer'] target_predicate=is_target_hub,
FILTER edge._from == @node_id OR edge._to == @node_id limit=limit,
LET neighbor_id = edge._from == @node_id ? edge._to : edge._from max_expansions=Query.MAX_EXPANSIONS,
LET neighbor = DOCUMENT(neighbor_id) )
FILTER neighbor != null
RETURN {
neighbor_key: neighbor._key,
neighbor_doc: neighbor,
transport_type: edge.transport_type,
distance_km: edge.distance_km,
travel_time_seconds: edge.travel_time_seconds
}
"""
cursor = db.aql.execute(
aql,
bind_vars={'node_id': f"nodes/{node_key}"},
)
return list(cursor)
queue = []
counter = 0
heapq.heappush(queue, (0, counter, source_uuid))
visited = {}
node_docs = {source_uuid: start}
found = []
expansions = 0
while queue and len(found) < limit and expansions < Query.MAX_EXPANSIONS:
cost, _, node_key = heapq.heappop(queue)
if node_key in visited and cost > visited[node_key]:
continue
visited[node_key] = cost
node_doc = node_docs.get(node_key)
if node_doc and is_target_hub(node_doc):
found.append(node_doc)
if len(found) >= limit:
break
neighbors = fetch_neighbors(node_key)
expansions += 1
for neighbor in neighbors:
neighbor_key = neighbor.get('neighbor_key')
if not neighbor_key:
continue
node_docs[neighbor_key] = neighbor.get('neighbor_doc')
step_cost = neighbor.get('travel_time_seconds') or neighbor.get('distance_km') or 0
new_cost = cost + step_cost
if neighbor_key in visited and new_cost >= visited[neighbor_key]:
continue
counter += 1
heapq.heappush(queue, (new_cost, counter, neighbor_key))
hubs = [] hubs = []
for node in found: for match in matches:
node = match.get('node') or {}
hubs.append(NodeType( hubs.append(NodeType(
uuid=node.get('_key'), uuid=node.get('_key'),
name=node.get('name'), name=node.get('name'),
@@ -1449,78 +1245,80 @@ class Query(graphene.ObjectType):
synced_at=node.get('synced_at'), synced_at=node.get('synced_at'),
transport_types=node.get('transport_types') or [], transport_types=node.get('transport_types') or [],
edges=[], edges=[],
distance_km=match.get('distance_km'),
)) ))
return hubs return hubs
if product_uuid: if product_uuid:
return self.resolve_hubs_for_product_graph(info, product_uuid, limit=limit) return self.resolve_hubs_for_product_graph(info, product_uuid, limit=limit)
# Simple nearest hubs search (no radius filtering) start_hub = _resolve_start_hub(db, lat=lat, lon=lon)
aql = """ if not start_hub:
FOR hub IN nodes
FILTER hub.node_type == 'logistics' OR hub.node_type == null
FILTER hub.product_uuid == null
LET types = hub.transport_types != null ? hub.transport_types : []
FILTER ('rail' IN types) OR ('sea' IN types)
FILTER hub.latitude != null AND hub.longitude != null
LET dist = DISTANCE(hub.latitude, hub.longitude, @lat, @lon) / 1000
SORT dist ASC
LIMIT @limit
RETURN MERGE(hub, {distance_km: dist})
"""
bind_vars = {'lat': lat, 'lon': lon, 'limit': limit}
try:
cursor = db.aql.execute(aql, bind_vars=bind_vars)
hubs = []
for node in cursor:
hubs.append(NodeType(
uuid=node['_key'],
name=node.get('name'),
latitude=node.get('latitude'),
longitude=node.get('longitude'),
country=node.get('country'),
country_code=node.get('country_code'),
synced_at=node.get('synced_at'),
transport_types=node.get('transport_types') or [],
edges=[],
distance_km=node.get('distance_km'),
))
logger.info("Found %d hubs near (%.3f, %.3f)", len(hubs), lat, lon)
return hubs
except Exception as e:
logger.error("Error finding nearest hubs: %s", e)
return [] return []
start_uuid = start_hub.get('_key')
def is_target_hub(doc):
if doc.get('_key') == start_uuid:
return False
if doc.get('node_type') not in ('logistics', None):
return False
types = doc.get('transport_types') or []
return ('rail' in types) or ('sea' in types)
matches = _graph_find_targets(
db,
start_uuid=start_uuid,
target_predicate=is_target_hub,
limit=max(limit - 1, 0),
max_expansions=Query.MAX_EXPANSIONS,
)
hubs = [
NodeType(
uuid=start_hub.get('_key'),
name=start_hub.get('name'),
latitude=start_hub.get('latitude'),
longitude=start_hub.get('longitude'),
country=start_hub.get('country'),
country_code=start_hub.get('country_code'),
synced_at=start_hub.get('synced_at'),
transport_types=start_hub.get('transport_types') or [],
edges=[],
distance_km=0,
)
]
for match in matches:
node = match.get('node') or {}
hubs.append(NodeType(
uuid=node.get('_key'),
name=node.get('name'),
latitude=node.get('latitude'),
longitude=node.get('longitude'),
country=node.get('country'),
country_code=node.get('country_code'),
synced_at=node.get('synced_at'),
transport_types=node.get('transport_types') or [],
edges=[],
distance_km=match.get('distance_km'),
))
logger.info("Found %d hubs via graph near (%.3f, %.3f)", len(hubs), lat, lon)
return hubs[:limit]
def resolve_nearest_offers(self, info, lat, lon, radius=500, product_uuid=None, hub_uuid=None, limit=50): def resolve_nearest_offers(self, info, lat, lon, radius=500, product_uuid=None, hub_uuid=None, limit=50):
"""Find nearest offers to coordinates, optionally filtered by product. If hub_uuid provided, calculates routes.""" """Find nearest offers to coordinates, optionally filtered by product. If hub_uuid provided, calculates routes."""
db = get_db() db = get_db()
ensure_graph()
try: try:
nodes_col = db.collection('nodes') nodes_col = db.collection('nodes')
# If no hub_uuid provided, snap to nearest hub by coordinates. start_hub = _resolve_start_hub(db, source_uuid=hub_uuid, lat=lat, lon=lon)
if not hub_uuid: if not start_hub:
aql_hub = """ logger.info("No hub found near coordinates (%.3f, %.3f)", lat, lon)
FOR hub IN nodes return []
FILTER hub.node_type == 'logistics' OR hub.node_type == null
FILTER hub.product_uuid == null
LET types = hub.transport_types != null ? hub.transport_types : []
FILTER ('rail' IN types) OR ('sea' IN types)
FILTER hub.latitude != null AND hub.longitude != null
LET dist = DISTANCE(hub.latitude, hub.longitude, @lat, @lon) / 1000
SORT dist ASC
LIMIT 1
RETURN hub
"""
cursor = db.aql.execute(aql_hub, bind_vars={'lat': lat, 'lon': lon})
hubs = list(cursor)
if not hubs:
logger.info("No hub found near coordinates (%.3f, %.3f)", lat, lon)
return []
hub_uuid = hubs[0]['_key']
hub_uuid = start_hub.get('_key')
expanded_limit = max(limit * 5, limit) expanded_limit = max(limit * 5, limit)
route_options = Query.resolve_offers_by_hub( route_options = Query.resolve_offers_by_hub(
Query, info, hub_uuid, product_uuid, expanded_limit Query, info, hub_uuid, product_uuid, expanded_limit
@@ -1723,33 +1521,49 @@ class Query(graphene.ObjectType):
def resolve_route_to_coordinate(self, info, offer_uuid, lat, lon): def resolve_route_to_coordinate(self, info, offer_uuid, lat, lon):
"""Get route from offer to target coordinates (finds nearest hub automatically).""" """Get route from offer to target coordinates (finds nearest hub automatically)."""
db = get_db() db = get_db()
nodes_col = db.collection('nodes')
# Find nearest hub to target coordinates
aql_hub = """
FOR hub IN nodes
FILTER hub.node_type == 'logistics' OR hub.node_type == null
FILTER hub.product_uuid == null
FILTER hub.latitude != null AND hub.longitude != null
LET dist = DISTANCE(hub.latitude, hub.longitude, @lat, @lon) / 1000
SORT dist ASC
LIMIT 1
RETURN hub
"""
try: try:
cursor = db.aql.execute(aql_hub, bind_vars={'lat': lat, 'lon': lon}) offer = nodes_col.get(offer_uuid)
hubs = list(cursor) if not offer:
if not hubs: logger.info("Offer %s not found", offer_uuid)
return None
nearest_hub = _snap_to_nearest_hub(db, lat, lon)
if not nearest_hub:
logger.info("No hub found near coordinates (%.3f, %.3f)", lat, lon) logger.info("No hub found near coordinates (%.3f, %.3f)", lat, lon)
return None return None
nearest_hub = hubs[0]
hub_uuid = nearest_hub['_key'] hub_uuid = nearest_hub['_key']
logger.info("Found nearest hub %s to coordinates (%.3f, %.3f)", hub_uuid, lat, lon) logger.info("Found nearest hub %s to coordinates (%.3f, %.3f)", hub_uuid, lat, lon)
# Use existing offer_to_hub logic matches = _graph_find_targets(
# Note: in graphene, self is None (root value), so we call as class method db,
return Query.resolve_offer_to_hub(Query, info, offer_uuid, hub_uuid) start_uuid=hub_uuid,
target_predicate=lambda doc: doc.get('_key') == offer_uuid,
limit=1,
max_expansions=Query.MAX_EXPANSIONS,
)
if not matches:
return None
match = matches[0]
route = match.get('route')
distance_km = match.get('distance_km')
if distance_km is None:
src_lat = offer.get('latitude')
src_lon = offer.get('longitude')
if src_lat is not None and src_lon is not None:
distance_km = _distance_km(src_lat, src_lon, nearest_hub.get('latitude'), nearest_hub.get('longitude'))
return ProductRouteOptionType(
source_uuid=offer.get('_key'),
source_name=offer.get('name') or offer.get('product_name'),
source_lat=offer.get('latitude'),
source_lon=offer.get('longitude'),
distance_km=distance_km,
routes=[route] if route else [],
)
except Exception as e: except Exception as e:
logger.error("Error finding route to coordinates: %s", e) logger.error("Error finding route to coordinates: %s", e)
return None return None
@@ -2025,3 +1839,199 @@ def _distance_km(lat1, lon1, lat2, lon2):
Query._distance_km = _distance_km Query._distance_km = _distance_km
def _graph_allowed_next_phase(current_phase, transport_type):
"""
Phase-based routing: auto → rail* → auto.
- end_auto: allow one auto, rail, or offer
- end_auto_done: auto used — rail or offer
- rail: any number of rail, then one auto or offer
- start_auto_done: auto used — only offer
"""
if current_phase == 'end_auto':
if transport_type == 'offer':
return 'offer'
if transport_type == 'auto':
return 'end_auto_done'
if transport_type == 'rail':
return 'rail'
return None
if current_phase == 'end_auto_done':
if transport_type == 'offer':
return 'offer'
if transport_type == 'rail':
return 'rail'
return None
if current_phase == 'rail':
if transport_type == 'offer':
return 'offer'
if transport_type == 'rail':
return 'rail'
if transport_type == 'auto':
return 'start_auto_done'
return None
if current_phase == 'start_auto_done':
if transport_type == 'offer':
return 'offer'
return None
return None
def _graph_allowed_types_for_phase(phase):
if phase == 'end_auto':
return ['auto', 'rail', 'offer']
if phase == 'end_auto_done':
return ['rail', 'offer']
if phase == 'rail':
return ['rail', 'auto', 'offer']
if phase == 'start_auto_done':
return ['offer']
return ['offer']
def _graph_fetch_neighbors(db, node_key, allowed_types):
aql = """
FOR edge IN edges
FILTER edge.transport_type IN @types
FILTER edge._from == @node_id OR edge._to == @node_id
LET neighbor_id = edge._from == @node_id ? edge._to : edge._from
LET neighbor = DOCUMENT(neighbor_id)
FILTER neighbor != null
RETURN {
neighbor_key: neighbor._key,
neighbor_doc: neighbor,
from_id: edge._from,
to_id: edge._to,
transport_type: edge.transport_type,
distance_km: edge.distance_km,
travel_time_seconds: edge.travel_time_seconds
}
"""
cursor = db.aql.execute(
aql,
bind_vars={'node_id': f"nodes/{node_key}", 'types': allowed_types},
)
return list(cursor)
def _graph_find_targets(db, start_uuid, target_predicate, limit=10, max_expansions=20000):
"""Unified graph traversal: auto → rail* → auto, returns routes for target nodes."""
ensure_graph()
nodes_col = db.collection('nodes')
start = nodes_col.get(start_uuid)
if not start:
return []
queue = []
counter = 0
heapq.heappush(queue, (0, counter, start_uuid, 'end_auto'))
visited = {}
predecessors = {}
node_docs = {start_uuid: start}
found = []
expansions = 0
while queue and len(found) < limit and expansions < max_expansions:
cost, _, node_key, phase = heapq.heappop(queue)
if (node_key, phase) in visited and cost > visited[(node_key, phase)]:
continue
visited[(node_key, phase)] = cost
node_doc = node_docs.get(node_key)
if node_doc and target_predicate(node_doc):
path_edges = []
state = (node_key, phase)
current_key = node_key
while state in predecessors:
prev_state, edge_info = predecessors[state]
prev_key = prev_state[0]
path_edges.append((current_key, prev_key, edge_info))
state = prev_state
current_key = prev_key
route = _build_route_from_edges(path_edges, node_docs)
distance_km = route.total_distance_km if route else None
found.append({
'node': node_doc,
'route': route,
'distance_km': distance_km,
'cost': cost,
})
continue
neighbors = _graph_fetch_neighbors(db, node_key, _graph_allowed_types_for_phase(phase))
expansions += 1
for neighbor in neighbors:
transport_type = neighbor.get('transport_type')
next_phase = _graph_allowed_next_phase(phase, transport_type)
if next_phase is None:
continue
travel_time = neighbor.get('travel_time_seconds')
distance_km = neighbor.get('distance_km')
neighbor_key = neighbor.get('neighbor_key')
if not neighbor_key:
continue
node_docs[neighbor_key] = neighbor.get('neighbor_doc')
step_cost = travel_time if travel_time is not None else (distance_km or 0)
new_cost = cost + step_cost
state_key = (neighbor_key, next_phase)
if state_key in visited and new_cost >= visited[state_key]:
continue
counter += 1
heapq.heappush(queue, (new_cost, counter, neighbor_key, next_phase))
predecessors[state_key] = ((node_key, phase), neighbor)
return found
def _snap_to_nearest_hub(db, lat, lon):
aql = """
FOR hub IN nodes
FILTER hub.node_type == 'logistics' OR hub.node_type == null
FILTER hub.product_uuid == null
LET types = hub.transport_types != null ? hub.transport_types : []
FILTER ('rail' IN types) OR ('sea' IN types)
FILTER hub.latitude != null AND hub.longitude != null
LET dist = DISTANCE(hub.latitude, hub.longitude, @lat, @lon) / 1000
SORT dist ASC
LIMIT 1
RETURN hub
"""
cursor = db.aql.execute(aql, bind_vars={'lat': lat, 'lon': lon})
hubs = list(cursor)
return hubs[0] if hubs else None
def _resolve_start_hub(db, source_uuid=None, lat=None, lon=None):
nodes_col = db.collection('nodes')
if source_uuid:
node = nodes_col.get(source_uuid)
if not node:
return None
if node.get('node_type') in ('logistics', None):
types = node.get('transport_types') or []
if ('rail' in types) or ('sea' in types):
return node
node_lat = node.get('latitude')
node_lon = node.get('longitude')
if node_lat is None or node_lon is None:
return None
return _snap_to_nearest_hub(db, node_lat, node_lon)
if lat is None or lon is None:
return None
return _snap_to_nearest_hub(db, lat, lon)