محلل ملفات السجل

محلل ملفات السجل — Log Analyzer

في هذا التطبيق الموجّه، ستبني محلل سجلات حقيقياً من الصفر. السجلات (Logs) هي الشريان الذي يُخبر المطوّر بكل ما يحدث في تطبيقه — متى نجحت العمليات، متى وقعت أخطاء، ومن الذي استخدم النظام.

ستستخدم معاً: Counter لعدّ مستويات الخطورة، datetime لفلترة السجلات حسب النطاق الزمني، وpathlib لكتابة تقرير الملخّص.

البيانات — السجلات الخام

هذه هي بيانات الإدخال التي سنعمل عليها طوال التطبيق. كل سطر يمثّل حدثاً واحداً بتنسيق ثابت: [التاريخ والوقت] LEVEL: الرسالة

2024-03-15 08:01:23 INFO: بدأ التطبيق بنجاح
2024-03-15 08:01:25 INFO: اتصال بقاعدة البيانات نجح
2024-03-15 08:03:10 DEBUG: طلب جديد من 192.168.1.1
2024-03-15 08:05:44 INFO: تسجيل دخول ناجح للمستخدم [email protected]
2024-03-15 08:07:02 WARNING: محاولة دخول فاشلة ثلاث مرات
2024-03-15 08:09:18 ERROR: فشل الاتصال بخادم البريد
2024-03-15 08:12:30 INFO: تم معالجة 50 طلب بنجاح
2024-03-15 08:15:05 WARNING: استخدام الذاكرة تجاوز 80%
2024-03-15 08:18:22 INFO: تم إرسال 20 بريد إلكتروني
2024-03-15 08:20:40 ERROR: خطأ في قاعدة البيانات: connection timeout
2024-03-15 08:22:15 CRITICAL: انقطع الاتصال بالخادم الرئيسي
2024-03-15 08:25:00 INFO: تم استعادة الاتصال بالخادم
2024-03-15 09:00:00 INFO: بدأ المهمة المجدولة اليومية
2024-03-15 09:05:33 WARNING: ملف السجل يقترب من الحجم الأقصى
2024-03-15 09:10:00 INFO: انتهت المهمة المجدولة بنجاح

الخطوة 1 — تحميل وتحليل السجلات

أول مهمة: تحويل هذا النص الخام إلى قائمة من القواميس المنظمة. كل سطر يُحلَّل إلى: وقت، مستوى، رسالة.

main.go

# الخطوة 1: تحليل السجلات الخام — Step 1: Parse raw logs
from datetime import datetime

نص_السجل = """2024-03-15 08:01:23 INFO: بدأ التطبيق بنجاح
2024-03-15 08:01:25 INFO: اتصال بقاعدة البيانات نجح
2024-03-15 08:03:10 DEBUG: طلب جديد من 192.168.1.1
2024-03-15 08:05:44 INFO: تسجيل دخول ناجح للمستخدم ahmed@example.com
2024-03-15 08:07:02 WARNING: محاولة دخول فاشلة ثلاث مرات
2024-03-15 08:09:18 ERROR: فشل الاتصال بخادم البريد
2024-03-15 08:12:30 INFO: تم معالجة 50 طلب بنجاح
2024-03-15 08:15:05 WARNING: استخدام الذاكرة تجاوز 80%
2024-03-15 08:18:22 INFO: تم إرسال 20 بريد إلكتروني
2024-03-15 08:20:40 ERROR: خطأ في قاعدة البيانات: connection timeout
2024-03-15 08:22:15 CRITICAL: انقطع الاتصال بالخادم الرئيسي
2024-03-15 08:25:00 INFO: تم استعادة الاتصال بالخادم
2024-03-15 09:00:00 INFO: بدأ المهمة المجدولة اليومية
2024-03-15 09:05:33 WARNING: ملف السجل يقترب من الحجم الأقصى
2024-03-15 09:10:00 INFO: انتهت المهمة المجدولة بنجاح"""

def حلّل_سطر(سطر):
    """حوّل سطر سجل إلى قاموس منظم — Parse a log line into a structured dict"""
    # التنسيق: "2024-03-15 08:01:23 INFO: الرسالة"
    # نقسم عند أول مسافتين للحصول على التاريخ والوقت
    جزء_التاريخ = سطر[:19]          # "2024-03-15 08:01:23"
    باقي = سطر[20:]                  # "INFO: الرسالة"
    
    المستوى, _, الرسالة = باقي.partition(": ")
    
    وقت = datetime.strptime(جزء_التاريخ, "%Y-%m-%d %H:%M:%S")
    
    return {
        "وقت": وقت,
        "مستوى": المستوى.strip(),
        "رسالة": الرسالة.strip(),
    }

