-- 挂载一个 SQLite 数据库（最简单示例）
ATTACH 'path/to/file.db' AS my_db (TYPE SQLITE);

-- 现在可以跨库查询了
SELECT *
FROM my_db.some_table AS a
JOIN main.public.another_table AS b ON a.id = b.id;

#!/usr/bin/env python3
"""
DuckDB 跨库 JOIN 实战演示
模拟场景：电商公司数据分散在三个数据源，一条 SQL 完成联查

前置条件: pip install duckdb openpyxl
"""

import duckdb
import os

# ====== 第1步：创建模拟数据 ======

# 模拟 MySQL 订单表 (CSV 文件)
orders_csv = """order_id,customer_id,product_id,amount,order_date
1001,201,5001,299.00,2026-05-01
1002,202,5002,159.00,2026-05-01
1003,201,5003,899.00,2026-05-02
1004,203,5001,299.00,2026-05-02
1005,204,5004,459.00,2026-05-03
1006,202,5002,159.00,2026-05-03
1007,205,5005,1299.00,2026-05-04
1008,203,5003,899.00,2026-05-04
1009,206,5001,299.00,2026-05-05
1010,201,5004,459.00,2026-05-05
"""

# 模拟 PostgreSQL 用户表 (CSV 文件)
users_csv = """customer_id,name,city,member_level,register_date
201,张三,北京,金牌,2025-01-15
202,李四,上海,银牌,2025-03-20
203,王五,广州,金牌,2025-02-01
204,赵六,深圳,普通,2025-06-10
205,陈七,杭州,银牌,2025-04-05
206,孙八,成都,普通,2025-08-15
"""

# 模拟商品目录 (Excel 将会生成)
products_csv = """product_id,product_name,category,unit_price,cost_price
5001,无线蓝牙耳机,数码,299.00,180.00
5002,保温杯,家居,159.00,80.00
5003,智能手表,数码,899.00,550.00
5004,运动鞋,服饰,459.00,280.00
5005,平板支架,数码,1299.00,800.00
"""

# 写入临时文件
os.makedirs("day07_data", exist_ok=True)
with open("day07_data/orders.csv", "w") as f:
    f.write(orders_csv)
with open("day07_data/users.csv", "w") as f:
    f.write(users_csv)
with open("day07_data/products.csv", "w") as f:
    f.write(products_csv)

# ====== 第2步：用 DuckDB 模拟跨源 JOIN ======

# 启动 DuckDB 内存数据库
con = duckdb.connect()

# 挂载 CSV 目录，模拟"不同数据源"
con.execute("""
    CREATE VIEW orders AS 
    SELECT * FROM read_csv_auto('day07_data/orders.csv')
""")
con.execute("""
    CREATE VIEW users AS 
    SELECT * FROM read_csv_auto('day07_data/users.csv')
""")
con.execute("""
    CREATE VIEW products AS 
    SELECT * FROM read_csv_auto('day07_data/products.csv')
""")

print("=" * 60)
print("📊 跨源分析：金牌会员购买了哪些商品？")
print("=" * 60)

# 一条 SQL 跨三个"数据源" JOIN
result = con.execute("""
    SELECT 
        u.name AS 客户名称,
        u.city AS 城市,
        u.member_level AS 会员等级,
        p.product_name AS 商品名称,
        p.category AS 品类,
        o.amount AS 单价,
        (o.amount - p.cost_price) AS 毛利
    FROM orders o
    JOIN users u ON o.customer_id = u.customer_id
    JOIN products p ON o.product_id = p.product_id
    WHERE u.member_level IN ('金牌', '银牌')
    ORDER BY o.amount DESC
""").fetchdf()

print(result.to_string(index=False))

print("\n" + "=" * 60)
print("💰 每日销售额汇总（含毛利）")
print("=" * 60)

