ما الذي ستبنيه

يبني هذا الدرس travel_journal_analyzer، وهو تطبيق Python صغير لكن مصمم بعقلية إنتاجية. يقبل التطبيق ملفًا نصيًا أو Markdown أو CSV، أو مجلدًا يحتوي على هذه الصيغ. ويحوّل كل مصدر إلى كيان JournalEntry متسق، ثم يرسل الإدخال إلى نموذج دردشة من OpenAI، ويتحقق محليًا من JSON المُعاد، ويكتب تقرير JSON واحدًا للبرمجيات اللاحقة.

للتقرير غرض محدد عمدًا: تحديد المدن المذكورة في الإدخال، والمطاعم المسماة صراحةً، والأطباق، والتقييمات حيثما وردت، والانطباع العام، والنصائح العملية، وملخص موجز. لا يتعامل التطبيق مع بنية JSON الصالحة بوصفها دليلًا على صحة الادعاء. إذ يوجّه الطلب النموذج إلى استخراج الحقائق من الإدخال المزوّد فقط، بينما يتحقق الفحص المحلي من عقد البيانات قبل تصدير النتائج.

يجمع الاستخراج المنظّم الموثوق بين حدّين. الأول هو مخطط JSON المطلوب، الذي يعرّف الحقول التي يتوقعها التطبيق. والثاني هو التحقق المحلي باستخدام Pydantic، الذي يرفض القيم المشوهة، مثل تقييم خارج نطاق خمس نقاط. تصف الأبحاث المتعلقة بالتوليد المنظّم أهمية المخرجات الموثوقة والمحددة الأنواع للتطبيقات التي تحتاج إلى بيانات قابلة للتنبؤ بدلًا من نص نثري غير مقيّد. وفي العمل العملي باستخدام Python، تجعل هذه الحدود الاختبار والصيانة أسهل بكثير.

المتطلبات والإعداد

• Python ٣.١٠ أو إصدار أحدث.
• مفتاح OpenAI API وإمكانية الوصول إلى نموذج دردشة مُعدّ عبر متغير بيئة.
• إلمام أساسي بالطرفية والملفات ودوال Python.

أنشئ مشروعًا وبيئة افتراضية معزولة:

mkdir travel-journal-analyzer
cd travel-journal-analyzer
python -m venv .venv

# macOS or Linux
source .venv/bin/activate

# Windows PowerShell
# .venvScriptsActivate.ps1

python -m pip install --upgrade pip
python -m pip install "openai>=1.0.0" "pydantic>=2.7.0" "python-dotenv>=1.0.1" "pytest>=8.0.0"
mkdir data output tests

أنشئ ملف .env. احفظ هذا الملف خارج نظام التحكم بالمصادر. يجب أن تكون مفاتيح API في متغيرات البيئة محليًا، وفي مدير أسرار النشر ضمن بيئة الإنتاج.

OPENAI_API_KEY=your-api-key
OPENAI_MODEL=your-chat-model
MAX_ENTRY_CHARACTERS=12000
REQUEST_TIMEOUT_SECONDS=45

أنشئ ملف .gitignore:

.env
.venv/
__pycache__/
.pytest_cache/
output/
*.pyc

يستخدم هذا الدرس نمط عميل OpenAI الحديث في Python: from openai import OpenAI، ثم client.chat.completions.create(...). لا تستخدم استدعاءات الإكمال القديمة على مستوى الوحدة البرمجية.

الخطوة ١: تعريف عقد البيانات

أنشئ ملف models.py. تمثل الحقول القابلة للقيم الفارغة الحقائق التي لم يوفرها المصدر. وهذا أفضل من اختلاق مدينة أو نوع مطبخ أو تقييم.

from __future__ import annotations

from typing import Literal
from pydantic import BaseModel, Field, field_validator


class JournalEntry(BaseModel):
    source_name: str = Field(min_length=1, max_length=255)
    entry_id: str = Field(min_length=1, max_length=100)
    text: str = Field(min_length=1)

    @field_validator("text")
    @classmethod
    def validate_text(cls, value: str) -> str:
        cleaned = value.strip()
        if not cleaned:
            raise ValueError("Journal entry text cannot be blank.")
        return cleaned