# تحميل كل السجلات — Load all logs
سجلات = []
for سطر in نص_السجل.strip().splitlines():
    if سطر.strip():  # تجاهل الأسطر الفارغة — skip blank lines
        سجلات.append(حلّل_سطر(سطر))

print(f"تم تحميل {len(سجلات)} سجلاً")
print("\nأول سجلين:")
for س in سجلات[:2]:
    print(f"  [{س['وقت'].strftime('%H:%M:%S')}] {س['مستوى']}: {س['رسالة']}")

Output:

الخطوة 2 — عدّ مستويات الخطورة

الآن بعد أن صارت السجلات منظمة، نستخدم Counter لمعرفة توزيع مستويات الخطورة دفعة واحدة:

تحدي — Challenge

from collections import Counter
from datetime import datetime

نص_السجل = """2024-03-15 08:01:23 INFO: بدأ التطبيق بنجاح
2024-03-15 08:01:25 INFO: اتصال بقاعدة البيانات نجح
2024-03-15 08:03:10 DEBUG: طلب جديد من 192.168.1.1
2024-03-15 08:05:44 INFO: تسجيل دخول ناجح للمستخدم ahmed@example.com
2024-03-15 08:07:02 WARNING: محاولة دخول فاشلة ثلاث مرات
2024-03-15 08:09:18 ERROR: فشل الاتصال بخادم البريد
2024-03-15 08:12:30 INFO: تم معالجة 50 طلب بنجاح
2024-03-15 08:15:05 WARNING: استخدام الذاكرة تجاوز 80%
2024-03-15 08:18:22 INFO: تم إرسال 20 بريد إلكتروني
2024-03-15 08:20:40 ERROR: خطأ في قاعدة البيانات: connection timeout
2024-03-15 08:22:15 CRITICAL: انقطع الاتصال بالخادم الرئيسي
2024-03-15 08:25:00 INFO: تم استعادة الاتصال بالخادم
2024-03-15 09:00:00 INFO: بدأ المهمة المجدولة اليومية
2024-03-15 09:05:33 WARNING: ملف السجل يقترب من الحجم الأقصى
2024-03-15 09:10:00 INFO: انتهت المهمة المجدولة بنجاح"""

def حلّل_سطر(سطر):
    جزء_التاريخ = سطر[:19]
    باقي = سطر[20:]
    المستوى, _, الرسالة = باقي.partition(": ")
    وقت = datetime.strptime(جزء_التاريخ, "%Y-%m-%d %H:%M:%S")
    return {"وقت": وقت, "مستوى": المستوى.strip(), "رسالة": الرسالة.strip()}

سجلات = [حلّل_سطر(س) for س in نص_السجل.strip().splitlines() if س.strip()]

# عُدّ المستويات واطبع من الأكثر للأقل تكراراً
# Count levels and print from most to least common
# المطلوب — Expected:
# INFO: 8
# WARNING: 3
# ERROR: 2
# DEBUG: 1
# CRITICAL: 1

الخطوة 3 — استخراج التواقيت وفلترة نطاق زمني

السجلات كثيرة. في الواقع العملي نريد غالباً رؤية سجلات الساعة الأخيرة فقط — أو نطاق زمني محدد:

تحدي — Challenge

from datetime import datetime
from collections import Counter

نص_السجل = """2024-03-15 08:01:23 INFO: بدأ التطبيق بنجاح
2024-03-15 08:01:25 INFO: اتصال بقاعدة البيانات نجح
2024-03-15 08:03:10 DEBUG: طلب جديد من 192.168.1.1
2024-03-15 08:05:44 INFO: تسجيل دخول ناجح للمستخدم ahmed@example.com
2024-03-15 08:07:02 WARNING: محاولة دخول فاشلة ثلاث مرات
2024-03-15 08:09:18 ERROR: فشل الاتصال بخادم البريد
2024-03-15 08:12:30 INFO: تم معالجة 50 طلب بنجاح
2024-03-15 08:15:05 WARNING: استخدام الذاكرة تجاوز 80%
2024-03-15 08:18:22 INFO: تم إرسال 20 بريد إلكتروني
2024-03-15 08:20:40 ERROR: خطأ في قاعدة البيانات: connection timeout
2024-03-15 08:22:15 CRITICAL: انقطع الاتصال بالخادم الرئيسي
2024-03-15 08:25:00 INFO: تم استعادة الاتصال بالخادم
2024-03-15 09:00:00 INFO: بدأ المهمة المجدولة اليومية
2024-03-15 09:05:33 WARNING: ملف السجل يقترب من الحجم الأقصى
2024-03-15 09:10:00 INFO: انتهت المهمة المجدولة بنجاح"""

