modules/preprocess/preprocess.py

#!/usr/bin/env python3
import json
import logging
from typing import Dict, Any
from pathlib import Path

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("DataFormatUpdater")

# Standard attributes to add to all files
STANDARD_ADDITIONS = {
    "secondary_data": [],
    "variables": {
      "width": 600,
      "height": 600,
      "has_rounded_corners": False,
      "has_shadow": False,
      "has_spacing": False,
      "has_gradient": False,
      "has_stroke": False
    },
    "typography": {
      "title": {
        "font_family": "Arial",
        "font_size": "28px",
        "font_weight": 700
      },
      "description": {
        "font_family": "Arial",
        "font_size": "16px",
        "font_weight": 500
      },
      "label": {
        "font_family": "Arial",
        "font_size": "16px",
        "font_weight": 500
      },
      "annotation": {
        "font_family": "Arial",
        "font_size": "12px",
        "font_weight": 400
      }
    }
}
from typing import Dict, List, Tuple
import re
from datetime import datetime
import logging

logger = logging.getLogger(__name__)

def process_temporal_data(data: Dict) -> None:
    """处理时间类型的数据"""
    for column in data["data"]["columns"]:
        if column["data_type"] == "temporal":
            has_valid_temporal = False
            for row in data["data"]["data"]:
                value = str(row.get(column["name"], ""))
                
                try:
                    if value.isdigit():
                        if len(value) == 4:
                            has_valid_temporal = True
                            continue
                        else:
                            has_valid_temporal = False
                            break
                    
                    if "." in value:
                        parts = value.split(".")
                        if len(parts) == 2 and parts[0].isdigit() and parts[1].isdigit():
                            year, month = parts
                            month = month.zfill(2)
                            row[column["name"]] = f"{year}-{month}"
                            has_valid_temporal = True
                        elif len(parts) == 3 and all(part.isdigit() for part in parts):
                            year, month, day = parts
                            month = month.zfill(2)
                            day = day.zfill(2)
                            row[column["name"]] = f"{year}-{month}-{day}"
                            has_valid_temporal = True
                        else:
                            continue
                        continue
                    
                    if " " in value:
                        try:
                            # 尝试解析完整的月份名称
                            date_obj = datetime.strptime(value, "%B %Y")
                        except ValueError:
                            try:
                                # 尝试解析缩写的月份名称
                                date_obj = datetime.strptime(value, "%b %Y")
                            except ValueError:
                                # 尝试其他常见格式
                                try:
                                    # 处理 "YYYY-MM" 或 "YYYY/MM" 格式
                                    if "-" in value or "/" in value:
                                        separator = "-" if "-" in value else "/"
                                        parts = value.split(separator)
                                        if len(parts) == 2 and parts[0].isdigit() and parts[1].isdigit():
                                            year = parts[0]
                                            month = parts[1].zfill(2)
                                            row[column["name"]] = f"{year}-{month}"
                                            has_valid_temporal = True
                                            continue
                                except Exception:
                                    continue
                                continue
                        
                        # 转换为 "YYYY-MM" 格式
                        row[column["name"]] = date_obj.strftime("%Y-%m")
                        has_valid_temporal = True
                        continue
                        
                except Exception as e:
                    logger.warning(f"Failed to parse temporal value '{value}': {str(e)}")
                    continue
            
            # 如果没有找到任何有效的时间数据，将类型改为categorical
            if not has_valid_temporal:
                column["data_type"] = "categorical"
                data["data"]["type_combination"] = " + ".join([col["data_type"] for col in data["data"]["columns"]])
                logger.info(f"Changed column '{column['name']}' from temporal to categorical due to invalid temporal data")

def process_numerical_data(data: Dict) -> None:
    """处理数值类型的数据"""
    for column in data["data"]["columns"]:
        if column["data_type"] == "numerical":
            for row in data["data"]["data"]:
                value = row.get(column["name"])
                
                # 处理 null 或 None
                if value is None or value == "null" or value == "":
                    row[column["name"]] = 0
                    continue
                
                # 转换为字符串以进行处理
                value_str = str(value)
                
                # 提取数字（包括负号和小数点）
                numeric_chars = re.findall(r'-?\d*\.?\d+', value_str)
                if numeric_chars:
                    # 使用第一个匹配的数字
                    try:
                        row[column["name"]] = float(numeric_chars[0])
                    except ValueError:
                        row[column["name"]] = 0
                else:
                    row[column["name"]] = 0 

