انتقل إلى المحتوى الرئيسي

قيادة Elasticsearch وNeo4j من بايثون

يريد Karim، مطوِّر Veille، أن يربط واجهة المنتج بالمحرِّكَين دون إعادة كتابة الاستعلامات بـHTTP خامًا. أمّا Sami فيريد أن يُؤتمت تقريرًا شهريًّا: «أعطني قائمة أعلى المؤلِّفين خلال الأشهر الستّة الأخيرة وفئاتهم المفضَّلة». تُبيِّن هذه الوحدة كيف يُكتَب كلّ ذلك في خمسة عشر سطرًا من بايثون، داخل إحدى حاويات الحقيبة حيث العميلان الرّسميّان مثبَّتان سلفًا.

الحاوية veille-python: بايثون دون بايثون​

توفّر الحقيبة حاوية veille-python جاهزة للاستعمال. يُثبِّت ملفّها Dockerfile حزمتَين فحسب: elasticsearch>=9,<10 وneo4j>=5.28,<6. مجلّد python/ الخاصّ بالحقيبة مركَّب على /work: أيّ سكربت python/mon_script.py تكتبه ينفَّذ فورًا داخل الحاوية. متغيّرات البيئة ES_URL وES_USER وES_PASSWORD وNEO4J_URI (bolt://neo4j:7687) وNEO4J_USER وNEO4J_PASSWORD يحقنها docker-compose.yml: يقرؤها السكربت بلا أيّ ترميز صلب.

يكفي أمران اثنان:

./lab.sh python veille.py "climate change"   # ينفّذ python/veille.py بوسيطة
./lab.sh python-shell                        # يفتح مفسِّر بايثون تفاعليًّا

المكافئ على Windows PowerShell هو .\lab.ps1 python veille.py "climate change". لا pip install على الجهاز: هذا مبدأ الحقيبة، وهو ما يجعل الدّورة قابلة للتّكرار من مقعد إلى آخر.

أين نكتب؟

ملفّاتك .py تُوضَع في مجلّد python/ من الحقيبة. تراه الحاوية تحت /work. احفظ الملفّ من محرِّرك، ثمّ أعِد تشغيل ./lab.sh python …: تُؤخَذ النّسخة الجديدة في الحسبان فورًا، دون إعادة بناء الصّورة.

عميل elasticsearch 9 في خمس حركات​

يُحاكي عميل بايثون لـElasticsearch واجهة REST. تكتب بايثون اصطلاحيًّا، والمكتبة تُصنّع طلبات HTTP.

الاتّصال​

from elasticsearch import Elasticsearch
import os

es = Elasticsearch(
    os.environ["ES_URL"],
    basic_auth=(os.environ["ES_USER"], os.environ["ES_PASSWORD"]),
    request_timeout=30,
)
info = es.info()
print(f"Elasticsearch {info['version']['number']} — cluster « {info['cluster_name']} »")

المخرَج المنتظَر:

Elasticsearch 9.5.3 — cluster « veille »

Elasticsearch(url, basic_auth=(u, p)) هي الصّيغة الاعتياديّة في الإصدار 9. يُعيد العميل استعمال تجمُّع اتّصالات HTTP؛ لا تُنشئ عميلًا لكلّ طلب.

البحث​

كلّ استعلام Query DSL يمرّ عبر es.search(index=..., query=..., size=..., source=[...]). لاحِظ: بدءًا من العميل 9، تُكتَب الوسائط مباشرةً (query=...)، ولم يعد لازمًا تغليفها في body={"query": ...}.

rep = es.search(
    index="news",
    size=3,
    query={
        "multi_match": {
            "query": "climate change",
            "fields": ["headline^3", "short_description"],
        }
    },
    source=["headline", "date", "category"],
)
for h in rep["hits"]["hits"]:
    print(round(h["_score"], 2), h["_source"]["headline"])

المخرَج المنتظَر (قد تتفاوت الدّرجات قليلًا حسب BM25):

19.12 Change Is Here. Climate Change.
15.83 Ellen DeGeneres Warns Climate Change Will Be 'Dangerous'
14.44 Climate Change: Time for Action Is Now

يُقرأ إجماليّ النّتائج في rep["hits"]["total"]["value"] — 2 834 لعبارة climate change.

القراءة والفهرسة والتّحديث​

doc = es.get(index="news", id="1")               # وثيقة واحدة عبر _id
es.index(index="news", id="200854", document={   # إنشاء أو استبدال
    "headline": "Veille lance sa v2",
    "short_description": "Un moteur combinant Elasticsearch et Neo4j.",
    "category": "TECH",
    "authors": "Inès Bouraoui",
    "date": "2026-09-09",
})
es.update(index="news", id="200854", doc={"category": "BUSINESS"})

تُعيد كلّ دالّة قاموسًا يحوي _id و_version وresult (created، updated، noop).

الفهرسة الجماعيّة: helpers.bulk​

إرسال خمسين ألف استدعاء es.index منفصلًا مكلف جدًّا. توفّر الحزمة elasticsearch.helpers الدّالّة bulk التي تُعدّ صيغة NDJSON وتدير المحاولات المُعادة.

from elasticsearch import helpers

actions = [
    {"_index": "news", "_id": str(200_855 + i), "_source": {
        "headline": f"Article de test numéro {i}",
        "short_description": "Généré par le module 13.",
        "category": "TECH",
        "authors": "Sami Karray",
        "date": "2026-09-09",
    }}
    for i in range(200)
]
ok, erreurs = helpers.bulk(es, actions, chunk_size=100, request_timeout=60)
print(f"{ok} documents indexés, {len(erreurs)} erreurs")

المخرَج المنتظَر:

200 documents indexés, 0 erreurs

chunk_size=100 يُرسل الوثائق في حزم من مئة. على مجموعة News، يتّبع مُستورِد الحقيبة (importer/import_news.py) هذا المنطق بالضّبط، بدفعات من ألفين.

إدارة الأخطاء​

يعود استثناءان أكثر من غيرهما. لا تدعهما يصعدان بلا رسالة واضحة.

from elasticsearch import Elasticsearch, AuthenticationException, NotFoundError, TransportError

try:
    es = Elasticsearch(os.environ["ES_URL"], basic_auth=(os.environ["ES_USER"], os.environ["ES_PASSWORD"]))
    es.info()
except AuthenticationException:
    print("401 : identifiants invalides. Vérifiez ELASTIC_PASSWORD dans .env,")
    print("et si vous l'avez changé après le premier démarrage, faites ./lab.sh reset.")
    raise
except TransportError as e:
    print(f"Elasticsearch injoignable ({e}). Le service est-il « healthy » ? ./lab.sh status")
    raise

يقابل AuthenticationException رمزَ HTTP 401 «security_exception ... unable to authenticate user [elastic]». يُستعمل NotFoundError عند es.get على _id غير موجود. أمّا TransportError فيغطّي مشكلات الشّبكة.

عميل neo4j 5 في خمس حركات​

يتّبع العميل الرّسميّ لـNeo4j نموذجًا مختلفًا قليلًا: سائق (driver)، وجلسة (session)، ومعاملات — ضمنيّة عبر session.run، أو صريحة عبر execute_read وexecute_write.

الاتّصال​

from neo4j import GraphDatabase
import os

pilote = GraphDatabase.driver(
    os.environ["NEO4J_URI"],
    auth=(os.environ["NEO4J_USER"], os.environ["NEO4J_PASSWORD"]),
)
pilote.verify_connectivity()
print("Neo4j accessible via", os.environ["NEO4J_URI"])

يختبر verify_connectivity() الاتّصال دون إطلاق استعلام؛ مفيد عند بدء تشغيل تطبيق للفشل فورًا إن عجز السّائق عن الوصول إلى الخادم.

استعلام مع مُعامِلات​

with pilote.session() as session:
    resultat = session.run(
        "MATCH (au:Auteur {nom: $nom})<-[:ECRIT_PAR]-(a:Article) "
        "RETURN a.titre AS titre, a.date AS date "
        "ORDER BY a.date DESC LIMIT 3",
        nom="Lee Moran",
    )
    for enregistrement in resultat:
        print(enregistrement["date"], enregistrement["titre"])

المخرَج المنتظَر (قد يختلف رقمك بحسب المقالات الحديثة):

2018-05-25 Trump Cracks Terrible Joke About Assassinated Journalist
2018-05-24 Ivanka Trump's Latest Bit Of Advice Doesn't Sit Well With Some Twitter Users
2018-05-23 Rudy Giuliani Sparks Confusion With Baffling Answers

استعمل المُعامِلات دائمًا ($nom)، وليس أبدًا f"…{nom}…": أسرع (يحفظ Cypher الاستعلام في التّخبئة) ويجنّبك أيّ حقن.

المعاملات الصّريحة​

للحصول على شيفرة متينة، الممارسة الحسنة هي تغليف الاستعلام في دالّة وتمريرها إلى execute_read أو execute_write. يُدير السّائق المحاولات المُعادة عند انقطاع الشّبكة.

def top_auteurs(tx, limite: int):
    rep = tx.run(
        "MATCH (au:Auteur)<-[:ECRIT_PAR]-(a:Article) "
        "RETURN au.nom AS auteur, count(a) AS articles "
        "ORDER BY articles DESC LIMIT $limite",
        limite=limite,
    )
    return [dict(r) for r in rep]

with pilote.session() as session:
    for ligne in session.execute_read(top_auteurs, 5):
        print(f"{ligne['articles']:>5}  {ligne['auteur']}")

المخرَج المنتظَر:

 4954  Reuters
 2433  Lee Moran
 1915  Ron Dicker
 1328  Ed Mazza
 1145  Cole Delbyck

هذه الأسطر الخمسة هي المؤلِّفون الأكثر غزارةً في مجموعة News، كما حمَّلها ./lab.sh cypher 11-charger-news.cypher.

الإغلاق النّظيف​

pilote.close()

أو، وهو الأفضل، استعمال with GraphDatabase.driver(...) as pilote: — يُغلَق الاتّصال حتّى عند وقوع استثناء.

إدارة الأخطاء​

from neo4j.exceptions import ServiceUnavailable, AuthError

try:
    pilote = GraphDatabase.driver(os.environ["NEO4J_URI"], auth=(os.environ["NEO4J_USER"], os.environ["NEO4J_PASSWORD"]))
    pilote.verify_connectivity()
except AuthError:
    print("Neo4j : mot de passe refusé. Si vous avez modifié NEO4J_PASSWORD après le premier démarrage,")
    print("le mot de passe vit dans le volume : ./lab.sh reset puis ./lab.sh up.")
    raise
except ServiceUnavailable as e:
    print(f"Neo4j injoignable ({e}). Vérifiez ./lab.sh status.")
    raise

يغطّي AuthError رمز HTTP 401 الذي يُعيده خادم Bolt. أمّا ServiceUnavailable فيشمل انقضاء المُهَل والخدمة التي لم تصبح جاهزة بعد.

قراءة موجَّهة لـpython/veille.py​

تتضمّن الحقيبة python/veille.py، وهو سكربت يجمع بين المحرّكَين: يعثر Elasticsearch على المقالات ذات الصّلة باستعلام، فينطلق Neo4j من مؤلِّفيها ليصعد إلى مقالات أخرى للتّوصية بها. هذا هو نواة الوحدة 16.

rechercher(): الجزء الخاصّ بـElasticsearch​

def rechercher(es: Elasticsearch, texte: str, taille: int = 5) -> list[dict]:
    rep = es.search(
        index="news",
        size=taille,
        query={
            "multi_match": {
                "query": texte,
                "fields": ["headline^3", "short_description"],
                "type": "best_fields",
            }
        },
        source=["headline", "authors", "category", "date"],
    )
    return [{"id": int(h["_id"]), "score": round(h["_score"], 2), **h["_source"]} for h in rep["hits"]["hits"]]

ثلاث نقاط جديرة بالمُلاحظة. headline^3 يمنح العنوان وزنًا يفوق الوصف بثلاث مرّات: كلمة في العنوان تثقل الدّرجة أكثر. type: "best_fields" يحتفظ لكلّ وثيقة بأفضل الحقلَين بدلًا من جمعهما. source=[...] يُصفّي الحقول المُعادة لتوفير عرض النّطاق — لا يحتاج السّكربت إلى link ولا إلى النّصّ الكامل.

يدمج المُخرَج المُسطَّح _score و_id (رقم السّطر، محوَّلًا إلى int لاستعماله من جهة Cypher) و_source في قاموس واحد لكلّ مقال.

recommander(): الجزء الخاصّ بـNeo4j​

def recommander(session, ids: list[int], limite: int = 5) -> list[dict]:
    cypher = """
        MATCH (a:Article)-[:ECRIT_PAR]->(au:Auteur)<-[:ECRIT_PAR]-(reco:Article)-[:PUBLIE_DANS]->(c:Categorie)
        WHERE a.id IN $ids AND NOT reco.id IN $ids AND au.nom <> 'Reuters'
        RETURN reco.titre AS titre, au.nom AS auteur, c.nom AS categorie, reco.date AS date
        ORDER BY date DESC
        LIMIT $limite
    """
    return [dict(r) for r in session.run(cypher, ids=ids, limite=limite)]

يُقرأ النّمط كأنّه جملة: «أنطلق من المقالات التي عثرتُ عليها، أصعد إلى مؤلِّفيها، ثمّ أنزل مجدّدًا نحو مقالات أخرى كتبها هؤلاء المؤلِّفون أنفسهم، وأسترجع الفئة». يمنعان الحاجزان — NOT reco.id IN $ids وau.nom <> 'Reuters' — التّوصية بمقال قرأه المستخدم لتوّه، وإغراقَ النّتيجة بوكالة Reuters التي وحدها تحوز خمسة آلاف مقال.

main(): الغراء بين الاثنين​

def main() -> None:
    es = Elasticsearch(ES_URL, basic_auth=(ES_USER, ES_PASSWORD))
    info = es.info()
    print(f"Elasticsearch {info['version']['number']} — cluster « {info['cluster_name']} »")

    resultats = rechercher(es, REQUETE)
    print(f"\n{len(resultats)} articles pour « {REQUETE} » :")
    for r in resultats:
        print(f"  [{r['score']:>5}] {r['date']}  {r['headline']}  — {r['authors'] or 'sans auteur'} ({r['category']})")

    pilote = GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USER, NEO4J_PASSWORD))
    with pilote.session() as session:
        recos = recommander(session, [r["id"] for r in resultats])
    pilote.close()

    print(f"\n{len(recos)} recommandations (mêmes auteurs, autres articles) :")
    for r in recos:
        print(f"  {r['date']}  {r['titre']}  — {r['auteur']} ({r['categorie']})")

