"""تقارير الامتثال — أقوى ورقة بيع: تحوّل أحداث الجهاز إلى تقرير مؤرّخ بأدلة
مصوّرة، مربوط بالجهة الرقابية السعودية لكل قطاع. HTML عربي RTL قابل للطباعة PDF.
"""
from datetime import timedelta
from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import HTMLResponse
from sqlalchemy import select
from sqlalchemy.orm import Session
from .db import get_db
from .models import Branch, Camera, Event, EventType, utcnow
router = APIRouter(prefix="/reports", tags=["reports"])
# القطاع → الجهة الرقابية السعودية (زاوية البيع القاتلة)
REGULATOR = {
"restaurant": "البلدية + الهيئة العامة للغذاء والدواء (SFDA)",
"factory": "الهيئة العليا للأمن الصناعي (HCIS) + SFDA",
"education": "وزارة التعليم",
"health": "وزارة الصحة + المركز السعودي لاعتماد المنشآت الصحية (CBAHI)",
"farm": "الهيئة العامة للغذاء والدواء (SFDA)",
"hotel": "وزارة السياحة + البلدية (وزارة الشؤون البلدية والقروية والإسكان)",
"retail": "البلدية + وزارة التجارة",
"mosque": "وزارة الشؤون الإسلامية + الحج والعمرة + الدفاع المدني (SDAIA)",
"warehouse": "الدفاع المدني + الهيئة العامة للنقل (HCIS للمواد الخطرة)",
"kids": "البلدية + الهيئة العامة للترفيه (GEA) + الدفاع المدني",
"sports": "وزارة الرياضة + الدفاع المدني + الأمن العام",
"general": "اشتراطات السلامة العامة",
}
# ربط ميزات التعليم ببنود اشتراطات وزارة التعليم للأمن والسلامة المدرسية
MOE_DOMAINS = {
"الأمن والحماية من التهديدات": ["weapon_detection", "emergency_lockdown",
"visitor_unknown", "after_hours_intrusion",
"wrong_direction", "perimeter_intrusion"],
"السلوك ومكافحة العنف والتنمّر": ["fight_detection", "bullying"],
"الصحة والسلامة الجسدية": ["stair_fall", "student_sleeping", "vape_detection",
"smoking", "patient_fall"],
"تنظيم الدخول والحضور": ["gate_attendance", "restricted_zone", "crowding",
"queue_count", "entry_exit_count"],
}
def branch_sector(db: Session, branch: Branch) -> str:
"""قطاع الفرع = أكثر قطاع للمزايا المفعّلة على كاميراته (تقريب عملي)."""
counts = {}
for cam in branch.cameras:
for a in cam.assignments:
if a.enabled and a.event_type.feature_key:
s = a.event_type.sector or "general"
counts[s] = counts.get(s, 0) + 1
non_general = {k: v for k, v in counts.items() if k != "general"}
pool = non_general or counts
return max(pool, key=pool.get) if pool else "general"
@router.get("/branch/{branch_id}", response_class=HTMLResponse)
def branch_report(branch_id: str, days: int = 30, db: Session = Depends(get_db)):
branch = db.get(Branch, branch_id)
if not branch:
raise HTTPException(404, "branch not found")
since = utcnow() - timedelta(days=days)
cam_ids = [c.id for c in branch.cameras]
events = []
if cam_ids:
events = db.scalars(
select(Event).where(Event.camera_id.in_(cam_ids),
Event.detected_at >= since)
.order_by(Event.detected_at.desc())).all()
names = {et.feature_key: et.name_ar for et in db.scalars(select(EventType)).all()}
# تجميع حسب الميزة + عدّ المخالفات (severity=high)
by_feature: dict[str, dict] = {}
for e in events:
f = by_feature.setdefault(e.event_type, {"total": 0, "violations": 0, "samples": []})
f["total"] += 1
if e.severity == "high":
f["violations"] += 1
if e.snapshot_url and len(f["samples"]) < 4:
f["samples"].append(e)
violations = sum(f["violations"] for f in by_feature.values())
# ponytail: درجة تقريبية — 100 ناقص وزن المخالفات، بحد أدنى 0. يُعاير لاحقاً بالمعدّل/الفرع.
score = max(0, 100 - min(100, violations * 3))
sector = branch_sector(db, branch)
org = branch.organization.name if branch.organization else "—"
# مزايا مفعّلة على كاميرات الفرع (لتحديد التغطية مقابل بنود الجهة)
enabled_keys = {a.event_type.feature_key for cam in branch.cameras
for a in cam.assignments
if a.enabled and a.event_type.feature_key}
moe_html = _moe_block(enabled_keys, by_feature, names) if sector == "education" else ""
return HTMLResponse(_render(branch, org, sector, days, score, violations,
len(events), by_feature, names, moe_html))
def _moe_block(enabled_keys, by_feature, names) -> str:
"""جدول بنود اشتراطات وزارة التعليم: التغطية + المخالفات المرصودة لكل مجال."""
rows = ""
for domain, keys in MOE_DOMAINS.items():
covered = [k for k in keys if k in enabled_keys]
viol = sum(by_feature.get(k, {}).get("violations", 0) for k in keys)
status = ("✅ مغطّى" if covered else "⚠️ غير مفعّل")
color = "#0b7a4b" if covered else "#9a6700"
items = "، ".join(names.get(k, k) for k in covered) or "لا مزايا مفعّلة"
rows += f"""
{domain}
{items}
{status}
{viol}
"""
return f"""
اشتراطات وزارة التعليم للأمن والسلامة المدرسية
المجال
حالة التغطية
مخالفات الفترة
{rows}
مصفوفة تغطية آلية مقابل مجالات الأمن والسلامة والصحة المدرسية —
دليل جاهز لزيارة المشرف/التقييم الذاتي.
"""
def _badge(score: int) -> tuple[str, str]:
if score >= 85:
return "#0b7a4b", "ملتزم"
if score >= 60:
return "#9a6700", "يحتاج تحسيناً"
return "#c0392b", "غير ملتزم"
def _render(branch, org, sector, days, score, violations, total, by_feature, names,
extra_html=""):
color, label = _badge(score)
rows = ""
for key, f in sorted(by_feature.items(), key=lambda kv: -kv[1]["violations"]):
thumbs = "".join(
f'' for e in f["samples"])
rows += f"""
{names.get(key, key)}
{f['total']}
{f['violations']}
{thumbs or '—'}
"""
if not rows:
rows = '
لا أحداث في الفترة
'
date = utcnow().strftime("%Y-%m-%d")
return f"""
تقرير امتثال — {branch.name}
Veedon.ai
تقرير الامتثال التشغيلي
تاريخ التقرير: {date} الفترة: آخر {days} يوماً
{org} — {branch.name}
الجهة الرقابية: {REGULATOR.get(sector, REGULATOR['general'])}
| القطاع: {sector}
درجة الالتزام: {score}/100 — {label}
إجمالي الأحداث المرصودة: {total} | المخالفات (تحتاج إجراء): {violations}
بند المراقبة
مرصود
مخالفات
أدلة مصوّرة
{rows}
{extra_html}
تقرير آلي من منصة Veedon.ai — الأدلة مؤرّخة ومحفوظة وفق سياسة الاحتفاظ.
هذا التقرير مؤشّر تشغيلي مساعد ولا يُغني عن التفتيش الرسمي.