-- =============================================
-- 物流轨迹表分区与索引补充
-- 适用于 MySQL 8.0+
-- 注意：分区表不支持直接ALTER，需要重建表操作
-- 执行前请在测试环境验证，并在低峰期执行
-- =============================================

USE erp_db;

-- =============================================
-- Step 1: 创建带分区的影子表
-- =============================================
CREATE TABLE IF NOT EXISTS logistics_trace_partitioned (
    id BIGINT AUTO_INCREMENT,
    waybill_no VARCHAR(50) NOT NULL COMMENT '运单号',
    carrier VARCHAR(20) NOT NULL COMMENT '物流商编码',
    status VARCHAR(30) DEFAULT '' COMMENT '轨迹节点状态',
    status_label VARCHAR(50) DEFAULT '' COMMENT '状态标签',
    location VARCHAR(200) DEFAULT '' COMMENT '轨迹发生地点',
    description VARCHAR(500) DEFAULT '' COMMENT '轨迹描述',
    trace_time DATETIME NOT NULL COMMENT '轨迹发生时间',
    raw_status_code VARCHAR(50) DEFAULT '' COMMENT '物流商返回的原始状态码',
    raw_data JSON DEFAULT NULL COMMENT '原始轨迹数据（JSON格式）',
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
    deleted TINYINT(1) DEFAULT 0 COMMENT '逻辑删除：0-未删除, 1-已删除',
    PRIMARY KEY (id, trace_time),  -- ⚠️ 分区字段必须在主键中
    INDEX idx_waybill_no (waybill_no),
    INDEX idx_carrier (carrier),
    INDEX idx_trace_time (trace_time),
    INDEX idx_waybill_trace (waybill_no, trace_time),
    INDEX idx_waybill_status (waybill_no, status),
    INDEX idx_status (status)
)
ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
COMMENT='物流轨迹记录表（按月RANGE分区）'
PARTITION BY RANGE (YEAR(trace_time) * 100 + MONTH(trace_time)) (
    PARTITION p202601 VALUES LESS THAN (202602) COMMENT '2026年1月',
    PARTITION p202602 VALUES LESS THAN (202603) COMMENT '2026年2月',
    PARTITION p202603 VALUES LESS THAN (202604) COMMENT '2026年3月',
    PARTITION p202604 VALUES LESS THAN (202605) COMMENT '2026年4月',
    PARTITION p202605 VALUES LESS THAN (202606) COMMENT '2026年5月',
    PARTITION p202606 VALUES LESS THAN (202607) COMMENT '2026年6月',
    PARTITION p202607 VALUES LESS THAN (202608) COMMENT '2026年7月',
    PARTITION p202608 VALUES LESS THAN (202609) COMMENT '2026年8月',
    PARTITION p202609 VALUES LESS THAN (202610) COMMENT '2026年9月',
    PARTITION p202610 VALUES LESS THAN (202611) COMMENT '2026年10月',
    PARTITION p202611 VALUES LESS THAN (202612) COMMENT '2026年11月',
    PARTITION p202612 VALUES LESS THAN (202701) COMMENT '2026年12月',
    PARTITION p202701 VALUES LESS THAN (202702) COMMENT '2027年1月',
    PARTITION p202702 VALUES LESS THAN (202703) COMMENT '2027年2月',
    PARTITION p202703 VALUES LESS THAN (202704) COMMENT '2027年3月',
    PARTITION p202704 VALUES LESS THAN (202705) COMMENT '2027年4月',
    PARTITION p_future  VALUES LESS THAN MAXVALUE COMMENT '未来数据'
);

-- =============================================
-- Step 2: 迁移数据（从旧表到分区表）
-- =============================================
-- 仅迁移未删除的数据
INSERT INTO logistics_trace_partitioned
    (id, waybill_no, carrier, status, status_label, location, description,
     trace_time, raw_status_code, raw_data, created_at, deleted)
SELECT id, waybill_no, carrier, status, status_label, location, description,
       trace_time, raw_status_code, raw_data, created_at, deleted
FROM logistics_trace
WHERE deleted = 0
ON DUPLICATE KEY KEY UPDATE id=id;  -- 避免ID冲突跳过

-- =============================================
-- Step 3: 验证数据一致性
-- =============================================
-- SELECT COUNT(*) AS old_count FROM logistics_trace WHERE deleted = 0;
-- SELECT COUNT(*) AS new_count FROM logistics_trace_partitioned;

-- =============================================
-- Step 4: 表切换（低峰期执行，示例命令）
-- =============================================
-- RENAME TABLE logistics_trace TO logistics_trace_old_bak,
--              logistics_trace_partitioned TO logistics_trace;