لا شيء أكثر: استدعاءان للعميلَين، وحلقة عرض. يقوم كامل منطق الأعمال على دالّتَين ونقطة دخول.

التّنفيذ الفعليّ​

./lab.sh python veille.py climate change

المخرَج المُلاحَظ على المجموعة الكاملة (قد تتفاوت الدّرجات قليلًا):

Elasticsearch 9.5.3 — cluster « veille »

5 articles pour « climate change » :
  [19.12] 2015-06-24  Change Is Here. Climate Change.  — Dominique Browning (ENVIRONMENT)
  [15.83] 2016-04-25  Ellen DeGeneres Warns Climate Change Will Be 'Dangerous'  — Kate Sheppard (ENVIRONMENT)
  [15.44] 2013-12-04  Climate Change Isn't Real, Say Fewer Than Ever  — Kate Sheppard (POLITICS)
  [14.44] 2014-11-19  Climate Change: Time for Action Is Now  — Karen Feridun (POLITICS)
  [13.98] 2016-08-15  Louisiana Flooding Is Climate Change  — Kate Sheppard (ENVIRONMENT)

5 recommandations (mêmes auteurs, autres articles) :
  2018-04-19  The E.P.A. Rolls Back A Regulation On Coal Ash  — Kate Sheppard (ENVIRONMENT)
  2017-11-08  A Vote For Common Sense And Public Health  — Karen Feridun (POLITICS)
  2017-06-01  Trump Just Pulled Out Of The Paris Agreement  — Kate Sheppard (POLITICS)
  2016-11-14  What Do We Tell The Children?  — Dominique Browning (ENVIRONMENT)
  2016-09-08  The State Of The Air We Breathe  — Karen Feridun (ENVIRONMENT)