result2 = con.execute("""
    SELECT 
        o.order_date AS 日期,
        COUNT(DISTINCT o.order_id) AS 订单数,
        SUM(o.amount) AS 销售额,
        SUM(o.amount - p.cost_price) AS 总毛利,
        ROUND(AVG(o.amount - p.cost_price), 2) AS 平均每单毛利
    FROM orders o
    JOIN products p ON o.product_id = p.product_id
    GROUP BY o.order_date
    ORDER BY o.order_date
""").fetchdf()

print(result2.to_string(index=False))

print("\n" + "=" * 60)
print("🏆 商品品类毛利分析")
print("=" * 60)

result3 = con.execute("""
    SELECT 
        p.category AS 品类,
        COUNT(*) AS 销量,
        SUM(o.amount) AS 销售额,
        SUM(o.amount - p.cost_price) AS 总毛利,
        ROUND(AVG(o.amount - p.cost_price), 2) AS 平均毛利,
        ROUND(SUM(o.amount - p.cost_price) / SUM(o.amount) * 100, 1) AS 毛利率_percent
    FROM orders o
    JOIN products p ON o.product_id = p.product_id
    GROUP BY p.category
    ORDER BY 总毛利 DESC
""").fetchdf()

print(result3.to_string(index=False))

# ====== 第3步：输出为 Excel 报表 ======

from duckdb import connect as duck_connect

try:
    con.execute("INSTALL spatial; LOAD spatial;")
    con.execute("""
        COPY (
            SELECT 
                o.order_id,
                u.name,
                u.city,
                u.member_level,
                p.product_name,
                p.category,
                o.amount,
                (o.amount - p.cost_price) AS profit
            FROM orders o
            JOIN users u ON o.customer_id = u.customer_id
            JOIN products p ON o.product_id = p.product_id
        ) TO 'day07_data/跨源分析报表.xlsx' 
        WITH (FORMER XLSX);
    """)
    print("\n✅ 报表已导出: day07_data/跨源分析报表.xlsx")
except Exception as e:
    print(f"\n⚠️ 导出 XLSX 需要 spatial 扩展: {e}")
    print("可以改用 CSV 导出:")
    con.execute("""
        COPY (
            SELECT 
                o.order_id,
                u.name,
                u.city,
                u.member_level,
                p.product_name,
                p.category,
                o.amount,
                (o.amount - p.cost_price) AS profit
            FROM orders o
            JOIN users u ON o.customer_id = u.customer_id
            JOIN products p ON o.product_id = p.product_id
        ) TO 'day07_data/跨源分析报表.csv' (HEADER, DELIMITER ',');
    """)
    print("✅ 报表已导出: day07_data/跨源分析报表.csv")

# ====== 第4步：验证结果 ======
print("\n" + "=" * 60)
print("📈 验证：所有订单是否都关联到用户和商品？")
print("=" * 60)

validation = con.execute("""
    SELECT 
        '订单总数' AS 指标, CAST(COUNT(*) AS VARCHAR) AS 值 FROM orders
    UNION ALL
    SELECT '匹配到用户', CAST(COUNT(*) AS VARCHAR) FROM orders o 
        JOIN users u ON o.customer_id = u.customer_id
    UNION ALL
    SELECT '匹配到商品', CAST(COUNT(*) AS VARCHAR) FROM orders o 
        JOIN products p ON o.product_id = p.product_id
    UNION ALL
    SELECT '完全匹配', CAST(COUNT(*) AS VARCHAR) FROM orders o 
        JOIN users u ON o.customer_id = u.customer_id
        JOIN products p ON o.product_id = p.product_id
""").fetchdf()

print(validation.to_string(index=False))

con.close()
print("\n🎉 跨库 JOIN 演示完成！")

-- 安装扩展（只需一次）
INSTALL mysql_scanner;
LOAD mysql_scanner;
INSTALL postgres_scanner;
LOAD postgres_scanner;

-- 挂载 MySQL 订单库
ATTACH 'host=localhost port=3306 dbname=orders_db user=analyst password=xxx' 
    AS mysql_db (TYPE MYSQL);

