#!/usr/bin/env python3
"""
CDC Streams：用 faust-streaming 把 cdc.mysql.orders 清洗后写到 orders.enriched

清洗规则：
1. 抽 after 字段（Debezium envelope 的精华）
2. 过滤 op == 'd'（删除）
3. 加 city：根据 user_id 简单 mock；生产应查 Redis 维表
4. 时区统一 UTC+8

启动：
    python cdc_streams.py worker  -l info

依赖：pip install faust-streaming  (注意：用 fork 而非已停更的 faust)
"""

import os
import json
import faust

BOOTSTRAP = os.getenv("BOOTSTRAP", "kafka://localhost:9092")
SRC_TOPIC = "cdc.mysql.orders"
DST_TOPIC = "orders.enriched"

app = faust.App(
    "cdc-streams",
    broker=BOOTSTRAP,
    value_serializer="json",     # 简化示例：用 JSON。生产改 Avro
    consumer_max_fetch_size=10485760,
    topic_replication_factor=3,
    processing_guarantee="at_least_once",
    store="memory://",
)

src = app.topic(SRC_TOPIC, value_type=bytes)
dst = app.topic(DST_TOPIC)


CITY_BY_USER = {  # mock 维表
    1001: "beijing", 1002: "shanghai", 1003: "shenzhen",
    1004: "hangzhou", 1005: "chengdu",
}


@app.agent(src)
async def process(stream):
    async for raw in stream:
        try:
            env = json.loads(raw)
        except Exception as e:
            print(f"[streams][skip] decode fail: {e}")
            continue

        op = env.get("op") or env.get("payload", {}).get("op")
        after = env.get("after") or env.get("payload", {}).get("after")
        if op == "d" or after is None:
            continue

        # enrichment
        user_id = after.get("user_id")
        if "city" not in after or after["city"] is None:
            after["city"] = CITY_BY_USER.get(user_id, "unknown")

        # 时区：MySQL 默认 UTC，转成 +08:00（这里只是示例）
        # after["created_at"] 假设是字符串
        out = {
            "order_id":   after["order_id"],
            "user_id":    after["user_id"],
            "amount":     float(after["amount"]),
            "currency":   after.get("currency", "CNY"),
            "status":     after.get("status", "CREATED"),
            "city":       after["city"],
            "items_json": after.get("items_json"),
            "created_at": after.get("created_at"),
            "updated_at": after.get("updated_at"),
        }
        await dst.send(value=out)


if __name__ == "__main__":
    app.main()