From 30b9825365c36924fd2f9b5ff8cf424cef20d2c8 Mon Sep 17 00:00:00 2001 From: Jeroen Schweitzer Date: Wed, 10 Dec 2025 01:16:24 +0100 Subject: [PATCH] feat(library-desk): add entity management to graph and ingestion services Graph Service: - Add get_all_entities() to retrieve entities with wiki page paths - Add create_entity_mentions() for MENTIONS relationship creation - Support entity-to-document linking via title matching Ingestion Service: - Add _link_existing_entities() for automatic entity linking - Auto-link entities during page ingestion - Support skip_entity_linking parameter for granular control --- .../src/services/graph_service.py | 1181 +++++++++++++++++ .../src/services/ingestion_service.py | 415 ++++++ 2 files changed, 1596 insertions(+) create mode 100644 services/library-desk/src/services/graph_service.py create mode 100644 services/library-desk/src/services/ingestion_service.py diff --git a/services/library-desk/src/services/graph_service.py b/services/library-desk/src/services/graph_service.py new file mode 100644 index 0000000..c32798e --- /dev/null +++ b/services/library-desk/src/services/graph_service.py @@ -0,0 +1,1181 @@ +""" +Graph service for Library Desk Neo4j operations. + +Handles knowledge graph management with user-scoped operations. +""" + +import re +import time +from typing import List, Dict, Any, Optional +from datetime import datetime +import logging + +from src.clients.neo4j_client import Neo4jClient +from src.clients.wikijs_client import WikiJSClient +from src.core.multi_tenancy import get_neo4j_user_label +from src.models.graph import ( + GraphNode, GraphRelationship, GraphNodeDetail, + CypherQueryResponse, GraphUpdateSummary, EntityMention, + NodeListResponse, MindMapNode, MindMapLink, MindMapResponse +) + +logger = logging.getLogger(__name__) + + +def _serialize_neo4j_types(obj: Any) -> Any: + """ + Convert Neo4j types to JSON-serializable types. + + Args: + obj: Object that may contain Neo4j types + + Returns: + JSON-serializable object + """ + if obj is None: + return None + + # Handle Neo4j DateTime + if hasattr(obj, 'to_native'): # Neo4j temporal types have to_native() + return obj.to_native().isoformat() + + # Handle dict recursively + if isinstance(obj, dict): + return {k: _serialize_neo4j_types(v) for k, v in obj.items()} + + # Handle list recursively + if isinstance(obj, list): + return [_serialize_neo4j_types(item) for item in obj] + + # Return as-is for basic types + return obj + + +class GraphService: + """ + Service for Neo4j knowledge graph operations. + + Responsibilities: + - User-scoped graph queries + - Entity extraction from wiki pages + - Graph updates from page content + - Mind map generation + """ + + def __init__(self, neo4j_client: Neo4jClient, wikijs_client: WikiJSClient): + """ + Initialize graph service. + + Args: + neo4j_client: Neo4j database client + wikijs_client: Wiki.js client for fetching pages + """ + self.neo4j = neo4j_client + self.wiki = wikijs_client + + async def execute_query( + self, + query: str, + parameters: Dict[str, Any], + user: str + ) -> CypherQueryResponse: + """ + Execute user-scoped Cypher query. + + Automatically injects user label into query for security. + + Args: + query: Cypher query + parameters: Query parameters + user: User identifier + + Returns: + Query results with metadata + """ + start_time = time.time() + + # Get user-specific label + user_label = get_neo4j_user_label(user) + + # Inject user label into query for scoping + # This ensures users can only query their own data + scoped_query = self._scope_query_to_user(query, user_label) + + try: + results = await self.neo4j.execute_query(scoped_query, parameters) + query_time_ms = (time.time() - start_time) * 1000 + + return CypherQueryResponse( + results=results, + count=len(results), + query_time_ms=query_time_ms + ) + + except Exception as e: + logger.error(f"Cypher query failed: {e}", exc_info=True) + raise ValueError(f"Query execution failed: {str(e)}") + + def _scope_query_to_user(self, query: str, user_label: str) -> str: + """ + Inject user label into Cypher query for multi-tenancy. + + Simple implementation: adds user label to node patterns. + Production version would use proper query parsing. + + Args: + query: Original Cypher query + user_label: User-specific label + + Returns: + Scoped query + """ + # For now, return query as-is + # TODO: Implement proper query scoping with label injection + logger.warning("Query scoping not yet implemented - returning unscoped query") + return query + + async def list_nodes( + self, + user: str, + node_type: Optional[str] = None, + limit: int = 100 + ) -> NodeListResponse: + """ + List nodes for a user. + + Args: + user: User identifier + node_type: Optional node type filter (Document, Person, etc.) + limit: Maximum nodes to return + + Returns: + List of nodes + """ + user_label = get_neo4j_user_label(user) + + # Build query based on filters + # Note: We need to explicitly return elementId and labels since result.data() converts nodes to dicts + if node_type: + query = f""" + MATCH (n:{user_label}:{node_type}) + RETURN elementId(n) as id, labels(n) as labels, properties(n) as props + LIMIT $limit + """ + else: + query = f""" + MATCH (n:{user_label}) + RETURN elementId(n) as id, labels(n) as labels, properties(n) as props + LIMIT $limit + """ + + results = await self.neo4j.execute_query(query, {"limit": limit}) + + nodes = [ + GraphNode( + id=str(record["id"]), + labels=record["labels"], + properties=_serialize_neo4j_types(record["props"]) + ) + for record in results + ] + + return NodeListResponse( + nodes=nodes, + total=len(nodes), + user=user + ) + + async def get_node(self, node_id: str, user: str) -> Optional[GraphNodeDetail]: + """ + Get node details with relationships. + + Args: + node_id: Node element ID + user: User identifier + + Returns: + Node details or None if not found + """ + user_label = get_neo4j_user_label(user) + + query = f""" + MATCH (n:{user_label}) + WHERE elementId(n) = $node_id + OPTIONAL MATCH (n)-[r]-(m) + RETURN + elementId(n) as node_id, + labels(n) as node_labels, + properties(n) as node_props, + collect({{ + id: elementId(r), + type: type(r), + start_node: elementId(startNode(r)), + end_node: elementId(endNode(r)), + props: properties(r) + }}) as rels, + collect({{ + id: elementId(m), + labels: labels(m), + props: properties(m) + }}) as related + """ + + results = await self.neo4j.execute_query(query, {"node_id": node_id}) + + if not results: + return None + + record = results[0] + node = GraphNode( + id=node_id, + labels=record["node_labels"], + properties=_serialize_neo4j_types(record["node_props"]) + ) + + # Parse relationships + relationships = [] + related_nodes = [] + + for rel in record["rels"]: + if rel and rel["id"]: # Check if relationship exists (not null from OPTIONAL MATCH) + relationships.append(GraphRelationship( + id=rel["id"], + type=rel["type"], + start_node=rel["start_node"], + end_node=rel["end_node"], + properties=_serialize_neo4j_types(rel["props"] or {}) + )) + + for rel_node in record["related"]: + if rel_node and rel_node["id"]: # Check if node exists + related_nodes.append(GraphNode( + id=rel_node["id"], + labels=rel_node["labels"], + properties=_serialize_neo4j_types(rel_node["props"] or {}) + )) + + return GraphNodeDetail( + node=node, + relationships=relationships, + related_nodes=related_nodes + ) + + def _extract_entities(self, content: str) -> List[EntityMention]: + """ + Extract entities from page content. + + Simple regex-based extraction for now. + TODO: Use spaCy or similar NLP library for better extraction. + + Args: + content: Markdown content + + Returns: + List of extracted entities + """ + entities = [] + + # Extract mentions in format: @username, #tag, [[WikiLink]] + # Person mentions: @username + person_pattern = r'@([a-zA-Z0-9_-]+)' + for match in re.finditer(person_pattern, content): + entities.append(EntityMention( + text=match.group(1), + type="Person", + confidence=0.8 + )) + + # Project/Concept mentions: [[WikiLink]] + wikilink_pattern = r'\[\[([^\]]+)\]\]' + for match in re.finditer(wikilink_pattern, content): + entities.append(EntityMention( + text=match.group(1), + type="Concept", + confidence=0.9 + )) + + # Technology mentions: docker, kubernetes, python, etc. + tech_keywords = ['docker', 'kubernetes', 'python', 'neo4j', 'qdrant', + 'wikijs', 'fastapi', 'ollama', 'redis'] + content_lower = content.lower() + for tech in tech_keywords: + if tech in content_lower: + entities.append(EntityMention( + text=tech.capitalize(), + type="Technology", + confidence=0.7 + )) + + return entities + + async def update_from_page( + self, + page_id: int, + user: str, + force_refresh: bool = False + ) -> GraphUpdateSummary: + """ + Update knowledge graph from a wiki page. + + Extracts entities and creates/updates graph nodes and relationships. + + Args: + page_id: Wiki page ID + user: User identifier + force_refresh: Force re-extraction even if unchanged + + Returns: + Summary of update operation + """ + start_time = time.time() + + try: + # Fetch page from Wiki.js + page = await self.wiki.get_page(page_id) + if not page: + raise ValueError(f"Page {page_id} not found") + + # PROTECTION: Skip entity extraction on auto-generated entity stub pages + tags = page.get("tags", []) + if "entity-stub" in tags or "auto-generated" in tags: + logger.info(f"Skipping entity extraction for auto-generated page {page_id}") + return GraphUpdateSummary( + page_id=page_id, + page_title=page.get("title", ""), + processing_time_ms=(time.time() - start_time) * 1000, + success=True + ) + + # Extract entities from content + content = page.get("content", "") + entities = self._extract_entities(content) + + # Get user labels for namespacing + from src.core.multi_tenancy import get_neo4j_user_base_label + user_base_label = get_neo4j_user_base_label(user) # For entities + user_doc_label = get_neo4j_user_label(user) # For documents + + # Create/update Document node + doc_query = f""" + MERGE (d:{user_doc_label}:Document {{page_id: $page_id}}) + SET d.title = $title, + d.path = $path, + d.updated_at = datetime(), + d.content_length = $content_length + RETURN d + """ + + await self.neo4j.execute_query(doc_query, { + "page_id": page_id, + "title": page.get("title"), + "path": page.get("path"), + "content_length": len(content) + }) + + nodes_created = 1 # Document node + nodes_updated = 0 + relationships_created = 0 + + # Create entity nodes and relationships + for entity in entities: + entity_query = f""" + MERGE (e:{user_base_label}:{entity.type} {{name: $name}}) + WITH e + MATCH (d:{user_doc_label}:Document {{page_id: $page_id}}) + MERGE (d)-[r:MENTIONS]->(e) + SET r.confidence = $confidence + RETURN e, r + """ + + result = await self.neo4j.execute_query(entity_query, { + "name": entity.text, + "page_id": page_id, + "confidence": entity.confidence + }) + + if result: + relationships_created += 1 + + processing_time_ms = (time.time() - start_time) * 1000 + + logger.info(f"Updated graph from page {page_id}: {len(entities)} entities") + + return GraphUpdateSummary( + page_id=page_id, + page_title=page.get("title", ""), + nodes_created=nodes_created, + nodes_updated=nodes_updated, + relationships_created=relationships_created, + entities_extracted=entities, + processing_time_ms=processing_time_ms, + success=True + ) + + except Exception as e: + processing_time_ms = (time.time() - start_time) * 1000 + logger.error(f"Failed to update graph from page {page_id}: {e}", exc_info=True) + + return GraphUpdateSummary( + page_id=page_id, + page_title="Unknown", + processing_time_ms=processing_time_ms, + success=False, + error_message=str(e) + ) + + async def delete_page( + self, + page_id: int, + user: str + ) -> int: + """ + Delete a page's Document node and all its relationships from the graph. + + Args: + page_id: Wiki page ID to delete + user: User identifier + + Returns: + Number of nodes deleted (should be 1 if successful, 0 if not found) + """ + from src.core.multi_tenancy import get_neo4j_user_label + + user_doc_label = get_neo4j_user_label(user) + + # Delete Document node and all its relationships + # DETACH DELETE removes the node and all relationships connected to it + delete_query = f""" + MATCH (d:{user_doc_label}:Document {{page_id: $page_id}}) + DETACH DELETE d + RETURN count(d) as deleted_count + """ + + try: + result = await self.neo4j.execute_query( + delete_query, + {"page_id": page_id} + ) + + deleted_count = result[0]["deleted_count"] if result else 0 + + if deleted_count > 0: + logger.info(f"Deleted Document node for page {page_id} from graph") + else: + logger.warning(f"No Document node found for page {page_id}") + + return deleted_count + + except Exception as e: + logger.error(f"Failed to delete page {page_id} from graph: {e}", exc_info=True) + return 0 + + async def generate_mindmap( + self, + center_node_id: str, + user: str, + depth: int = 2 + ) -> MindMapResponse: + """ + Generate mind map data for visualization. + + Args: + center_node_id: Central node ID + user: User identifier + depth: Traversal depth + + Returns: + Mind map nodes and links + """ + user_label = get_neo4j_user_label(user) + + # Traverse graph from center node + query = f""" + MATCH path = (center:{user_label})-[*1..{depth}]-(related:{user_label}) + WHERE elementId(center) = $center_id + WITH center, collect(distinct related) as nodes, collect(relationships(path)) as rels_list + UNWIND rels_list as rels_in_path + UNWIND rels_in_path as rel + WITH center, nodes, collect(distinct rel) as all_rels + RETURN + {{ + id: elementId(center), + labels: labels(center), + props: properties(center) + }} as center, + [node in nodes | {{ + id: elementId(node), + labels: labels(node), + props: properties(node) + }}] as nodes, + [r in all_rels | {{ + id: elementId(r), + type: type(r), + start_node: elementId(startNode(r)), + end_node: elementId(endNode(r)), + props: properties(r) + }}] as all_rels + """ + + results = await self.neo4j.execute_query(query, {"center_id": center_node_id}) + + if not results: + return MindMapResponse(nodes=[], links=[], center_node=center_node_id, depth=depth) + + record = results[0] + + # Build mind map nodes + mind_nodes = [] + + # Center node + center = record["center"] + center_props = _serialize_neo4j_types(center["props"]) + mind_nodes.append(MindMapNode( + id=center["id"], + label=center_props.get("title") or center_props.get("name", "Unknown"), + type=center["labels"][0] if center["labels"] else "Node", + size=20, + color="#FF6B6B" + )) + + # Related nodes + for node in record["nodes"]: + node_props = _serialize_neo4j_types(node["props"]) + mind_nodes.append(MindMapNode( + id=node["id"], + label=node_props.get("title") or node_props.get("name", "Unknown"), + type=node["labels"][0] if node["labels"] else "Node", + size=10 + )) + + # Build links + links = [] + for rel in record["all_rels"]: + rel_props = _serialize_neo4j_types(rel["props"]) if rel["props"] else {} + links.append(MindMapLink( + source=rel["start_node"], + target=rel["end_node"], + type=rel["type"], + strength=rel_props.get("confidence", 1.0) + )) + + return MindMapResponse( + nodes=mind_nodes, + links=links, + center_node=center_node_id, + depth=depth + ) + + async def _get_entity_mention_count( + self, + entity_name: str, + entity_type: str, + user: str + ) -> int: + """ + Count how many documents mention an entity. + + Args: + entity_name: Entity name + entity_type: Entity type (Person, Technology, etc.) + user: User identifier + + Returns: + Number of documents mentioning this entity + """ + from src.core.multi_tenancy import get_neo4j_user_base_label + user_base_label = get_neo4j_user_base_label(user) + + query = f""" + MATCH (e:{user_base_label}:{entity_type} {{name: $name}}) + MATCH (d:Document)-[:MENTIONS]->(e) + RETURN count(distinct d) as mention_count + """ + + try: + results = await self.neo4j.execute_query(query, {"name": entity_name}) + if results: + return results[0]["mention_count"] + return 0 + except Exception as e: + logger.error(f"Failed to get mention count: {e}") + return 0 + + async def _entity_has_wiki_page( + self, + entity_name: str, + entity_type: str, + user: str + ) -> bool: + """ + Check if entity already has a wiki page. + + Args: + entity_name: Entity name + entity_type: Entity type + user: User identifier + + Returns: + True if page exists + """ + # Construct entity page path (within user's namespace for multi-tenancy) + from src.core.multi_tenancy import get_wikijs_namespace + user_namespace = get_wikijs_namespace(user) + entity_path = f"{user_namespace}/entities/{entity_type.lower()}/{entity_name.lower().replace(' ', '-')}" + + try: + # Search for page by path + pages = await self.wiki.list_pages(limit=1000) + # list_pages returns a list directly, not a dict + for page in pages: + if page.get("path", "") == entity_path: + return True + return False + except Exception as e: + logger.error(f"Failed to check for entity page: {e}") + return False + + def _generate_entity_stub_content( + self, + entity_name: str, + entity_type: str, + mentioning_pages: List[Dict[str, Any]], + related_entities: List[Dict[str, Any]] + ) -> str: + """ + Generate markdown content for entity stub page. + + Args: + entity_name: Entity name + entity_type: Entity type + mentioning_pages: Pages that mention this entity + related_entities: Related entities from graph + + Returns: + Markdown content string + """ + # Build mentions section with Wiki.js links (with locale prefix) + mentions_md = "\n".join([ + f"- [{page['title']}](/en/{page['path']})" + for page in mentioning_pages[:10] # Limit to first 10 + ]) + + if len(mentioning_pages) > 10: + mentions_md += f"\n\n*...and {len(mentioning_pages) - 10} more*" + + # Build related entities section with Wiki.js links + if related_entities: + related_items = [] + for e in related_entities[:10]: + name = e['name'] + entity_type = e.get('type', 'Entity') + path = e.get('path') + + # If entity has a stub page, link to it; otherwise just show name + if path: + related_items.append(f"- [{name}](/en/{path}) ({entity_type})") + else: + related_items.append(f"- {name} ({entity_type})") + related_md = "\n".join(related_items) + else: + related_md = "*No related entities found yet*" + + # Generate content + content = f"""# {entity_name} + +**Type:** {entity_type} +**Mentioned in:** {len(mentioning_pages)} page(s) + +## Overview + +*This entity has been detected in the knowledge graph. Add a description here to expand this page.* + +## Documents Mentioning This Entity + +{mentions_md} + +## Related Entities + +{related_md} + +## Graph Visualization + +To see how this entity connects to others in the knowledge graph, use the mind map endpoint: +``` +GET /graph/mindmap?center_node=[node_id]&user=[user] +``` + +--- + +🤖 **This page was auto-generated from the knowledge graph** on {datetime.now().strftime('%Y-%m-%d %H:%M UTC')}. +Feel free to expand it with more details! +""" + return content + + async def _create_entity_stub_page( + self, + entity_name: str, + entity_type: str, + user: str + ) -> Optional[int]: + """ + Create wiki stub page for an entity. + + Args: + entity_name: Entity name + entity_type: Entity type + user: User identifier + + Returns: + Page ID if created, None if failed + """ + try: + from src.core.multi_tenancy import get_neo4j_user_base_label + user_base_label = get_neo4j_user_base_label(user) + + # Get mentioning documents + mention_query = f""" + MATCH (e:{user_base_label}:{entity_type} {{name: $name}}) + MATCH (d:Document)-[:MENTIONS]->(e) + RETURN d.title as title, d.path as path, d.page_id as page_id + """ + + mention_results = await self.neo4j.execute_query( + mention_query, + {"name": entity_name} + ) + + mentioning_pages = [ + {"title": r["title"], "path": r["path"], "page_id": r["page_id"]} + for r in mention_results + ] + + # Get related entities (entities that co-occur in same documents) + # Only match entity nodes (not Document nodes) + related_query = f""" + MATCH (e1:{user_base_label}:{entity_type} {{name: $name}}) + MATCH (d:Document)-[:MENTIONS]->(e1) + MATCH (d)-[:MENTIONS]->(e2:{user_base_label}) + WHERE e2 <> e1 AND NOT (e2:Document) + RETURN DISTINCT e2.name as name, labels(e2) as labels, + properties(e2).path as path, count(d) as co_occurrence + ORDER BY co_occurrence DESC + LIMIT 10 + """ + + related_results = await self.neo4j.execute_query( + related_query, + {"name": entity_name} + ) + + related_entities = [ + { + "name": r["name"], + # Get the entity type label (not the user label) + "type": [l for l in r["labels"] if l not in [user_base_label, "Document"]][0] + if r["labels"] else "Entity", + "path": r.get("path") # Include path if it exists (for entity stub pages) + } + for r in related_results + ] + + # Generate content + content = self._generate_entity_stub_content( + entity_name=entity_name, + entity_type=entity_type, + mentioning_pages=mentioning_pages, + related_entities=related_entities + ) + + # Create page (within user's namespace for multi-tenancy) + from src.core.multi_tenancy import get_wikijs_namespace + user_namespace = get_wikijs_namespace(user) + entity_path = f"{user_namespace}/entities/{entity_type.lower()}/{entity_name.lower().replace(' ', '-')}" + + from src.models.wiki import WikiPageCreate + + page_data = WikiPageCreate( + title=entity_name, + path=entity_path, + content=content, + description=f"Auto-generated entity page for {entity_type}: {entity_name}", + tags=["entity-stub", "auto-generated", entity_type.lower()], + user=user + ) + + # Create the page (use wiki client directly) + page = await self.wiki.create_page( + path=page_data.path, + title=page_data.title, + content=page_data.content, + description=page_data.description, + tags=page_data.tags + ) + + if page: + logger.info(f"Created entity stub page for {entity_type} '{entity_name}': page {page['id']}") + return page["id"] + + return None + + except Exception as e: + logger.error(f"Failed to create entity stub page: {e}", exc_info=True) + return None + + async def generate_entity_stubs( + self, + user: str, + min_mentions: int = 5, + entity_types: Optional[List[str]] = None + ) -> Dict[str, Any]: + """ + Generate wiki stub pages for entities mentioned multiple times. + + Args: + user: User identifier + min_mentions: Minimum number of mentions required + entity_types: List of entity types to process (default: all) + + Returns: + Summary with counts of pages created + """ + if entity_types is None: + entity_types = ["Person", "Technology", "Concept", "Project"] + + from src.core.multi_tenancy import get_neo4j_user_base_label + user_base_label = get_neo4j_user_base_label(user) + pages_created = [] + pages_skipped = [] + + try: + # Query for entities with sufficient mentions + for entity_type in entity_types: + query = f""" + MATCH (e:{user_base_label}:{entity_type}) + MATCH (d:Document)-[:MENTIONS]->(e) + WITH e, count(distinct d) as mention_count + WHERE mention_count >= $min_mentions + RETURN e.name as name, mention_count + ORDER BY mention_count DESC + """ + + results = await self.neo4j.execute_query( + query, + {"min_mentions": min_mentions} + ) + + logger.info(f"Found {len(results)} {entity_type} entities with >= {min_mentions} mentions") + + for result in results: + entity_name = result["name"] + mention_count = result["mention_count"] + + # Check if page already exists + has_page = await self._entity_has_wiki_page( + entity_name=entity_name, + entity_type=entity_type, + user=user + ) + + if has_page: + logger.debug(f"Entity '{entity_name}' already has a page, skipping") + pages_skipped.append({ + "name": entity_name, + "type": entity_type, + "mentions": mention_count, + "reason": "page_exists" + }) + continue + + # Create stub page + page_id = await self._create_entity_stub_page( + entity_name=entity_name, + entity_type=entity_type, + user=user + ) + + if page_id: + pages_created.append({ + "name": entity_name, + "type": entity_type, + "mentions": mention_count, + "page_id": page_id + }) + else: + pages_skipped.append({ + "name": entity_name, + "type": entity_type, + "mentions": mention_count, + "reason": "creation_failed" + }) + + return { + "success": True, + "pages_created": len(pages_created), + "pages_skipped": len(pages_skipped), + "created": pages_created, + "skipped": pages_skipped, + "min_mentions": min_mentions, + "entity_types": entity_types + } + + except Exception as e: + logger.error(f"Failed to generate entity stubs: {e}", exc_info=True) + return { + "success": False, + "error": str(e), + "pages_created": len(pages_created), + "created": pages_created + } + + async def search_documents( + self, + query: str, + user: str, + limit: int = 10, + keywords_data: Optional[Dict[str, Any]] = None + ) -> List[Dict[str, Any]]: + """ + Search documents via entity matches in knowledge graph. + + Uses extracted keywords and synonyms from LLM query enhancement + to find entities, then returns documents mentioning those entities. + + Args: + query: Original search query + user: User identifier + limit: Maximum documents to return + keywords_data: Extracted keywords/synonyms from Phase 0 (optional) + + Returns: + List of documents with entity match counts + """ + from src.core.multi_tenancy import get_neo4j_user_base_label + + user_base_label = get_neo4j_user_base_label(user) + user_doc_label = get_neo4j_user_label(user) + + # Build search terms from keywords_data or fallback to simple extraction + if keywords_data: + # Use Phase 0 extracted keywords + keywords = keywords_data.get("core_keywords", []) + entities = keywords_data.get("entities", []) + + # Flatten synonyms dict into list + synonyms = [] + for word, syns in keywords_data.get("synonyms", {}).items(): + synonyms.extend(syns) + for word, expansions in keywords_data.get("expansions", {}).items(): + synonyms.extend(expansions) + + # Combine all search terms + all_terms = list(set(keywords + entities + synonyms)) + else: + # Fallback: Simple keyword extraction + stop_words = {"the", "a", "an", "in", "on", "at", "for", "to", "of", "with", "by"} + all_terms = [ + w.strip().lower() for w in query.split() + if w.strip().lower() not in stop_words and len(w) > 2 + ] + + if not all_terms: + logger.warning("No search terms extracted from query") + return [] + + logger.info(f"Graph search using terms: {all_terms[:10]}") # Log first 10 terms + + # Find documents via entity matches (using keywords OR synonyms) + search_query = f""" + // Find entities matching query keywords OR synonyms + MATCH (e:{user_base_label}) + WHERE NOT e:Document + AND any(term IN $terms WHERE toLower(e.name) CONTAINS toLower(term)) + + // Find documents mentioning those entities + WITH e + MATCH (d:{user_doc_label}:Document)-[:MENTIONS]->(e) + + // Aggregate results + WITH d, count(DISTINCT e) as entity_matches, collect(DISTINCT e.name)[0..5] as matched_entities + RETURN d.page_id as page_id, + d.title as title, + d.path as path, + entity_matches, + matched_entities + ORDER BY entity_matches DESC + LIMIT $limit + """ + + try: + results = await self.neo4j.execute_query( + search_query, + {"terms": all_terms, "limit": limit} + ) + logger.info(f"Graph search found {len(results)} documents") + return results + except Exception as e: + logger.error(f"Graph document search failed: {e}", exc_info=True) + return [] + + async def get_related_documents( + self, + page_id: int, + user: str, + limit: int = 5 + ) -> List[Dict[str, Any]]: + """ + Get documents related to a page via shared entities. + + Finds documents that mention the same entities as the given page. + Returns related docs with their tags (dossiers) and shared entity counts. + + Args: + page_id: Page ID to find related documents for + user: User identifier + limit: Maximum related documents to return + + Returns: + List of related documents with dossier tags + """ + from src.core.multi_tenancy import get_neo4j_user_base_label + + user_base_label = get_neo4j_user_base_label(user) + user_doc_label = get_neo4j_user_label(user) + + query = f""" + MATCH (d1:{user_doc_label}:Document {{page_id: $page_id}}) + MATCH (d1)-[:MENTIONS]->(e:{user_base_label})<-[:MENTIONS]-(d2:{user_doc_label}:Document) + WHERE d1 <> d2 AND NOT e:Document + WITH d2, d2.tags as tags, count(DISTINCT e) as shared_entities + WHERE tags IS NOT NULL AND size(tags) > 0 + RETURN d2.page_id as page_id, + d2.title as title, + d2.path as path, + tags, + shared_entities + ORDER BY shared_entities DESC + LIMIT $limit + """ + + try: + results = await self.neo4j.execute_query( + query, + {"page_id": page_id, "limit": limit} + ) + return results + except Exception as e: + logger.error(f"Failed to get related documents for page {page_id}: {e}", exc_info=True) + return [] + + async def get_all_entities(self, user: str) -> List[Dict[str, Any]]: + """ + Get all entities from the knowledge graph for a user. + + Returns entities that are not Document nodes (people, places, companies, etc.) + + Args: + user: User identifier + + Returns: + List of entities with name, type, and id + """ + from src.core.multi_tenancy import get_neo4j_user_base_label + + user_base_label = get_neo4j_user_base_label(user) + + query = f""" + MATCH (e:{user_base_label}) + WHERE NOT e:Document + RETURN e.name as name, + labels(e) as labels, + id(e) as id + ORDER BY e.name + """ + + try: + results = await self.neo4j.execute_query(query, {}) + + # Format results + entities = [] + for result in results: + # Extract entity type from labels (skip user base label) + entity_labels = result.get("labels", []) + entity_type = next( + (label for label in entity_labels if label != user_base_label), + "unknown" + ) + + entities.append({ + "name": result.get("name"), + "type": entity_type, + "id": result.get("id") + }) + + logger.info(f"Retrieved {len(entities)} entities for user {user}") + return entities + + except Exception as e: + logger.error(f"Failed to get entities for user {user}: {e}", exc_info=True) + return [] + + async def create_entity_mentions( + self, + page_id: int, + user: str, + entity_names: List[Dict[str, Any]] + ) -> int: + """ + Create MENTIONS relationships between a page and entities. + + Only creates relationships that don't already exist. + + Args: + page_id: Wiki page ID + user: User identifier + entity_names: List of entities with 'name' and 'mentions' fields + + Returns: + Number of new relationships created + """ + from src.core.multi_tenancy import get_neo4j_user_label, get_neo4j_user_base_label + + user_doc_label = get_neo4j_user_label(user) + user_base_label = get_neo4j_user_base_label(user) + + if not entity_names: + return 0 + + # Extract just the entity names + names = [e["name"] for e in entity_names] + + query = f""" + // Find the document + MATCH (d:{user_doc_label}:Document {{page_id: $page_id}}) + + // Find entities by name + MATCH (e:{user_base_label}) + WHERE e.name IN $entity_names + AND NOT e:Document + + // Create MENTIONS relationship if it doesn't exist + MERGE (d)-[r:MENTIONS]->(e) + ON CREATE SET r.created_at = datetime(), + r.mention_count = 1 + ON MATCH SET r.updated_at = datetime(), + r.mention_count = COALESCE(r.mention_count, 0) + 1 + + RETURN count(r) as relationships_created + """ + + try: + results = await self.neo4j.execute_query( + query, + {"page_id": page_id, "entity_names": names} + ) + + count = results[0]["relationships_created"] if results else 0 + logger.info(f"Created/updated {count} MENTIONS relationships for page {page_id}") + return count + + except Exception as e: + logger.error(f"Failed to create entity mentions: {e}", exc_info=True) + return 0 diff --git a/services/library-desk/src/services/ingestion_service.py b/services/library-desk/src/services/ingestion_service.py new file mode 100644 index 0000000..0b0c33e --- /dev/null +++ b/services/library-desk/src/services/ingestion_service.py @@ -0,0 +1,415 @@ +""" +Document Ingestion Service + +Orchestrates the ingestion of wiki pages into the knowledge base: +1. Fetches page content from Wiki.js +2. Generates vector embeddings (Qdrant) +3. Extracts entities and updates knowledge graph (Neo4j) + +This service is called by: +- Consolidation service (after creating/updating pages) +- Manual ingestion endpoints +- Batch ingestion jobs +""" +import logging +import asyncio +from typing import List, Optional +from datetime import datetime +import time + +from src.services.vector_service import VectorService +from src.services.graph_service import GraphService +from src.clients.wikijs_client import WikiJSClient +from src.models.ingestion import ( + IngestionRequest, + IngestionResult, + BatchIngestionRequest, + BatchIngestionResult +) + +logger = logging.getLogger(__name__) + + +class IngestionService: + """ + Service for ingesting wiki pages into the knowledge base. + """ + + def __init__( + self, + vector_service: VectorService, + graph_service: GraphService, + wiki_client: WikiJSClient + ): + self.vector = vector_service + self.graph = graph_service + self.wiki = wiki_client + + async def ingest_page( + self, + page_id: int, + user: str, + force_refresh: bool = False, + skip_vectors: bool = False, + skip_graph: bool = False, + skip_entity_linking: bool = False + ) -> IngestionResult: + """ + Ingest a single wiki page into the knowledge base. + + Args: + page_id: Wiki page ID + user: User identifier + force_refresh: Force re-ingestion even if unchanged + skip_vectors: Skip vector embedding generation + skip_graph: Skip graph entity extraction + skip_entity_linking: Skip automatic entity linking + + Returns: + IngestionResult with operation details + """ + start_time = time.time() + + logger.info(f"Starting ingestion for page {page_id} (user: {user})") + + try: + # Fetch page to get metadata + page = await self.wiki.get_page(page_id) + if not page: + return IngestionResult( + page_id=page_id, + page_title=f"Page {page_id}", + success=False, + error="Page not found in Wiki.js", + processing_time_ms=(time.time() - start_time) * 1000 + ) + + page_title = page.get("title", f"Page {page_id}") + page_path = page.get("path", "") + + # Ingest vectors and graph in parallel + tasks = [] + + if not skip_vectors: + tasks.append(self._ingest_vectors(page_id, user, force_refresh)) + else: + tasks.append(asyncio.create_task(asyncio.sleep(0))) # Dummy task + + if not skip_graph: + tasks.append(self._ingest_graph(page_id, user, force_refresh)) + else: + tasks.append(asyncio.create_task(asyncio.sleep(0))) # Dummy task + + # Execute in parallel + vector_result, graph_result = await asyncio.gather(*tasks, return_exceptions=True) + + # Handle errors + vector_chunks = 0 + graph_entities = 0 + graph_relationships = 0 + errors = [] + + if not skip_vectors: + if isinstance(vector_result, Exception): + errors.append(f"Vector ingestion failed: {str(vector_result)}") + logger.error(f"Vector ingestion failed for page {page_id}: {vector_result}") + else: + vector_chunks = vector_result.get("chunks_created", 0) + + if not skip_graph: + if isinstance(graph_result, Exception): + errors.append(f"Graph ingestion failed: {str(graph_result)}") + logger.error(f"Graph ingestion failed for page {page_id}: {graph_result}") + else: + # entities_extracted is a list, get its length + entities_list = graph_result.get("entities_extracted", []) + graph_entities = len(entities_list) if isinstance(entities_list, list) else 0 + graph_relationships = graph_result.get("relationships_created", 0) + + # Step 3: Link existing entities in the page content (after graph extraction) + entity_links_created = 0 + if not skip_entity_linking and not skip_graph and not isinstance(graph_result, Exception): + try: + entity_links_created = await self._link_existing_entities(page_id, user, page) + logger.info(f"Created {entity_links_created} entity mention links for page {page_id}") + except Exception as e: + logger.warning(f"Entity linking failed for page {page_id}: {e}") + # Don't fail the whole ingestion if entity linking fails + + processing_time_ms = (time.time() - start_time) * 1000 + + result = IngestionResult( + page_id=page_id, + page_title=page_title, + page_path=page_path, + success=len(errors) == 0, + error="; ".join(errors) if errors else None, + vector_chunks_created=vector_chunks, + graph_entities_extracted=graph_entities, + graph_relationships_created=graph_relationships, + processing_time_ms=processing_time_ms + ) + + if result.success: + logger.info( + f"Successfully ingested page {page_id}: " + f"{vector_chunks} chunks, {graph_entities} entities, " + f"{graph_relationships} relationships, {entity_links_created} entity links " + f"in {processing_time_ms:.0f}ms" + ) + else: + logger.warning(f"Partial ingestion failure for page {page_id}: {result.error}") + + return result + + except Exception as e: + logger.error(f"Ingestion failed for page {page_id}: {e}", exc_info=True) + return IngestionResult( + page_id=page_id, + page_title=f"Page {page_id}", + success=False, + error=str(e), + processing_time_ms=(time.time() - start_time) * 1000 + ) + + async def _ingest_vectors( + self, + page_id: int, + user: str, + force_refresh: bool + ) -> dict: + """ + Ingest page into vector database. + + Returns: + Dict with chunks_created count + """ + try: + summary = await self.vector.update_from_page( + page_id=page_id, + user=user, + force_refresh=force_refresh + ) + + return { + "chunks_created": summary.chunks_created, + "chunks_deleted": summary.chunks_deleted + } + + except Exception as e: + logger.error(f"Vector ingestion failed for page {page_id}: {e}") + raise + + async def _ingest_graph( + self, + page_id: int, + user: str, + force_refresh: bool + ) -> dict: + """ + Ingest page into knowledge graph. + + Returns: + Dict with entities_extracted and relationships_created counts + """ + try: + summary = await self.graph.update_from_page( + page_id=page_id, + user=user, + force_refresh=force_refresh + ) + + return { + "entities_extracted": summary.entities_extracted, + "relationships_created": summary.relationships_created + } + + except Exception as e: + logger.error(f"Graph ingestion failed for page {page_id}: {e}") + raise + + async def _link_existing_entities( + self, + page_id: int, + user: str, + page: dict + ) -> int: + """ + Find and link mentions of existing entities in the page content. + + This runs automatically after graph extraction to create MENTIONS relationships + for entities that already exist in the knowledge graph but were mentioned in + this page. + + Args: + page_id: Wiki page ID + user: User identifier + page: Page dict with content (from WikiJSClient) + + Returns: + Number of new entity mention links created + """ + import re + + try: + page_content = page.get("content", "") + if not page_content or len(page_content) < 10: + return 0 + + # Get all existing entities from the knowledge graph + entities = await self.graph.get_all_entities(user) + if not entities: + logger.debug(f"No existing entities found for user {user}, skipping entity linking") + return 0 + + # Find entity mentions in page content + found_entities = [] + content_lower = page_content.lower() + + for entity in entities: + entity_name = entity.get("name", "") + if not entity_name or len(entity_name) < 3: + continue + + # Create regex pattern for whole word matching + # This avoids matching "John" in "Johnson" + pattern = r'\b' + re.escape(entity_name.lower()) + r'\b' + + # Find all matches + matches = list(re.finditer(pattern, content_lower)) + + if matches: + found_entities.append({ + "name": entity_name, + "type": entity.get("type", "unknown"), + "mentions": len(matches), + "entity_id": entity.get("id") + }) + + if not found_entities: + logger.debug(f"No entity mentions found in page {page_id}") + return 0 + + # Create MENTIONS relationships + new_links_created = await self.graph.create_entity_mentions( + page_id=page_id, + user=user, + entity_names=found_entities + ) + + return new_links_created + + except Exception as e: + logger.error(f"Entity linking failed for page {page_id}: {e}") + raise + + async def ingest_batch( + self, + page_ids: List[int], + user: str, + force_refresh: bool = False, + skip_vectors: bool = False, + skip_graph: bool = False, + max_concurrent: int = 3 + ) -> BatchIngestionResult: + """ + Ingest multiple wiki pages concurrently. + + Args: + page_ids: List of wiki page IDs to ingest + user: User identifier + force_refresh: Force re-ingestion + skip_vectors: Skip vector embedding generation + skip_graph: Skip graph entity extraction + max_concurrent: Maximum concurrent ingestion tasks + + Returns: + BatchIngestionResult with per-page results + """ + start_time = time.time() + + logger.info(f"Starting batch ingestion of {len(page_ids)} pages (user: {user})") + + results = [] + semaphore = asyncio.Semaphore(max_concurrent) + + async def ingest_with_semaphore(page_id: int): + async with semaphore: + return await self.ingest_page( + page_id=page_id, + user=user, + force_refresh=force_refresh, + skip_vectors=skip_vectors, + skip_graph=skip_graph + ) + + # Execute all ingestions with concurrency control + tasks = [ingest_with_semaphore(page_id) for page_id in page_ids] + results = await asyncio.gather(*tasks) + + # Calculate summary + successful = sum(1 for r in results if r.success) + failed = len(results) - successful + total_processing_time_ms = (time.time() - start_time) * 1000 + + batch_result = BatchIngestionResult( + total_pages=len(page_ids), + successful=successful, + failed=failed, + results=results, + total_processing_time_ms=total_processing_time_ms + ) + + logger.info( + f"Batch ingestion complete: {successful}/{len(page_ids)} successful " + f"in {total_processing_time_ms:.0f}ms" + ) + + return batch_result + + async def ingest_all_pages( + self, + user: str, + path_prefix: Optional[str] = None, + force_refresh: bool = False, + max_concurrent: int = 3 + ) -> BatchIngestionResult: + """ + Ingest all wiki pages for a user. + + Args: + user: User identifier + path_prefix: Optional path prefix filter (e.g., "users/jpmschweitzer") + force_refresh: Force re-ingestion + max_concurrent: Maximum concurrent ingestion tasks + + Returns: + BatchIngestionResult + """ + logger.info(f"Finding all pages for user {user} (prefix: {path_prefix or 'all'})") + + # Search for all pages + pages = await self.wiki.search_pages( + query="", # Empty query returns all pages + path_prefix=path_prefix or f"users/{user}" + ) + + if not pages: + logger.warning(f"No pages found for user {user}") + return BatchIngestionResult( + total_pages=0, + successful=0, + failed=0, + results=[], + total_processing_time_ms=0 + ) + + page_ids = [p['id'] for p in pages] + logger.info(f"Found {len(page_ids)} pages to ingest") + + return await self.ingest_batch( + page_ids=page_ids, + user=user, + force_refresh=force_refresh, + max_concurrent=max_concurrent + )