مشروع ختامي: أداة سطر أوامر لتحليل السجلات — Python المتقدم

مشروع ختامي: أداة سطر أوامر لتحليل السجلات

ستبني أداة يستخدمها المبرمجون فعلاً: تقرأ ملف سجلات خادم، وتستخرج منه إحصائيات، ويُتحكم بها من سطر الأوامر. المشروع يجمع أهم مفاهيم هذه الدورة.

الاستخدام المطلوب

python logtool.py server.log --level ERROR --top 5

- عدد الرسائل لكل مستوى (INFO، WARNING، ERROR).

- أكثر N رسائل خطأ تكراراً.

- الساعة الأكثر نشاطاً.

- تجاهل الأسطر التي لا تطابق الصيغة مع تسجيل تحذير.

صيغة السطر

2026-03-15 14:02:11 ERROR [db] connection timeout

خريطة المفاهيم

| الجزء | المفهوم من الدورة |

|---|---|

| parse_line | تعبير نمطي بمجموعات مسماة، يرجع dataclass أو None |

| read_entries | مولّد: يقرأ الملف سطراً سطراً (يعمل مع ملفات ضخمة) |

| filter_level | تعبير مولّد في خط معالجة |

| summarize | Counter و dataclass للنتيجة، وتلميحات أنواع |

| الأسطر المعطوبة | logging بمستوى WARNING |

| واجهة الأوامر | argparse |

| الثقة بالنتائج | اختبارات unittest للدوال النقية |

argparse باختصار

import argparse
parser = argparse.ArgumentParser(description="تحليل السجلات")
parser.add_argument("path")
parser.add_argument("--level", default=None)
parser.add_argument("--top", type=int, default=3)
args = parser.parse_args(["server.log", "--top", "5"])   # في الأداة الحقيقية: parse_args() تقرأ سطر الأوامر
print(args.path, args.level, args.top)

تولّد تلقائياً رسالة --help ورسائل خطأ عند المدخلات الخاطئة.

تصميم قابل للاختبار

الدوال التي تحلل وتحسب لا تقرأ ملفات ولا تطبع: تأخذ أسطراً (أي iterable) وترجع بيانات. لذلك يمكن اختبارها بقائمة نصوص صغيرة، ويمكن تمرير ملف حقيقي أو sys.stdin لها دون تعديل. main(argv) تأخذ قائمة المعاملات بدل قراءة sys.argv مباشرة، فيمكن استدعاؤها من الاختبارات أيضاً.

في المثال

لكي يعمل المثال داخل المحرر دون ملف خارجي، يكتب ملف سجل تجريبياً، ويشغّل main بمعاملات ثابتة، ثم يشغّل الاختبارات.

أفكار للتوسعة

- --since و --until لتحديد فترة زمنية (datetime).

- إخراج JSON بخيار --json.

- معالجة عدة ملفات بالتوازي بـ ProcessPoolExecutor.

- قراءة ملفات مضغوطة .gz بـ gzip.open.

ما بعد الدورة

أتممت مسار Python: من أول print إلى أداة حقيقية مختبرة. خيارات التخصص بعدها: تطوير الويب (Django / FastAPI)، تحليل البيانات (pandas)، أو تعلم الآلة (مسار الذكاء الاصطناعي في Kodax).

import argparse
import logging
import re
import unittest
from collections import Counter
from collections.abc import Iterable, Iterator
from dataclasses import dataclass, field
from pathlib import Path

logger = logging.getLogger("logtool")

LINE = re.compile(
    r"^(?P<date>\d{4}-\d{2}-\d{2}) (?P<hour>\d{2}):\d{2}:\d{2} "
    r"(?P<level>[A-Z]+) \[(?P<source>[\w-]+)\] (?P<message>.+)$"
)

@dataclass(frozen=True)
class Entry:
    hour: int
    level: str
    source: str
    message: str

@dataclass
class Summary:
    by_level: Counter = field(default_factory=Counter)
    top_errors: list = field(default_factory=list)
    busiest_hour: int | None = None
    skipped: int = 0

def parse_line(line: str) -> Entry | None:
    m = LINE.match(line.strip())
    if not m:
        return None
    return Entry(int(m["hour"]), m["level"], m["source"], m["message"])

def read_entries(lines: Iterable[str], stats: Summary) -> Iterator[Entry]:
    for number, line in enumerate(lines, start=1):
        entry = parse_line(line)
        if entry is None:
            stats.skipped += 1
            logger.warning("سطر %d غير مطابق للصيغة", number)
            continue
        yield entry

def summarize(lines: Iterable[str], level: str | None = None, top: int = 3) -> Summary:
    stats = Summary()
    hours = Counter()
    errors = Counter()
    for e in read_entries(lines, stats):
        if level and e.level != level:
            continue
        stats.by_level[e.level] += 1
        hours[e.hour] += 1
        if e.level == "ERROR":
            errors[e.message] += 1
    stats.top_errors = errors.most_common(top)
    stats.busiest_hour = hours.most_common(1)[0][0] if hours else None
    return stats

def main(argv: list[str]) -> Summary:
    parser = argparse.ArgumentParser(description="تحليل ملفات السجلات")
    parser.add_argument("path")
    parser.add_argument("--level")
    parser.add_argument("--top", type=int, default=3)
    args = parser.parse_args(argv)
    with open(args.path, encoding="utf-8") as f:
        result = summarize(f, args.level, args.top)
    print("حسب المستوى:", dict(result.by_level))
    print("أكثر الأخطاء:", result.top_errors)
    print("أنشط ساعة:", result.busiest_hour, "| أسطر متجاهلة:", result.skipped)
    return result

class TestLogTool(unittest.TestCase):
    SAMPLE = [
        "2026-03-15 14:02:11 ERROR [db] connection timeout",
        "2026-03-15 14:05:00 INFO [api] request ok",
        "سطر معطوب",
        "2026-03-15 15:00:01 ERROR [db] connection timeout",
    ]

    def test_parse_valid_and_invalid(self):
        self.assertEqual(parse_line(self.SAMPLE[0]).source, "db")
        self.assertIsNone(parse_line("hello"))

    def test_summary(self):
        s = summarize(self.SAMPLE)
        self.assertEqual(s.by_level["ERROR"], 2)
        self.assertEqual(s.top_errors[0], ("connection timeout", 2))
        self.assertEqual(s.skipped, 1)
        self.assertEqual(s.busiest_hour, 14)

if __name__ == "__main__":
    logging.basicConfig(level=logging.WARNING, format="%(levelname)s: %(message)s")
    demo = Path("demo.log")
    demo.write_text("\n".join(TestLogTool.SAMPLE) + "\n", encoding="utf-8")
    main([str(demo), "--top", "2"])
    demo.unlink()
    unittest.main(argv=["tests"], exit=False)