"""Device API — يتصل به veedon-agent من داخل الفرع (Bearer device_token). يقرأ ويكتب في قاعدة بيانات لوحة Beyond Senses مباشرة: الإعدادات من cameras + camera_event_assignments + event_types، والأحداث تُدرج في events فتظهر لحظياً في اللوحة (supabase realtime). """ import json import os from typing import Optional from fastapi import (APIRouter, BackgroundTasks, Depends, File, Form, Header, HTTPException, UploadFile) from fastapi.responses import FileResponse from sqlalchemy.orm import Session from . import escalation, notifier, reid, vlm from .db import DATA_DIR, SessionLocal, get_db from .models import (Branch, Camera, CameraEventAssignment, Event, EventType, utcnow) router = APIRouter(prefix="/api/device", tags=["device"]) PUBLIC_URL = os.environ.get("VEEDON_PUBLIC_URL", "http://localhost:8600").rstrip("/") DEFAULT_ENGINE_IMAGE = os.environ.get( "VEEDON_ENGINE_IMAGE", "veedon-ai/edge-engine:stable") def get_branch(authorization: str = Header(default=""), db: Session = Depends(get_db)) -> Branch: token = authorization.removeprefix("Bearer ").strip() branch = db.query(Branch).filter(Branch.device_token == token).first() if token else None if not branch: raise HTTPException(401, "invalid device token") return branch def gate(branch: Branch, fingerprint: str) -> tuple[bool, str]: """هل يُسمح لهذا الجهاز بالعمل؟ يُرجع (مسموح، السبب). الترتيب مقصود: السداد ثم التسجيل ثم تطابق البصمة ثم VM ثم موافقة الأدمن. الرفض في أي مرحلة = المحرك لا يعمل والإعدادات/النماذج لا تُسلَّم. """ if branch.is_suspended: return False, "suspended" # عدم سداد if not branch.device_fingerprint: return False, "pending" # لم يُسجّل جهاز بعد if fingerprint and fingerprint != branch.device_fingerprint: return False, "fingerprint_mismatch" # نسخة على جهاز آخر / VMware if branch.is_vm and not branch.allow_vm: return False, "vm_blocked" # جهاز افتراضي غير مسموح if branch.device_status != "approved": return False, branch.device_status # pending / rejected / blocked return True, "active" @router.post("/activate") def activate(payload: dict, branch: Branch = Depends(get_branch), db: Session = Depends(get_db)): """تسجيل بصمة الجهاز عند أول إقلاع. البصمة تُثبّت مرة واحدة؛ أي جهاز آخر بنفس الـ token (نسخة/VMware) يُرفض ولا يستبدل البصمة المعتمدة.""" fp = (payload.get("fingerprint") or "").strip() if not fp: raise HTTPException(400, "fingerprint required") hw = payload.get("hardware_info") or {} is_vm = bool(payload.get("is_vm")) if not branch.device_fingerprint: # أول تسجيل → بانتظار موافقة الأدمن branch.device_fingerprint = fp branch.hardware_info = hw branch.is_vm = is_vm branch.device_first_seen = utcnow() if branch.device_status not in ("rejected", "blocked"): branch.device_status = "pending" elif fp == branch.device_fingerprint: # نفس الجهاز → حدّث بيانات العتاد فقط branch.hardware_info = hw branch.is_vm = is_vm else: # جهاز مختلف بنفس الـ token → سجّل المحاولة info = dict(branch.agent_info or {}) info["last_rejected_fingerprint"] = fp info["last_rejected_hw"] = hw info["last_rejected_at"] = utcnow().isoformat() branch.agent_info = info db.commit() allowed, reason = gate(branch, fp) return {"authorized": allowed, "status": reason} @router.get("/config") def device_config(branch: Branch = Depends(get_branch), x_device_fingerprint: str = Header(default="")): allowed, reason = gate(branch, x_device_fingerprint) if not allowed: return {"status": reason, "config_version": branch.config_version} cameras, features = [], [] for cam in branch.cameras: if not cam.enabled or not cam.rtsp_url: continue cameras.append({"name": cam.frigate_name, "rtsp_url": cam.rtsp_url}) for a in cam.assignments: et = a.event_type if a.enabled and et.available and et.enabled and et.feature_key: cfg = dict(et.default_config or {}) cfg.update(a.config or {}) cfg["camera"] = cam.frigate_name if a.confidence_threshold is not None: cfg["confidence"] = a.confidence_threshold features.append({"key": et.feature_key, "kind": et.kind, "config": cfg}) return { "status": "active", "config_version": branch.config_version, "branch": branch.name, "cameras": cameras, "features": features, # تغيير الصورة من اللوحة = تحديث نسخة المحرك في الفرع عن بُعد "engine": {"image": branch.engine_image or DEFAULT_ENGINE_IMAGE}, } @router.post("/heartbeat") def heartbeat(payload: dict, branch: Branch = Depends(get_branch), db: Session = Depends(get_db), x_device_fingerprint: str = Header(default="")): branch.last_seen = utcnow() # يُسجَّل دائماً حتى يرى الأدمن الجهاز المعلّق info = dict(branch.agent_info or {}) info.update(payload) branch.agent_info = info db.commit() allowed, reason = gate(branch, x_device_fingerprint) return {"action": "run" if allowed else "stop", "reason": reason, "config_version": branch.config_version} def _snapshot_path(event_id) -> str: return os.path.join(DATA_DIR, "snapshots", f"{event_id}.jpg") def notify_event(event_id): """تنبيه واتساب (Evolution API) لأرقام الفرع — مهمة خلفية.""" db = SessionLocal() try: event = db.get(Event, event_id) if not event or event.severity != "high": return camera = db.get(Camera, event.camera_id) if event.camera_id else None branch = camera.branch if camera else None if not branch or not branch.alert_numbers: return et = (db.query(EventType) .filter_by(feature_key=event.event_type).first()) lines = [ "🔴 تنبيه Veedon.ai", f"العميل: {branch.organization.name if branch.organization else '—'}", f"الفرع: {branch.name}", f"الحدث: {et.name_ar if et else event.event_type}" + (f" — كاميرا {camera.name}" if camera else ""), f"الوقت: {event.detected_at.strftime('%Y-%m-%d %H:%M')}", ] note = (event.meta or {}).get("ai_note") if note: lines.append(f"ملاحظة الذكاء الاصطناعي: {note}") path = _snapshot_path(event_id) if event.snapshot_url else None notifier.send_alert(branch.alert_numbers, "\n".join(lines), path) finally: db.close() def _vlm_task(event_id, snapshot_path: str, feature_key: str): with open(snapshot_path, "rb") as f: verdict = vlm.analyze(feature_key, f.read()) db = SessionLocal() try: event = db.get(Event, event_id) if event: meta = dict(event.meta or {}) meta["ai_note"] = verdict["note"] meta["score"] = verdict["score"] event.meta = meta event.severity = "high" if verdict["flagged"] else "info" event.reviewed = not verdict["flagged"] # السليم لا يحتاج مراجعة db.commit() finally: db.close() notify_event(event_id) # يُرسل فقط إذا ثبّت التحليل المشكلة @router.post("/events") def ingest_event(background: BackgroundTasks, feature_key: str = Form(...), camera: str = Form(""), score: float = Form(0.0), meta: str = Form("{}"), flagged: bool = Form(True), snapshot: Optional[UploadFile] = File(None), branch: Branch = Depends(get_branch), db: Session = Depends(get_db), x_device_fingerprint: str = Header(default="")): allowed, reason = gate(branch, x_device_fingerprint) if not allowed: # طبقة حماية: نسخة غير معتمدة لا تُدرج أحداثاً raise HTTPException(403, reason) cam = next((c for c in branch.cameras if c.frigate_name == camera), None) try: extra = json.loads(meta or "{}") except json.JSONDecodeError: extra = {} import uuid as _uuid et = db.query(EventType).filter_by(feature_key=feature_key).first() event = Event( id=_uuid.uuid4(), # مطلوب قبل الإدراج لتسمية ملف الصورة camera_id=cam.id if cam else None, event_type=feature_key, severity="high" if flagged else "info", meta={"feature_key": feature_key, "score": score, "branch_id": str(branch.id), "source": "veedon-edge", **extra}, ) if snapshot is not None: # الصورة قبل الإدراج → INSERT واحد كامل data = snapshot.file.read() mode = reid.redaction_enabled(db, cam.id) if cam else None if mode: # تعتيم الخصوصية قبل التخزين (PDPL) data = reid.redact(data, mode) with open(_snapshot_path(event.id), "wb") as f: f.write(data) event.snapshot_url = f"{PUBLIC_URL}/snapshots/{event.id}.jpg" if flagged: # درجة الإنذار الحقيقي للقواعد (تصعيد لاحق) escalation.score_event(db, event, vlm_flagged=False) db.add(event) db.commit() # ملاحظة: الكشف السلوكي (VLM) يمرّ عبر /analyze (لقطات متتالية). هذا المسار # للقواعد الفورية والعدّ؛ ويحلّل لقطة vlm مفردة كاحتياطي إن وردت. if et and et.kind == "vlm" and event.snapshot_url: background.add_task(_vlm_task, event.id, _snapshot_path(event.id), feature_key) elif flagged: # قواعد الكشف الفورية → واتساب مباشرة background.add_task(notify_event, event.id) if cam and event.snapshot_url: # re-ID: يفهرس مظهر الشخص إن كانت الميزة مفعّلة background.add_task(_reid_task, event.id, _snapshot_path(event.id), str(branch.id), camera) return {"id": str(event.id)} def _reid_task(event_id, snap_path: str, branch_id: str, camera: str): db = SessionLocal() try: event = db.get(Event, event_id) if event and event.camera_id and reid.reid_enabled(db, event.camera_id): with open(snap_path, "rb") as f: reid.index_event(db, event, f.read(), branch_id, camera) except Exception: pass # فشل الفهرسة لا يُسقط استقبال الحدث finally: db.close() @router.post("/analyze") def analyze_behavior(background: BackgroundTasks, feature_key: str = Form(...), camera: str = Form(""), frames: list[UploadFile] = File(...), branch: Branch = Depends(get_branch), db: Session = Depends(get_db), x_device_fingerprint: str = Header(default="")): """كشف سلوكي على السيرفر السحابي: الحافة ترسل **لقطات متتالية** وتنتظر النتيجة. الاستدلال (نموذج مدرب / LM Studio / Ollama / Claude) يعمل هنا لا على الحافة. عند ثبوت المخالفة يُنشأ حدث بأفضل لقطة + إشعار، ويُعاد الحكم للحافة.""" allowed, reason = gate(branch, x_device_fingerprint) if not allowed: raise HTTPException(403, reason) imgs = [f.file.read() for f in frames] cam = next((c for c in branch.cameras if c.frigate_name == camera), None) # custom_vlm: نص الكشف يكتبه الأدمن في إعداد الميزة («اكشف أي شيء تصفه بالكلمات») prompt = "" if cam: a = next((x for x in cam.assignments if x.event_type and x.event_type.feature_key == feature_key and x.enabled), None) if a: prompt = (a.config or {}).get("prompt", "") verdict = vlm.analyze_frames(feature_key, imgs, prompt_override=prompt) if verdict["flagged"] and imgs: import uuid as _uuid event = Event( id=_uuid.uuid4(), camera_id=cam.id if cam else None, event_type=feature_key, severity="high", meta={"feature_key": feature_key, "score": verdict["score"], "branch_id": str(branch.id), "source": "veedon-edge", "ai_note": verdict["note"], "frames": len(imgs)}) escalation.score_event(db, event, vlm_flagged=True) # درجة الإنذار الحقيقي best = imgs[-1] # آخر لقطة كدليل mode = reid.redaction_enabled(db, cam.id) if cam else None if mode: # تعتيم الخصوصية (PDPL) best = reid.redact(best, mode) with open(_snapshot_path(event.id), "wb") as fh: fh.write(best) event.snapshot_url = f"{PUBLIC_URL}/snapshots/{event.id}.jpg" db.add(event) db.commit() background.add_task(notify_event, event.id) verdict["event_id"] = str(event.id) verdict["true_alarm_score"] = (event.meta or {}).get("true_alarm_score") return verdict @router.get("/models/{name}") def download_model(name: str, branch: Branch = Depends(get_branch), x_device_fingerprint: str = Header(default="")): """سجل النماذج — الوكيل ينزّل حزم edge_model من هنا (OTA). نسخة غير معتمدة لا تحصل على النماذج (حماية الملكية الفكرية).""" allowed, reason = gate(branch, x_device_fingerprint) if not allowed: raise HTTPException(403, reason) safe = os.path.basename(name) path = os.path.join(DATA_DIR, "models", safe) if not os.path.isfile(path): raise HTTPException(404, "model not published yet") return FileResponse(path, filename=safe)