diff --git a/trustgraph-base/trustgraph/api/flow.py b/trustgraph-base/trustgraph/api/flow.py index 74b7a117..58cf5c80 100644 --- a/trustgraph-base/trustgraph/api/flow.py +++ b/trustgraph-base/trustgraph/api/flow.py @@ -492,12 +492,148 @@ class FlowInstance: "service/structured-query", input ) - + # Check for system-level error if "error" in response and response["error"]: error_type = response["error"].get("type", "unknown") error_message = response["error"].get("message", "Unknown error") raise ProtocolException(f"{error_type}: {error_message}") - + + return response + + def detect_type(self, sample): + """ + Detect the data type of a structured data sample. + + Args: + sample: Data sample to analyze (string content) + + Returns: + dict with detected_type, confidence, and optional metadata + """ + + input = { + "operation": "detect-type", + "sample": sample + } + + response = self.request( + "service/structured-diag", + input + ) + + # Check for system-level error + if "error" in response and response["error"]: + error_type = response["error"].get("type", "unknown") + error_message = response["error"].get("message", "Unknown error") + raise ProtocolException(f"{error_type}: {error_message}") + + return response + + def generate_descriptor(self, sample, data_type, schema_name, options=None): + """ + Generate a descriptor for structured data mapping to a specific schema. + + Args: + sample: Data sample to analyze (string content) + data_type: Data type (csv, json, xml) + schema_name: Target schema name for descriptor generation + options: Optional parameters (e.g., delimiter for CSV) + + Returns: + dict with descriptor and metadata + """ + + input = { + "operation": "generate-descriptor", + "sample": sample, + "type": data_type, + "schema-name": schema_name + } + + if options: + input["options"] = options + + response = self.request( + "service/structured-diag", + input + ) + + # Check for system-level error + if "error" in response and response["error"]: + error_type = response["error"].get("type", "unknown") + error_message = response["error"].get("message", "Unknown error") + raise ProtocolException(f"{error_type}: {error_message}") + + return response + + def diagnose_data(self, sample, schema_name=None, options=None): + """ + Perform combined data diagnosis: detect type and generate descriptor. + + Args: + sample: Data sample to analyze (string content) + schema_name: Optional target schema name for descriptor generation + options: Optional parameters (e.g., delimiter for CSV) + + Returns: + dict with detected_type, confidence, descriptor, and metadata + """ + + input = { + "operation": "diagnose", + "sample": sample + } + + if schema_name: + input["schema-name"] = schema_name + + if options: + input["options"] = options + + response = self.request( + "service/structured-diag", + input + ) + + # Check for system-level error + if "error" in response and response["error"]: + error_type = response["error"].get("type", "unknown") + error_message = response["error"].get("message", "Unknown error") + raise ProtocolException(f"{error_type}: {error_message}") + + return response + + def schema_selection(self, sample, options=None): + """ + Select matching schemas for a data sample using prompt analysis. + + Args: + sample: Data sample to analyze (string content) + options: Optional parameters + + Returns: + dict with schema_matches array and metadata + """ + + input = { + "operation": "schema-selection", + "sample": sample + } + + if options: + input["options"] = options + + response = self.request( + "service/structured-diag", + input + ) + + # Check for system-level error + if "error" in response and response["error"]: + error_type = response["error"].get("type", "unknown") + error_message = response["error"].get("message", "Unknown error") + raise ProtocolException(f"{error_type}: {error_message}") + return response diff --git a/trustgraph-base/trustgraph/schema/services/diagnosis.py b/trustgraph-base/trustgraph/schema/services/diagnosis.py index 2bd6caf0..1bd6d3ed 100644 --- a/trustgraph-base/trustgraph/schema/services/diagnosis.py +++ b/trustgraph-base/trustgraph/schema/services/diagnosis.py @@ -1,4 +1,4 @@ -from pulsar.schema import Record, String, Map, Double +from pulsar.schema import Record, String, Map, Double, Array from ..core.primitives import Error ############################################################################ @@ -6,7 +6,7 @@ from ..core.primitives import Error # Structured data diagnosis services class StructuredDataDiagnosisRequest(Record): - operation = String() # "detect-type", "generate-descriptor", or "diagnose" + operation = String() # "detect-type", "generate-descriptor", "diagnose", or "schema-selection" sample = String() # Data sample to analyze (text content) type = String() # Data type (csv, json, xml) - optional, required for generate-descriptor schema_name = String() # Target schema name for descriptor generation - optional @@ -27,4 +27,7 @@ class StructuredDataDiagnosisResponse(Record): # JSON encoded additional metadata (e.g., field count, sample records) metadata = Map(String()) + # Array of matching schema IDs (for schema-selection operation) - optional + schema_matches = Array(String()) + ############################################################################ \ No newline at end of file diff --git a/trustgraph-flow/trustgraph/retrieval/structured_diag/service.py b/trustgraph-flow/trustgraph/retrieval/structured_diag/service.py index 75af6dc3..8d0bd647 100644 --- a/trustgraph-flow/trustgraph/retrieval/structured_diag/service.py +++ b/trustgraph-flow/trustgraph/retrieval/structured_diag/service.py @@ -143,10 +143,12 @@ class Processor(FlowProcessor): response = await self.generate_descriptor_operation(request, flow) elif request.operation == "diagnose": response = await self.diagnose_operation(request, flow) + elif request.operation == "schema-selection": + response = await self.schema_selection_operation(request, flow) else: error = Error( type="InvalidOperation", - message=f"Unknown operation: {request.operation}. Supported: detect-type, generate-descriptor, diagnose" + message=f"Unknown operation: {request.operation}. Supported: detect-type, generate-descriptor, diagnose, schema-selection" ) response = StructuredDataDiagnosisResponse( error=error, @@ -155,7 +157,7 @@ class Processor(FlowProcessor): # Send response await flow("response").send( - id, response, properties={"id": id} + response, properties={"id": id} ) except Exception as e: @@ -172,7 +174,7 @@ class Processor(FlowProcessor): ) await flow("response").send( - id, response, properties={"id": id} + response, properties={"id": id} ) async def detect_type_operation(self, request: StructuredDataDiagnosisRequest, flow) -> StructuredDataDiagnosisResponse: @@ -307,6 +309,92 @@ class Processor(FlowProcessor): metadata=metadata ) + async def schema_selection_operation(self, request: StructuredDataDiagnosisRequest, flow) -> StructuredDataDiagnosisResponse: + """Handle schema-selection operation""" + logger.info("Processing schema-selection operation") + + # Prepare all schemas for the prompt + all_schemas = [] + for schema_name, row_schema in self.schemas.items(): + schema_info = { + "name": row_schema.name, + "description": row_schema.description, + "fields": [ + { + "name": f.name, + "type": f.type, + "description": f.description, + "required": f.required, + "primary_key": f.primary, + "indexed": f.indexed, + "enum_values": f.enum_values if f.enum_values else [] + } + for f in row_schema.fields + ] + } + all_schemas.append(schema_info) + + # Create prompt variables - schemas array contains ALL schemas + variables = { + "sample": request.sample, + "schemas": all_schemas, + "options": request.options or {} + } + + # Call prompt service (assuming there's a schema-selection prompt template) + terms = {k: json.dumps(v) for k, v in variables.items()} + prompt_request = PromptRequest( + id="schema-selection", # This prompt template needs to exist + terms=terms + ) + + try: + logger.info("Calling prompt service for schema selection") + response = await flow("prompt-request").request(prompt_request) + + if response.error: + logger.error(f"Prompt service error: {response.error.message}") + error = Error( + type="PromptServiceError", + message="Failed to select schemas using prompt service" + ) + return StructuredDataDiagnosisResponse(error=error, operation=request.operation) + + if not response.text or not response.text.strip(): + logger.error("Empty response from prompt service") + error = Error( + type="PromptServiceError", + message="Empty response from prompt service" + ) + return StructuredDataDiagnosisResponse(error=error, operation=request.operation) + + # Parse the response as JSON array of schema IDs + try: + schema_matches = json.loads(response.text.strip()) + if not isinstance(schema_matches, list): + raise ValueError("Response must be an array") + except (json.JSONDecodeError, ValueError) as e: + logger.error(f"Failed to parse schema matches response: {e}") + error = Error( + type="ParseError", + message="Failed to parse schema selection response as JSON array" + ) + return StructuredDataDiagnosisResponse(error=error, operation=request.operation) + + return StructuredDataDiagnosisResponse( + error=None, + operation=request.operation, + schema_matches=schema_matches + ) + + except Exception as e: + logger.error(f"Error calling prompt service: {e}", exc_info=True) + error = Error( + type="PromptServiceError", + message="Failed to select schemas using prompt service" + ) + return StructuredDataDiagnosisResponse(error=error, operation=request.operation) + async def generate_descriptor_with_prompt( self, sample: str, data_type: str, target_schema: RowSchema, options: Dict[str, str], flow diff --git a/trustgraph-flow/trustgraph/retrieval/structured_diag/type_detector.py b/trustgraph-flow/trustgraph/retrieval/structured_diag/type_detector.py index ccd6bf8b..a291d5cc 100644 --- a/trustgraph-flow/trustgraph/retrieval/structured_diag/type_detector.py +++ b/trustgraph-flow/trustgraph/retrieval/structured_diag/type_detector.py @@ -31,28 +31,13 @@ def detect_data_type(sample: str) -> Tuple[Optional[str], float]: sample = sample.strip() - # Try each format and calculate confidence scores - json_confidence = _check_json_format(sample) - xml_confidence = _check_xml_format(sample) - csv_confidence = _check_csv_format(sample) - - logger.debug(f"Format confidence scores - JSON: {json_confidence}, XML: {xml_confidence}, CSV: {csv_confidence}") - - # Find the format with highest confidence - scores = { - "json": json_confidence, - "xml": xml_confidence, - "csv": csv_confidence - } - - best_format = max(scores, key=scores.get) - best_confidence = scores[best_format] - - # Only return a result if confidence is above threshold - if best_confidence < 0.3: - return None, best_confidence - - return best_format, best_confidence + # Simple pattern matching + if sample.startswith(' float: @@ -83,33 +68,20 @@ def _check_json_format(sample: str) -> float: def _check_xml_format(sample: str) -> float: """Check if sample is valid XML format""" - try: - # Quick heuristic checks first - if not sample.startswith('<'): - return 0.0 - - if not ('>' in sample and ' 10: - return 0.95 - elif child_count > 5: - return 0.9 - elif child_count > 0: - return 0.8 + # XML declaration or starts with tag + if sample.startswith('' in sample: + try: + # Quick parse test + ET.fromstring(sample) + return 0.9 # Valid XML + except ET.ParseError: + return 0.3 # Looks like XML but malformed else: - return 0.6 + return 0.1 # Incomplete XML - except ET.ParseError: - # Check for common XML characteristics even if not well-formed - xml_indicators = [' float: