Backend: hodisalar bazasi, panel va Telegram signal

kameralar ──RTSP──> tahlilchi (YOLO + ByteTrack + qoidalar) ──HTTP API──> Django ──> Postgres
                                                                              │
                                    signal_yubor (alohida jarayon) <──────────┤──> panel (Django admin)
                                          │
                                          └──> Telegram

sudo apt-get update && sudo apt-get install -y ca-certificates curl
sudo install -m 0755 -d /etc/apt/keyrings
sudo curl -fsSL https://download.docker.com/linux/ubuntu/gpg -o /etc/apt/keyrings/docker.asc
echo "deb [arch=$(dpkg --print-architecture) signed-by=/etc/apt/keyrings/docker.asc] https://download.docker.com/linux/ubuntu \
  $(. /etc/os-release && echo "$VERSION_CODENAME") stable" | sudo tee /etc/apt/sources.list.d/docker.list
sudo apt-get update && sudo apt-get install -y docker-ce docker-ce-cli containerd.io docker-compose-plugin
sudo usermod -aG docker $USER            # docker ni sudo siz ishlatish uchun (keyin qayta kiring)
docker run --rm hello-world              # tekshiruv

export POSTGRES_PASSWORD="$(python3 -c 'import secrets; print(secrets.token_urlsafe(24))')"
docker run -d --rm --name yolo-kitob-pg \
  -e POSTGRES_PASSWORD -e POSTGRES_USER=kitob -e POSTGRES_DB=kuzatuv \
  -p 127.0.0.1:25432:5432 \
  postgres:16-alpine

# Xizmatlar: Postgres konteyneri, fon jarayonlari, RTSP server va soxta kameralar. Hammasi `with` bilan — xato bo'lsa ham yopiladi.
import contextlib
import os
import secrets
import shutil
import subprocess
import sys
import time
from pathlib import Path

import psycopg


def docker_buyrugi():
    """Linux serverda — docker. WSL da Docker Desktop integratsiyasi yoqilmagan bo'lsa — Windows dagi docker.exe."""
    for nom in ("docker", "docker.exe"):
        if shutil.which(nom) and subprocess.run([nom, "version"], capture_output=True).returncode == 0:
            return nom
    raise RuntimeError("Docker topilmadi: Docker Engine yoki Docker Desktop o'rnating (20-bob)")


def xost_manzili(docker):
    """Konteyner porti qaysi manzilda ochiladi. docker.exe (WSL, NAT) da Windows ning 127.0.0.1 i WSL dan ko'rinmaydi —
    WSL virtual tarmog'idagi Windows manzili olinadi: u faqat shu kompyuter ichida, tashqi tarmoqqa ochiq emas."""
    if docker != "docker.exe":
        return "127.0.0.1"
    return subprocess.run(["ip", "route", "show", "default"], capture_output=True, text=True).stdout.split()[2]


@contextlib.contextmanager
def postgres(port=25432, nom="yolo-kitob-pg"):
    """Vaqtinchalik Postgres 16 konteyneri: tasodifiy parol, ish tugagach o'chiriladi (--rm)."""
    docker = docker_buyrugi()
    xost, parol = xost_manzili(docker), secrets.token_urlsafe(24)
    subprocess.run([docker, "rm", "-f", nom], capture_output=True)       # oldingi yiqilgan ishdan qolgan bo'lsa
    r = subprocess.run([docker, "run", "-d", "--rm", "--name", nom, "-e", "POSTGRES_PASSWORD", "-e", "POSTGRES_USER=kitob",
                        "-e", "POSTGRES_DB=kuzatuv", "-p", f"{xost}:{port}:5432", "postgres:16-alpine"],
                       env={**os.environ, "POSTGRES_PASSWORD": parol}, capture_output=True, text=True)   # parol buyruq qatorida emas
    if r.returncode != 0:
        raise RuntimeError(f"docker run xato: {r.stderr.strip()}")
    try:
        for _ in range(60):
            try:
                psycopg.connect(host=xost, port=port, user="kitob", password=parol, dbname="kuzatuv", connect_timeout=2).close()
                break
            except psycopg.OperationalError:
                time.sleep(1)
        else:
            raise RuntimeError("Postgres 60 s ichida tayyor bo'lmadi")
        yield {"PG_HOST": xost, "PG_PORT": str(port), "PG_PAROL": parol}
    finally:
        subprocess.run([docker, "stop", nom], capture_output=True)


@contextlib.contextmanager
def jarayon(buyruq, env=None, **kw):
    """Fon jarayoni: blokdan chiqishda (xato bo'lsa ham) to'xtatiladi."""
    p = subprocess.Popen(buyruq, env={**os.environ, **(env or {})}, **kw)
    try:
        yield p
    finally:
        p.terminate()
        try:
            p.wait(timeout=10)
        except subprocess.TimeoutExpired:
            p.kill()


def rtsp_sozlama(fayl, kamera_parol, tahlil_parol, manzil="127.0.0.1:8554"):
    """19-bobdagi xavfsiz MediaMTX sozlamasi: bitta manzil, faqat TCP, faqat RTSP, ikki foydalanuvchi."""
    Path(fayl).parent.mkdir(parents=True, exist_ok=True)
    Path(fayl).write_text(f"""logLevel: error
rtsp: true
rtspTransports: [tcp]
rtspAddress: {manzil}
rtmp: false
hls: false
webrtc: false
srt: false
moq: false
api: false
metrics: false
pprof: false
playback: false
authInternalUsers:
  - user: kamera
    pass: {kamera_parol}
    ips: ["127.0.0.1"]
    permissions:
      - action: publish
  - user: tahlil
    pass: {tahlil_parol}
    ips: ["127.0.0.1"]
    permissions:
      - action: read
paths:
  all_others:
""")