def deduplicate_combinations(data: Dict) -> None:
    """检查并去重temporal和categorical属性的组合
    
    Args:
        data: 包含数据的字典，格式为 {"data": {"columns": [...], "data": [...]}}
    """
    # 找出所有temporal和categorical列
    temporal_categorical_cols = [
        col["name"] for col in data["data"]["columns"]
        if col["data_type"] in ["temporal", "categorical"]
    ]
    
    if not temporal_categorical_cols:
        return
    
    # 用于存储已见过的组合
    seen_combinations = set()
    # 用于存储要保留的行索引
    rows_to_keep = []
    
    # 检查每一行
    for idx, row in enumerate(data["data"]["data"]):
        # 获取当前行的temporal和categorical值组合
        combination = tuple(str(row.get(col, "")) for col in temporal_categorical_cols)
        
        # 如果这个组合还没见过，就保留这行
        if combination not in seen_combinations:
            seen_combinations.add(combination)
            rows_to_keep.append(idx)
    
    # 只保留不重复的行
    data["data"]["data"] = [data["data"]["data"][i] for i in rows_to_keep]
    
    # 记录去重信息
    removed_count = len(data["data"]["data"]) - len(rows_to_keep)
    #if removed_count > 0:
    #    logger.info(f"Removed {removed_count} duplicate combinations of temporal/categorical attributes") 
def remove_unnecessary_fields(data: Any) -> Any:
    """
    Recursively remove unnecessary fields from any level of the data structure
    """
    unnecessary_fields = {
        "discarded_data_points",
        "missing_percentage",
        "zero_percentage",
        "transformed_columns"
    }
    
    if isinstance(data, dict):
        return {
            k: remove_unnecessary_fields(v)
            for k, v in data.items()
            if k not in unnecessary_fields
        }
    elif isinstance(data, list):
        return [remove_unnecessary_fields(item) for item in data]
    else:
        return data

def update_data_format(data: Dict[str, Any]) -> Dict[str, Any]:
    """
    Update the data format to match the new requirements
    """
    # First, remove unnecessary fields at all levels
    updated_data = remove_unnecessary_fields(data.copy())
    
    # Extract columns and data from the nested structure
    if "data" in updated_data and "data" in updated_data["data"] and "columns" in updated_data["data"]:
        pass
    else:
        columns = updated_data["columns"]
        data = updated_data["data"]
        updated_data["data"] = {
            "data": data,
            "columns": columns
        }
        del updated_data["columns"]

    try:
        if "title" in updated_data and "description" in updated_data and "main_insight" in updated_data:
            title = updated_data["title"]
            description = updated_data["description"]
            main_insight = updated_data["main_insight"]
            updated_data["metadata"] = {
                "title": title,
                "description": description,
                "main_insight": main_insight
            }
        elif "description" in updated_data and "titles" in updated_data and "main_title" in updated_data["titles"]:
            description = updated_data["description"]
            main_title = updated_data["titles"]["main_title"]
            main_insight = updated_data["metadata"]["main_insight"]
            datafact = updated_data["metadata"]["datafact"]
            updated_data["metadata"] = {
                "title": main_title,
                "description": description,
                "main_insight": main_insight,
                "datafact": datafact
            }
    except Exception as e:
        pass
    
    if "data" in updated_data and "type_combinations" in updated_data["data"]:
        updated_data["data"]["type_combination"] = updated_data["data"]["type_combinations"]
        del updated_data["data"]["type_combinations"]
    # Add standard attributes
    for key, value in STANDARD_ADDITIONS.items():
        if key not in updated_data:
            updated_data[key] = value
    
    return updated_data

def process(input: str, output: str = None) -> None:
    """
    Pipeline入口函数，处理单个文件的数据预处理
    
    Args:
        input (str): 输入JSON文件路径
        output (str): 输出JSON文件路径，如果为None则原地修改输入文件
    """
    try:
        # 如果没有指定输出路径，则原地修改
        if output is None:
            output = input
            
        logger.info(f"处理文件: {input}")
        
        # 检查是否需要处理
        if Path(output).exists():
            with open(output) as f:
                data = json.load(f)
                #if "metadata" in data and "data" in data and "variables" in data and "processed" in data:
                #    logger.info(f"跳过处理: {output} 已包含必要字段")
                #    return
        
        # 读取输入数据
        with open(input, 'r', encoding='utf-8') as f:
            data = json.load(f)
        
        # 更新数据格式
        updated_data = update_data_format(data)
        
        # 处理时间类型数据
        process_temporal_data(updated_data)
        
        # 处理数值类型数据
        process_numerical_data(updated_data)
        
        # 去重temporal和categorical属性的组合
        deduplicate_combinations(updated_data)
        updated_data["processed"] = True
        
        # 保存更新后的数据
        with open(output, 'w', encoding='utf-8') as f:
            json.dump(updated_data, f, indent=2, ensure_ascii=False)
            
        logger.info(f"处理完成: {output}")
            
    except Exception as e:
        logger.error(f"处理失败: {str(e)}")
        raise