app.py

import os
import decimal
from datetime import datetime, date
from statistics import mean
import requests
import tempfile
from openpyxl import load_workbook
from frictionless import Resource, Detector
from frictionless.formats import ExcelControl
import logging
import gradio as gr

# --- 0. 日志配置 ---
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# --- 2. 核心逻辑函数 ---
def get_sheets_logic(file_url):
    with tempfile.NamedTemporaryFile(suffix=".xlsx", delete=False) as f_in:
        resp = requests.get(file_url, timeout=60)
        f_in.write(resp.content)
        tmp_path = f_in.name
    wb = load_workbook(tmp_path, read_only=True)
    sheets = wb.sheetnames
    wb.close()
    os.unlink(tmp_path)
    return sheets

def profile_data_logic(file_url, sheet_name, detail_level, sample_rows=5):
    """
    分析 Excel/CSV 文件并返回结构化画像。
    """
    src_tmp_name = None
    normalized_tmp_name = None

    try:
        logger.info(f"正在分析, 模式: {detail_level}")

        # 1. 下载文件到临时空间
        resp = requests.get(file_url, timeout=60)
        resp.raise_for_status()
        
        # 使用 .xlsx 后缀确保 openpyxl 和 frictionless 能正确识别格式
        with tempfile.NamedTemporaryFile(suffix=".xlsx", delete=False) as f_in:
            f_in.write(resp.content)
            src_tmp_name = f_in.name

        # 2. 规范化处理 (处理 Excel 公式、合并单元格)
        with tempfile.NamedTemporaryFile(suffix=".xlsx", delete=False) as f_out:
            normalized_tmp_name = f_out.name


        # 3. Frictionless 探测配置
        # 加载计算后的值 (data_only=True)
        wb = load_workbook(src_tmp_name, data_only=True)
        if sheet_name and sheet_name not in wb.sheetnames:
            return {"status": "error", "message": f"Sheet '{sheet_name}' 不存在"}
        wb.save(normalized_tmp_name)

        # 2. 核心探测逻辑
        control = ExcelControl(sheet=sheet_name, fill_merged_cells=True) if sheet_name else ExcelControl(fill_merged_cells=True)
        
        detector = Detector(sample_size=1000, field_confidence=0.9)
        resource = Resource(path=normalized_tmp_name, control=control, detector=detector)
        resource.infer(stats=True)

        with resource:
            schema = resource.schema
            all_rows = resource.read_rows()
            total_rows = len(all_rows)
            
            fields_analysis = []
            for index, field in enumerate(schema.fields):
                # --- [Basic Level] 基础信息 ---
                analysis = {
                    "name": field.name,
                    "type": field.type
                }

                # 如果级别是 basic，直接跳过后续重型计算
                if detail_level == "basic":
                    fields_analysis.append(analysis)
                    continue

                # 提取非空数据用于后续分析
                col_data = []
                null_count = 0
                for row in all_rows:
                    # Frictionless Row object behaves like a dictionary for field access
                    # row[index] works in older versions but it seems safer to use field name or direct property
                    # Let's try to access by field name which is safer
                    val = row.get(field.name)
                    if val is not None:
                        col_data.append(val)
                    else:
                        null_count += 1

                # --- [Standard Level] 质量与数值统计 ---
                analysis["missing_rate"] = f"{(null_count / total_rows) * 100:.2f}%" if total_rows > 0 else "0%"
                
                if field.type in ['integer', 'number'] and col_data:
                    # 确保是可计算的数值
                    numeric_values = [
                        float(v) for v in col_data 
                        if isinstance(v, (int, float, decimal.Decimal))
                    ]
                    if numeric_values:
                        analysis["numeric_stats"] = {
                            "min": min(numeric_values),
                            "max": max(numeric_values),
                            "avg": round(mean(numeric_values), 2)
                        }

                # --- [Full Level] 枚举与样本 ---
                if detail_level == "full":
                    # 枚举检测：如果唯一值较少（<15）且有数据，则视为枚举
                    unique_vals = list(set(col_data))
                    if 0 < len(unique_vals) <= 15:
                        analysis["is_enumeration"] = True
                        analysis["enum_values"] = [str(v) for v in unique_vals]
                    else:
                        analysis["is_enumeration"] = False
                    
                    # 样本提取：转换日期等特殊对象为字符串
                    samples = []
                    for v in col_data[:sample_rows]:
                        if isinstance(v, (datetime, date)):
                            samples.append(v.isoformat())
                        else:
                            samples.append(str(v))
                    analysis["samples"] = samples

                fields_analysis.append(analysis)

            # 4. 组装最终报告
            report = {
                "file_summary": {
                    "total_rows": total_rows,
                    "total_fields": len(fields_analysis),
                    "detail_level_applied": detail_level
                },
                "schema_analysis": fields_analysis
            }

            logger.info(f"分析成功: 识别到 {len(fields_analysis)} 列")
            return report

    except Exception as e:
        logger.error(f"分析异常: {str(e)}")
        return {
            "status": "failed",
            "error_type": type(e).__name__,
            "message": str(e)
        }

    finally:
        # 5. 清理临时文件，防止磁盘溢出
        for path in (src_tmp_name, normalized_tmp_name):
            if path and os.path.exists(path):
                try:
                    os.unlink(path)
                except Exception as cleanup_err:
                    logger.warning(f"清理临时文件失败: {cleanup_err}")

