File size: 5,079 Bytes
c6497cf 6d856e6 c6497cf 7d15cae 6d856e6 c6497cf 6d856e6 c6497cf 6d856e6 7d15cae 6d856e6 c6497cf 6d856e6 e1cc896 6d856e6 7d15cae 6d856e6 c6497cf 6d856e6 7d15cae 6d856e6 7d15cae 6d856e6 7d15cae 6d856e6 c6497cf 6d856e6 7d15cae 6d856e6 7d15cae 6d856e6 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 |
import os
import re
from datetime import datetime
from flask import Flask, request, jsonify
from flask_cors import CORS
import psycopg2
from psycopg2 import pool
from typing import Optional, Dict, List
app = Flask(__name__)
CORS(app)
# إعدادات اتصال قاعدة البيانات
DB_URL = os.getenv('SUPABASE_DB_URL')
connection_pool = psycopg2.pool.SimpleConnectionPool(
minconn=1,
maxconn=5,
dsn=DB_URL
)
class QueryBuilder:
"""فئة مسؤولة عن بناء استعلامات SQL الآمنة"""
@staticmethod
def get_last_office_visit(cam_mac: str) -> str:
"""استعلام عن آخر زيارة للمكتب"""
return f"""
SELECT caption, created_at, image_url
FROM data
WHERE cam_mac = '{cam_mac}'
AND (caption LIKE '%مكتب%' OR caption LIKE '%عمل%')
ORDER BY created_at DESC
LIMIT 1
"""
@staticmethod
def get_last_five_places(cam_mac: str) -> str:
"""استعلام عن آخر 5 أماكن مميزة"""
return f"""
SELECT DISTINCT ON (caption) caption, created_at, latitude, longitude, image_url
FROM data
WHERE cam_mac = '{cam_mac}'
AND caption IS NOT NULL
ORDER BY caption, created_at DESC
LIMIT 5
"""
@staticmethod
def check_garden_visit(cam_mac: str) -> str:
"""استعلام عن زيارة الحديقة"""
return f"""
SELECT caption, created_at, image_url
FROM data
WHERE cam_mac = '{cam_mac}'
AND (caption LIKE '%حديقة%' OR caption LIKE '%حدائق%' OR caption LIKE '%نباتات%')
ORDER BY created_at DESC
LIMIT 1
"""
@staticmethod
def check_classroom_visit(cam_mac: str) -> str:
"""استعلام عن زيارة صف دراسي"""
return f"""
SELECT caption, created_at, image_url
FROM data
WHERE cam_mac = '{cam_mac}'
AND (caption LIKE '%صف%' OR caption LIKE '%قاعة%' OR caption LIKE '%محاضرة%'
OR caption LIKE '%جامعة%' OR caption LIKE '%مدرسة%')
ORDER BY created_at DESC
LIMIT 1
"""
class DatabaseExecutor:
"""فئة مسؤولة عن تنفيذ الاستعلامات وإدارة الاتصالات"""
@staticmethod
def execute_query(sql: str) -> List[Dict]:
"""تنفيذ استعلام SQL وإرجاع النتائج"""
conn = None
try:
conn = connection_pool.getconn()
cursor = conn.cursor()
cursor.execute(sql)
if cursor.description:
columns = [desc[0] for desc in cursor.description]
return [dict(zip(columns, row)) for row in cursor.fetchall()]
return []
finally:
if conn:
connection_pool.putconn(conn)
@app.route('/query', methods=['POST'])
def handle_query():
try:
data = request.get_json()
cam_mac = data.get('cam_mac')
question = data.get('question')
if not cam_mac or not question:
return jsonify({"error": "يجب تقديم cam_mac و question"}), 400
# تنظيف المدخلات
cam_mac = re.sub(r'[^a-fA-F0-9:]', '', cam_mac)
question = question.strip().lower()
# تحديد الاستعلام المناسب
sql = None
if 'مكتب' in question or 'عمل' in question:
sql = QueryBuilder.get_last_office_visit(cam_mac)
elif 'خمس أماكن' in question or 'آخر أماكن' in question:
sql = QueryBuilder.get_last_five_places(cam_mac)
elif 'حديقة' in question or 'حدائق' in question:
sql = QueryBuilder.check_garden_visit(cam_mac)
elif 'صف دراسي' in question or 'محاضرة' in question or 'قاعة' in question:
sql = QueryBuilder.check_classroom_visit(cam_mac)
if not sql:
return jsonify({"error": "نوع السؤال غير معروف"}), 400
# تنفيذ الاستعلام
results = DatabaseExecutor.execute_query(sql)
return jsonify({
"question": question,
"results": results,
"timestamp": datetime.now().isoformat()
})
except psycopg2.Error as e:
return jsonify({"error": "خطأ في قاعدة البيانات", "details": str(e)}), 500
except Exception as e:
return jsonify({"error": "خطأ غير متوقع", "details": str(e)}), 500
@app.route('/health', methods=['GET'])
def health_check():
try:
conn = connection_pool.getconn()
conn.close()
return jsonify({"status": "يعمل بنجاح"})
except Exception as e:
return jsonify({"status": "غير متصل", "error": str(e)}), 500
if __name__ == '__main__':
app.run(host='0.0.0.0', port=int(os.environ.get('PORT', 5000))) |