Haicheng Qu

from abc import ABC, abstractmethod
from datetime import datetime, timezone
from dataclasses import asdict
import hashlib
import json
import inspect
from typing import Any, Dict, List, Union
from redb import settings
import traceback
from contextlib import contextmanager

class DatabaseExporter(ABC):
    @abstractmethod
    def export(self, data: Any, **kwargs) -> bool:
        """Export data to the database"""
        pass

class ElasticsearchExporter(DatabaseExporter):
    def __init__(self, logger: Any, index_prefix: str):
        self.client = settings.get_elasticsearch_client()
        self.index_prefix = index_prefix
        self.log = logger

    def export(self, data: Any, **kwargs) -> bool:
        self.log.debug(inspect.currentframe().f_code.co_name)
        if not isinstance(data, list):
            data = [data]
            
        now_t = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
        index = kwargs.get('index')
        tag = kwargs.get('tag')
        hashes = kwargs.get('hashes')
        
        for dataclass_ in data:
            try:
                # Convert dataclass to dict and filter out None values
                document = {k: v for k, v in asdict(dataclass_).items() if v is not None}
                
                if tag:
                    document["tag"] = tag
                document |= hashes
                document["timestamp_utc"] = now_t
                document["known_benign"] = kwargs.get('known_benign', False)
                document["known_malicious"] = kwargs.get('known_malicious', False)

                if "_id" in document:
                    tmp_id = document.pop("_id") + document["sha256"]
                    _id = hashlib.sha256(tmp_id.encode()).hexdigest()
                else:
                    _id = document["sha256"]

                doc_dump = json.dumps(document)
                self.client.index(index=index, id=_id, document=doc_dump)
                
            except Exception as e:
                self.log.error(f"Failed to export to Elasticsearch: {e}")
                return False
                
        return True

class PrintExporter(DatabaseExporter):
    """Exporter that prints results instead of uploading to database (for dry-run mode)"""
    def __init__(self, logger: Any, index_prefix: str):
        self.index_prefix = index_prefix
        self.log = logger
        self.log.debug(f"PrintExporter initialized with index_prefix: {index_prefix}")

    def export(self, data: Any, **kwargs) -> bool:
        try:
            import json
            from dataclasses import asdict, is_dataclass

            self.log.debug(f"PrintExporter.export called with data type: {type(data)}")

            if data is None:
                self.log.debug("PrintExporter.export: data is None, returning True")
                return True

            if not isinstance(data, list):
                data = [data]

            for i, item in enumerate(data):
                if is_dataclass(item) and not isinstance(item, type):
                    # Convert dataclass instance to dict
                    try:
                        item_dict = asdict(item)
                        print(json.dumps(item_dict, indent=2, default=str))
                    except Exception as e:
                        self.log.error(f"PrintExporter.export: Error converting dataclass: {e}")
                        print(str(item))
                elif isinstance(item, dict):
                    # Already a dict, print as JSON
                    print(json.dumps(item, indent=2, default=str))
                else:
                    # Fallback for other types
                    print(json.dumps(item, indent=2, default=str) if not isinstance(item, str) else item)

            self.log.debug("PrintExporter.export: completed successfully")
            return True
        except Exception as e:
            self.log.error(f"PrintExporter.export: Failed to print data: {e}")
            return False