# --- 3. Gradio Interface (Will be auto-converted to MCP Tools) ---

def get_excel_structure(file_url: str):
    """
    获取Excel所有工作表名称。
    
    Args:
        file_url: Excel 文件的公网 URL 地址
    
    Returns:
        list: 工作表名称列表
    """
    return get_sheets_logic(file_url)

def profile_sheet_data(file_url: str, sheet_name: str = None, detail_level: str = "standard", sample_rows: int = 5):
    """
    深度分析指定工作表的数据画像。
    
    Args:
        file_url: Excel 文件的公网 URL 地址
        sheet_name: 要分析的工作表名称 (可选)
        detail_level: 分析详细程度 ('basic', 'standard', 'full')
        sample_rows: 当 detail_level 为 'full' 时，返回的样本行数
    
    Returns:
        dict: 包含数据画像的 JSON 对象
    """
    return profile_data_logic(file_url, sheet_name, detail_level, sample_rows)

# 构建 UI
with gr.Blocks(title="Excel Analyzer Pro") as demo:
    gr.Markdown("# 📊 Excel Analyzer Pro")
    gr.Markdown("提供 Excel 文件的结构解析与深度数据画像功能。支持 MCP 协议。")
    
    with gr.Tab("获取工作表结构"):
        with gr.Row():
            url_input_structure = gr.Textbox(label="Excel 文件 URL", placeholder="https://example.com/file.xlsx")
            btn_structure = gr.Button("获取工作表列表", variant="primary")
        output_structure = gr.JSON(label="工作表列表")
        
        btn_structure.click(get_excel_structure, inputs=[url_input_structure], outputs=[output_structure])

    with gr.Tab("生成数据画像"):
        with gr.Row():
            url_input_profile = gr.Textbox(label="Excel 文件 URL", placeholder="https://example.com/file.xlsx")
            sheet_input_profile = gr.Textbox(label="工作表名称 (可选)")
        with gr.Row():
            detail_input_profile = gr.Dropdown(["basic", "standard", "full"], label="详细程度", value="standard")
            sample_input_profile = gr.Number(value=5, label="样本行数 (仅 full 模式)")
        
        btn_profile = gr.Button("生成数据画像", variant="primary")
        output_profile = gr.JSON(label="数据画像报告")
        
        btn_profile.click(
            profile_sheet_data, 
            inputs=[url_input_profile, sheet_input_profile, detail_input_profile, sample_input_profile], 
            outputs=[output_profile]
        )

# Hugging Face Spaces 启动入口
if __name__ == "__main__":
    demo.launch(server_name="0.0.0.0", server_port=7860, mcp_server=True)