@contextlib.contextmanager
def rtsp(soni=1, kenglik=1920, manzil="127.0.0.1:8554", **kamera_sozlama):
    """MediaMTX va `soni` ta soxta kamera (19-bob). Qaytaradi: o'qish manzilining asosi va kamera jarayoni.
    kamera_sozlama — soxta kameraning qo'shimcha parametrlari, masalan video="...", fps=30, bitreyt="8000k"."""
    kp, tp, sozlama = secrets.token_urlsafe(16), secrets.token_urlsafe(16), Path("out/rtsp/mediamtx.yml")
    rtsp_sozlama(sozlama, kp, tp, manzil)
    kamera_sozlama.setdefault("bitreyt", "4000k" if kenglik >= 1920 else "512k")
    qoshimcha = []
    for k, v in kamera_sozlama.items():                                   # bir_marta=True -> --bir-marta
        qoshimcha += [f"--{k.replace('_', '-')}"] + ([] if v is True else [str(v)])
    try:
        with jarayon(["tools/mediamtx/mediamtx", str(sozlama)]):
            time.sleep(1)
            with jarayon([sys.executable, "kod/19_soxta_kamera.py", "--url", f"rtsp://kamera:{kp}@{manzil}/", "--soni", str(soni),
                          "--kenglik", str(kenglik), *qoshimcha]) as kamera:
                time.sleep(3)
                yield {"asos": f"rtsp://tahlil:{tp}@{manzil}/", "kamera": kamera}
    finally:
        sozlama.unlink(missing_ok=True)                                     # parollar yozilgan fayl qolmaydi

import json
import os
import secrets
import signal
import subprocess
import sys
import time
import urllib.request
from pathlib import Path

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import psycopg

sys.path.insert(0, "kod")
from tizim import docker_buyrugi, jarayon, postgres, rtsp, xost_manzili

docker = docker_buyrugi()
print("Docker buyrug'i:", docker, "| port ochiladigan manzil:", xost_manzili(docker))
print(subprocess.run([docker, "version", "--format", "Docker: klient {{.Client.Version}}, server {{.Server.Version}}"],
                     capture_output=True, text=True).stdout.strip())
with postgres() as pg:
    with psycopg.connect(host=pg["PG_HOST"], port=pg["PG_PORT"], user="kitob", password=pg["PG_PAROL"], dbname="kuzatuv") as db:
        print(db.execute("select current_user, current_database(), split_part(version(), ' on ', 1)").fetchone())
for _ in range(20):                                                       # --rm o'chirishni bir oz keyin tugatadi
    nomlar = subprocess.run([docker, "ps", "-a", "--format", "{{.Names}}"], capture_output=True, text=True).stdout.split()
    if "yolo-kitob-pg" not in nomlar:
        break
    time.sleep(0.5)
print("konteyner o'chirildimi:", "yolo-kitob-pg" not in nomlar)

for papka in ("loyiha/kuzatuv", "loyiha/hodisalar", "loyiha/hodisalar/migrations", "loyiha/hodisalar/management",
              "loyiha/hodisalar/management/commands"):
    Path(papka).mkdir(parents=True, exist_ok=True)
    (Path(papka) / "__init__.py").touch()
print(sorted(str(p) for p in Path("loyiha").rglob("*.py") if "migrations/0" not in str(p)))

import os
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent.parent
SECRET_KEY = os.environ["DJANGO_SECRET_KEY"]                            # bo'lmasa — ishga tushmaydi
DEBUG = os.environ.get("DJANGO_DEBUG") == "1"
ALLOWED_HOSTS = os.environ.get("DJANGO_ALLOWED_HOSTS", "127.0.0.1,localhost").split(",")

INSTALLED_APPS = [
    "django.contrib.admin",
    "django.contrib.auth",
    "django.contrib.contenttypes",
    "django.contrib.sessions",
    "django.contrib.messages",
    "django.contrib.staticfiles",
    "hodisalar",
]
MIDDLEWARE = [
    "django.middleware.security.SecurityMiddleware",
    "django.contrib.sessions.middleware.SessionMiddleware",
    "django.middleware.common.CommonMiddleware",
    "django.middleware.csrf.CsrfViewMiddleware",
    "django.contrib.auth.middleware.AuthenticationMiddleware",
    "django.contrib.messages.middleware.MessageMiddleware",
    "django.middleware.clickjacking.XFrameOptionsMiddleware",
]
ROOT_URLCONF = "kuzatuv.urls"
TEMPLATES = [{
    "BACKEND": "django.template.backends.django.DjangoTemplates",
    "APP_DIRS": True,
    "OPTIONS": {"context_processors": ["django.template.context_processors.request", "django.contrib.auth.context_processors.auth",
                                       "django.contrib.messages.context_processors.messages"]},
}]
WSGI_APPLICATION = "kuzatuv.wsgi.application"

DATABASES = {"default": {
    "ENGINE": "django.db.backends.postgresql",
    "HOST": os.environ.get("PG_HOST", "127.0.0.1"),
    "PORT": os.environ.get("PG_PORT", "5432"),
    "NAME": os.environ.get("PG_BAZA", "kuzatuv"),
    "USER": os.environ.get("PG_FOYDALANUVCHI", "kitob"),
    "PASSWORD": os.environ["PG_PAROL"],
    "CONN_MAX_AGE": 60,
}}
AUTH_PASSWORD_VALIDATORS = [{"NAME": f"django.contrib.auth.password_validation.{n}"}
                            for n in ("UserAttributeSimilarityValidator", "MinimumLengthValidator", "CommonPasswordValidator",
                                      "NumericPasswordValidator")]