يتفوّق المقال الأوّل على الثّاني بأربع نقاط تقريبًا: الكلمتان «climate» و«change» موجودتان في عنوانه، وheadline^3 يُثقّل. تصعد التّوصيات إلى مقالات لاحقة لتلك التي عُثِر عليها، كتبها المؤلِّفون أنفسهم — هذه هي قيمة اقتران Elasticsearch وNeo4j.

اكتب سكربتك الخاصّ: python/top_auteurs.py​

يريد Sami، من أجل تقرير شهريّ، معرفة المؤلِّفين الأكثر كتابةً خلال الأشهر الستّة الأخيرة من 2017، ولكلّ منهم فئاته الرّئيسيّة الثّلاث. يقوم Elasticsearch بالتّجميع، ويُكمل Neo4j بواسطة الرّسم البيانيّ.

أنشئ الملفّ python/top_auteurs.py:

#!/usr/bin/env python3
"""Top auteurs sur une période + leurs catégories favorites."""

from __future__ import annotations

import os
from elasticsearch import Elasticsearch
from neo4j import GraphDatabase

ES = Elasticsearch(os.environ["ES_URL"], basic_auth=(os.environ["ES_USER"], os.environ["ES_PASSWORD"]))
PILOTE = GraphDatabase.driver(os.environ["NEO4J_URI"], auth=(os.environ["NEO4J_USER"], os.environ["NEO4J_PASSWORD"]))