class RestaurantFinding(BaseModel):
    name: str = Field(min_length=1, max_length=200)
    city: str | None = Field(default=None, max_length=120)
    country: str | None = Field(default=None, max_length=120)
    cuisine: str | None = Field(default=None, max_length=120)
    dishes: list[str] = Field(default_factory=list)
    rating_out_of_five: float | None = Field(default=None, ge=0, le=5)
    sentiment: Literal["positive", "neutral", "negative"]
    recommendation_reason: str = Field(min_length=1, max_length=600)


class JournalAnalysis(BaseModel):
    entry_id: str = Field(min_length=1, max_length=100)
    cities_mentioned: list[str] = Field(default_factory=list)
    restaurants: list[RestaurantFinding] = Field(default_factory=list)
    travel_tips: list[str] = Field(default_factory=list)
    concise_summary: str = Field(min_length=1, max_length=1000)


class AnalysisReport(BaseModel):
    generated_at_utc: str
    model: str
    total_entries: int = Field(ge=0)
    successful_analyses: int = Field(ge=0)
    failed_entries: list[str] = Field(default_factory=list)
    analyses: list[JournalAnalysis] = Field(default_factory=list)

هذه النماذج ليست مجرد توثيق. إذ تحوّل JournalAnalysis.model_validate_json() استجابة النموذج إلى كائن تم التحقق منه. وتفشل الاستجابة التي تحتوي على تسمية انطباع غير صالحة أو تقييم من ست نقاط قبل أن تصل إلى قاعدة بيانات أو جدول بيانات أو واجهة موجهة للعملاء.

الخطوة ٢: تحميل المدخلات النصية وCSV

أنشئ ملف journal_loader.py. تنشئ ملفات النص العادي وMarkdown إدخالًا واحدًا لكل ملف. ويتطلب ملف CSV عمود text وينشئ إدخالًا واحدًا لكل صف غير فارغ. يمنع حد الأحرف الطلبات الكبيرة على نحو غير متوقع.

from __future__ import annotations

import csv
from pathlib import Path
from models import JournalEntry

SUPPORTED_SUFFIXES = {".txt", ".md", ".csv"}


def load_journal_entries(path_value: str, max_characters: int) -> list[JournalEntry]:
    path = Path(path_value).expanduser().resolve()
    if not path.exists():
        raise FileNotFoundError(f"Input path does not exist: {path}")

    if path.is_dir():
        entries: list[JournalEntry] = []
        for child in sorted(path.iterdir()):
            if child.is_file() and child.suffix.lower() in SUPPORTED_SUFFIXES:
                entries.extend(load_journal_entries(str(child), max_characters))
        if not entries:
            raise ValueError("Directory contains no supported input files.")
        return entries

    if path.suffix.lower() in {".txt", ".md"}:
        text = path.read_text(encoding="utf-8").strip()
        _check_length(text, path.name, max_characters)
        return [JournalEntry(source_name=path.name, entry_id=path.stem, text=text)]

    if path.suffix.lower() != ".csv":
        raise ValueError("Use a .txt, .md, .csv file, or directory.")

    entries = []
    with path.open("r", encoding="utf-8-sig", newline="") as handle:
        reader = csv.DictReader(handle)
        if not reader.fieldnames or "text" not in reader.fieldnames:
            raise ValueError("CSV must contain a column named 'text'.")
        for row_number, row in enumerate(reader, start=2):
            text = (row.get("text") or "").strip()
            if not text:
                continue
            _check_length(text, f"{path.name} row {row_number}", max_characters)
            entries.append(JournalEntry(
                source_name=(row.get("source_name") or path.name).strip(),
                entry_id=(row.get("entry_id") or f"{path.stem}-{row_number}").strip(),
                text=text,
            ))
    if not entries:
        raise ValueError("CSV contains no non-empty text rows.")
    return entries


def _check_length(text: str, label: str, maximum: int) -> None:
    if not text:
        raise ValueError(f"{label} is empty.")
    if len(text) > maximum:
        raise ValueError(f"{label} exceeds the {maximum}-character limit.")

احفظ هذا المثال باسم data/journal.txt:

We arrived in Lisbon on a rainy Thursday.
For dinner, we booked Taberna da Rua das Flores. The grilled octopus was tender,
and I would rate the experience 4.5 out of 5. Reserve ahead because it is small.

Two days later in Porto, Cafe Santiago served a memorable francesinha.
The sandwich was rich but the service was slow, so I felt neutral overall.

الخطوة ٣: طلب JSON والتحقق منه