LANGUAGE_CODE = "uz"
TIME_ZONE = "Asia/Tashkent"
USE_I18N = True
USE_TZ = True
STATIC_URL = "static/"
STATIC_ROOT = BASE_DIR / "static"
DEFAULT_AUTO_FIELD = "django.db.models.BigAutoField"
DATA_UPLOAD_MAX_MEMORY_SIZE = 64 * 1024                                 # hodisa — kichik JSON; katta so'rovlar rad etiladi

if os.environ.get("DJANGO_HTTPS") == "1":                              # HTTPS orqasida (21-bob)
    SESSION_COOKIE_SECURE = CSRF_COOKIE_SECURE = True
    SECURE_HSTS_SECONDS = 31_536_000
    SECURE_SSL_REDIRECT = True
    SECURE_REDIRECT_EXEMPT = [r"^api/"]                                # ishchi ichki tarmoqdan HTTP bilan keladi; tashqaridan — nginx, HTTPS
    SECURE_PROXY_SSL_HEADER = ("HTTP_X_FORWARDED_PROTO", "https")

ISHCHI_TOKENLAR = [t for t in os.environ.get("ISHCHI_TOKENLAR", "").split(",") if t]
API_DAQIQAGA = int(os.environ.get("API_DAQIQAGA", "600"))              # bitta token uchun daqiqasiga so'rovlar chegarasi
TELEGRAM_TOKEN = os.environ.get("TELEGRAM_TOKEN", "")
TELEGRAM_CHAT = os.environ.get("TELEGRAM_CHAT", "")
TELEGRAM_API = os.environ.get("TELEGRAM_API", "https://api.telegram.org")
SIGNAL_TURLARI = ["kaskasiz", "zona", "kamera_jim", "kamera_tiklandi"]  # qolganlari (masalan, chiziq) — faqat statistika

from django.contrib import admin
from django.urls import path

from hodisalar import views

urlpatterns = [
    path("panel/", admin.site.urls),                                    # "admin/" emas: taxmin qilinadigan manzil kamroq hujum oladi
    path("api/hodisalar/", views.hodisa_qabul),
    path("api/salomatlik/", views.salomatlik),
]

import os

from django.core.wsgi import get_wsgi_application

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "kuzatuv.settings")
application = get_wsgi_application()

import os
import sys

if __name__ == "__main__":
    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "kuzatuv.settings")
    from django.core.management import execute_from_command_line

    execute_from_command_line(sys.argv)

from django.apps import AppConfig


class HodisalarConfig(AppConfig):
    name = "hodisalar"
    verbose_name = "Kuzatuv hodisalari"

from django.db import models


class Kamera(models.Model):
    nom = models.CharField("nomi", max_length=64, unique=True)
    joy = models.CharField("joylashuvi", max_length=200, blank=True)

    class Meta:
        verbose_name_plural = "kameralar"

    def __str__(self):
        return self.nom


class Hodisa(models.Model):
    class Tur(models.TextChoices):
        CHIZIQ = "chiziq", "chiziqni kesib o'tdi"
        ZONA = "zona", "taqiqlangan zonaga kirdi"
        KASKASIZ = "kaskasiz", "kaskasiz odam"
        KAMERA_JIM = "kamera_jim", "kamera jim"
        KAMERA_TIKLANDI = "kamera_tiklandi", "kamera tiklandi"

    class Signal(models.TextChoices):
        KUTMOQDA = "kutmoqda", "kutmoqda"
        YUBORILDI = "yuborildi", "yuborildi"
        XATO = "xato", "yuborilmadi"
        KERAK_EMAS = "kerak_emas", "signal kerak emas"

    kamera = models.ForeignKey(Kamera, on_delete=models.PROTECT, related_name="hodisalar")
    tur = models.CharField(max_length=20, choices=Tur.choices)
    vaqt = models.DateTimeField("kadr vaqti")
    trek = models.IntegerField(null=True, blank=True)
    yonalish = models.CharField(max_length=20, blank=True)
    malumot = models.JSONField(default=dict, blank=True)
    kalit = models.CharField(max_length=100, unique=True)                  # ishchi bir hodisani qayta yuborsa ham bitta yozuv
    yaratildi = models.DateTimeField(auto_now_add=True)
    signal = models.CharField(max_length=12, choices=Signal.choices, default=Signal.KUTMOQDA)
    urinishlar = models.PositiveSmallIntegerField(default=0)
    keyingi_urinish = models.DateTimeField(null=True, blank=True)
    yuborildi = models.DateTimeField(null=True, blank=True)

    class Meta:
        verbose_name_plural = "hodisalar"
        ordering = ["-vaqt"]
        indexes = [models.Index(fields=["kamera", "vaqt"]), models.Index(fields=["signal", "keyingi_urinish"])]

    def __str__(self):
        return f"{self.kamera} · {self.get_tur_display()} · {self.vaqt:%Y-%m-%d %H:%M:%S}"

from django.contrib import admin

from .models import Hodisa, Kamera


@admin.register(Kamera)
class KameraAdmin(admin.ModelAdmin):
    list_display = ["nom", "joy"]


@admin.register(Hodisa)
class HodisaAdmin(admin.ModelAdmin):
    list_display = ["vaqt", "kamera", "tur", "trek", "yonalish", "signal"]
    list_filter = ["tur", "kamera", "signal"]
    date_hierarchy = "vaqt"
    readonly_fields = [f.name for f in Hodisa._meta.fields]

    def has_add_permission(self, request):                               # hodisalarni faqat ishchi yozadi
        return False

    def has_change_permission(self, request, obj=None):                  # tarix o'zgarmaydi
        return False

import hmac
import json
import threading
import time

from django import forms
from django.conf import settings
from django.db import IntegrityError, transaction
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_GET, require_POST

from .models import Hodisa, Kamera


def javob(ok, data=None, xato=None, status=200):
    """Hamma javoblar bir xil ko'rinishda: {"ok", "data", "xato"}."""
    return JsonResponse({"ok": ok, "data": data, "xato": xato}, status=status)