def top_auteurs_es(du: str, au: str, taille: int = 5) -> list[tuple[str, int]]:
    """Agrégation terms sur authors.raw filtrée par date."""
    rep = ES.search(
        index="news",
        size=0,
        query={"range": {"date": {"gte": du, "lte": au}}},
        aggs={
            "auteurs": {
                "terms": {"field": "authors.raw", "size": taille + 1, "exclude": ["", "Reuters"]}
            }
        },
    )
    seaux = rep["aggregations"]["auteurs"]["buckets"]
    return [(s["key"], s["doc_count"]) for s in seaux[:taille]]


def categories_neo4j(nom: str, limite: int = 3) -> list[tuple[str, int]]:
    """Pour un auteur donné, ses catégories les plus fréquentes."""
    cypher = """
        MATCH (au:Auteur {nom: $nom})<-[:ECRIT_PAR]-(a:Article)-[:PUBLIE_DANS]->(c:Categorie)
        RETURN c.nom AS categorie, count(a) AS articles
        ORDER BY articles DESC LIMIT $limite
    """
    with PILOTE.session() as session:
        return [(r["categorie"], r["articles"]) for r in session.run(cypher, nom=nom, limite=limite)]


def main() -> None:
    du, au = "2017-07-01", "2017-12-31"
    print(f"Top 5 auteurs entre {du} et {au} (hors Reuters)\n")
    for auteur, articles in top_auteurs_es(du, au):
        cats = categories_neo4j(auteur)
        etiquettes = ", ".join(f"{c} ({n})" for c, n in cats)
        print(f"  {articles:>4}  {auteur:<25}  {etiquettes}")
    PILOTE.close()


