import duckdb

con = duckdb.connect(":memory:")

# 模拟脏数据
con.execute("""
    CREATE TABLE customers AS
    SELECT * FROM (VALUES
        ('张三'),
        ('张三（北京分公司）'),
        ('[email protected]'),
        ('Zhang San'),
        ('zhang san'),
        ('李四'),
        ('李四（上海分部）'),
        ('[email protected]'),
        ('王五'),
        ('王五_技术部'),
        ('赵六'),
        ('赵六（广州）')
    ) AS t(name)
""")

# 核心去重逻辑：标准化 → 取前两个有效字符 → 去重
result = con.execute("""
    SELECT
        MIN(name) AS canonical_name,
        COUNT(*) AS duplicate_count,
        ARRAY_AGG(name) AS all_variants
    FROM (
        SELECT
            name,
            LOWER(REGEXP_REPLACE(name, '[^a-z\u4e00-\u9fff\s]', '')) AS normalized,
            CASE
                WHEN REGEXP_MATCHES(LOWER(REGEXP_REPLACE(name, '[^a-z\u4e00-\u9fff\s]', '')), '[\u4e00-\u9fff]')
                THEN LEFT(REGEXP_REPLACE(LOWER(REGEXP_REPLACE(name, '[^a-z\u4e00-\u9fff\s]', '')), '\s+', ''), 2)
                ELSE LEFT(REGEXP_REPLACE(LOWER(REGEXP_REPLACE(name, '[^a-z\u4e00-\u9fff\s]', '')), '\s+', ''), 2)
            END AS fingerprint
        FROM customers
    )
    GROUP BY fingerprint
    HAVING COUNT(*) > 1
""").fetchdf()

print(result)

import duckdb

con = duckdb.connect(":memory:")

# 创建客户公司表
con.execute("""
    CREATE TABLE companies AS
    SELECT * FROM (VALUES
        ('阿里巴巴网络技术有限公司'),
        ('阿里巴巴网络科技'),
        ('阿里网络'),
        ('腾讯科技有限公司'),
        ('腾讯科技'),
        ('字节跳动科技有限公司'),
        ('字节跳动'),
        ('字节'),
        ('美团科技有限公司'),
        ('美团')
    ) AS t(company_name)
""")

# 核心逻辑：找出编辑距离 <= 5 的所有配对
result = con.execute("""
    SELECT
        a.company_name AS company_a,
        b.company_name AS company_b,
        LEVENSHTEIN(a.company_name, b.company_name) AS distance
    FROM companies a
    JOIN companies b ON a.rowid < b.rowid
    WHERE LEVENSHTEIN(a.company_name, b.company_name) <= 5
    ORDER BY distance
""").fetchdf()

print(result)

import duckdb

con = duckdb.connect(":memory:")

# 自定义 N-Gram 函数
con.execute("""
    CREATE OR REPLACE FUNCTION ngrams(text VARCHAR, n INT)
    RETURNS TABLE (gram VARCHAR)
    AS $$
        SELECT SUBSTR(text, i, n)
        FROM generate_series(1, LENGTH(text) - n + 1) AS s(i)
    $$
    LANGUAGE sql IMMUTABLE;
""")

# 计算 Jaccard 相似度（2-gram）
result = con.execute("""
    SELECT
        '阿里巴巴' AS name_a,
        '阿里' AS name_b,
        ROUND(
            (ARRAY_LENGTH(ARRAY_INTERSECT(
                ARRAY_AGG(DISTINCT SUBSTR('阿里巴巴', i, 2)),
                ARRAY_AGG(DISTINCT SUBSTR('阿里', i, 2))
            )))::DOUBLE /
            NULLIF(ARRAY_LENGTH(ARRAY_UNION(
                ARRAY_AGG(DISTINCT SUBSTR('阿里巴巴', i, 2)),
                ARRAY_AGG(DISTINCT SUBSTR('阿里', i, 2))
            )), 0),
            2
        ) AS similarity
    FROM generate_series(1, 4) AS s(i)
""").fetchdf()

print(result)

import duckdb
import pandas as pd

def deduplicate_customers(df: pd.DataFrame, name_col: str, threshold: int = 5) -> pd.DataFrame:
    """
    对任意 DataFrame 进行模糊去重

    Args:
        df: 原始数据
        name_col: 需要去重的列名
        threshold: 编辑距离阈值

    Returns:
        去重后的 DataFrame，每行保留一个代表性记录
    """
    con = duckdb.connect(":memory:")

    # Step 1: 导入并标准化
    con.register('raw_data', df)
    con.execute("""
        CREATE TABLE standardized AS
        SELECT
            name_col,
            LOWER(REGEXP_REPLACE(name_col, r'[^\w\u4e00-\u9fff\s\-（）()]', '')) AS clean_name
        FROM raw_data
    """)

    # Step 2: 编辑距离匹配
    con.execute("""
        CREATE TABLE matches AS
        SELECT
            a.rowid AS row_a,
            b.rowid AS row_b,
            LEVENSHTEIN(a.clean_name, b.clean_name) AS dist
        FROM standardized a
        JOIN standardized b ON a.rowid < b.rowid
        WHERE LEVENSHTEIN(a.clean_name, b.clean_name) <= ?
        ORDER BY dist
    """, [threshold])

    # Step 3: 构建连通分量（传递闭包）
    # 如果 A≈B 且 B≈C，则 A≈C
    con.execute("""
        CREATE TABLE clusters AS
        WITH RECURSIVE connected AS (
            SELECT MIN(row_a) AS root, row_a, row_b FROM matches
            UNION
            SELECT c.root, m.row_a, c.row_b
            FROM connected c
            JOIN matches m ON m.row_b = c.row_a
        )
        SELECT root, MAX(row_b) AS rep_rowid
        FROM connected
        GROUP BY root
    """)

    # Step 4: 获取代表性记录
    result = con.execute("""
        SELECT DISTINCT ON (c.root)
            r.*
        FROM raw_data r
        JOIN clusters c ON r.rowid = c.rep_rowid
        ORDER BY c.root, r.rowid
    """).fetchdf()

    con.close()
    return result

# 使用示例
df = pd.DataFrame({
    'name': ['张三', '张三（北京）', 'zhangsan', '李四', '李思', '王五'],
    'email': ['[email protected]', '[email protected]', '[email protected]', '[email protected]', '[email protected]', '[email protected]'],
    'phone': ['138001', '138001', '138001', '139001', '139002', '137001']
})

deduped = deduplicate_customers(df, 'name', threshold=3)
print(deduped)