def حلّل_سطر(سطر):
    جزء_التاريخ = سطر[:19]
    باقي = سطر[20:]
    المستوى, _, الرسالة = باقي.partition(": ")
    وقت = datetime.strptime(جزء_التاريخ, "%Y-%m-%d %H:%M:%S")
    return {"وقت": وقت, "مستوى": المستوى.strip(), "رسالة": الرسالة.strip()}

سجلات = [حلّل_سطر(س) for س in نص_السجل.strip().splitlines() if س.strip()]

# صفّ السجلات بين 08:15:00 و 08:25:00 فقط
# Filter logs between 08:15:00 and 08:25:00
# اطبع عدد السجلات المُصفّاة فقط — print only the count
بداية = datetime(2024, 3, 15, 8, 15, 0)
نهاية = datetime(2024, 3, 15, 8, 25, 0)
# اكتب الكود هنا — write your code here

الخطوة 4 — كتابة تقرير ملخّص

الآن ندمج كل شيء: نكتب تقريراً نصياً إلى /tmp باستخدام pathlib:

تحدي — Challenge

from pathlib import Path
from datetime import datetime
from collections import Counter

نص_السجل = """2024-03-15 08:01:23 INFO: بدأ التطبيق بنجاح
2024-03-15 08:01:25 INFO: اتصال بقاعدة البيانات نجح
2024-03-15 08:03:10 DEBUG: طلب جديد من 192.168.1.1
2024-03-15 08:05:44 INFO: تسجيل دخول ناجح للمستخدم ahmed@example.com
2024-03-15 08:07:02 WARNING: محاولة دخول فاشلة ثلاث مرات
2024-03-15 08:09:18 ERROR: فشل الاتصال بخادم البريد
2024-03-15 08:12:30 INFO: تم معالجة 50 طلب بنجاح
2024-03-15 08:15:05 WARNING: استخدام الذاكرة تجاوز 80%
2024-03-15 08:18:22 INFO: تم إرسال 20 بريد إلكتروني
2024-03-15 08:20:40 ERROR: خطأ في قاعدة البيانات: connection timeout
2024-03-15 08:22:15 CRITICAL: انقطع الاتصال بالخادم الرئيسي
2024-03-15 08:25:00 INFO: تم استعادة الاتصال بالخادم
2024-03-15 09:00:00 INFO: بدأ المهمة المجدولة اليومية
2024-03-15 09:05:33 WARNING: ملف السجل يقترب من الحجم الأقصى
2024-03-15 09:10:00 INFO: انتهت المهمة المجدولة بنجاح"""

def حلّل_سطر(سطر):
    جزء_التاريخ = سطر[:19]
    باقي = سطر[20:]
    المستوى, _, الرسالة = باقي.partition(": ")
    وقت = datetime.strptime(جزء_التاريخ, "%Y-%m-%d %H:%M:%S")
    return {"وقت": وقت, "مستوى": المستوى.strip(), "رسالة": الرسالة.strip()}

سجلات = [حلّل_سطر(س) for س in نص_السجل.strip().splitlines() if س.strip()]
عدّاد = Counter(س["مستوى"] for س in سجلات)

# اكتب تقريراً بـ pathlib إلى /tmp/log_report.txt
# ثم اطبع: "تم كتابة التقرير: True"
# Write a report using pathlib to /tmp/log_report.txt
# then print: "تم كتابة التقرير: True"
# اكتب الكود هنا — write your code here

الخطوة 5 — خط المعالجة الكامل

الآن ندمج كل الخطوات في دالة واحدة منظمة — هذا هو النمط الذي ترى في قواعد كود حقيقية:

main.go

# خط المعالجة الكامل — Full pipeline
from collections import Counter
from datetime import datetime
from pathlib import Path

نص_السجل = """2024-03-15 08:01:23 INFO: بدأ التطبيق بنجاح
2024-03-15 08:01:25 INFO: اتصال بقاعدة البيانات نجح
2024-03-15 08:03:10 DEBUG: طلب جديد من 192.168.1.1
2024-03-15 08:05:44 INFO: تسجيل دخول ناجح للمستخدم ahmed@example.com
2024-03-15 08:07:02 WARNING: محاولة دخول فاشلة ثلاث مرات
2024-03-15 08:09:18 ERROR: فشل الاتصال بخادم البريد
2024-03-15 08:12:30 INFO: تم معالجة 50 طلب بنجاح
2024-03-15 08:15:05 WARNING: استخدام الذاكرة تجاوز 80%
2024-03-15 08:18:22 INFO: تم إرسال 20 بريد إلكتروني
2024-03-15 08:20:40 ERROR: خطأ في قاعدة البيانات: connection timeout
2024-03-15 08:22:15 CRITICAL: انقطع الاتصال بالخادم الرئيسي
2024-03-15 08:25:00 INFO: تم استعادة الاتصال بالخادم
2024-03-15 09:00:00 INFO: بدأ المهمة المجدولة اليومية
2024-03-15 09:05:33 WARNING: ملف السجل يقترب من الحجم الأقصى
2024-03-15 09:10:00 INFO: انتهت المهمة المجدولة بنجاح"""


