II qism loyihasi: uchtasini bitta quvurga yig'ish

diagramma(
    {
        "Postgres\n(RostUy)": (0, 1.8, 2.0, 1.1, "manba"),
        "CDC\n(WAL slot)": (2.8, 1.8, 1.9, 1.1, "qayta"),
        "Kafka\n(jurnal)": (5.4, 1.8, 1.8, 1.1, "qayta"),
        "Sifat\n+ karantin": (8.0, 1.8, 2.0, 1.1, "qayta"),
        "Karantin\nfayli": (8.0, 0.1, 2.0, 0.8, "kuzatuv"),
        "Iceberg\nbronza": (10.8, 3.2, 1.9, 0.9, "saqlash"),
        "Iceberg\nkumush": (10.8, 1.9, 1.9, 0.9, "saqlash"),
        "Oltin\n(panel)": (10.8, 0.6, 1.9, 0.9, "xizmat"),
    },
    [
        ("Postgres\n(RostUy)", "CDC\n(WAL slot)"),
        ("CDC\n(WAL slot)", "Kafka\n(jurnal)"),
        ("Kafka\n(jurnal)", "Sifat\n+ karantin"),
        ("Sifat\n+ karantin", "Karantin\nfayli", "yomon"),
        ("Sifat\n+ karantin", "Iceberg\nbronza", "toza"),
        ("Iceberg\nbronza", "Iceberg\nkumush"),
        ("Iceberg\nkumush", "Oltin\n(panel)"),
    ],
    nom="de_loyiha.png", kenglik=14,
    izoh="II qismning barcha boblari bitta quvurda. Har quti - bitta bob.",
)

docker rm -f kitob-pg kitob-kafka > /dev/null 2>&1
docker run -d --name kitob-pg -p 15455:5432 \
  -e POSTGRES_PASSWORD=kitob -e POSTGRES_DB=rostuy \
  postgres:16 -c wal_level=logical -c max_replication_slots=4 -c max_wal_senders=4 > /dev/null
docker run -d --name kitob-kafka -p 19092:9092 \
  -e KAFKA_NODE_ID=1 -e KAFKA_PROCESS_ROLES=broker,controller \
  -e KAFKA_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 \
  -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:19092 \
  -e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \
  -e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \
  -e KAFKA_CONTROLLER_QUORUM_VOTERS=1@localhost:9093 \
  -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
  -e KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1 \
  -e KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1 \
  -e KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0 \
  apache/kafka:3.9.0 > /dev/null

for i in $(seq 1 60); do
  docker exec kitob-pg pg_isready -U postgres > /dev/null 2>&1 && { echo "postgres ${i}s da tayyor"; break; }
  sleep 1
done
for i in $(seq 1 60); do
  python3 -c "from kafka import KafkaAdminClient; KafkaAdminClient(bootstrap_servers='localhost:19092')" > /dev/null 2>&1 \
    && { echo "kafka ${i}s da tayyor"; break; }
  sleep 1
done

"""II qism loyihasi: CDC -> Kafka -> sifat -> Iceberg -> oltin."""
import json
from typing import Any, Iterator

import pandas as pd

USTUNLAR = ["elon_id", "tuman", "narx_mln", "xonalar", "holat"]

# 6-bob: chegaralar. 20-bob: buzilgani karantinga.
QOIDA = {
    "narx_mln": lambda s: s.between(0.5, 200),
    "xonalar": lambda s: s.between(1, 10),
    "tuman": lambda s: s.notna() & (s.str.len() > 2),
}


def hodisadan_qator(hodisa: dict[str, Any]) -> dict[str, Any] | None:
    """CDC hodisasini jadval qatoriga aylantiradi. DELETE -> holat='ochirildi'."""
    q = hodisa["qiymat"]
    if hodisa["amal"] == "DELETE":
        return {"elon_id": q.get("elon_id"), "tuman": None, "narx_mln": None,
                "xonalar": None, "holat": "ochirildi"}
    return {
        "elon_id": q.get("elon_id"),
        "tuman": q.get("tuman"),
        "narx_mln": float(q["narx_mln"]) if q.get("narx_mln") else None,
        "xonalar": int(q["xonalar"]) if q.get("xonalar") else None,
        "holat": q.get("holat") or "faol",
    }


def sifat_ajrat(d: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]:
    """Toza va karantin qatorlarga ajratadi. O'chirilganlar tekshirilmaydi."""
    ochirilgan = d["holat"] == "ochirildi"
    yaxshi = pd.Series(True, index=d.index)
    for ustun, tekshir in QOIDA.items():
        yaxshi &= tekshir(d[ustun]) | ochirilgan
    return d[yaxshi], d[~yaxshi]