class HodisaForma(forms.Form):
    kamera = forms.SlugField(max_length=64)
    tur = forms.ChoiceField(choices=Hodisa.Tur.choices)
    vaqt = forms.DateTimeField()
    trek = forms.IntegerField(required=False, min_value=0)
    yonalish = forms.CharField(required=False, max_length=20)
    malumot = forms.JSONField(required=False)
    kalit = forms.CharField(max_length=100)


_hisob, _qulf = {}, threading.Lock()


def chegaradan_oshdimi(token):
    """Bitta token daqiqasiga API_DAQIQAGA tadan ko'p so'rov yubora olmaydi. Hisob — shu jarayon xotirasida."""
    daqiqa = int(time.time() // 60)
    with _qulf:
        kalit = (token, daqiqa)
        _hisob[kalit] = _hisob.get(kalit, 0) + 1
        for eski in [k for k in _hisob if k[1] < daqiqa]:
            del _hisob[eski]
        return _hisob[kalit] > settings.API_DAQIQAGA


def token_ol(request):
    token = request.headers.get("Authorization", "").removeprefix("Bearer ").strip()
    if token and any(hmac.compare_digest(token, t) for t in settings.ISHCHI_TOKENLAR):
        return token
    return None


@csrf_exempt                                                             # cookie emas, token — CSRF xavfi yo'q
@require_POST
def hodisa_qabul(request):
    token = token_ol(request)
    if token is None:
        return javob(False, xato="token noto'g'ri yoki yo'q", status=401)
    if chegaradan_oshdimi(token):
        return javob(False, xato="so'rovlar juda ko'p", status=429)
    try:
        forma = HodisaForma(json.loads(request.body))
    except (ValueError, TypeError):
        return javob(False, xato="JSON emas", status=400)
    if not forma.is_valid():
        return javob(False, xato=forma.errors.get_json_data(), status=400)
    d = forma.cleaned_data
    kamera, _ = Kamera.objects.get_or_create(nom=d["kamera"])
    signal = Hodisa.Signal.KUTMOQDA if d["tur"] in settings.SIGNAL_TURLARI else Hodisa.Signal.KERAK_EMAS
    try:
        with transaction.atomic():
            h = Hodisa.objects.create(kamera=kamera, tur=d["tur"], vaqt=d["vaqt"], trek=d["trek"], yonalish=d["yonalish"],
                                      malumot=d["malumot"] or {}, kalit=d["kalit"], signal=signal)
        return javob(True, {"id": h.id, "yangi": True}, status=201)
    except IntegrityError:                                               # shu kalitli hodisa bor — qayta yuborilgan
        return javob(True, {"id": Hodisa.objects.get(kalit=d["kalit"]).id, "yangi": False})


@require_GET
def salomatlik(request):
    return javob(True, {"hodisalar": Hodisa.objects.count()})

import json
import urllib.error
import urllib.request

from django.conf import settings


def yubor(matn):
    """Telegram Bot API: sendMessage. Qaytaradi: (muvaffaqiyat, xato turi)."""
    if not settings.TELEGRAM_TOKEN or not settings.TELEGRAM_CHAT:
        return False, "TELEGRAM_TOKEN yoki TELEGRAM_CHAT sozlanmagan"
    so_rov = urllib.request.Request(f"{settings.TELEGRAM_API}/bot{settings.TELEGRAM_TOKEN}/sendMessage", method="POST",
                                    data=json.dumps({"chat_id": settings.TELEGRAM_CHAT, "text": matn}).encode(),
                                    headers={"Content-Type": "application/json"})
    try:
        with urllib.request.urlopen(so_rov, timeout=10) as r:
            return bool(json.load(r).get("ok")), None
    except (urllib.error.URLError, TimeoutError, ValueError) as e:
        return False, type(e).__name__                                    # xato matnida token bo'lishi mumkin — faqat turini

import time
from datetime import timedelta

from django.core.management.base import BaseCommand
from django.db import transaction
from django.db.models import Q
from django.utils import timezone

from hodisalar.models import Hodisa
from hodisalar.telegram import yubor

MAX_URINISH = 8


def matn(h):
    qatorlar = [h.get_tur_display().capitalize(), f"Kamera: {h.kamera.nom}", f"Vaqt: {timezone.localtime(h.vaqt):%Y-%m-%d %H:%M:%S}"]
    if h.trek is not None:
        qatorlar.append(f"Trek: {h.trek}")
    return "\n".join(qatorlar)


def bir_aylanish(soni=20):
    """Kutayotgan signallarni yuboradi. Bir nechta nusxa parallel ishlasa ham bitta hodisa ikki marta yuborilmaydi."""
    hozir = timezone.now()
    with transaction.atomic():
        navbat = list(Hodisa.objects.select_for_update(skip_locked=True).select_related("kamera")
                      .filter(signal__in=[Hodisa.Signal.KUTMOQDA, Hodisa.Signal.XATO], urinishlar__lt=MAX_URINISH)
                      .filter(Q(keyingi_urinish__isnull=True) | Q(keyingi_urinish__lte=hozir)).order_by("vaqt")[:soni])
        for h in navbat:
            ok, _ = yubor(matn(h))
            h.urinishlar += 1
            if ok:
                h.signal, h.yuborildi, h.keyingi_urinish = Hodisa.Signal.YUBORILDI, timezone.now(), None
            else:
                h.signal, h.keyingi_urinish = Hodisa.Signal.XATO, timezone.now() + timedelta(seconds=2 ** h.urinishlar)
            h.save(update_fields=["signal", "urinishlar", "keyingi_urinish", "yuborildi"])
            if not ok:                                                     # Telegram ishlamayapti — qolganini keyinroq
                break
    return len(navbat)


class Command(BaseCommand):
    help = "Kutayotgan hodisalar uchun Telegram signallarini yuboradi"

    def add_arguments(self, parser):
        parser.add_argument("--bir-marta", action="store_true")

    def handle(self, *args, **opt):
        while True:
            n = bir_aylanish()
            if opt["bir_marta"]:
                self.stdout.write(f"yuborishga urinildi: {n}")
                return
            if n == 0:
                time.sleep(1)

import json
from unittest import mock

from django.test import TestCase, override_settings
from django.utils import timezone

from . import views
from .management.commands import signal_yubor
from .models import Hodisa, Kamera

TOKEN = "sinov-token-123"


def hodisa(**qism):
    return {"kamera": "k1", "tur": "zona", "vaqt": "2026-09-22T10:00:00+05:00", "trek": 7, "kalit": "k1:zona:7:1", **qism}


@override_settings(ISHCHI_TOKENLAR=[TOKEN], API_DAQIQAGA=1000)
class ApiSinovi(TestCase):
    def setUp(self):
        views._hisob.clear()                                              # cheklov hisobi testlar orasida qolmasin

    def yubor(self, data, token=TOKEN):
        return self.client.post("/api/hodisalar/", data=json.dumps(data), content_type="application/json",
                                headers={"Authorization": f"Bearer {token}"} if token else {})

    def test_tokensiz_va_notogri_token(self):
        self.assertEqual(self.yubor(hodisa(), token=None).status_code, 401)
        self.assertEqual(self.yubor(hodisa(), token="xato").status_code, 401)
        self.assertEqual(Hodisa.objects.count(), 0)

    def test_notogri_malumot(self):
        r = self.yubor(hodisa(tur="yoq_tur", vaqt="kecha"))
        self.assertEqual(r.status_code, 400)
        self.assertFalse(r.json()["ok"])
        self.assertEqual(set(r.json()["xato"]), {"tur", "vaqt"})

    def test_json_emas(self):
        r = self.client.post("/api/hodisalar/", data="salom", content_type="application/json",
                             headers={"Authorization": f"Bearer {TOKEN}"})
        self.assertEqual(r.status_code, 400)

    def test_katta_sorov(self):
        self.assertEqual(self.yubor(hodisa(malumot={"x": "a" * 70_000})).status_code, 400)

    def test_yaratish_va_signal_holati(self):
        r = self.yubor(hodisa())
        self.assertEqual(r.status_code, 201)
        self.assertEqual(r.json(), {"ok": True, "data": {"id": r.json()["data"]["id"], "yangi": True}, "xato": None})
        self.yubor(hodisa(tur="chiziq", kalit="k1:chiziq:7:1", yonalish="chapga"))
        self.assertEqual(Hodisa.objects.get(tur="zona").signal, Hodisa.Signal.KUTMOQDA)
        self.assertEqual(Hodisa.objects.get(tur="chiziq").signal, Hodisa.Signal.KERAK_EMAS)

    def test_qayta_yuborish_bitta_yozuv(self):
        birinchi, ikkinchi = self.yubor(hodisa()), self.yubor(hodisa())
        self.assertEqual((birinchi.status_code, ikkinchi.status_code), (201, 200))
        self.assertFalse(ikkinchi.json()["data"]["yangi"])
        self.assertEqual(Hodisa.objects.count(), 1)

    @override_settings(API_DAQIQAGA=3)
    def test_sorovlar_chegarasi(self):
        kodlar = [self.yubor(hodisa(kalit=f"k{i}")).status_code for i in range(5)]
        self.assertEqual(kodlar, [201, 201, 201, 429, 429])


class SignalSinovi(TestCase):
    def setUp(self):
        k = Kamera.objects.create(nom="k1")
        self.h = Hodisa.objects.create(kamera=k, tur="zona", vaqt=timezone.now(), kalit="a")
        Hodisa.objects.create(kamera=k, tur="chiziq", vaqt=timezone.now(), kalit="b", signal=Hodisa.Signal.KERAK_EMAS)

    def test_xato_keyin_qayta_urinish(self):
        with mock.patch.object(signal_yubor, "yubor", return_value=(False, "URLError")):
            signal_yubor.bir_aylanish()
        self.h.refresh_from_db()
        self.assertEqual((self.h.signal, self.h.urinishlar), (Hodisa.Signal.XATO, 1))
        self.assertGreater(self.h.keyingi_urinish, timezone.now())
        with mock.patch.object(signal_yubor, "yubor", return_value=(True, None)) as m:
            self.assertEqual(signal_yubor.bir_aylanish(), 0)             # pauza hali tugamagan
            Hodisa.objects.filter(pk=self.h.pk).update(keyingi_urinish=timezone.now())
            self.assertEqual(signal_yubor.bir_aylanish(), 1)
        self.h.refresh_from_db()
        self.assertEqual((self.h.signal, m.call_count), (Hodisa.Signal.YUBORILDI, 1))

    def test_urinishlar_chegarasi(self):
        Hodisa.objects.filter(pk=self.h.pk).update(urinishlar=signal_yubor.MAX_URINISH, signal=Hodisa.Signal.XATO)
        with mock.patch.object(signal_yubor, "yubor", return_value=(True, None)) as m:
            signal_yubor.bir_aylanish()
        self.assertEqual(m.call_count, 0)

PY = sys.executable
MANAGE = ["loyiha/manage.py"]

def django(buyruq, env):
    r = subprocess.run([PY, *MANAGE, *buyruq], env={**os.environ, **env}, capture_output=True, text=True)
    return (r.stdout + r.stderr).strip()

with postgres() as pg:
    env = {**pg, "DJANGO_SECRET_KEY": secrets.token_urlsafe(50)}
    print(django(["makemigrations", "hodisalar"], env).splitlines()[0])
    natija = django(["test", "hodisalar"], env)
    print("\n".join(q for q in natija.splitlines() if q.startswith(("Ran", "OK", "FAIL", "ERROR"))))
    print(django(["check", "--deploy"], {**env, "DJANGO_HTTPS": "1"}))

# Ishchi: RTSP kameralar -> YOLO11n (TensorRT) -> ByteTrack -> qoidalar (chiziq, zona, kamera holati) -> Django API.
# Maxfiy sozlamalar muhitdan: RTSP_ASOS (rtsp://login:parol@server:port/), ISHCHI_TOKEN.
# Misol: RTSP_ASOS=... ISHCHI_TOKEN=... python kod/20_tahlilchi.py --kameralar k1,k2
import argparse
import collections
import json
import os
import sys
import threading
import time
import urllib.error
import urllib.request
from datetime import datetime, timezone

import cv2
import numpy as np

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from oqim import Oqim  # noqa: E402
from ultralytics import YOLO  # noqa: E402
from ultralytics.trackers import BYTETracker  # noqa: E402
from ultralytics.utils import LOGGER, YAML, IterableSimpleNamespace  # noqa: E402
from ultralytics.utils.checks import check_yaml  # noqa: E402

LOGGER.setLevel("ERROR")
kodni_oqi = __import__("19_soxta_kamera").kodni_oqi
CHIZIQ = ((960, 500), (960, 1080))                                        # 14-bob: x = 960 (1920x1080 kadrda)
ZONA = np.array([(150, 600), (760, 600), (820, 800), (100, 800)], np.float32)   # 14-bob: do'kon eshigi ichkarisi


def c(a, b, p):
    return (b[0] - a[0]) * (p[1] - a[1]) - (b[1] - a[1]) * (p[0] - a[0])


def kesib_otdimi(p1, p2, a, b):
    return c(a, b, p1) * c(a, b, p2) < 0 and c(p1, p2, a) * c(p1, p2, b) < 0


class Yuboruvchi(threading.Thread):
    """Hodisalarni navbat bilan API ga yuboradi. API ishlamasa — kutadi va qayta urinadi, hodisa yo'qolmaydi."""
    def __init__(self, api, token):
        super().__init__(daemon=True)
        self.api, self.token, self.navbat, self.shart = api, token, collections.deque(maxlen=100_000), threading.Condition()
        self.hisob = collections.Counter()

    def qosh(self, hodisa):
        with self.shart:
            self.navbat.append(hodisa)
            self.shart.notify()

    def run(self):
        pauza = 1.0
        while True:
            with self.shart:
                while not self.navbat:
                    self.shart.wait()
                hodisa = self.navbat[0]
            kod = self.post(hodisa)
            if kod in (200, 201, 400):                                     # 400 — noto'g'ri hodisa: qayta yuborish foyda bermaydi
                with self.shart:
                    self.navbat.popleft()
                self.hisob[kod] += 1
                pauza = 1.0
            else:
                self.hisob["qayta urinish"] += 1
                time.sleep(pauza)
                pauza = min(pauza * 2, 30)

    def post(self, hodisa):
        so_rov = urllib.request.Request(self.api, data=json.dumps(hodisa).encode(), method="POST",
                                        headers={"Content-Type": "application/json", "Authorization": f"Bearer {self.token}"})
        try:
            with urllib.request.urlopen(so_rov, timeout=5) as r:
                return r.status
        except urllib.error.HTTPError as e:
            return e.code
        except (urllib.error.URLError, TimeoutError):
            return None


def vaqt_iso(ms):
    return datetime.fromtimestamp(ms / 1000, timezone.utc).isoformat()


def main():
    p = argparse.ArgumentParser()
    p.add_argument("--kameralar", default="k1")
    p.add_argument("--api", default="http://127.0.0.1:18000/api/hodisalar/")
    p.add_argument("--model", default="weights/yolo11n_384x640_b16.engine")
    p.add_argument("--jim", type=float, default=10, help="shuncha soniya kadr kelmasa — kamera_jim")
    p.add_argument("--soniya", type=float, default=0, help="0 — to'xtovsiz")
    p.add_argument("--vaqt-kodi", action="store_true", help="soxta kameralar kadridagi vaqtni o'qish (kechikishni o'lchash)")
    a = p.parse_args()
    asos, token = os.environ["RTSP_ASOS"], os.environ["ISHCHI_TOKEN"]        # buyruq qatorida emas: ps da ko'rinadi

    model = YOLO(a.model, task="detect")
    for b in range(1, 17):                                                # qizdirish: birinchi kadrlar kechikmasin
        model.predict([np.zeros((360, 640, 3), np.uint8)] * b, imgsz=(384, 640), device=0, verbose=False)
    bt = IterableSimpleNamespace(**YAML.load(check_yaml("bytetrack.yaml")))
    yuboruvchi = Yuboruvchi(a.api, token)
    yuboruvchi.start()
    kameralar = {}
    for nom in a.kameralar.split(","):
        o = Oqim(nom, asos + nom, jurnal=lambda m: print(m, flush=True))
        o.start()
        kameralar[nom] = {"oqim": o, "tracker": BYTETracker(bt, frame_rate=15), "oxirgi": {}, "zonada": set(), "jim": False}

    def hodisa(nom, tur, ms, **qism):
        yuboruvchi.qosh({"kamera": nom, "tur": tur, "vaqt": vaqt_iso(ms), "kalit": f"{nom}:{tur}:{qism.get('trek', '-')}:{ms}", **qism})

    t0 = time.time()
    while not a.soniya or time.time() - t0 < a.soniya:
        hozir = int(time.time() * 1000)
        for nom, k in kameralar.items():                                    # kamera holati
            jimlik = k["oqim"].jimlik()
            if not k["jim"] and jimlik > a.jim and k["oqim"].kadr_vaqti:
                k["jim"] = True
                hodisa(nom, "kamera_jim", hozir, malumot={"jimlik_s": round(jimlik, 1)})
            elif k["jim"] and jimlik < 1:
                k["jim"] = False
                hodisa(nom, "kamera_tiklandi", hozir)
        tayyor = [(nom, r) for nom, k in kameralar.items() if (r := k["oqim"].ol()) is not None]
        if not tayyor:
            time.sleep(0.005)
            continue
        for j in range(0, len(tayyor), 16):
            qism = tayyor[j:j + 16]
            for (nom, rasm), r in zip(qism, model.predict([r for _, r in qism], imgsz=(384, 640), device=0, verbose=False, classes=[0])):
                k = kameralar[nom]
                ms = int(time.time() * 1000)
                if a.vaqt_kodi:                                            # 24 bitli kod — to'liq vaqtga qaytaramiz
                    ms = hozir - ((hozir - kodni_oqi(rasm)) % 2 ** 24)
                s = 1920 / rasm.shape[1]
                for t in k["tracker"].update(r.boxes.cpu().numpy(), rasm):
                    tid, oyoq = int(t[4]), (float(t[0] + t[2]) / 2 * s, float(t[3]) * s)
                    oldingi = k["oxirgi"].get(tid)
                    if oldingi and kesib_otdimi(oldingi, oyoq, *CHIZIQ):
                        hodisa(nom, "chiziq", ms, trek=tid, yonalish="chapga" if c(*CHIZIQ, oyoq) > 0 else "o'ngga")
                    if tid not in k["zonada"] and cv2.pointPolygonTest(ZONA, oyoq, False) > 0:
                        k["zonada"].add(tid)                                # bitta trek — bitta signal
                        hodisa(nom, "zona", ms, trek=tid, malumot={"oyoq": [round(oyoq[0]), round(oyoq[1])]})
                    k["oxirgi"][tid] = oyoq
    for k in kameralar.values():
        k["oqim"].toxtat()
    for k in kameralar.values():                                            # o'quvchilar cv2 ichida qolmasin
        k["oqim"].join(timeout=5)
    kutish = time.time() + 30
    while yuboruvchi.navbat and time.time() < kutish:
        time.sleep(0.2)
    print(json.dumps({"yuborildi": yuboruvchi.hisob, "navbatda_qoldi": len(yuboruvchi.navbat),
                      "kadrlar": {n: k["oqim"].olindi for n, k in kameralar.items()}}), flush=True)


if __name__ == "__main__":
    main()

# Soxta Telegram Bot API (faqat sinov uchun): /bot<TOKEN>/sendMessage ni qabul qiladi, xabarlarni faylga yozadi.
# Misol: TELEGRAM_TOKEN=... python kod/20_soxta_telegram.py --xato 3
import argparse
import json
import os
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path


def main():
    p = argparse.ArgumentParser()
    p.add_argument("--manzil", default="127.0.0.1", help="konteyner ichida: 0.0.0.0 (boshqa konteynerlar ko'rishi uchun)")
    p.add_argument("--port", type=int, default=18081)
    p.add_argument("--jurnal", default="out/telegram/xabarlar.jsonl")
    p.add_argument("--xato", type=int, default=0, help="birinchi N ta so'rovga 502 qaytaradi (qayta urinishni sinash uchun)")
    a = p.parse_args()
    token, jurnal = os.environ["TELEGRAM_TOKEN"], Path(a.jurnal)
    jurnal.parent.mkdir(parents=True, exist_ok=True)
    holat, qulf = {"sorovlar": 0}, threading.Lock()

    class Ishlovchi(BaseHTTPRequestHandler):
        def do_POST(self):
            with qulf:
                holat["sorovlar"] += 1
                n = holat["sorovlar"]
            if self.path != f"/bot{token}/sendMessage":
                return self.javob(401, {"ok": False, "description": "Unauthorized"})
            if n <= a.xato:
                return self.javob(502, {"ok": False, "description": "Bad Gateway"})
            data = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
            with qulf, jurnal.open("a", encoding="utf-8") as f:
                f.write(json.dumps({"qabul": time.time(), "chat_id": data["chat_id"], "text": data["text"]}, ensure_ascii=False) + "\n")
            self.javob(200, {"ok": True, "result": {"message_id": n}})

        def javob(self, kod, body):
            b = json.dumps(body).encode()
            self.send_response(kod)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(b)))
            self.end_headers()
            self.wfile.write(b)

        def log_message(self, *args):
            pass

    ThreadingHTTPServer((a.manzil, a.port), Ishlovchi).serve_forever()