if __name__ == "__main__":
    main()

شغِّله:

./lab.sh python top_auteurs.py

المخرَج المنتظَر (قد تختلف عدّادات النّصف الثّاني من 2017 قليلًا حسب إصدار المجموعة):

Top 5 auteurs entre 2017-07-01 et 2017-12-31 (hors Reuters)

   312  Lee Moran                  COMEDY (779), WEIRD NEWS (391), ENTERTAINMENT (387)
   246  Ed Mazza                   POLITICS (546), ENTERTAINMENT (240), COMEDY (155)
   198  Ron Dicker                 SPORTS (612), COMEDY (355), ENTERTAINMENT (198)
   187  Cole Delbyck               ENTERTAINMENT (886), STYLE & BEAUTY (76), POLITICS (54)
   142  Nina Golgowski             POLITICS (298), U.S. NEWS (188), CRIME (161)

تأتي الأرقام إلى اليمين بين قوسَين من الرّسم البيانيّ: تعدّ جميع مقالات المؤلِّف، لا فقط تلك من النّصف الثّاني من 2017. حسب حاجتك المهنيّة، يمكنك التّصفية من جانب Cypher بإضافة عبارة WHERE a.date >= date($du).

لماذا نفصل بين المحرِّكَين؟

يبرع Elasticsearch في التّصفية والتّجميع على ملايين الوثائق؛ ويبرع Neo4j في اجتياز العلاقات. هذا التّقسيم ثابت: حين يخلط سؤال بين الاثنين، فحلّه على مرحلتَين يكون دائمًا تقريبًا أوضح وأسرع من حشر كلّ المنطق في محرِّك واحد.