def oltin_qatlam(kumush: pd.DataFrame) -> pd.DataFrame:
    """Panel uchun oldindan hisoblangan javob (19-bob)."""
    faol = kumush[kumush["holat"] != "ochirildi"]
    return (faol.groupby("tuman")
                .agg(elonlar=("elon_id", "count"),
                     ort_narx=("narx_mln", "mean"))
                .round(3)
                .reset_index()
                .sort_values("ort_narx", ascending=False))

python3 -c "
import psycopg2, pandas as pd

c = psycopg2.connect(host='localhost', port=15455, user='postgres', password='kitob', dbname='rostuy')
c.autocommit = True
k = c.cursor()
k.execute('''DROP TABLE IF EXISTS elon;
CREATE TABLE elon (elon_id text PRIMARY KEY, tuman text, narx_mln numeric,
                   xonalar int, holat text DEFAULT 'faol');''')

k.execute(\"SELECT count(*) FROM pg_replication_slots WHERE slot_name='loyiha_slot'\")
if k.fetchone()[0]:
    k.execute(\"SELECT pg_drop_replication_slot('loyiha_slot')\")
k.execute(\"SELECT pg_create_logical_replication_slot('loyiha_slot','test_decoding')\")
print('1. slot ochildi (snapshotdan OLDIN)')

d = pd.read_csv('../datasets/ijara.csv').drop_duplicates('elon_id').head(4000)
k.executemany('INSERT INTO elon (elon_id,tuman,narx_mln,xonalar) VALUES (%s,%s,%s,%s)',
              [(r.elon_id, r.tuman, float(r.narx_mln), int(r.xonalar)) for r in d.itertuples()])
print(f'2. snapshot yuklandi: {k.execute(\"SELECT count(*) FROM elon\") or c.cursor()}', end='')
k.execute('SELECT count(*) FROM elon')
print(f' {k.fetchone()[0]:,} qator')
c.close()
"

python3 -c "
import psycopg2
c = psycopg2.connect(host='localhost', port=15455, user='postgres', password='kitob', dbname='rostuy')
c.autocommit = True
k = c.cursor()

k.execute(\"UPDATE elon SET narx_mln = narx_mln * 1.15 WHERE tuman = 'Yunusobod'\")
n1 = k.rowcount
k.execute(\"DELETE FROM elon WHERE elon_id IN (SELECT elon_id FROM elon ORDER BY elon_id LIMIT 120)\")
n2 = k.rowcount
k.execute(\"\"\"INSERT INTO elon (elon_id,tuman,narx_mln,xonalar)
              SELECT 'YANGI'||g, 'Chilonzor', 6.5, 2 FROM generate_series(1,80) g\"\"\")
n3 = k.rowcount
# ataylab buzuq: manfiy narx va imkonsiz xona soni
k.execute(\"UPDATE elon SET narx_mln = -3 WHERE elon_id LIKE 'YANGI1%'\")
n4 = k.rowcount
k.execute(\"UPDATE elon SET xonalar = 99 WHERE elon_id LIKE 'YANGI2%'\")
n5 = k.rowcount

print(f'narx oshirildi : {n1:,}')
print(f'ochirildi      : {n2:,}')
print(f'yangi qoshildi : {n3:,}')
print(f'BUZUQ (manfiy narx): {n4}')
print(f'BUZUQ (99 xona)    : {n5}')
print(f'jami ozgarish  : {n1+n2+n3+n4+n5:,}')
c.close()
"

python3 -c "
import sys, collections; sys.path.insert(0, 'src')
import psycopg2
from kafka import KafkaProducer
from kafka.admin import KafkaAdminClient, NewTopic
from quvur.cdc import oqim, kafka_xabari
from quvur.chidamlilik import qayta_urin

a = KafkaAdminClient(bootstrap_servers='localhost:19092')
if 'elon-cdc' not in a.list_topics():
    a.create_topics([NewTopic(name='elon-cdc', num_partitions=3, replication_factor=1)])

c = psycopg2.connect(host='localhost', port=15455, user='postgres', password='kitob', dbname='rostuy')
c.autocommit = True
k = c.cursor()

p = qayta_urin(lambda: KafkaProducer(bootstrap_servers='localhost:19092'), urinish=3)
tur = collections.Counter()
for h in oqim(k, 'loyiha_slot'):
    kalit, qiymat = kafka_xabari(h)
    p.send('elon-cdc', key=kalit.encode(), value=qiymat)
    tur[h['amal']] += 1
p.flush()
print('Kafka ga yuborildi:', dict(tur))
print('jami:', sum(tur.values()))
c.close()
" 2>&1 | grep -v Warning

python3 -c "
import sys, json; sys.path.insert(0, 'src')
import pandas as pd
from kafka import KafkaConsumer
from quvur.loyiha import hodisadan_qator, sifat_ajrat

c = KafkaConsumer('elon-cdc', bootstrap_servers='localhost:19092',
                  auto_offset_reset='earliest', group_id='loyiha',
                  consumer_timeout_ms=8000)
qatorlar = []
for m in c:
    q = hodisadan_qator(json.loads(m.value))
    if q and q['elon_id']:
        qatorlar.append(q)
c.close()

d = pd.DataFrame(qatorlar)
toza, karantin = sifat_ajrat(d)
toza.to_parquet('bronza.parquet', index=False)
karantin.to_parquet('karantin.parquet', index=False)

ulush = len(karantin) / len(d) * 100
print(f'Kafka dan oqildi : {len(d):,} hodisa')
print(f'toza             : {len(toza):,}')
print(f'karantin         : {len(karantin):,} ({ulush:.2f}%)')
print()
print(f'chegara 5% -> {\"DAVOM ETADI\" if ulush < 5 else \"TOXTAYDI\"}')
if len(karantin):
    print()
    print('karantin namunasi:')
    print(karantin.head(3).to_string(index=False))
" 2>&1 | grep -v Warning

python3 -c "
import sys; sys.path.insert(0, 'src')
import pandas as pd
from quvur.loyiha import oltin_qatlam

bronza = pd.read_parquet('bronza.parquet')

# KUMUSH: bir elon_id bo'yicha OXIRGI holat (idempotent - 20-bob)
kumush = bronza.drop_duplicates('elon_id', keep='last')
kumush.to_parquet('kumush.parquet', index=False)

oltin = oltin_qatlam(kumush)
oltin.to_parquet('oltin.parquet', index=False)

print(f'bronza (hamma hodisa) : {len(bronza):,}')
print(f'kumush (oxirgi holat) : {len(kumush):,}')
print(f'oltin  (panel javobi) : {len(oltin):,}')
print()
ochirilgan = (kumush['holat'] == 'ochirildi').sum()
print(f'ochirilgan elonlar    : {ochirilgan:,}  (panelda KORINMAYDI)')
print(f'faol elonlar          : {len(kumush) - ochirilgan:,}')
print()
print(oltin.head(5).to_string(index=False))
"

python3 -c "
import sys, json; sys.path.insert(0, 'src')
import pandas as pd
from kafka import KafkaConsumer
from quvur.loyiha import hodisadan_qator, sifat_ajrat, oltin_qatlam

oldingi = pd.read_parquet('oltin.parquet')

# butun quvurni BOSHIDAN qayta ishga tushiramiz (yangi group_id = hammasini qayta oqiydi)
c = KafkaConsumer('elon-cdc', bootstrap_servers='localhost:19092',
                  auto_offset_reset='earliest', group_id='loyiha-qayta',
                  consumer_timeout_ms=8000)
qatorlar = [hodisadan_qator(json.loads(m.value)) for m in c]
c.close()
d = pd.DataFrame([q for q in qatorlar if q and q['elon_id']])
toza, _ = sifat_ajrat(d)
kumush = toza.drop_duplicates('elon_id', keep='last')
yangi = oltin_qatlam(kumush)

print(f'1-ishga tushirish: {len(oldingi)} qator')
print(f'2-ishga tushirish: {len(yangi)} qator')
print()
bir_xil = oldingi.reset_index(drop=True).equals(yangi.reset_index(drop=True))
print(f'natijalar AYNAN bir xilmi: {bir_xil}')
print()
if bir_xil:
    print('Quvur IDEMPOTENT: qayta ishga tushirish xavfsiz.')
else:
    print('FARQ BOR:')
    print(oldingi.head(3).to_string(index=False))
    print(yangi.head(3).to_string(index=False))
" 2>&1 | grep -v Warning

import pandas as pd
show(pd.DataFrame([
    {"Talab": "O'zgarish darhol ko'rinadi", "Qanday": "CDC WAL dan o'qiydi, polling emas",
     "Bob": "11"},
    {"Talab": "Hech narsa yo'qolmaydi", "Qanday": "Slot snapshotdan oldin ochildi",
     "Bob": "11"},
    {"Talab": "O'chirilgan e'lon panelda yo'q", "Qanday": "DELETE hodisasi holat='ochirildi'",
     "Bob": "11, 17"},
    {"Talab": "Buzuq ma'lumot tushmaydi", "Qanday": "Sifat tekshiruvi + karantin",
     "Bob": "6, 20"},
    {"Talab": "Quvur to'xtamaydi", "Qanday": "Karantin chegarasi 5%",
     "Bob": "20"},
    {"Talab": "Panel tez", "Qanday": "Oltin qatlam oldindan hisoblangan",
     "Bob": "19"},
    {"Talab": "Qayta ishga tushirish xavfsiz", "Qanday": "drop_duplicates(keep='last')",
     "Bob": "20"},
]))

docker rm -f kitob-pg kitob-kafka > /dev/null 2>&1 && echo "konteynerlar ochirildi"
rm -f bronza.parquet kumush.parquet oltin.parquet karantin.parquet
echo "vaqtinchalik fayllar tozalandi"
docker ps --format '{{.Names}}'