if __name__ == "__main__":
    main()

ISHCHI_TOKEN = secrets.token_urlsafe(24)
TELEGRAM_TOKEN = f"{secrets.randbelow(10 ** 9)}:{secrets.token_urlsafe(24)}"
xabarlar = Path("out/telegram/xabarlar.jsonl")
xabarlar.unlink(missing_ok=True)
vaqt = {}
with postgres() as pg:
    env = {**pg, "DJANGO_SECRET_KEY": secrets.token_urlsafe(50), "ISHCHI_TOKENLAR": ISHCHI_TOKEN, "TELEGRAM_TOKEN": TELEGRAM_TOKEN,
           "TELEGRAM_CHAT": "-1001234567890", "TELEGRAM_API": "http://127.0.0.1:18081"}
    django(["migrate", "--noinput"], env)
    tinch = {"stdout": subprocess.DEVNULL, "stderr": subprocess.DEVNULL}
    with jarayon([PY, *MANAGE, "runserver", "127.0.0.1:18000", "--noreload"], env, **tinch), \
         jarayon([PY, "kod/20_soxta_telegram.py", "--xato", "3"], env, **tinch), \
         jarayon([PY, *MANAGE, "signal_yubor"], env, **tinch), rtsp(2, 1920) as r:
        for _ in range(50):                                                # server tayyor bo'lguncha
            try:
                print("API:", urllib.request.urlopen("http://127.0.0.1:18000/api/salomatlik/", timeout=2).read().decode())
                break
            except OSError:
                time.sleep(0.3)
        with jarayon([PY, "kod/20_tahlilchi.py", "--kameralar", "k1,k2", "--soniya", "70", "--vaqt-kodi"],
                     {"RTSP_ASOS": r["asos"], "ISHCHI_TOKEN": ISHCHI_TOKEN}, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True) as w:
            time.sleep(30)
            vaqt["qotdi"] = time.time()
            os.kill(r["kamera"].pid, signal.SIGSTOP)
            time.sleep(20)
            vaqt["davom"] = time.time()
            os.kill(r["kamera"].pid, signal.SIGCONT)
            print("tahlilchi:", w.communicate(timeout=180)[0].strip().splitlines()[-1])
        time.sleep(8)                                                      # signal_yubor oxirgi xabarlarni yuborsin
        with psycopg.connect(host=pg["PG_HOST"], port=pg["PG_PORT"], user="kitob", password=pg["PG_PAROL"], dbname="kuzatuv") as db:
            holat = pd.DataFrame(db.execute("select tur, signal, count(*), sum(urinishlar) from hodisalar_hodisa group by 1, 2 order by 1, 2")
                                 .fetchall(), columns=["tur", "signal", "hodisalar", "yuborish urinishlari"])
            hammasi = pd.DataFrame(db.execute("select tur, extract(epoch from vaqt), extract(epoch from yaratildi), "
                                              "extract(epoch from yuborildi) from hodisalar_hodisa").fetchall(),
                                   columns=["tur", "vaqt", "yaratildi", "yuborildi"]).astype({"vaqt": float, "yaratildi": float, "yuborildi": float})