أنشئ ملف analyzer.py. يُمرر المخطط أدناه بصفته صيغة الاستجابة المطلوبة. يجب أن يدعم النموذج الذي أعددته قدرة صيغة الاستجابة المختارة؛ تحقق من هذا الدعم في وثائق OpenAI الحالية الخاصة بالنموذج المتاح لحسابك قبل النشر. وبصرف النظر عن قيد الاستجابة من المصدر، ينفذ الكود تحقق Pydantic محليًا.

from __future__ import annotations

from openai import OpenAI
from pydantic import ValidationError

from models import JournalAnalysis, JournalEntry

SCHEMA = {
    "name": "travel_journal_analysis",
    "strict": True,
    "schema": {
        "type": "object",
        "additionalProperties": False,
        "properties": {
            "entry_id": {"type": "string"},
            "cities_mentioned": {"type": "array", "items": {"type": "string"}},
            "restaurants": {"type": "array", "items": {"type": "object", "additionalProperties": False, "properties": {
                "name": {"type": "string"}, "city": {"type": ["string", "null"]},
                "country": {"type": ["string", "null"]}, "cuisine": {"type": ["string", "null"]},
                "dishes": {"type": "array", "items": {"type": "string"}},
                "rating_out_of_five": {"type": ["number", "null"]},
                "sentiment": {"type": "string", "enum": ["positive", "neutral", "negative"]},
                "recommendation_reason": {"type": "string"}
            }, "required": ["name", "city", "country", "cuisine", "dishes", "rating_out_of_five", "sentiment", "recommendation_reason"]}},
            "travel_tips": {"type": "array", "items": {"type": "string"}},
            "concise_summary": {"type": "string"}
        },
        "required": ["entry_id", "cities_mentioned", "restaurants", "travel_tips", "concise_summary"]
    }
}

SYSTEM_PROMPT = """Extract facts only from the journal entry.
Include restaurants only when explicitly named. Never invent facts.
Use null for unknown scalar values. Return only schema-conforming JSON."""


class JournalAnalyzer:
    def __init__(self, api_key: str, model: str, timeout: float) -> None:
        self.model = model
        self.client = OpenAI(api_key=api_key, timeout=timeout, max_retries=0)

    def analyze(self, entry: JournalEntry) -> JournalAnalysis:
        completion = self.client.chat.completions.create(
            model=self.model,
            temperature=0,
            response_format={"type": "json_schema", "json_schema": SCHEMA},
            messages=[
                {"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content": f"Required entry_id: {entry.entry_id}nnJournal:n{entry.text}"},
            ],
        )
        content = completion.choices[0].message.content
        if not content:
            raise RuntimeError("The model returned an empty response.")
        try:
            result = JournalAnalysis.model_validate_json(content)
        except ValidationError as error:
            raise RuntimeError(f"Local schema validation failed: {error}") from error
        if result.entry_id != entry.entry_id:
            raise RuntimeError("Response entry_id does not match the input entry.")
        return result

سياسة الاستخراج متحفظة عن قصد. قد تناقش اليومية كشكًا غير مسمى بشكل سلبي، لكن لا ينبغي أن يتحول إلى سجل مطعم مختلق. إذا كانت الأماكن غير المسماة مهمة للمنتج، فأضف حقلًا مستقلًا ووضح بجلاء أنه يمثل إشارة غير مسماة، وليس هوية رسمية للمكان.

الخطوة ٤: إنشاء تطبيق سطر الأوامر

from __future__ import annotations

import argparse
import os
from datetime import datetime, timezone
from pathlib import Path
from dotenv import load_dotenv

