#!/usr/bin/env python3 """ Knowledge Graph Validation Script Validates nodes and relationships in Neo4j database """ from neo4j import GraphDatabase import logging import os # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class KGValidator: def __init__(self, uri=None, user=None, password=None): # Đọc từ env (NEO4J_URI/USER/PASSWORD) để validate được cả AuraDB uri = uri or os.getenv("NEO4J_URI", "bolt://localhost:7687") user = user or os.getenv("NEO4J_USER", "neo4j") password = password or os.getenv("NEO4J_PASSWORD", "tourism123") logger.info(f"Connecting to Neo4j at: {uri}") self.driver = GraphDatabase.driver(uri, auth=(user, password)) def close(self): self.driver.close() def get_database_overview(self): """Get overview of database structure""" logger.info("=== DATABASE OVERVIEW ===") with self.driver.session() as session: # Count nodes by label result = session.run("CALL db.labels()") labels = [record["label"] for record in result] for label in labels: count_result = session.run(f"MATCH (n:{label}) RETURN count(n) as count") count = count_result.single()["count"] logger.info(f"{label}: {count} nodes") # Count relationships by type result = session.run("CALL db.relationshipTypes()") rel_types = [record["relationshipType"] for record in result] logger.info("\n=== RELATIONSHIPS ===") for rel_type in rel_types: count_result = session.run(f"MATCH ()-[r:{rel_type}]->() RETURN count(r) as count") count = count_result.single()["count"] logger.info(f"{rel_type}: {count} relationships") def find_orphaned_nodes(self): """Find nodes without any relationships""" logger.info("\n=== ORPHANED NODES ===") with self.driver.session() as session: # Find orphaned locations result = session.run(""" MATCH (l:Location) WHERE NOT (l)-[]-() RETURN l.id as id, l.name as name LIMIT 10 """) orphaned_locations = list(result) if orphaned_locations: logger.warning(f"Found {len(orphaned_locations)} orphaned locations:") for loc in orphaned_locations: logger.warning(f" - {loc['name']} (ID: {loc['id']})") else: logger.info("No orphaned locations found") # Find orphaned wards result = session.run(""" MATCH (w:Ward) WHERE NOT (w)-[]-() RETURN w.id as id, w.name as name LIMIT 10 """) orphaned_wards = list(result) if orphaned_wards: logger.warning(f"Found {len(orphaned_wards)} orphaned wards:") for ward in orphaned_wards: logger.warning(f" - {ward['name']} (ID: {ward['id']})") else: logger.info("No orphaned wards found") def validate_location_ward_relationships(self): """Validate that all locations are properly connected to wards""" logger.info("\n=== LOCATION-WARD RELATIONSHIPS ===") with self.driver.session() as session: # Check locations without ward relationships result = session.run(""" MATCH (l:Location) WHERE NOT (l)-[:LOCATED_IN]->(:Ward) RETURN count(l) as count """) unconnected_locations = result.single()["count"] if unconnected_locations > 0: logger.warning(f"Found {unconnected_locations} locations not connected to wards") else: logger.info("All locations are properly connected to wards") # Check wards without province relationships result = session.run(""" MATCH (w:Ward) WHERE NOT (w)-[:BELONGS_TO]->(:Province) RETURN count(w) as count """) unconnected_wards = result.single()["count"] if unconnected_wards > 0: logger.warning(f"Found {unconnected_wards} wards not connected to provinces") else: logger.info("All wards are properly connected to provinces") def analyze_proximity_relationships(self): """Analyze proximity relationships between locations""" logger.info("\n=== PROXIMITY ANALYSIS ===") with self.driver.session() as session: # Find locations with most nearby locations result = session.run(""" MATCH (l:Location)-[r:NEAR]->(nearby:Location) RETURN l.name as location, count(nearby) as nearby_count ORDER BY nearby_count DESC LIMIT 10 """) logger.info("Top 10 locations with most nearby places:") for record in result: logger.info(f" - {record['location']}: {record['nearby_count']} nearby places") # Average distance of NEAR relationships result = session.run(""" MATCH ()-[r:NEAR]->() WHERE r.distance IS NOT NULL RETURN avg(r.distance) as avg_distance, min(r.distance) as min_distance, max(r.distance) as max_distance """) stats = result.single() if stats and stats['avg_distance']: logger.info(f"Distance statistics:") logger.info(f" - Average: {stats['avg_distance']:.2f} meters") logger.info(f" - Minimum: {stats['min_distance']:.2f} meters") logger.info(f" - Maximum: {stats['max_distance']:.2f} meters") def sample_data_quality(self): """Sample some data to check quality""" logger.info("\n=== DATA QUALITY SAMPLES ===") with self.driver.session() as session: # Sample locations with their ward and province info result = session.run(""" MATCH (l:Location)-[:LOCATED_IN]->(w:Ward)-[:BELONGS_TO]->(p:Province) RETURN l.name as location, w.name as ward, p.name as province LIMIT 5 """) logger.info("Sample location hierarchy:") for record in result: logger.info(f" - {record['location']} → {record['ward']} → {record['province']}") # Check for locations with coordinates result = session.run(""" MATCH (l:Location) WITH count(l) as total MATCH (l2:Location) WHERE l2.lat IS NOT NULL AND l2.lng IS NOT NULL RETURN count(l2) as with_coords, total """) coords_info = result.single() logger.info(f"Locations with coordinates: {coords_info['with_coords']}/{coords_info['total']}") def run_full_validation(self): """Run complete validation""" logger.info("Starting Knowledge Graph validation...") self.get_database_overview() self.find_orphaned_nodes() self.validate_location_ward_relationships() self.analyze_proximity_relationships() self.sample_data_quality() logger.info("\nValidation completed!") def main(): """Main execution function""" validator = KGValidator() try: # Test connection with validator.driver.session() as session: result = session.run("RETURN 'Connection successful' as message") logger.info(result.single()['message']) # Run validation validator.run_full_validation() except Exception as e: logger.error(f"Validation failed: {e}") raise finally: validator.close() if __name__ == "__main__": main()