جرّب 1 — تحويل استعلام Kibana إلى بايثون​

خذ الاستعلام التّالي، المكتوب بـQuery DSL في Kibana Dev Tools:

GET news/_search
{
  "size": 3,
  "query": {
    "bool": {
      "must":   [ { "match": { "headline": "election" } } ],
      "filter": [ { "range": { "date": { "gte": "2016-11-01", "lte": "2016-11-30" } } } ]
    }
  },
  "_source": ["headline", "date"]
}

أعِد كتابته في سكربت python/election_novembre.py، وشغِّله بـ./lab.sh python election_novembre.py واعرض العناوين الثّلاثة.

الحلّ
import os
from elasticsearch import Elasticsearch

es = Elasticsearch(os.environ["ES_URL"], basic_auth=(os.environ["ES_USER"], os.environ["ES_PASSWORD"]))

rep = es.search(
    index="news",
    size=3,
    query={
        "bool": {
            "must":   [{"match": {"headline": "election"}}],
            "filter": [{"range": {"date": {"gte": "2016-11-01", "lte": "2016-11-30"}}}],
        }
    },
    source=["headline", "date"],
)
for h in rep["hits"]["hits"]:
    print(h["_source"]["date"], h["_source"]["headline"])

ترتيب bool في بايثون مطابق لترتيبه في JSON: يكتفي العميل بالتّسلسل. التّغيير الوحيد: يُصبح _source وسيطًا source (البادئة _ محفوظة للبيانات الوصفيّة على جانب الخادم).

جرّب 2 — التّوصية بـexecute_read​

اكتب python/reco_auteur.py يأخذ اسم مؤلِّف كوسيطة ويعرض المقالات الثّلاثة الأحدث لمؤلِّفين يشاركونه فئتَين على الأقلّ. استعمل session.execute_read.

الحلّ
import os, sys
from neo4j import GraphDatabase

nom = " ".join(sys.argv[1:]) or "Lee Moran"

pilote = GraphDatabase.driver(os.environ["NEO4J_URI"], auth=(os.environ["NEO4J_USER"], os.environ["NEO4J_PASSWORD"]))

def voisins(tx, nom_auteur, limite=3):
    cypher = """
        MATCH (moi:Auteur {nom: $nom})<-[:ECRIT_PAR]-(:Article)-[:PUBLIE_DANS]->(c:Categorie)
              <-[:PUBLIE_DANS]-(:Article)-[:ECRIT_PAR]->(voisin:Auteur)
        WHERE voisin <> moi
        WITH voisin, count(DISTINCT c) AS communes
        WHERE communes >= 2
        MATCH (voisin)<-[:ECRIT_PAR]-(a:Article)
        RETURN voisin.nom AS auteur, a.titre AS titre, a.date AS date
        ORDER BY a.date DESC LIMIT $limite
    """
    return [dict(r) for r in tx.run(cypher, nom=nom_auteur, limite=limite)]

with pilote.session() as session:
    for r in session.execute_read(voisins, nom):
        print(f"{r['date']}  [{r['auteur']}]  {r['titre']}")
pilote.close()

يعبر MATCH المزدوج علاقة :PUBLIE_DANS مرّتَين للعثور على المؤلِّفين الذين يتقاسمون فئات مع الشّخص المطلوب. يدير execute_read المحاولات المُعادة إن استجاب Neo4j ببطء.

جرّب 3 — فهرسة مجموعة صغيرة بـhelpers.bulk​