def حلّل_سطر(سطر: str) -> dict:
    """حوّل سطر سجل إلى قاموس — Parse a single log line"""
    جزء_التاريخ = سطر[:19]
    باقي = سطر[20:]
    المستوى, _, الرسالة = باقي.partition(": ")
    return {
        "وقت": datetime.strptime(جزء_التاريخ, "%Y-%m-%d %H:%M:%S"),
        "مستوى": المستوى.strip(),
        "رسالة": الرسالة.strip(),
    }


def حمّل_سجلات(نص: str) -> list:
    """حمّل النص وحوّله لقائمة سجلات — Load text into list of log dicts"""
    return [حلّل_سطر(س) for س in نص.strip().splitlines() if س.strip()]


def صفّ_بنطاق(سجلات: list, بداية: datetime, نهاية: datetime) -> list:
    """صفّ السجلات بنطاق زمني — Filter by time range"""
    return [س for س in سجلات if بداية <= س["وقت"] <= نهاية]


def اكتب_تقرير(سجلات: list, مسار: Path) -> None:
    """اكتب تقرير ملخّص إلى ملف — Write summary report to file"""
    عدّاد = Counter(س["مستوى"] for س in سجلات)
    
    أسطر = [
        "=" * 50,
        "تقرير تحليل السجلات — Log Analysis Report",
        f"وقت الإنشاء: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
        f"إجمالي السجلات: {len(سجلات)}",
        "=" * 50,
        "",
        "توزيع مستويات الخطورة:",
    ]
    
    for مستوى, عدد in عدّاد.most_common():
        نسبة = (عدد / len(سجلات)) * 100
        أسطر.append(f"  {مستوى:10s}: {عدد:3d} ({نسبة:.1f}%)")
    
    # السجلات الحرجة والأخطاء — Critical and error logs
    حرجة = [س for س in سجلات if س["مستوى"] in ("ERROR", "CRITICAL")]
    if حرجة:
        أسطر.extend(["", "السجلات الحرجة والأخطاء:"])
        for س in حرجة:
            أسطر.append(f"  [{س['وقت'].strftime('%H:%M:%S')}] {س['مستوى']}: {س['رسالة']}")
    
    أسطر.append("=" * 50)
    
    مسار.write_text("\n".join(أسطر), encoding="utf-8")


# --- التشغيل الكامل — Full run ---

# 1. تحميل — Load
سجلات = حمّل_سجلات(نص_السجل)
print(f"تم تحميل {len(سجلات)} سجلاً")

# 2. عدّ — Count
عدّاد = Counter(س["مستوى"] for س in سجلات)
print("\nالتوزيع:")
for مستوى, عدد in عدّاد.most_common():
    print(f"  {مستوى}: {عدد}")

# 3. فلترة — Filter (النطاق الصباحي المبكر)
صباحية = صفّ_بنطاق(
    سجلات,
    datetime(2024, 3, 15, 8, 0, 0),
    datetime(2024, 3, 15, 8, 30, 0),
)
print(f"\nسجلات 08:00-08:30: {len(صباحية)} سجل")

# 4. كتابة التقرير — Write report
مسار_التقرير = Path("/tmp/log_report.txt")
اكتب_تقرير(سجلات, مسار_التقرير)
print(f"\nتم كتابة التقرير: {مسار_التقرير.exists()}")
print(f"حجم التقرير: {مسار_التقرير.stat().st_size} بايت")

# 5. عرض التقرير — Display report
print("\n" + "=" * 50)
print(مسار_التقرير.read_text(encoding="utf-8"))

# تنظيف — Cleanup
مسار_التقرير.unlink()

Output:

ما تعلّمته في هذا التطبيق

لاحظ كيف تكاملت الأدوات الثلاث بشكل طبيعي:

datetime.strptime حوّل النص الخام إلى كائنات قابلة للمقارنة والفلترة
Counter أعطانا توزيع المستويات بسطر واحد بدلاً من حلقات ومتغيرات عدّ
pathlib.Path.write_text كتبت التقرير بأناقة دون الحاجة لفتح/إغلاق يدوي

هذا هو قلب “batteries-included” في Python — كل أداة تحل مشكلتها بدقة، وتتكامل مع الأخريات بسلاسة لبناء أنظمة أكبر.