-- 挂载 PostgreSQL 用户库
ATTACH 'host=localhost port=5432 dbname=users_db user=analyst password=xxx' 
    AS pg_db (TYPE POSTGRES);

-- 挂载本地商品 CSV
CREATE VIEW products AS 
SELECT * FROM read_csv_auto('products.csv');

-- 一条 SQL 跨三个数据源
SELECT 
    u.name,
    u.city,
    p.product_name,
    SUM(o.amount) AS total_spent
FROM mysql_db.orders AS o
JOIN pg_db.public.customers AS u ON o.customer_id = u.customer_id
JOIN products AS p ON o.product_id = p.product_id
WHERE o.order_date >= '2026-04-01'
GROUP BY u.name, u.city, p.product_name
ORDER BY total_spent DESC;

-- DuckDB 会在 MySQL 端执行这个 GROUP BY
-- 只回传聚合后的少量结果，而不是全表扫描
SELECT customer_id, COUNT(*) 
FROM mysql_db.orders 
WHERE order_date >= '2026-01-01'
GROUP BY customer_id;

-- ❌ 不推荐：全表拉取后再过滤
SELECT * FROM mysql_db.orders;  -- 可能拉几百万行

-- ✅ 推荐：推下过滤 + 限制
SELECT * FROM mysql_db.orders 
WHERE order_date >= '2026-05-01' 
LIMIT 1000;

-- ✅ 复杂查询：先聚合再 JOIN
WITH daily_stats AS (
    -- 这个聚合在 MySQL 端执行
    SELECT customer_id, DATE(order_date) AS day, SUM(amount) AS daily_total
    FROM mysql_db.orders
    WHERE order_date >= '2026-04-01'
    GROUP BY customer_id, DATE(order_date)
)
-- 再和本地数据 JOIN
SELECT u.name, d.day, d.daily_total
FROM daily_stats d
JOIN pg_db.public.customers u ON d.customer_id = u.customer_id
ORDER BY d.daily_total DESC
LIMIT 20;

-- MySQL → DuckDB 本地表（一次性快照）
CREATE TABLE local_orders AS 
SELECT * FROM mysql_db.orders 
WHERE order_date >= '2026-01-01';

-- DuckDB → PostgreSQL（写回）
CREATE TABLE pg_db.public.report AS 
SELECT * FROM local_analytics;

-- 同时挂载生产和测试库
ATTACH 'prod_conn' AS prod (TYPE POSTGRES);
ATTACH 'staging_conn' AS staging (TYPE POSTGRES);

-- 对比两个环境的数据差异
SELECT 
    COALESCE(p.order_id, s.order_id) AS order_id,
    p.amount AS prod_amount,
    s.amount AS staging_amount,
    (p.amount - s.amount) AS diff
FROM prod.public.orders p
FULL OUTER JOIN staging.public.orders s 
    ON p.order_id = s.order_id
WHERE p.amount IS DISTINCT FROM s.amount;

# 配合 cron 定时执行，每天自动出报表
import duckdb
import smtplib

con = duckdb.connect()

# ATTACH 各数据源（同上）
con.execute("ATTACH '...' AS mysql_db (TYPE MYSQL)")
con.execute("ATTACH '...' AS pg_db (TYPE POSTGRES)")

# 生成日报
con.execute("""
    COPY (
        -- 跨源查询...
    ) TO '/tmp/daily_report.csv' (HEADER, DELIMITER ',');
""")

# 发送邮件（伪代码，需要配置 SMTP）
# send_email(to='boss@company.com', attachment='/tmp/daily_report.csv')
print("✅ 日报已生成")

# 确保 MySQL 客户端库已安装
apt-get install -y default-libmysqlclient-dev  # Ubuntu/Debian
# 或
brew install mysql-client  # macOS

INSTALL mysql_scanner;
LOAD mysql_scanner;

rm -rf day07_data/