اكتب python/ajouter_notes.py يُدرج في فهرس news خمسة مقالات وهميّة تقابل مذكّرات داخليّة لـVeille (فئة INTERNE، مؤلِّف Inès Bouraoui)، ثمّ تحقّق بـGET news/_count من أنّ الفهرس نما بخمسة.

الحلّ
import os
from elasticsearch import Elasticsearch, helpers

es = Elasticsearch(os.environ["ES_URL"], basic_auth=(os.environ["ES_USER"], os.environ["ES_PASSWORD"]))

sujets = [
    "Point hebdomadaire produit — semaine 36",
    "Ateliers Kibana : nouveaux tableaux pour Léa",
    "Revue de sécurité mensuelle",
    "Feuille de route Q4 : reco temps réel",
    "Onboarding Sami : bilan à 30 jours",
]
actions = [
    {"_index": "news", "_id": f"veille-{i}", "_source": {
        "headline": s,
        "short_description": f"Note interne du {i} septembre 2026.",
        "category": "INTERNE",
        "authors": "Inès Bouraoui",
        "date": "2026-09-09",
    }}
    for i, s in enumerate(sujets, start=1)
]
ok, erreurs = helpers.bulk(es, actions)
print(f"{ok} documents indexés")
es.indices.refresh(index="news")
print("Total news :", es.count(index="news")["count"])

المخرَج المنتظَر (المجموع يعتمد على حالتك؛ على مجموعة مُستورَدة حديثًا، ينتقل من 200 853 إلى 200 858):

5 documents indexés
Total news : 200858

يضمن refresh الصّريح أن يرى count الفوريّ الوثائقَ فعلًا؛ في الإنتاج، تكفي فترة الإنعاش الافتراضيّة بثانية واحدة.

النقاط الأساسيّة​

  • تحوي الحاوية veille-python عميلَي elasticsearch 9 وneo4j 5 مُثبَّتَين سلفًا؛ سكربتاتك تعيش في مجلّد python/ من الحقيبة وتُطلَق بـ./lab.sh python <script>.py.
  • كلّ المتغيّرات (ES_URL وES_USER وES_PASSWORD وNEO4J_URI وNEO4J_USER وNEO4J_PASSWORD) محقونة سلفًا؛ لا تُرمِّز كلمة مرور صلبًا أبدًا.
  • في جانب Elasticsearch، الحركات المفتاحيّة هي es.info() وes.search(index=, query=, size=, source=) وes.get وes.index وes.update، وhelpers.bulk للاستيراد الضّخم.
  • في جانب Neo4j، سائق (GraphDatabase.driver)، وجلسة (with pilote.session())، واستعلام (session.run أو session.execute_read / execute_write)، ومُعامِلات مُسمّاة $x دائمًا.
  • ثلاثة استثناءات تُعالَج بنظافة: AuthenticationException (401 من Elasticsearch)، وAuthError (Neo4j)، وServiceUnavailable (Neo4j غير جاهز).
  • اقتران المحرِّكَين، كما يوضّحه veille.py، يقوم على دالّتَين: يعثر Elasticsearch، ويوسّع Neo4j.

استكشاف الأخطاء​

  • elasticsearch.AuthenticationException: 401 security_exception ← كلمة المرور تغيّرت بعد إنشاء المجلَّد. ./lab.sh reset ثمّ ./lab.sh up.
  • neo4j.exceptions.ServiceUnavailable ← Neo4j ليس جاهزًا أو السّائق يُشير إلى localhost بدل neo4j. تحقّق بـ./lab.sh status وتأكّد من أنّ NEO4J_URI تساوي bolt://neo4j:7687 داخل الحاوية.
  • ImportError: No module named elasticsearch ← تُنفِّذ السّكربت على جهازك المضيف بدل الحاوية. استعمل دائمًا ./lab.sh python <script>.py.
  • ModuleNotFoundError: No module named 'elasticsearch.helpers' بينما يمرّ الاستيراد ← تعارض مع حزمة بايثون قديمة مثبَّتة في venv على جهازك. الحاوية لا تعاني من هذه المشكلة: مُرّ عبر ./lab.sh python.

للاستزادة​