import os
import sys
import shutil
import logging
import argparse
import datetime
import urllib.parse
import warnings
import pandas as pd
import numpy as np
from sqlalchemy import create_engine, text
from sqlalchemy.engine import URL
from sqlalchemy.types import NVARCHAR, Integer, Date

# 导入配置
from log_utils import (log_error, log_warning, log_info, log_processing, log_file, log_sync,
                         log_start, log_complete, LoggerManager)
from config import DB_CONFIG, CACHE_DIR, EXCEL_CONFIGS, BATCH_SIZE, TABLE_SCHEMA

# ================= 抑制 openpyxl 的数据验证警告 =================
warnings.filterwarnings('ignore', category=UserWarning, module='openpyxl')

class DataSynchronizer:
    def __init__(self, force_sync=False):
        self.force_sync = force_sync
        self.engine = self._get_db_connection()
        self.cache_dir = CACHE_DIR
        
        if not os.path.exists(self.cache_dir):
            os.makedirs(self.cache_dir)

    def _get_db_connection(self):
        connection_string = (
            f"DRIVER={{{DB_CONFIG['driver']}}};"
            f"SERVER={DB_CONFIG['server']};"
            f"DATABASE={DB_CONFIG['database']};"
            f"UID={DB_CONFIG['username']};"
            f"PWD={DB_CONFIG['password']};"
            f"TrustServerCertificate={DB_CONFIG.get('TrustServerCertificate', 'yes')};"
        )
        connection_url = URL.create("mssql+pyodbc", query={"odbc_connect": connection_string})
        return create_engine(connection_url, fast_executemany=True)

    def _should_process_file(self, remote_path, local_path):
        if self.force_sync:
            return True, "强制同步"
        
        if not os.path.exists(local_path):
            return True, "缓存不存在"
        
        try:
            remote_mtime = os.path.getmtime(remote_path)
            local_mtime = os.path.getmtime(local_path)
            if remote_mtime > local_mtime + 1:
                return True, f"源文件更新"
        except OSError as e:
            log_error(f"无法访问源文件: {remote_path}, Error: {e}")
            return False, "源文件无法访问"
            
        return False, "文件未变更"

    def _clean_dataframe(self, df, contract_year):
        """主表数据清洗与验证"""
        # 1. 设置合同年份
        df['合同年份'] = contract_year

        # 2. 移除总排号为空的行
        if '总排号' in df.columns:
            df = df.dropna(subset=['总排号'])
            df = df[df['总排号'].astype(str).str.strip() != '']
        else:
            log_error("数据源中找不到映射后的[总排号]列，跳过此 sheet")
            return None, None

        # ★ 新增：去除重复的总排号（保留第一条）
        if '总排号' in df.columns:
            df['总排号'] = df['总排号'].astype(str).str.strip()
            duplicates = df[df.duplicated(subset=['总排号'], keep='first')]
            if not duplicates.empty:
                log_warning(f"发现 {len(duplicates)} 条重复的总排号，已自动去重。重复的总排号: {duplicates['总排号'].tolist()[:10]}")
                df = df.drop_duplicates(subset=['总排号'], keep='first')

        # 3. 补全列
        for col in TABLE_SCHEMA.keys():
            if col not in df.columns:
                df[col] = None

        # 用于存储每一列的 SQL 类型
        dtype_dict = {}

        # 4. 字段清洗
        for col, rules in TABLE_SCHEMA.items():
            if col not in df.columns: 
                continue
            
            if rules['type'] == 'int':
                # ★ 修改：先转换为数值，然后四舍五入到整数
                df[col] = pd.to_numeric(df[col], errors='coerce')
                # 将浮点数四舍五入为整数（处理如 123.5 这样的值）
                df[col] = df[col].round(0)
                # 转换为可空整数类型
                df[col] = df[col].astype('Int64')
                # 将 NaN 替换为 None
                df[col] = df[col].replace({pd.NA: None})
                dtype_dict[col] = Integer()
                
            elif rules['type'] == 'date':
                df[col] = pd.to_datetime(df[col], errors='coerce')
                df[col] = df[col].apply(lambda x: x.date() if pd.notnull(x) else None)
                dtype_dict[col] = Date()
                
            elif rules['type'] == 'str':
                # 先转换为字符串
                df[col] = df[col].fillna('').astype(str)
                # 替换各种空值表示
                df[col] = df[col].replace({'nan': '', 'None': '', '<NA>': ''})
                
                # 强制截断
                max_len = rules.get('max_len', 255)
                df[col] = df[col].str.slice(0, max_len)
                # 将空字符串转为 None
                df[col] = df[col].replace('', None)
                
                dtype_dict[col] = NVARCHAR(max_len)

        final_cols = list(TABLE_SCHEMA.keys())
        
        return df[final_cols], dtype_dict

    def _sync_to_db(self, df, dtype_dict):
        """同步主表数据 - 使用更稳健的方法"""
        if df is None or df.empty:
            return

        target_table = "[warehouseOutbound].[executionCardData]"
        
        with self.engine.connect() as conn:
            existing_ids = pd.read_sql(f"SELECT [总排号] FROM {target_table}", conn)
        
        existing_id_set = set(existing_ids['总排号'].astype(str))
        df['总排号'] = df['总排号'].astype(str).str.strip()
        
        df_update = df[df['总排号'].isin(existing_id_set)].copy()
        df_insert = df[~df['总排号'].isin(existing_id_set)].copy()

        log_info(f"分析结果: 需插入 {len(df_insert)} 条, 需更新 {len(df_update)} 条")

        # 1. 插入新数据
        if not df_insert.empty:
            log_info("正在执行批量插入...")
            df_insert.to_sql('executionCardData', self.engine, schema='warehouseOutbound', 
                             if_exists='append', index=False, chunksize=BATCH_SIZE,
                             dtype=dtype_dict) 
            log_info("批量插入完成。")

        # 2. 更新现有数据 - 改用逐条或小批量 UPDATE
        if not df_update.empty:
            log_info("正在执行批量更新...")
            
            cols = [c for c in df.columns if c != '总排号']
            set_clause = ", ".join([f"[{c}] = :{c}" for c in cols])
            update_sql = f"""
                UPDATE [warehouseOutbound].[executionCardData]
                SET {set_clause}
                WHERE [总排号] = :总排号
            """
            
            with self.engine.begin() as conn:
                batch_size = 1000
                total_rows = len(df_update)
                update_count = 0

                for i in range(0, total_rows, batch_size):
                    batch = df_update.iloc[i:i+batch_size]
                    records = batch.to_dict('records')

                    result = conn.execute(text(update_sql), records)
                    update_count += result.rowcount

                    if (i + batch_size) % 5000 == 0:
                        log_info(f"已更新 {i + batch_size}/{total_rows} 条记录...")

            # 修复 SQL Server executemany 返回负数 rowcount 的问题
            affected_rows = abs(update_count) if update_count < 0 else total_rows
            log_info(f"批量更新完成，共影响 {affected_rows} 行。")

    def process_excel_files(self):
        for cfg in EXCEL_CONFIGS:
            remote_path = cfg['file_path']
            filename = os.path.basename(remote_path)
            local_path = os.path.join(self.cache_dir, filename)

            should_sync, reason = self._should_process_file(remote_path, local_path)
            
            if should_sync:
                log_info(f"开始处理文件: {filename} ({reason})")
                try:
                    # 复制文件到本地缓存（只复制一次）
                    if os.path.exists(remote_path):
                        shutil.copy2(remote_path, local_path)
                    
                    # 遍历该文件的所有指定 sheet
                    for sheet_name in cfg['sheet_names']:
                        log_info(f"  → 处理工作表: {sheet_name} (合同年份: {cfg['contract_year']})")
                        try:
                            df = pd.read_excel(local_path, sheet_name=sheet_name, header=0, engine='openpyxl')
                            df.columns = [str(c).strip() for c in df.columns]
                            df.rename(columns=cfg['field_mapping'], inplace=True)

                            cleaned_df, dtype_mapping = self._clean_dataframe(df, cfg['contract_year'])

                            if cleaned_df is not None:
                                self._sync_to_db(cleaned_df, dtype_mapping)
                                log_info(f"    工作表 {sheet_name} 同步成功。")
                            else:
                                log_warning(f"    工作表 {sheet_name} 清洗失败，跳过。")
                        except Exception as e:
                            log_error(f"    处理工作表 {sheet_name} 时发生错误: {str(e)}", exc_info=True)
                    
                    log_info(f"文件 {filename} 所有工作表处理完成。")
                    
                except Exception as e:
                    log_error(f"处理文件 {filename} 时发生错误: {str(e)}", exc_info=True)
            else:
                log_info(f"跳过文件: {filename} ({reason})")

    def generate_contract_data(self):
        log_info("开始生成/更新 contractData 表...")
        
        merge_sql = """
        WITH SourceData AS (
            SELECT 
                CAST(ISNULL([合同年份], '') AS NVARCHAR(10)) AS [合同年份],
                CAST(ISNULL([车间号], '') AS NVARCHAR(20)) AS [车间号],
                CAST(ISNULL([工令号], '') AS NVARCHAR(200)) AS [工令号],
                CAST([订单号] AS NVARCHAR(150)) AS [订单号],
                CAST([客户名称] AS NVARCHAR(200)) AS [客户名称],
                CAST([产品名称] AS NVARCHAR(200)) AS [产品型号], 
                CAST([量程] AS NVARCHAR(150)) AS [量程],
                TRY_CAST([数量] AS INT) AS [数量],
                CAST(NULL AS INT) AS [单价],
                TRY_CAST([序号] AS INT) AS [ID],
                CAST([位号] AS NVARCHAR(500)) AS [位号],
                ROW_NUMBER() OVER (
                    PARTITION BY [合同年份], [车间号], [工令号] 
                    ORDER BY [总排号] DESC
                ) as rn
            FROM [warehouseOutbound].[executionCardData]
            WHERE 
                [车间号] IS NOT NULL AND [车间号] <> '' 
                AND [工令号] IS NOT NULL AND [工令号] <> ''
        )
        
        MERGE INTO [warehouseOutbound].[contractData] AS Target
        USING (SELECT * FROM SourceData WHERE rn = 1) AS Source
        ON (
            Target.[合同年份] = Source.[合同年份] 
            AND Target.[车间号] = Source.[车间号] 
            AND Target.[工令号] = Source.[工令号]
        )
        
        WHEN MATCHED THEN
            UPDATE SET 
                Target.[订单号] = Source.[订单号],
                Target.[客户名称] = Source.[客户名称],
                Target.[产品型号] = Source.[产品型号],
                Target.[量程] = Source.[量程],
                Target.[数量] = Source.[数量],
                Target.[ID] = Source.[ID],
                Target.[位号] = Source.[位号]
                
        WHEN NOT MATCHED BY TARGET THEN
            INSERT (
                [合同年份], [车间号], [工令号], 
                [订单号], [客户名称], [产品型号], 
                [量程], [数量], [单价], [ID], [位号]
            )
            VALUES (
                Source.[合同年份], Source.[车间号], Source.[工令号], 
                Source.[订单号], Source.[客户名称], Source.[产品型号], 
                Source.[量程], Source.[数量], Source.[单价], Source.[ID], Source.[位号]
            )
        ;
        """
        
        try:
            with self.engine.begin() as conn:
                result = conn.execute(text(merge_sql))
                log_info(f"ContractData 表同步完成 (SQL Server 内部处理)。rowcount: {result.rowcount}")
                
        except Exception as e:
            log_error(f"生成 ContractData 失败: {e}", exc_info=True)

def main():
    # 初始化日志管理器
    LoggerManager("etl_manager", log_prefix="sync")

    parser = argparse.ArgumentParser(description="Excel数据同步至SQL Server")
    parser.add_argument('--force', action='store_true', help='强制同步所有文件')
    args = parser.parse_args()

    syncer = DataSynchronizer(force_sync=args.force)

    if args.force:
        log_start("Excel 同步任务 (强制模式)")
    else:
        log_start("Excel 同步任务 (增量模式)")

    syncer.process_excel_files()
    syncer.generate_contract_data()

    log_complete("Excel 同步任务已完成")

if __name__ == "__main__":
    main()