-- =============================================
-- Step 5: 旧表清理（确认新表运行正常后删除）
-- =============================================
-- DROP TABLE IF EXISTS logistics_trace_old_bak;

-- =============================================
-- logistics_waybill_status 索引补充
-- =============================================
-- 复合索引：加速按运单号+物流商+状态查询
CREATE INDEX idx_waybill_carrier_status
    ON logistics_waybill_status (waybill_no, carrier, status);

-- 同步优先级索引：need_sync + sync_status + sync_retry_count
CREATE INDEX idx_sync_need
    ON logistics_waybill_status (need_sync, sync_status, sync_retry_count);

-- 时间索引：用于查找长期无更新的运单
CREATE INDEX idx_last_trace_time_status
    ON logistics_waybill_status (last_trace_time, status, deleted);

-- =============================================
-- 异常检测视图（方便异常查询）
-- =============================================
CREATE OR REPLACE VIEW v_logistics_exception AS
SELECT
    w.id,
    w.waybill_no,
    w.carrier,
    CASE w.carrier
        WHEN 'SF' THEN '顺丰速运'
        WHEN 'YTO' THEN '圆通速递'
        WHEN 'ZTO' THEN '中通快递'
        WHEN 'YUNDA' THEN '韵达快递'
        ELSE w.carrier
    END AS carrier_name,
    w.status,
    w.status_label,
    w.location,
    w.description,
    w.last_trace_time,
    w.sync_status,
    CASE w.sync_status
        WHEN 0 THEN '未同步'
        WHEN 1 THEN '同步中'
        WHEN 2 THEN '已同步'
        WHEN 3 THEN '同步失败'
        ELSE '未知'
    END AS sync_status_label,
    w.sync_fail_reason,
    w.order_id,
    w.order_no,
    w.receiver_name,
    w.receiver_phone,
    w.receiver_address,
    TIMESTAMPDIFF(HOUR, COALESCE(w.last_trace_time, w.created_at), NOW()) AS stale_hours,
    CASE
        WHEN w.status = 'EXCEPTION' THEN '运单异常'
        WHEN w.sync_status = 3 THEN '同步失败'
        WHEN w.status NOT IN ('SIGNED','RETURNED','RETURNING')
             AND TIMESTAMPDIFF(HOUR, COALESCE(w.last_trace_time, w.created_at), NOW()) > 48
             THEN '停滞预警'
        WHEN w.status = 'DELIVERING'
             AND TIMESTAMPDIFF(HOUR, COALESCE(w.last_trace_time, w.created_at), NOW()) > 24
             THEN '派送超时'
        ELSE '正常'
    END AS exception_type,
    CASE
        WHEN w.status = 'EXCEPTION' OR w.sync_status = 3 THEN 'P1'
        WHEN w.status NOT IN ('SIGNED','RETURNED','RETURNING')
             AND TIMESTAMPDIFF(HOUR, COALESCE(w.last_trace_time, w.created_at), NOW()) > 72
             THEN 'P1'
        WHEN w.status NOT IN ('SIGNED','RETURNED','RETURNING')
             AND TIMESTAMPDIFF(HOUR, COALESCE(w.last_trace_time, w.created_at), NOW()) > 48
             THEN 'P2'
        ELSE 'P3'
    END AS priority
FROM logistics_waybill_status w
WHERE w.deleted = 0;

-- =============================================
-- 物流统计视图
-- =============================================
CREATE OR REPLACE VIEW v_logistics_daily_stats AS
SELECT
    carrier,
    DATE(last_trace_time) AS stat_date,
    COUNT(*) AS total_waybills,
    SUM(CASE WHEN status = 'SIGNED' THEN 1 ELSE 0 END) AS signed_count,
    SUM(CASE WHEN status = 'IN_TRANSIT' THEN 1 ELSE 0 END) AS in_transit_count,
    SUM(CASE WHEN status = 'DELIVERING' THEN 1 ELSE 0 END) AS delivering_count,
    SUM(CASE WHEN status = 'EXCEPTION' THEN 1 ELSE 0 END) AS exception_count,
    SUM(CASE WHEN status = 'RETURNING' OR status = 'RETURNED' THEN 1 ELSE 0 END) AS return_count,
    ROUND(AVG(TIMESTAMPDIFF(HOUR, created_at, signed_time)), 1) AS avg_delivery_hours
FROM logistics_waybill_status
WHERE deleted = 0 AND last_trace_time IS NOT NULL
GROUP BY carrier, DATE(last_trace_time);
