apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-07-24 20:51:02 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Auto mode

Browse source
This commit is contained in:
Cyber MacGeddon 2025-09-06 12:19:48 +01:00
parent 91b512416f
commit c67ffef799
1 changed files with 59 additions and 51 deletions
Download patch file Download diff file Expand all files Collapse all files

110
trustgraph-cli/trustgraph/cli/load_structured_data.py
View file

@ -97,64 +97,72 @@ def load_structured_data(
logger.info("✅ Generated descriptor configuration") logger.info("✅ Generated descriptor configuration")
print("📝 Generated descriptor configuration") print("📝 Generated descriptor configuration")
# Step 3: Parse and preview data # Step 3: Parse and preview data using shared pipeline
logger.info("Step 3: Parsing and validating data...") logger.info("Step 3: Parsing and validating data...")
preview_records = _auto_parse_preview(input_file, auto_descriptor, min(sample_size, 5), logger)
if preview_records is None:
logger.error("Failed to parse data with generated descriptor")
print("❌ Could not parse data with generated descriptor.")
return None
# Show preview
print("📊 Data Preview (first few records):")
print("=" * 50)
for i, record in enumerate(preview_records[:3], 1):
print(f"Record {i}: {record}")
print("=" * 50)
# Step 4: Import (unless dry_run) # Create temporary descriptor file for validation
if dry_run: import tempfile
logger.info("✅ Dry run complete - data is ready for import") temp_descriptor = tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False)
print("✅ Dry run successful! Data is ready for import.") json.dump(auto_descriptor, temp_descriptor, indent=2)
print(f"💡 Run without --dry-run to import {len(preview_records)} records to TrustGraph.") temp_descriptor.close()
return None
else: try:
logger.info("Step 4: Importing data to TrustGraph...") # Use shared pipeline for preview (small sample)
print("🚀 Importing data to TrustGraph...") preview_objects, _ = _process_data_pipeline(input_file, temp_descriptor.name, user, collection, sample_size=5)
# Recursively call ourselves with the auto-generated descriptor # Show preview
# This reuses all the existing import logic print("📊 Data Preview (first few records):")
import tempfile print("=" * 50)
import os for i, obj in enumerate(preview_objects[:3], 1):
values = obj.get('values', {})
print(f"Record {i}: {values}")
print("=" * 50)
# Save auto-generated descriptor to temp file # Step 4: Import (unless dry_run)
temp_descriptor = tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) if dry_run:
json.dump(auto_descriptor, temp_descriptor, indent=2) logger.info("✅ Dry run complete - data is ready for import")
temp_descriptor.close() print("✅ Dry run successful! Data is ready for import.")
print(f"💡 Run without --dry-run to import data to TrustGraph.")
try: return None
# Call the full pipeline mode with our auto-generated descriptor else:
result = load_structured_data( logger.info("Step 4: Importing data to TrustGraph...")
api_url=api_url, print("🚀 Importing data to TrustGraph...")
input_file=input_file,
descriptor_file=temp_descriptor.name, # Use shared pipeline for full processing (no sample limit)
flow=flow, output_objects, descriptor = _process_data_pipeline(input_file, temp_descriptor.name, user, collection)
user=user,
collection=collection, # Get batch size from descriptor
dry_run=False, # We already handled dry_run above batch_size = descriptor.get('output', {}).get('options', {}).get('batch_size', 1000)
verbose=verbose
) # Send to TrustGraph using shared function
imported_count = _send_to_trustgraph(output_objects, api_url, flow, batch_size)
# Summary
format_info = descriptor.get('format', {})
format_type = format_info.get('type', 'csv').lower()
schema_name = descriptor.get('output', {}).get('schema_name', 'default')
print(f"\n🎉 Auto-Import Complete!")
print(f"- Input format: {format_type}")
print(f"- Target schema: {schema_name}")
print(f"- Records imported: {imported_count}")
print(f"- Flow used: {flow}")
print("✅ Auto-import completed successfully!")
logger.info("Auto-import pipeline completed successfully") logger.info("Auto-import pipeline completed successfully")
return result return imported_count
finally: except Exception as e:
# Clean up temp descriptor file logger.error(f"Auto-import failed: {e}")
try: print(f"❌ Auto-import failed: {e}")
os.unlink(temp_descriptor.name) return None
except:
pass finally:
# Clean up temp descriptor file
try:
import os
os.unlink(temp_descriptor.name)
except:
pass
elif discover_schema: elif discover_schema:
logger.info(f"Analyzing {input_file} to discover schemas...") logger.info(f"Analyzing {input_file} to discover schemas...")