| """ |
| File I/O utilities for QAFD-RAG. |
| |
| Provides JSON and XML file operations. |
| """ |
|
|
| import json |
| import os |
| import xml.etree.ElementTree as ET |
|
|
|
|
| def load_json(file_name: str): |
| """ |
| Load JSON data from a file. |
| |
| Parameters: |
| ----------- |
| file_name : str |
| Path to the JSON file |
| |
| Returns: |
| -------- |
| dict or None |
| Parsed JSON data, or None if file doesn't exist |
| """ |
| if not os.path.exists(file_name): |
| return None |
| with open(file_name, encoding="utf-8") as f: |
| return json.load(f) |
|
|
|
|
| def write_json(json_obj, file_name: str): |
| """ |
| Write JSON data to a file. |
| |
| Parameters: |
| ----------- |
| json_obj : Any |
| Data to write as JSON |
| file_name : str |
| Path to the output file |
| """ |
| with open(file_name, "w", encoding="utf-8") as f: |
| json.dump(json_obj, f, indent=2, ensure_ascii=False) |
|
|
|
|
| def save_data_to_file(data, file_name: str): |
| """ |
| Save data to a JSON file with pretty formatting. |
| |
| Parameters: |
| ----------- |
| data : Any |
| Data to save |
| file_name : str |
| Path to the output file |
| """ |
| with open(file_name, "w", encoding="utf-8") as f: |
| json.dump(data, f, ensure_ascii=False, indent=4) |
|
|
|
|
| def xml_to_json(xml_file: str): |
| """ |
| Convert a GraphML XML file to JSON format. |
| |
| Parameters: |
| ----------- |
| xml_file : str |
| Path to the GraphML file |
| |
| Returns: |
| -------- |
| dict or None |
| Dictionary with 'nodes' and 'edges' lists, or None on error |
| """ |
| try: |
| tree = ET.parse(xml_file) |
| root = tree.getroot() |
|
|
| print(f"Root element: {root.tag}") |
| print(f"Root attributes: {root.attrib}") |
|
|
| data = {"nodes": [], "edges": []} |
| namespace = {"": "http://graphml.graphdrawing.org/xmlns"} |
|
|
| for node in root.findall(".//node", namespace): |
| node_data = { |
| "id": node.get("id").strip('"'), |
| "entity_type": node.find("./data[@key='d0']", namespace).text.strip('"') |
| if node.find("./data[@key='d0']", namespace) is not None |
| else "", |
| "description": node.find("./data[@key='d1']", namespace).text |
| if node.find("./data[@key='d1']", namespace) is not None |
| else "", |
| "source_id": node.find("./data[@key='d2']", namespace).text |
| if node.find("./data[@key='d2']", namespace) is not None |
| else "", |
| } |
| data["nodes"].append(node_data) |
|
|
| for edge in root.findall(".//edge", namespace): |
| edge_data = { |
| "source": edge.get("source").strip('"'), |
| "target": edge.get("target").strip('"'), |
| "weight": float(edge.find("./data[@key='d3']", namespace).text) |
| if edge.find("./data[@key='d3']", namespace) is not None |
| else 0.0, |
| "description": edge.find("./data[@key='d4']", namespace).text |
| if edge.find("./data[@key='d4']", namespace) is not None |
| else "", |
| "keywords": edge.find("./data[@key='d5']", namespace).text |
| if edge.find("./data[@key='d5']", namespace) is not None |
| else "", |
| "source_id": edge.find("./data[@key='d6']", namespace).text |
| if edge.find("./data[@key='d6']", namespace) is not None |
| else "", |
| } |
| data["edges"].append(edge_data) |
|
|
| print(f"Found {len(data['nodes'])} nodes and {len(data['edges'])} edges") |
|
|
| return data |
| except ET.ParseError as e: |
| print(f"Error parsing XML file: {e}") |
| return None |
| except Exception as e: |
| print(f"An error occurred: {e}") |
| return None |
|
|
|
|
| __all__ = [ |
| "load_json", |
| "write_json", |
| "save_data_to_file", |
| "xml_to_json", |
| ] |
|
|