class ClickHouseExporter(DatabaseExporter):
    def __init__(self, logger: Any, index_prefix: str, client=None):
        self.index_prefix = index_prefix
        self.log = logger
        self.client = client

    @contextmanager
    def clickhouse_connection(self):
        client = settings.create_clickhouse_client()
        try:
            yield client
        finally:
            try:
                client.close()
            except:
                pass
    # def export(self, data: Any, **kwargs) -> bool:
    #     try:
    #         # Handle multi-table export case
    #         # if isinstance(data, dict) and 'multi_table' in data:
    #         #     for table_name, table_data in data.items():
    #         #         if table_name != 'multi_table' and isinstance(table_data, dict) and 'table' in table_data:
    #         #             self.log.debug(f"Exporting to table: {table_data['table']}")
    #         #             self.log.debug(f"Column names: {table_data['column_names']}")
    #         #             self.log.debug(f"Column types: {table_data['column_type_names']}")
    #         #             self.log.debug(f"Data sample: {table_data['data'][0] if table_data['data'] else 'No data'}")
                        
    #         #             try:
    #         #                 result = self.client.insert(
    #         #                     table_data['table'],
    #         #                     table_data['data'],
    #         #                     column_names=table_data['column_names'],
    #         #                     column_type_names=table_data['column_type_names']
    #         #                 )
    #         #                 if not result:
    #         #                     self.log.error(f"Failed to insert into table {table_data['table']}")
    #         #                     return False
    #         #             except Exception as e:
    #         #                 self.log.error(f"Error inserting into table {table_data['table']}: {e}")
    #         #                 self.log.error(f"Data that caused error: {table_data['data']}")
    #         #                 return False
    #         #     return True

    #         if isinstance(data, dict) and 'multi_table' in data:
    #             for table_name, table_data in data.items():
    #                 if table_name != 'multi_table' and isinstance(table_data, dict) and 'table' in table_data:
    #                     self.log.debug(f"Exporting to table: {table_data['table']}")
                        
    #                     # Validate each row before insert
    #                     for row in table_data['data']:
    #                         for val, col_name, col_type in zip(row, table_data['column_names'], 
    #                                                         table_data['column_type_names']):
    #                             # Check for empty strings
    #                             if isinstance(val, str) and not val:
    #                                 self.log.error(f"Empty string found for column {col_name}")
    #                                 raise ValueError(f"Empty string not allowed for column {col_name}")
                                
    #                             # Log all string values for certificate_serial_number
    #                             if col_name == 'certificate_serial_number':
    #                                 self.log.debug(f"Serial number value: '{val}', type: {type(val)}, "
    #                                             f"repr: {repr(val)}")
                        
    #                     try:
    #                         result = self.client.insert(
    #                             table_data['table'],
    #                             table_data['data'],
    #                             column_names=table_data['column_names'],
    #                             column_type_names=table_data['column_type_names']
    #                         )
    #                         if not result:
    #                             self.log.error(f"Failed to insert into table {table_data['table']}")
    #                             return False
    #                     except Exception as e:
    #                         self.log.error(f"Error inserting into table {table_data['table']}: {e}")
    #                         self.log.error(f"Data that caused error: {table_data['data']}")
    #                         return False
    #             return True

    #         # Handle single table case
    #         table = kwargs.get('table')
    #         if not table:
    #             raise ValueError("Table name must be provided for ClickHouse export")
            
    #         # Unpack the tuple returned by prepare_export_data
    #         if isinstance(data, tuple):
    #             if len(data) != 3:
    #                 raise ValueError("Data tuple must contain exactly 3 elements: (data, column_names, column_type_names)")
    #             insert_data, column_names, column_type_names = data
    #         else:
    #             # Use provided column names and types from kwargs
    #             insert_data = data
    #             column_names = kwargs.get('column_names')
    #             column_type_names = kwargs.get('column_type_names')
                
    #         if not (column_names and column_type_names):
    #             raise ValueError("column_names and column_type_names must be provided")

    #         # Ensure data is in list format for batch insert
    #         if not isinstance(insert_data, list):
    #             insert_data = [insert_data]
                
    #         self.client.insert(
    #             table,
    #             insert_data,
    #             column_names=column_names,
    #             column_type_names=column_type_names
    #         )
    #         return True
            
    #     except Exception as e:
    #         self.log.error(f"Failed to export to ClickHouse: {e}")
    #         return False

    def get_client(self):
        """Get existing client or create new one if needed"""
        if self.client:
            return self.client
        try:
            return settings.create_clickhouse_client() 
        except Exception as e:
            self.log.error(f"Failed to create ClickHouse client: {e}")
            return None
            
    # WORKING SIMPLE VERSION
    def export(self, data: Any, **kwargs) -> bool:
        try:
            # # Handle multi-table export case
            # # if isinstance(data, dict) and 'multi_table' in data:
            # #     for table_data in data.values():
            # #         if isinstance(table_data, dict) and 'table' in table_data:
            # #             result = self.client.insert(
            # #                 table_data['table'],
            # #                 table_data['data'],
            # #                 column_names=table_data['column_names'],
            # #                 column_type_names=table_data['column_type_names']
            # #             )
            # #             if not result:  # If any insert fails, return False
            # #                 return False
            # #     return True
            # if isinstance(data, dict) and 'multi_table' in data:
            #     for key, table_data in data.items():
            #         if key != 'multi_table' and isinstance(table_data, dict) and 'table' in table_data:
            #             # Add type hints
            #             typed_data = []
            #             for row in table_data['data']:
            #                 typed_row = []
            #                 for val, type_name in zip(row, table_data['column_type_names']):
            #                     if isinstance(val, str) and not val:
            #                         val = 'UNKNOWN'
            #                     if type_name == 'String' or 'LowCardinality(String)' in type_name:
            #                         val = str(val) if val is not None else 'UNKNOWN'
            #                     typed_row.append(val)
            #                 typed_data.append(typed_row)
                        
            #             table_data['data'] = typed_data
                        
            #             result = self.client.insert(
            #                 table_data['table'],
            #                 table_data['data'],
            #                 column_names=table_data['column_names'],
            #                 column_type_names=table_data['column_type_names'],
            #                 settings={'input_format_values_interpret_expressions': 0}  # Disable type inference
            #             )
            #             if not result:
            #                 return False
            #         return True

            # Get a new client connection for this export operation
            with self.clickhouse_connection() as client:
                if not client:
                    return False

                # LAST WORKING VERSION
                if isinstance(data, dict) and 'multi_table' in data:
                    for key, table_data in data.items():
                        if key != 'multi_table' and isinstance(table_data, dict) and 'table' in table_data:
                            # Debug logging
                            self.log.debug(f"Processing table: {table_data['table']}")
                            self.log.debug(f"Column names: {table_data['column_names']}")
                            self.log.debug(f"Column types: {table_data['column_type_names']}")
                            
                            typed_data = []
                            for row_idx, row in enumerate(table_data['data']):
                                typed_row = []
                                for val_idx, (val, type_name, col_name) in enumerate(
                                    zip(row, table_data['column_type_names'], table_data['column_names'])
                                ):
                                    try:
                                        # Handle different types
                                        if val is None:
                                            if 'Nullable' in type_name:
                                                val = None  # Keep None for Nullable columns
                                            elif 'Array' in type_name:
                                                val = []
                                            elif 'Int' in type_name or 'UInt' in type_name:
                                                val = 0
                                            elif 'FixedString(16)' in type_name:
                                                val = b'\x00' * 16  # Binary FixedString
                                            elif 'String' in type_name or 'LowCardinality' in type_name:
                                                val = 'UNKNOWN'
                                            elif 'DateTime' in type_name:
                                                val = datetime.now(timezone.utc)
                                            elif 'Boolean' in type_name:
                                                val = False
                                            else:
                                                self.log.error(f"Unhandled type {type_name} for null value")
                                                val = 'UNKNOWN'
                                        
                                        # Log problematic values
                                        if isinstance(val, str) and not val.strip():
                                            self.log.warning(f"Empty string found in table {table_data['table']}, "
                                                        f"column {col_name}, row {row_idx}")
                                        
                                        # Ensure array types are always lists
                                        if 'Array' in type_name and not isinstance(val, list):
                                            val = [val]
                                        
                                        typed_row.append(val)
                                    except Exception as e:
                                        self.log.error(f"Error processing value in table {table_data['table']}, "
                                                    f"column {col_name} ({type_name}), "
                                                    f"row {row_idx}, value: {repr(val)}")
                                        raise
                                        
                                typed_data.append(typed_row)
                            
                            table_data['data'] = typed_data

                            # Skip insert if there is nothing to insert
                            if not typed_data:
                                self.log.debug(f"No data to insert for table {table_data['table']}, skipping.")
                                continue

                            # Debug log the first row of data
                            self.log.debug(f"Sample row for {table_data['table']}: {typed_data[0]}")

                            result = client.insert(
                                table_data['table'],
                                table_data['data'],
                                column_names=table_data['column_names'],
                                column_type_names=table_data['column_type_names'],
                                settings={'input_format_values_interpret_expressions': 0}
                            )
                            if not result:
                                self.log.error(f"Insert failed for table {table_data['table']}")
                                return False
                    return True

                # Handle single table case
                table = kwargs.get('table')
                if not table:
                    raise ValueError("Table name must be provided for ClickHouse export")
                
                # Unpack the tuple returned by prepare_export_data
                if isinstance(data, tuple):
                    if len(data) != 3:
                        raise ValueError("Data tuple must contain exactly 3 elements: (data, column_names, column_type_names)")
                    insert_data, column_names, column_type_names = data
                else:
                    # Use provided column names and types from kwargs
                    insert_data = data
                    column_names = kwargs.get('column_names')
                    column_type_names = kwargs.get('column_type_names')
                    
                if not (column_names and column_type_names):
                    raise ValueError("column_names and column_type_names must be provided")

                # Ensure data is in list format for batch insert
                if not isinstance(insert_data, list):
                    insert_data = [insert_data]

                # Skip insert if there is nothing to insert
                if not insert_data:
                    self.log.debug(f"No data to insert for table {table}, skipping.")
                    return True

                client.insert(
                    table,
                    insert_data,
                    column_names=column_names,
                    column_type_names=column_type_names,
                    settings={'input_format_values_interpret_expressions': 0}  # Disable type inference
                )
                return True

        except TypeError as e:
            # Get the actual value causing the error from the exception traceback
            import sys
            exc_type, exc_value, exc_traceback = sys.exc_info()
            
            # Walk through the traceback to find the frame with the problematic value
            current = exc_traceback
            while current:
                if 'string.py' in current.tb_frame.f_code.co_filename and '_data_size' in current.tb_frame.f_code.co_name:
                    locals_dict = current.tb_frame.f_locals
                    problematic_value = locals_dict.get('x')
                    self.log.error(f"TypeError with value: {problematic_value} (type: {type(problematic_value)})")
                    self.log.error(f"Full locals at error: {locals_dict}")
                    break
                current = current.tb_next
            
            self.log.error(f"Failed to export to ClickHouse: {e}")
            self.log.error("Full traceback:")
            self.log.error(traceback.format_exc())
            return False
        except Exception as e:
            self.log.error(f"Failed to export to ClickHouse: {e}")
            self.log.error(traceback.format_exc())
            return False
        finally:
            # Only close if we created a new client
            if client and client != self.client:
                # Close the client connection
                try:
                    client.close()
                except:
                    pass