show(holat)

qatorlar = []
for tur in ("chiziq", "zona"):
    q = hammasi[hammasi["tur"] == tur]
    qatorlar.append({"hodisa": tur, "yo'l": "kadr → baza", "soni": len(q), "median, s": (q["yaratildi"] - q["vaqt"]).median(),
                     "95% i shundan kam, s": (q["yaratildi"] - q["vaqt"]).quantile(0.95)})
q = hammasi[(hammasi["tur"] == "zona") & hammasi["yuborildi"].notna()]
qatorlar.append({"hodisa": "zona", "yo'l": "kadr → Telegram", "soni": len(q), "median, s": (q["yuborildi"] - q["vaqt"]).median(),
                 "95% i shundan kam, s": (q["yuborildi"] - q["vaqt"]).quantile(0.95)})
show(pd.DataFrame(qatorlar).round(2))
kamera = hammasi[hammasi["tur"].isin(["kamera_jim", "kamera_tiklandi"])].sort_values("vaqt")
show(pd.DataFrame({"hodisa": kamera["tur"], "qotishdan keyin, s": kamera["vaqt"] - vaqt["qotdi"],
                   "davomdan keyin, s": kamera["vaqt"] - vaqt["davom"], "Telegram gacha, s": kamera["yuborildi"] - kamera["vaqt"]}).round(1))
