#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
第 19 章 综合实战：大屏查询 Demo
-------------------------------
跑 4 大业务查询并打印结果，验证：
  ① 实时 PV / UV / Top10 商品
  ② 渠道 × 国家 维度 Cube
  ③ 最近 1 小时热门商品
  ④ 用户行为路径（随机取 1 个用户）
"""
import argparse
import time
import clickhouse_connect


def header(title: str) -> None:
    bar = '=' * 68
    print(f'\n{bar}\n  {title}\n{bar}')


def timed(label: str, fn):
    t0 = time.time()
    rs = fn()
    dt = (time.time() - t0) * 1000
    print(f'[{label}]  {dt:7.1f} ms  rows={len(rs.result_rows)}')
    return rs


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument('--host', default='127.0.0.1')
    ap.add_argument('--port', default=8123, type=int)
    ap.add_argument('--user', default='default')
    ap.add_argument('--password', default='')
    ap.add_argument('--db', default='learn_ck')
    args = ap.parse_args()

    cli = clickhouse_connect.get_client(
        host=args.host, port=args.port,
        username=args.user, password=args.password,
        database=args.db,
    )

    header('① 今日 PV / UV（扫聚合表 events_agg_pv_uv）')
    rs = timed('pv_uv', lambda: cli.query("""
        SELECT
            event_date,
            sumMerge(pv)        AS pv,
            uniqExactMerge(uv)  AS uv
        FROM events_agg_pv_uv
        WHERE event_date >= today() - 2 AND event_type = 'view'
        GROUP BY event_date
        ORDER BY event_date DESC
    """))
    for r in rs.result_rows:
        print(f'  {r[0]}  pv={r[1]:>10}  uv={r[2]:>8}')

    header('② Top10 商品（走字典映射 + 聚合表）')
    rs = timed('top10', lambda: cli.query("""
        SELECT
            product_id,
            dictGet('learn_ck.dict_products', 'name',     product_id) AS pname,
            dictGet('learn_ck.dict_products', 'category', product_id) AS cat,
            sumMerge(pv)       AS pv,
            uniqExactMerge(uv) AS uv
        FROM events_agg_pv_uv
        WHERE event_date >= today() - 2 AND event_type = 'view'
        GROUP BY product_id
        ORDER BY pv DESC
        LIMIT 10
    """))
    print(f"  {'pid':>4}  {'name':<14}  {'cat':<8}  {'pv':>8}  {'uv':>6}")
    for r in rs.result_rows:
        print(f'  {r[0]:>4}  {r[1]:<14}  {r[2]:<8}  {r[3]:>8}  {r[4]:>6}')

    header('③ 渠道 × 国家 Cube（WITH ROLLUP）')
    rs = timed('cube', lambda: cli.query("""
        SELECT
            channel,
            country,
            sumMerge(pv)       AS pv,
            uniqExactMerge(uv) AS uv
        FROM events_agg_pv_uv
        WHERE event_date >= today() - 2 AND event_type = 'view'
        GROUP BY channel, country WITH ROLLUP
        ORDER BY pv DESC
        LIMIT 20
    """))
    print(f"  {'channel':<8}  {'country':<8}  {'pv':>10}  {'uv':>8}")
    for r in rs.result_rows:
        ch = r[0] if r[0] else '<ALL>'
        co = r[1] if r[1] else '<ALL>'
        print(f'  {ch:<8}  {co:<8}  {r[2]:>10}  {r[3]:>8}')

    header('④ 用户行为路径（随机取 1 个用户）')
    uid = cli.query("""
        SELECT any(user_id)
        FROM events_raw
        WHERE event_date = today()
        LIMIT 1
    """).result_rows
    if not uid or uid[0][0] is None:
        # 兜底：从最近几天任意挑一个
        uid = cli.query("""
            SELECT any(user_id)
            FROM events_raw
            WHERE event_date >= today() - 2
        """).result_rows
    user_id = uid[0][0] if uid and uid[0][0] is not None else 1
    print(f'  sampled user_id = {user_id}')

    rs = timed('path', lambda: cli.query(f"""
        SELECT
            event_time,
            event_type,
            page_url,
            product_id
        FROM events_raw
        WHERE user_id = {user_id}
          AND event_date >= today() - 2
        ORDER BY event_time
        LIMIT 30
    """))
    for r in rs.result_rows:
        print(f'  {r[0]}  {r[1]:<10}  pid={r[3]:>3}  {r[2]}')

    header('Done.')


if __name__ == '__main__':
    main()