45 lines
2.2 KiB
Python
45 lines
2.2 KiB
Python
"""Publication predicates applied before entity resolution, joins and LIMIT.
|
|
|
|
The query wrappers fail at import if their insertion point changes. Filtering
|
|
only final citations would still expose hidden facts and incorrect counts.
|
|
"""
|
|
|
|
|
|
def visible_node(node: str) -> str:
|
|
# Identifiers come only from the fixed query definitions, never user input.
|
|
return f'''CASE
|
|
WHEN {node}:Document THEN coalesce({node}.published, true)
|
|
WHEN {node}:Person OR ({node}:Topic AND {node}.source_document IS NOT NULL) THEN EXISTS {{
|
|
MATCH (public_source:Document)
|
|
WHERE public_source.graph_scope=$scope AND public_source.path={node}.source_document
|
|
AND coalesce(public_source.published, true)
|
|
}}
|
|
WHEN {node}:Thesis OR {node}:Work THEN EXISTS {{
|
|
MATCH ({node})-[:EVIDENCED_BY]->(public_source:Document)
|
|
WHERE public_source.graph_scope=$scope AND coalesce(public_source.published, true)
|
|
}} AND ({node}.source_document IS NULL OR EXISTS {{
|
|
MATCH (origin:Document) WHERE origin.graph_scope=$scope
|
|
AND origin.path={node}.source_document AND coalesce(origin.published, true)
|
|
}})
|
|
ELSE EXISTS {{
|
|
MATCH ({node})--(public_source:Document)
|
|
WHERE public_source.graph_scope=$scope AND coalesce(public_source.published, true)
|
|
}} END'''
|
|
|
|
|
|
def public_query(query: str, *, nodes: tuple[str, ...] = (), documents: tuple[str, ...] = (),
|
|
path: bool = False) -> str:
|
|
if path:
|
|
anchor = 'WHERE all(node IN nodes(p) WHERE node.graph_scope = $scope)'
|
|
predicate = f'all(public_node IN nodes(p) WHERE {visible_node("public_node")})'
|
|
if query.count(anchor) != 1:
|
|
raise ValueError('Missing unique path publication guard position')
|
|
return query.replace(anchor, anchor + f' AND (NOT $published_only OR ({predicate}))', 1)
|
|
predicates = [visible_node(node) for node in nodes]
|
|
predicates.extend(f'coalesce({node}.published, true)' for node in documents)
|
|
anchor = '\nWHERE '
|
|
if anchor not in query or not predicates:
|
|
raise ValueError('Missing publication guard position')
|
|
condition = ' AND '.join(f'({p})' for p in predicates)
|
|
return query.replace(anchor, f'\nWHERE (NOT $published_only OR ({condition})) AND ', 1)
|