birinchi = json.loads(xabarlar.read_text(encoding="utf-8").splitlines()[0])
print(f"soxta Telegram qabul qilgan xabarlar: {len(xabarlar.read_text(encoding='utf-8').splitlines())}\n--- birinchisi ---\n{birinchi['text']}")

from playwright.sync_api import sync_playwright

ADMIN_PAROL = secrets.token_urlsafe(18)
with postgres() as pg:
    env = {**pg, "DJANGO_SECRET_KEY": secrets.token_urlsafe(50), "ISHCHI_TOKENLAR": ISHCHI_TOKEN, "DJANGO_SUPERUSER_USERNAME": "mijoz",
           "DJANGO_SUPERUSER_PASSWORD": ADMIN_PAROL, "DJANGO_SUPERUSER_EMAIL": "mijoz@example.com"}
    django(["migrate", "--noinput"], env)
    django(["createsuperuser", "--noinput"], env)
    with jarayon([PY, *MANAGE, "runserver", "127.0.0.1:18000", "--noreload", "--insecure"], env, **tinch), rtsp(2, 1920) as r:
        time.sleep(3)
        with jarayon([PY, "kod/20_tahlilchi.py", "--kameralar", "k1,k2", "--soniya", "25"],
                     {"RTSP_ASOS": r["asos"], "ISHCHI_TOKEN": ISHCHI_TOKEN}, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) as w:
            w.wait(timeout=120)
        with sync_playwright() as p:
            brauzer = p.chromium.launch()
            sahifa = brauzer.new_page(viewport={"width": 1250, "height": 720})
            sahifa.goto("http://127.0.0.1:18000/panel/login/")
            sahifa.fill("#id_username", "mijoz")
            sahifa.fill("#id_password", ADMIN_PAROL)
            sahifa.click("input[type=submit]")
            sahifa.goto("http://127.0.0.1:18000/panel/hodisalar/hodisa/?tur__exact=zona")
            Path("out/panel").mkdir(parents=True, exist_ok=True)
            sahifa.screenshot(path="out/panel/hodisalar.png")
            brauzer.close()
_, ax = plt.subplots(figsize=(12.5, 7.2))
ax.imshow(plt.imread("out/panel/hodisalar.png"))
ax.axis("off")
fig("panel.png", "Django admin — mijozning paneli: zona hodisalari, kamera va holat bo'yicha filtr, sana bo'yicha navigatsiya. Hodisalarni o'zgartirib bo'lmaydi.")