from analyzer import JournalAnalyzer
from journal_loader import load_journal_entries
from models import AnalysisReport


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("input_path")
    parser.add_argument("--output", default="output/travel-analysis.json")
    args = parser.parse_args()

    load_dotenv()
    api_key = os.getenv("OPENAI_API_KEY", "").strip()
    model = os.getenv("OPENAI_MODEL", "").strip()
    if not api_key or not model:
        raise RuntimeError("Set OPENAI_API_KEY and OPENAI_MODEL in .env or your shell.")

    maximum = int(os.getenv("MAX_ENTRY_CHARACTERS", "12000"))
    timeout = float(os.getenv("REQUEST_TIMEOUT_SECONDS", "45"))
    entries = load_journal_entries(args.input_path, maximum)
    analyzer = JournalAnalyzer(api_key, model, timeout)

    analyses = []
    failed = []
    for entry in entries:
        try:
            analyses.append(analyzer.analyze(entry))
        except RuntimeError as error:
            failed.append(entry.entry_id)
            print(f"Failed {entry.entry_id}: {error}")

    report = AnalysisReport(
        generated_at_utc=datetime.now(timezone.utc).isoformat(),
        model=model,
        total_entries=len(entries),
        successful_analyses=len(analyses),
        failed_entries=failed,
        analyses=analyses,
    )
    destination = Path(args.output)
    destination.parent.mkdir(parents=True, exist_ok=True)
    temporary = destination.with_suffix(destination.suffix + ".tmp")
    temporary.write_text(report.model_dump_json(indent=2), encoding="utf-8")
    temporary.replace(destination)
    print(f"Wrote {report.successful_analyses}/{report.total_entries} analyses to {destination}")
    return 1 if failed else 0


if __name__ == "__main__":
    raise SystemExit(main())

شغّل الأداة وافحص المخرجات:

python main.py data/journal.txt --output output/journal-report.json
python -m json.tool output/journal-report.json

يمنع استبدال الملف المؤقت ترك تقرير نهائي مكتوب جزئيًا إذا توقفت العملية أثناء التسلسل. وتُرجع العملية حالة غير صفرية عند فشل إدخال واحد أو أكثر، وهو ما يفيد في المهام المجدولة وCI.

الاختبارات وفحوصات الإنتاج

اختبر المكونات الحتمية من دون استدعاء API. على سبيل المثال، تحقق من أن أداة التحميل ترفض ملف CSV بلا عمود text، وأنها تتجاوز الصفوف الفارغة، وأن التقييمات التي تتجاوز خمسة تفشل في تحقق Pydantic. استخدم محاكاة لعميل OpenAI في اختبارات الوحدة الخاصة بالمحلل، ثم نفّذ اختبار smoke صغيرًا ومقصودًا باستخدام بيانات اعتماد حقيقية.

from models import RestaurantFinding
from pydantic import ValidationError
import pytest


def test_invalid_rating_is_rejected():
    with pytest.raises(ValidationError):
        RestaurantFinding(
            name="Example", sentiment="positive",
            rating_out_of_five=6,
            recommendation_reason="Unsupported rating range.",
        )

صلاحية المخطط ليست صلاحية واقعية. راجع مخرجات ممثلة للتحقق من إسناد الكيانات المسماة، ودقة التقييمات، وصحة الانطباع، والتفاصيل غير المدعومة. حافظ على مجموعة تقييم مصرح بها عند تغيير الطلبات أو إعدادات النموذج أو المخططات أو إصدارات SDK. وتجنب تسجيل محتوى اليوميات الخام افتراضيًا، لأن ملاحظات السفر قد تتضمن بيانات شخصية.

بالنسبة إلى عمليات النشر في دول مجلس التعاون الخليجي والشرق الأوسط، طبّق الانضباط نفسه في التعامل مع البيانات على المدخلات العربية والإنجليزية، والتزامات الخصوصية المحلية، ومتطلبات إقامة البيانات، وقواعد الاحتفاظ المعتمدة لدى المؤسسة. لا يقدم هذا الدرس ادعاءات بشأن مزود إقليمي أو مبادرة أو ترتيب بنية تحتية محدد؛ وينبغي للفرق التحقق من هذه المتطلبات مع الجهات القانونية والأمنية وأصحاب المصلحة في المنصة قبل الاستخدام الإنتاجي.

الخطوات التالية

• خزّن النتائج المعتمدة في SQLite وأنشئ تجزئة لإدخالات المصدر لتجنب التحليل المتكرر.
• أضف خطوة مراجعة بشرية قبل نشر التوصيات.
• تتبّع حجم المدخلات ووقت الاستجابة وحالات الفشل وإعدادات النموذج ونتائج التقييم، من دون الاحتفاظ بنص يوميات غير ضروري.
• أضف مقتطفات صريحة من المصدر إلى المخطط إذا كان المراجعون يحتاجون إلى تحقق أسرع من الحقائق.

الدرس الأساسي بسيط: استخدم العقود المحددة الأنواع والمدخلات المضبوطة لتحويل مخرجات النموذج المرنة إلى بيانات تطبيق، ثم اختبر حدّ البرمجيات ومدى الفائدة الواقعية للنتيجة معًا.