"""
Cypher template library — deterministic, direction-aware query strategies.
Each strategy produces a ready-to-execute Cypher query from structured intent.
Relations in Neo4j are directed: (source)-[:AMENDS]->(target) means
"source amends target". Templates preserve this semantics.
"""
import logging
from dataclasses import dataclass
from typing import List, Optional
from .intent import Direction, RelationIntent
logger = logging.getLogger(__name__)
[docs]
@dataclass(frozen=True)
class CypherTemplateMatch:
"""Deterministic Cypher template selected for a relation intent."""
strategy: str
cypher: str
detail: str
# ── Helpers ───────────────────────────────────────────────────────────────────
def _escape(value: str) -> str:
return value.replace("\\", "\\\\").replace("'", "\\'")
def _type_filter(relation_types: List[str]) -> str:
if not relation_types:
return ""
return ":" + "|".join(relation_types)
def _directed_pattern(
*,
variable: str,
relation_types: List[str],
direction: Direction,
max_depth: Optional[int] = None,
) -> str:
"""Build a directed relationship pattern string."""
type_part = _type_filter(relation_types)
depth_part = f"*..{max(int(max_depth), 1)}" if max_depth is not None else ""
rel = f"[{variable}{type_part}{depth_part}]"
if direction == "outgoing":
return f"-{rel}->"
if direction == "incoming":
return f"<-{rel}-"
return f"-{rel}-"
_RETURN_CLAUSE = (
"RETURN "
"startNode(r).id AS source_act_id, "
"startNode(r).celex AS source_celex, "
"coalesce(startNode(r).title, '') AS source_title, "
"type(r) AS relation_type, "
"endNode(r).id AS target_act_id, "
"endNode(r).celex AS target_celex, "
"coalesce(endNode(r).title, '') AS target_title"
)
def _projection(row_limit: int) -> str:
return f"{_RETURN_CLAUSE}\nLIMIT {max(int(row_limit), 1)}"
# ── 2-CELEX strategies ───────────────────────────────────────────────────────
def _direct_edge_between_acts(
left: str,
right: str,
intent: RelationIntent,
row_limit: int,
) -> CypherTemplateMatch:
"""Direct edges between two identified acts (respecting direction)."""
left_esc, right_esc = _escape(left), _escape(right)
pat = _directed_pattern(
variable="r",
relation_types=intent.relation_types,
direction=intent.direction,
)
return CypherTemplateMatch(
strategy="template_direct_between_acts",
cypher=(
f"MATCH (a:Act {{celex: '{left_esc}'}}){pat}(b:Act {{celex: '{right_esc}'}})\n{_projection(row_limit)}"
),
detail="Relations directes entre deux actes (orientées)",
)
def _typed_path_between_acts(
left: str,
right: str,
intent: RelationIntent,
max_depth: int,
row_limit: int,
) -> CypherTemplateMatch:
"""Bounded directed path between two acts."""
left_esc, right_esc = _escape(left), _escape(right)
type_part = _type_filter(intent.relation_types)
bounded = max(int(max_depth), 1)
# shortestPath requires undirected or a single direction — use allShortestPaths
# with direction to get legally meaningful chains.
if intent.direction == "incoming":
path_pattern = f"(b:Act {{celex: '{right_esc}'}})-[{type_part}*..{bounded}]->(a:Act {{celex: '{left_esc}'}})"
elif intent.direction == "outgoing":
path_pattern = f"(a:Act {{celex: '{left_esc}'}})-[{type_part}*..{bounded}]->(b:Act {{celex: '{right_esc}'}})"
else:
# Both directions: use undirected shortest path
path_pattern = f"(a:Act {{celex: '{left_esc}'}})-[{type_part}*..{bounded}]-(b:Act {{celex: '{right_esc}'}})"
return CypherTemplateMatch(
strategy="template_path_between_acts",
cypher=(
f"MATCH path = shortestPath({path_pattern})\n"
"WITH path\n"
"UNWIND relationships(path) AS r\n"
f"{_projection(row_limit)}"
),
detail="Chemin borné orienté entre deux actes",
)
# ── 1-CELEX strategies ───────────────────────────────────────────────────────
def _outgoing_neighbors(
celex: str,
intent: RelationIntent,
row_limit: int,
) -> CypherTemplateMatch:
"""What does this act do to others? (a)-[r]->(b)"""
c = _escape(celex)
pat = _directed_pattern(
variable="r",
relation_types=intent.relation_types,
direction="outgoing",
)
return CypherTemplateMatch(
strategy="template_outgoing_neighbors",
cypher=(f"MATCH (a:Act {{celex: '{c}'}}){pat}(b:Act)\n{_projection(row_limit)}"),
detail="Relations sortantes : ce que cet acte modifie/abroge/etc.",
)
def _incoming_neighbors(
celex: str,
intent: RelationIntent,
row_limit: int,
) -> CypherTemplateMatch:
"""What do others do to this act? (b)-[r]->(a)"""
c = _escape(celex)
pat = _directed_pattern(
variable="r",
relation_types=intent.relation_types,
direction="outgoing", # b->a means incoming for a
)
return CypherTemplateMatch(
strategy="template_incoming_neighbors",
cypher=(f"MATCH (b:Act){pat}(a:Act {{celex: '{c}'}})\n{_projection(row_limit)}"),
detail="Relations entrantes : ce qui modifie/abroge cet acte",
)
def _all_neighbors(
celex: str,
intent: RelationIntent,
row_limit: int,
) -> CypherTemplateMatch:
"""All relations around an act, both directions, with direction info preserved."""
c = _escape(celex)
type_part = _type_filter(intent.relation_types)
return CypherTemplateMatch(
strategy="template_all_neighbors",
cypher=(f"MATCH (a:Act {{celex: '{c}'}})-[r{type_part}]-(b:Act)\n{_projection(row_limit)}"),
detail="Toutes les relations autour de cet acte (entrantes + sortantes)",
)
def _amendment_chain(
celex: str,
max_depth: int,
row_limit: int,
) -> CypherTemplateMatch:
"""Multi-hop amendment/repeal chain originating from an act."""
c = _escape(celex)
bounded = max(int(max_depth), 1)
return CypherTemplateMatch(
strategy="template_amendment_chain",
cypher=(
f"MATCH path = (a:Act {{celex: '{c}'}})-[:AMENDS|REPEALS|REPLACES*..{bounded}]->(b:Act)\n"
"WITH path\n"
"UNWIND relationships(path) AS r\n"
f"{_projection(row_limit)}"
),
detail="Chaîne d'amendements/abrogations depuis cet acte",
)
def _implementing_acts(
celex: str,
row_limit: int,
) -> CypherTemplateMatch:
"""Acts that implement/transpose a directive."""
c = _escape(celex)
return CypherTemplateMatch(
strategy="template_implementing_acts",
cypher=(f"MATCH (b:Act)-[r:IMPLEMENTS]->(a:Act {{celex: '{c}'}})\n{_projection(row_limit)}"),
detail="Actes qui transposent/implémentent cette directive",
)
# ── Template router ──────────────────────────────────────────────────────────
[docs]
def match_template(
intent: RelationIntent,
*,
max_graph_depth: int,
row_limit: int,
) -> Optional[CypherTemplateMatch]:
"""Select the best deterministic Cypher template from structured intent."""
celexes = intent.celex_candidates
# ── 2-CELEX: relationship between two acts ──
if len(celexes) >= 2:
left, right = celexes[0], celexes[1]
if intent.wants_direct:
return _direct_edge_between_acts(left, right, intent, row_limit)
return _typed_path_between_acts(left, right, intent, max_graph_depth, row_limit)
# ── 1-CELEX: relations around a single act ──
if len(celexes) == 1:
celex = celexes[0]
# IMPLEMENTS-specific: "qui transpose cette directive ?"
if "IMPLEMENTS" in intent.relation_types and intent.direction == "incoming":
return _implementing_acts(celex, row_limit)
# Amendment chain: outgoing chain of AMENDS/REPEALS/REPLACES
amend_types = {"AMENDS", "REPEALS", "REPLACES"}
if intent.direction == "outgoing" and amend_types.intersection(intent.relation_types):
return _amendment_chain(celex, max_graph_depth, row_limit)
# Direction-aware neighbors
if intent.direction == "outgoing":
return _outgoing_neighbors(celex, intent, row_limit)
if intent.direction == "incoming":
return _incoming_neighbors(celex, intent, row_limit)
return _all_neighbors(celex, intent, row_limit)
# ── 0-CELEX: no template possible, fall through to Text2Cypher ──
return None