# -*- coding: utf-8 -*-
"""
مستخرج المعرفة من نصوص عربية لبناء رسم بياني معرفي (Knowledge Graph)
====================================================================
يحوّل النص العربي الخام إلى رسم بياني {nodes, edges} عبر أربع مراحل:
١. استخراج قاعدي (`extract_entities_and_relations`) عبر المشي على
شجرة الاعتماد النحوي (CamelParser2.0 / CATiB) - المصدر الأساسي.
٢. حل الإحالة المرجعية عبر الجمل (`resolve_pronoun_antecedents`،
`build_antecedent_text_map`).
٣. استخراج هجين تكميلي عبر نموذج mT5 (`hybrid_relation_extraction_api`)
للأزواج غير المغطاة قاعدياً.
٤. تصنيف نوع الكيان وبناء الرسم البياني (`classify_node_type`،
`build_knowledge_graph_json`).
نقطة الدخول الموحَّدة: `process_full_document_pipeline_as_graph`.
الاعتماديات: CAMeL Tools، CamelParser2.0، خادم FastAPI اختياري لنموذج
mT5 (المسار القاعدي يعمل بمفرده بلا هذا الخادم).
حدود معروفة: علاقات من حروف جر مجردة حين لا يوجد فعل صريح؛ نموذج mT5
يحتاج بيانات تدريب أوسع؛ الجمل المتعاطفة معقَّدة الرابط قد تُخلَط أحياناً؛
قوائم تصنيف الكيان تحتاج توسعة دورية لمجالات ومصطلحات جديدة.
"""
from __future__ import annotations
import re
import json
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Set, Tuple
try:
from camel_tools.utils.dediac import dediac_ar # type: ignore
except ImportError:
_DIAC_RE = re.compile(
"[\u0610-\u061A\u064B-\u065F\u0670\u06D6-\u06DC\u06DF-\u06E8\u06EA-\u06ED\u0640]"
)
def dediac_ar(text: str) -> str:
return _DIAC_RE.sub("", text)
# --------------------------------------------------------------------------- #
# 1. CoNLL-X data model
# --------------------------------------------------------------------------- #
@dataclass
class Token:
tid: int
form: str
lemma: str
cpos: str
pos: str
feats: str
head: int
deprel: str
Sentence = List[Token]
def parse_conllx(conll_text: str) -> List[Sentence]:
sentences: List[Sentence] = []
current: Sentence = []
for raw_line in conll_text.splitlines():
line = raw_line.strip()
if not line:
if current:
sentences.append(current)
current = []
continue
if line.startswith("#"):
continue
cols = line.split("\t") if "\t" in line else line.split()
if len(cols) < 8:
continue
tid, form, lemma, cpos, pos, feats, head, deprel = cols[:8]
current.append(
Token(
tid=int(tid),
form=form,
lemma=dediac_ar(lemma) if lemma != "_" else "",
cpos=cpos,
pos=pos,
feats=feats,
head=int(head),
deprel=deprel,
)
)
if current:
sentences.append(current)
return sentences
def _children_map(sentence: Sentence) -> Dict[int, List[int]]:
children: Dict[int, List[int]] = {}
for tok in sentence:
children.setdefault(tok.head, []).append(tok.tid)
return children
def _by_id(sentence: Sentence) -> Dict[int, Token]:
return {tok.tid: tok for tok in sentence}
# --------------------------------------------------------------------------- #
# 2. POS / relation predicates
# --------------------------------------------------------------------------- #
def _is_nominal(tok: Token) -> bool:
if tok.cpos in {"NOM", "PROP"} or tok.pos in {"NOM", "PROP"}:
return True
# توكن بلا وسم إطلاقاً (يحدث مع أعداد مكتوبة مثل "الثلاثة") أقرب
# لاسم غير مصنَّف من أداة أو علامة ترقيم (هذه تُوسَم صراحة دائماً).
return not tok.cpos and not tok.pos
def _is_particle(tok: Token) -> bool:
return tok.cpos == "PRT" or tok.pos == "PRT"
_CONJUNCTIONS = {"و", "ف", "أو", "او", "ثم"}
def _is_conjunction(tok: Token) -> bool:
return _clean_form(tok.form) in _CONJUNCTIONS
def _clean_form(form: str) -> str:
return dediac_ar(form.strip("+").strip())
_NP_INTERNAL_RELS = {"MOD", "IDF"}
def _looks_like_pp_object(tok: Token) -> bool:
return _is_nominal(tok) and not _is_pronoun(tok) and tok.deprel in {"OBJ", "MOD"}
# ملاحظة تصميم: CATiB6 لا تملك فئة نحوية مستقلة للصفة (تُوسَم NOM كالاسم
# تماماً)، فالتمييز في هذا الملف يعتمد حصراً على العلاقة النحوية (deprel):
# "MOD" = نعت، "IDF" = مضاف إليه - وهذا ما تبني عليه _walk.
_PRONOUN_FORMS = {
"هو", "هي", "هما", "هم", "هن", "أنا", "انا", "نحن",
"أنت", "انت", "أنتما", "انتما", "أنتم", "انتم", "أنتن", "انتن",
"ها", "نا", "ك", "كم", "كما", "كن", "هما", "هم", "ه",
}
def _is_pronoun(tok: Token) -> bool:
"""وسم CATiB6 لا يملك فئة مستقلة للضمير - حروف الجر وحروف العطف
تُوسم PRT أيضاً، فالاعتماد على PRT وحدها يجعل كل حرف جر يبدو ضميراً.
القائمة المغلقة _PRONOUN_FORMS ووسم "PRON" الدقيق هما الدليلان
الموثوقان."""
clean = _clean_form(tok.form)
return (
tok.cpos == "PRON" or
tok.pos == "PRON" or
clean in _PRONOUN_FORMS or
tok.feats.find("PRON") != -1
)
# --------------------------------------------------------------------------- #
# 2ب. حل الإحالة المرجعية عبر المستند (Cross-Sentence Coreference)
# --------------------------------------------------------------------------- #
# الضمائر المتصلة توكنات مستقلة أصلاً بفعل تجزئة CamelParser2.0 (+ها،
# +ه...)، فجنس/عدد كل ضمير يُستخرَج من جدول ثابت، وجنس/عدد المرشحين من
# عمود FEATS مباشرة.
def _parse_feats(feats: str) -> Dict[str, str]:
"""يحلّل عمود FEATS الخام (gen=m|num=s|...) إلى قاموس مفتاح=قيمة."""
result: Dict[str, str] = {}
if not feats or feats == "_":
return result
for part in feats.split("|"):
if "=" in part:
k, v = part.split("=", 1)
result[k] = v
return result
def _token_gender_number(tok: Token) -> Tuple[Optional[str], Optional[str]]:
f = _parse_feats(tok.feats)
return f.get("gen"), f.get("num")
# ضمائر الغائب فقط (المتكلم/المخاطَب لا تُحيل إلى اسم سابق في النص).
_THIRD_PERSON_PRONOUN_AGREEMENT: Dict[str, Tuple[str, str]] = {
"ه": ("m", "s"), "ها": ("f", "s"),
"هما": ("m", "d"),
"هم": ("m", "p"), "هن": ("f", "p"),
"هو": ("m", "s"), "هي": ("f", "s"),
}
# أعداد مكتوبة - لا تصلح أبداً كمُحال إليه (هي نعت لشيء آخر دائماً، لا
# كيان بحد ذاتها)، رغم كونها قد تحمل صفات جنس/عدد نحوية تطابق ضميراً
# بالمصادفة، فتُستبعَد صراحة قبل قبول أي مرشح للإحالة.
_CARDINAL_NUMBER_WORDS = frozenset(dediac_ar(w) for w in {
"واحد", "واحدة", "اثنان", "اثنين", "اثنتان", "اثنتين",
"ثلاثة", "ثلاث", "أربعة", "اربعة", "أربع", "اربع",
"خمسة", "خمس", "ستة", "ست", "سبعة", "سبع",
"ثمانية", "ثماني", "تسعة", "تسع", "عشرة", "عشر",
})
def _is_definite_surface(word: str) -> bool:
"""فحص التعريف الصريح فقط (يبدأ بـ"ال" أو عالق ملتصق بها). تجزئتنا
تفصل عوالق و/ف/ب/ل/ك في توكنات مستقلة أصلاً، فلا حاجة لتجريدها من
بداية الكلمة هنا. هذا الفحص وحده لا يكفي: انظر _is_definite_token."""
return word.startswith("ال") or word.startswith(("بال", "لل", "كال", "وال", "فال"))
def _is_definite_token(tok: Token, tokens: Dict[int, Token],
children: Dict[int, List[int]], _depth: int = 0) -> bool:
"""التعريف الكامل: صريح ("ال")، أو مكتسَب عبر الإضافة - "أهرامات" في
"أهرامات الجيزة" معرَّفة رغم عدم بدئها بـ"ال"، لأن المضاف إليه معرَّف
(التعريف ينتقل من المضاف إليه إلى المضاف بأكمله في العربية)."""
surface = dediac_ar(_clean_form(tok.form))
if _is_definite_surface(surface):
return True
if _depth > 5:
return False
for cid in children.get(tok.tid, []):
child = tokens.get(cid)
if child is None or child.deprel != "IDF":
continue
if child.cpos == "PROP" or _is_definite_token(child, tokens, children, _depth + 1):
return True
return False
def _build_document_stream(
sentences: List[Sentence],
) -> Tuple[List[Tuple[int, Token]], List[Dict[int, Token]], List[Dict[int, List[int]]]]:
"""يُسطِّح جمل المستند في تسلسل واحد (رقم الجملة، التوكن) - ضروري
لأن الإحالة قد تعبر حدود الجملة. تُرجع أيضاً خرائط tokens/children
لكل جملة، إذ يحتاجها فحص التعريف عبر الإضافة."""
stream: List[Tuple[int, Token]] = []
sentence_tokens: List[Dict[int, Token]] = []
sentence_children: List[Dict[int, List[int]]] = []
for s_idx, sentence in enumerate(sentences):
sentence_tokens.append(_by_id(sentence))
sentence_children.append(_children_map(sentence))
for tok in sentence:
stream.append((s_idx, tok))
return stream, sentence_tokens, sentence_children
def _candidate_prominence_depth(tok: Token, tokens: Dict[int, Token], max_hops: int = 10) -> int:
"""عمق ابتعاد المرشح عن أقرب موضع نحوي بارز (فاعل، مبتدأ، جذر) - كلما
قلّ العمق كان أبرز نحوياً، بصرف النظر عن قربه النصي من الضمير."""
depth = 0
current = tok
while depth < max_hops:
if current.deprel in ("SBJ", "TPC") or current.head == 0:
return depth
governor = tokens.get(current.head)
if governor is None:
return depth
current = governor
depth += 1
return depth
def resolve_pronoun_antecedents(
sentences: List[Sentence],
) -> Dict[Tuple[int, int], Tuple[int, Token]]:
"""تُرجع خريطة: (رقم جملة الضمير، معرّفه) -> (رقم جملة المُحال إليه،
توكنه). لكل ضمير غائب، تبحث عن اسم معرَّف سابق عبر المستند كاملاً
يطابقه جنساً وعدداً، مع تفعيل قاعدة "جمع غير العاقل يُعامَل معاملة
المفرد المؤنث". الاختيار بين عدة مرشحين: الأبرز نحوياً أولاً (انظر
_candidate_prominence_depth)، لا الأقرب نصياً فقط."""
stream, sentence_tokens, sentence_children = _build_document_stream(sentences)
resolution: Dict[Tuple[int, int], Tuple[int, Token]] = {}
for i, (s_idx, tok) in enumerate(stream):
surface = dediac_ar(_clean_form(tok.form))
if surface not in _THIRD_PERSON_PRONOUN_AGREEMENT or not _is_pronoun(tok):
continue
target_gen, target_num = _THIRD_PERSON_PRONOUN_AGREEMENT[surface]
direct_candidates: List[Tuple[int, int, int, Token]] = [] # (depth, -j, cand_s_idx, cand)
nh_plural_candidates: List[Tuple[int, int, int, Token]] = []
for j in range(i - 1, -1, -1):
cand_s_idx, cand = stream[j]
if not _is_nominal(cand) or _is_pronoun(cand):
continue
# مرشح لا يصلح مرجعاً إن كان هو نفسه نعتاً تابعاً - "المخطط"
# في "الليف ... المخطط" صفة تصف "الليف"، لا كياناً مستقلاً.
if cand.deprel == "MOD":
continue
cand_surface = dediac_ar(_clean_form(cand.form))
bare_surface = cand_surface[2:] if cand_surface.startswith("ال") else cand_surface
if bare_surface in _CARDINAL_NUMBER_WORDS:
continue
if not _is_definite_token(cand, sentence_tokens[cand_s_idx], sentence_children[cand_s_idx]):
continue
cand_gen, cand_num = _token_gender_number(cand)
if cand_gen is None or cand_num is None:
continue
nh_plural_match = cand_num == "p" and target_gen == "f" and target_num == "s"
direct_match = cand_gen == target_gen and cand_num == target_num
if not (nh_plural_match or direct_match):
continue
depth = _candidate_prominence_depth(cand, sentence_tokens[cand_s_idx])
entry = (depth, -j, cand_s_idx, cand)
if nh_plural_match:
nh_plural_candidates.append(entry)
if direct_match:
direct_candidates.append(entry)
# سقف ٦ مرشحين يكفي لالتقاط الأبرز ضمن نطاق معقول بلا مسح
# المستند كاملاً لكل ضمير.
if len(direct_candidates) + len(nh_plural_candidates) >= 6:
break
# "ها"/"هي" تحتمل قراءتين (مؤنث مفرد، أو جمع غير عاقل)؛ نُفضّل
# قراءة الجمع لأنها الأشيع ("أكبرها" = "أكبر منهم"). ضمن كل
# مجموعة، الأبرز نحوياً يفوز، والأقرب نصياً يكسر التعادل.
if surface in ("ها", "هي"):
pool = nh_plural_candidates or direct_candidates
else:
pool = direct_candidates + nh_plural_candidates
if pool:
pool.sort(key=lambda e: (e[0], -e[1]))
_depth, _negj, best_s_idx, best_cand = pool[0]
resolution[(s_idx, tok.tid)] = (best_s_idx, best_cand)
return resolution
def build_antecedent_text_map(sentences: List[Sentence]) -> Dict[Tuple[int, int], str]:
"""خريطة (رقم جملة الضمير، معرّفه) -> النص الكامل للمُحال إليه (عبارته
الاسمية كاملة بنعوتها وإضافاتها، لا التوكن وحده). تُستهلَك مباشرة
داخل extract_entities_and_relations لإثراء الكيانات التي تفقد ضميرها.
"""
resolution = resolve_pronoun_antecedents(sentences)
text_map: Dict[Tuple[int, int], str] = {}
for (pron_s_idx, pron_tid), (ant_s_idx, antecedent) in resolution.items():
ant_tokens = _by_id(sentences[ant_s_idx])
ant_children = _children_map(sentences[ant_s_idx])
text = _walk(antecedent.tid, ant_tokens, ant_children, set(), [], root_id=antecedent.tid)
if text:
text_map[(pron_s_idx, pron_tid)] = text
return text_map
# --------------------------------------------------------------------------- #
# 3. قوائم كلمات التوقف
# --------------------------------------------------------------------------- #
# HARD: عديمة المعنى دائماً حتى بعد تخصيصها ("بعض الأنظمة" -> "الأنظمة").
# WEAK: عامة جداً بمفردها، لكن تبقى إن أُلحق بها نعت/إضافة حقيقية
# ("سياق متصل" تبقى كاملة، "سياق" وحدها تُسقط).
# --------------------------------------------------------------------------- #
_HARD_STOP_FORMS = {
# كمّيات ومحددات لا معنى لها أبداً
"بعض", "عض", "كل", "جميع", "كافة", "كافه", "عدة", "عده", "عديد",
"نوع", "نوعين", "كثير", "كثيره", "كثيرة", "قليل", "قليلة",
"مختلف", "مختلفه", "مختلفة", "غير", "سوى", "نفس", "ذات", "اخر",
# أسماء موصولة وإشارة - وصل نحوي بحت، ليست محتوى أبداً
"الذي", "التي", "والذي", "والتي", "اللذان", "اللتان", "الذين", "اللاتي",
"هذا", "هذه", "ذلك", "تلك", "هؤلاء", "أولئك",
# ظروف مكان/زمان تعمل كحرف جر نحوياً - العلاقة لا الكلمة هي المهمة
"بعد", "قبل", "وسط", "حول", "عند", "بين", "خلال", "داخل", "خارج",
"فوق", "تحت", "حيث", "هنا", "هناك", "قرب", "لدى",
# أدوات خطابية/مقارنة لا تحمل معنى كياني أبداً بذاتها
"اما", "أما", "مثل", "كمثل",
# أسماء تأطير خطابي: روابط نصية بحتة، لا تصلح رأساً لكيان حتى بعد
# وصفها بنعت ("سياق متصل" ليست كياناً أكثر من "سياق" وحدها)
"سياق", "اطار", "إطار", "نطاق", "صعيد",
# ظروف حال/تكرار جامدة
"كذلك", "ايضا", "موجود", "موجوده", "موجودة", "يوجد",
"جدا", "حقا", "تماما", "نسبيا", "غالبا", "فقط", "دائما", "احيانا",
"مختلف", "مختلفه", "مختلفة", "غير", "سوى", "نفس", "ذات", "اخر",
"سيما", "لا سيما", "خاصة", "بخاصة",
# أدوات ربط خطابية بصيغة اسمية - "إضافة إلى" (علاوة على) يُحلَّلها
# المحلل كأنها اسم متبوع بحرف جر؛ "وجود X" لا يحمل معنى مستقلاً عن X.
"اضافة", "إضافة", "وجود",
}
_WEAK_HEAD_FORMS = {
"أخرى", "اخرى", "رغم", "برغم", "رغما", "بواسطة", "بواسطه", "واسطة",
"واسطه", "فضل", "بفضل", "جهة", "طريق",
}
STOP_LEMMAS = frozenset(dediac_ar(w) for w in _HARD_STOP_FORMS | _WEAK_HEAD_FORMS)
HARD_STOP_LEMMAS = frozenset(dediac_ar(w) for w in _HARD_STOP_FORMS)
WEAK_HEAD_LEMMAS = frozenset(dediac_ar(w) for w in _WEAK_HEAD_FORMS)
# "نفس"/"ذات": لمتان مشتركتان مع كلمات انعكاسية مختلفة المعنى ("أنفسهم")،
# فتُفحَصان بالشكل السطحي فقط لا باللمة، تفادياً لالتباس المعنى.
_AMBIGUOUS_LEMMA_STOPWORDS = {"نفس", "ذات"}
def _is_stopword(tok: Token) -> bool:
surface = dediac_ar(_clean_form(tok.form))
if surface in STOP_LEMMAS:
return True
if tok.lemma and tok.lemma in STOP_LEMMAS and tok.lemma not in _AMBIGUOUS_LEMMA_STOPWORDS:
return True
return False
def _is_bare_stopword_result(tok: Token, text: str) -> bool:
"""True فقط حين يكون النص المُعاد هو شكل الكلمة-المتوقَّفة نفسه بلا
إضافة، أي لم يُنقَذ محتوى حقيقي من تحتها (بخلاف "كل عنصر" -> "عنصر")."""
return _is_stopword(tok) and text == dediac_ar(_clean_form(tok.form))
def _is_hard_stopword(tok: Token) -> bool:
surface = dediac_ar(_clean_form(tok.form))
if surface in HARD_STOP_LEMMAS:
return True
if tok.lemma and tok.lemma in HARD_STOP_LEMMAS and tok.lemma not in _AMBIGUOUS_LEMMA_STOPWORDS:
return True
return False
# --------------------------------------------------------------------------- #
# 4. المشي على شجرة الاعتماد
# --------------------------------------------------------------------------- #
@dataclass
class ExtractionResult:
entities: List[str] = field(default_factory=list)
relations: List[Tuple[str, str, str]] = field(default_factory=list)
def _walk(
head_id: int,
tokens: Dict[int, Token],
children: Dict[int, List[int]],
consumed: Set[int],
relations: List[Tuple[int, str, str]],
root_id: Optional[int] = None,
_depth: int = 0,
sentence_idx: int = 0,
antecedent_text_map: Optional[Dict[Tuple[int, int], str]] = None,
) -> str:
if root_id is None:
root_id = head_id
if _depth > 12 or head_id in consumed:
return _clean_form(tokens[head_id].form) if head_id in tokens else ""
consumed.add(head_id)
head_tok = tokens[head_id]
head_was_stopword = _is_stopword(head_tok)
head_is_hard = _is_hard_stopword(head_tok)
# HARD (بعض، فوق، الذي...): لا تُعرض الكلمة نفسها أبداً، حتى لو كان
# تحتها محتوى حقيقي - "بعض الأنظمة" يجب أن تصبح "الأنظمة"، ليس
# "عض الأنظمة". WEAK وغير كلمات التوقف: تُعرض الكلمة دائماً في البداية،
# ويُقرَّر إسقاطها لاحقاً بعد رؤية الناتج الكامل (انظر الفحص بعد الحلقة).
span_words = [] if head_is_hard else [_clean_form(head_tok.form)]
for child_id in sorted(children.get(head_id, [])):
child = tokens.get(child_id)
if child is None or child_id in consumed:
continue
if _is_pronoun(child):
consumed.add(child_id)
# الضمير المتصل غالباً يحمل استمرارية الموضوع عبر الجمل
# ("يحيط به غشاء الليف" - "به" يعود على رأس الجملة السابقة).
# إسقاطه بصمت يقطع الرابط بين الكيانات فينتج رسماً بيانياً
# مجزَّأً لجزر معزولة بدل بنية مترابطة - إن حُلَّت إحالته
# مسبقاً، نُلحق نص المُحال إليه بدل تجاهله كلياً.
if antecedent_text_map:
resolved = antecedent_text_map.get((sentence_idx, child_id))
if resolved and resolved not in span_words:
span_words.append(resolved)
continue
if _is_particle(child):
if _is_conjunction(child):
continue
_emit_pp_edge(child_id, tokens, children, consumed, relations, root_id, _depth,
sentence_idx, antecedent_text_map)
continue
if child.deprel in _NP_INTERNAL_RELS and _is_nominal(child):
if _is_stopword(child):
consumed.add(child_id)
for gc_id in sorted(children.get(child_id, [])):
gc = tokens.get(gc_id)
if gc is None or gc_id in consumed:
continue
if _is_particle(gc):
# استثناء حرف العطف هنا أيضاً، مطابقةً للمسار
# الرئيسي: كلمة "شفافة" قد يكون لها معطوف حقيقي
# عبر "و/ف" لا حرف جر - وكان هذا المسار يُعامل أي
# جسيم كحرف جر بلا تمييز، فيُصدر "و" كعلاقة وهمية.
if _is_conjunction(gc):
continue
_emit_pp_edge(gc_id, tokens, children, consumed, relations, root_id, _depth,
sentence_idx, antecedent_text_map)
elif gc.deprel in _NP_INTERNAL_RELS and _is_nominal(gc) and not _is_pronoun(gc):
# الكلمة المتوقَّفة نفسها اسم ظرفي شبه-حرف-جر (داخل/
# خارج/فوق/تحت/بين/لدى...) وما يكمّلها متصل بها
# مباشرة بعلاقة IDF/MOD (إضافة) لا عبر حرف جر منفصل
# - نفس معالجة _emit_pp_edge لكن لاسم لا لحرف جر.
_emit_nominal_edge(gc_id, tokens, children, consumed,
relations, root_id, _depth, child,
sentence_idx, antecedent_text_map)
elif _is_pronoun(gc) and antecedent_text_map:
# متمِّم الاسم الظرفي ضمير متصل ("داخله" = "داخل" +
# "ه") لا اسم صريح - نمط شائع جداً ("وفي داخله هيولى
# ونوى")، غالباً ما يربط جملة بموضوع جملة سابقة عبر
# الإحالة المرجعية.
resolved = antecedent_text_map.get((sentence_idx, gc.tid))
if resolved:
consumed.add(gc_id)
relations.append((root_id, dediac_ar(_clean_form(child.form)), resolved))
continue
# معطوفات النعت/الإضافة ("الخصائص الفيزيائية والكيميائية") -
# نفس علة _collect_conjunct_heads تماماً، لكن هنا داخل امتصاص
# نعت متداخل ضمن مشي كيان آخر، لا عند فاعل/مفعول فعل مباشرة.
# نجمع كل المعطوفات في نص واحد بدل الاكتفاء بالمعطوف الأول.
conjunct_ids = _collect_conjunct_heads(child_id, tokens, children, consumed)
sub_texts = []
for cid in sorted(conjunct_ids):
t = _walk(cid, tokens, children, consumed, relations,
root_id=root_id, _depth=_depth + 1,
sentence_idx=sentence_idx, antecedent_text_map=antecedent_text_map)
if t:
sub_texts.append(t)
if sub_texts:
span_words.append(" و".join(sub_texts))
continue
# WEAK فقط: إن لم يبقَ تحت الرأس أي محتوى حقيقي (لا يزال عارياً تماماً)،
# يُسقط بالكامل - هذا ما تعنيه "ضعيفة": لا معنى لها عند التجرد.
# HARD تُسقط دائماً بلا شرط (السطر الذي يبنيها لم يضع الكلمة أصلاً).
only_head_survived = len(span_words) == 1 and not head_is_hard
if head_was_stopword and (head_is_hard and not span_words):
return ""
if head_was_stopword and only_head_survived:
return ""
clean_span = []
for w in span_words:
if not clean_span or clean_span[-1] != w:
clean_span.append(w)
return " ".join(clean_span)
def _emit_pp_edge(
particle_id: int,
tokens: Dict[int, Token],
children: Dict[int, List[int]],
consumed: Set[int],
relations: List[Tuple[int, str, str]],
root_id: int,
depth: int,
sentence_idx: int = 0,
antecedent_text_map: Optional[Dict[Tuple[int, int], str]] = None,
) -> None:
if particle_id in consumed:
return
consumed.add(particle_id)
particle = tokens[particle_id]
obj_id = next(
(c for c in sorted(children.get(particle_id, []))
if tokens.get(c) and _looks_like_pp_object(tokens[c])),
None,
)
if obj_id is None:
# لا متمّم اسمي مباشر - قد يكون المتمّم ضميراً متصلاً يعود على
# كيان سابق ("يحيط به" = "يحيط بـ[الليف]")؛ حلّه عبر الإحالة
# المرجعية هو غالباً الرابط الذي يصل جملة بأخرى ويُبقي الرسم
# البياني متصلاً بدل مُجزَّأ.
pronoun_id = next(
(c for c in sorted(children.get(particle_id, []))
if tokens.get(c) and _is_pronoun(tokens[c])),
None,
)
if pronoun_id is not None and antecedent_text_map:
resolved = antecedent_text_map.get((sentence_idx, pronoun_id))
if resolved:
consumed.add(pronoun_id)
relations.append((root_id, dediac_ar(_clean_form(particle.form)), resolved))
return
# استخدام _walk_entities بدل _walk المباشرة: لو كان متمّم حرف الجر
# معطوفاً ("...في التربة والهواء") تُصدر علاقة مستقلة لكل معطوف بدل
# دمجها في نص واحد أو فقدان ما بعد الأول (انظر _collect_conjunct_heads).
for cid, target_text in _walk_entities(obj_id, tokens, children, consumed,
relations, _depth=depth + 1,
sentence_idx=sentence_idx,
antecedent_text_map=antecedent_text_map):
if not target_text or _is_bare_stopword_result(tokens[cid], target_text):
continue
# حرف الجر "ب" + متمّم عارٍ تماماً (لا نعت ولا إضافة) هو نمط ظرف
# حال/أداة فارغ المعنى ("بكفاءة"، "بدقة"، "بنجاح") لا علاقة حقيقية.
# حروف الجر المكانية/الموضوعية الأخرى (في/من/على) غالباً تحمل
# محتوى حقيقياً ("في الفيزياء")، فلا تُقيَّد بهذا الشرط.
is_bare = target_text == dediac_ar(_clean_form(tokens[cid].form))
# "بشكل X" (أساسي/عام/كبير...) ظرف حال جامد دائماً بصرف النظر عن
# وجود نعت يلحق بـ"شكل"، فلا يكفي فحص العَرْي وحده لهذه الحالة.
is_shakl_idiom = dediac_ar(_clean_form(tokens[cid].form)) == "شكل"
if dediac_ar(_clean_form(particle.form)) == "ب" and (is_bare or is_shakl_idiom):
continue
relations.append((root_id, dediac_ar(_clean_form(particle.form)), target_text))
def _emit_nominal_edge(
obj_id: int,
tokens: Dict[int, Token],
children: Dict[int, List[int]],
consumed: Set[int],
relations: List[Tuple[int, str, str]],
root_id: int,
depth: int,
governor: Token,
sentence_idx: int = 0,
antecedent_text_map: Optional[Dict[Tuple[int, int], str]] = None,
) -> None:
"""مثل _emit_pp_edge لكن للحالة التي يكون فيها "حرف الجر" اسماً ظرفياً
(داخل/خارج/فوق/تحت/بين/خلال/قرب/لدى...) لا حرفاً حقيقياً - CATiB يصل
مُتمِّم هذه الأسماء بعلاقة IDF/MOD (إضافة نحوية عادية) لا OBJ عبر حرف
جر مستقل، فتحتاج مساراً منفصلاً عن _emit_pp_edge التي تبحث عن حرف جر
حقيقي فقط بين الأحفاد (مثال: "داخل النواة" - "النواة" متمِّم لاسم
ظرفي "داخل"، لا حرف جر، فتحتاج هذه الدالة تحديداً لتصبح كياناً/علاقة)."""
if obj_id in consumed:
return
for cid, target_text in _walk_entities(obj_id, tokens, children, consumed,
relations, _depth=depth + 1,
sentence_idx=sentence_idx,
antecedent_text_map=antecedent_text_map):
if target_text and not _is_bare_stopword_result(tokens[cid], target_text):
relations.append((root_id, dediac_ar(_clean_form(governor.form)), target_text))
def _collect_conjunct_heads(
head_id: int,
tokens: Dict[int, Token],
children: Dict[int, List[int]],
consumed: Set[int],
) -> List[int]:
"""تُرجع [head_id] مع كل رؤوس المعطوفات المتسلسلة عليه عبر حرف عطف
(و/ف/أو/ثم)، متتبّعة السلسلة عبر أي عمق. بلا هذه الدالة، أبناء الرأس
المعطوفة (المُقصاة عمداً من _walk العادية) تُفقَد بعد أول حرف عطف.
"""
result = [head_id]
seen = {head_id}
frontier = [head_id]
while frontier:
cur = frontier.pop(0)
for cid in sorted(children.get(cur, [])):
if cid in consumed or cid in seen:
continue
ctok = tokens.get(cid)
if ctok is None:
continue
if _is_particle(ctok) and _is_conjunction(ctok):
consumed.add(cid)
for gcid in sorted(children.get(cid, [])):
if gcid in consumed or gcid in seen:
continue
gtok = tokens.get(gcid)
if gtok is None:
continue
if _is_nominal(gtok) and not _is_pronoun(gtok):
result.append(gcid)
seen.add(gcid)
frontier.append(gcid)
return result
def _walk_entities(
head_id: int,
tokens: Dict[int, Token],
children: Dict[int, List[int]],
consumed: Set[int],
relations: List[Tuple[int, str, str]],
_depth: int = 0,
sentence_idx: int = 0,
antecedent_text_map: Optional[Dict[Tuple[int, int], str]] = None,
) -> List[Tuple[int, str]]:
"""مثل _walk، لكنها تُرجع قائمة (رأس المعطوف، نصه) بدل نص واحد مُدمَج
- تُفكِّك أي تعداد معطوف إلى كيانات/علاقات مستقلة. نقطة العبور
المركزية أيضاً لحل الإحالة: إن كان الرأس ضميراً مستقلاً ("وهي التي
تتسطح...")، تُحاول حل إحالته هنا مباشرة بدل إرجاعه كنص حرفي غير مفيد.
"""
head_tok = tokens.get(head_id)
if head_tok is not None and _is_pronoun(head_tok) and head_id not in consumed:
consumed.add(head_id)
if antecedent_text_map:
resolved = antecedent_text_map.get((sentence_idx, head_id))
if resolved:
return [(head_id, resolved)]
return []
out: List[Tuple[int, str]] = []
for cid in _collect_conjunct_heads(head_id, tokens, children, consumed):
text = _walk(cid, tokens, children, consumed, relations, root_id=cid, _depth=_depth,
sentence_idx=sentence_idx, antecedent_text_map=antecedent_text_map)
text = re.sub(r"\s+", " ", text).strip()
if text:
out.append((cid, text))
return out
def extract_entities_and_relations(
sentence: Sentence,
sentence_idx: int = 0,
antecedent_text_map: Optional[Dict[Tuple[int, int], str]] = None,
) -> ExtractionResult:
"""sentence_idx وantecedent_text_map اختياريان؛ حين يُمرَّران، يُثريان
كيانات فقدت ضميرها المتصل بنص المُحال إليه (عبر
resolve_pronoun_antecedents) بدل بقائها ناقصة.
"""
tokens = _by_id(sentence)
children = _children_map(sentence)
consumed: Set[int] = set()
raw_relations: List[Tuple[int, str, str]] = []
comprehensive_relations: List[Tuple[str, str, str]] = []
entities: List[str] = []
# يُحسَب مرة واحدة لكل الجملة: أي فعل يحمل أداة نفي مباشرة (لم/لن/لا/
# ليس) لا يُصدر أي علاقة إطلاقاً، لأن أي علاقة مستخرَجة منه ستكون
# معكوسة تماماً لما تقوله الجملة فعلاً.
negated_verb_ids = _find_negated_verb_ids(sentence, tokens, children)
# -----------------------------------------------------------------------
# المرحلة الأولى: التقاط الهياكل النحوية الشاملة (أفعال وجمل اسمية)
# -----------------------------------------------------------------------
for tok in sentence:
if tok.tid in negated_verb_ids:
continue
# 0. جملة اسمية برابط محذوف مُمثَّلة بديلاً: اسم تفضيل ("أكبر"،
# "أصغر"...) يُوسَم خطأً VRB ويتصل بالخبر عبر SBJ بدل أن يُتصَل به
# الخبر عبر TPC. الإشارة الآمنة: توكن VRB يحمل هو نفسه علاقة SBJ -
# الأفعال الحقيقية لا تُصنَّف SBJ لغيرها أبداً، فهذا النمط لا
# يتداخل مع معالجة الأفعال العادية أدناه.
if tok.cpos == "VRB" and tok.deprel == "SBJ" and not _is_stopword(tok):
own_children = children.get(tok.tid, [])
has_own_object_or_prep = any(
tokens[c].deprel == "OBJ" or (tokens.get(c) and _is_particle(tokens[c]))
for c in own_children
)
head_tok = tokens.get(tok.head)
if not has_own_object_or_prep and head_tok and _is_nominal(head_tok):
temp_consumed = set(consumed)
topic_text = _walk(tok.tid, tokens, children, temp_consumed, [], root_id=tok.tid,
sentence_idx=sentence_idx, antecedent_text_map=antecedent_text_map)
pred_text = _walk(head_tok.tid, tokens, children, temp_consumed, [], root_id=head_tok.tid,
sentence_idx=sentence_idx, antecedent_text_map=antecedent_text_map)
if topic_text and pred_text:
comprehensive_relations.append((topic_text, "هو", pred_text))
entities.append(topic_text)
entities.append(pred_text)
continue
# ٠أ. جملة اسمية برابط محذوف باسم عادي - اسم (لا فعل) بعلاقة SBJ
# لرأس اسمي آخر نمط غير اعتيادي (الأفعال هي من تملك فاعلاً SBJ
# عادة)، يدل بذاته على جملة "X [هو] Y" ("الشبكية: طبقة حساسة").
# الاستهلاك محلي عمداً (لا يُدمَج) ليبقى الرأس متاحاً كفاعل ضمني
# كامل لفعل لاحق يتصل به مباشرة (كـ"تحتوي" أدناه).
if (tok.deprel == "SBJ" and _is_nominal(tok) and tok.cpos != "VRB"
and not _is_stopword(tok)):
head_tok = tokens.get(tok.head)
if head_tok and _is_nominal(head_tok) and head_tok.cpos != "VRB":
temp_consumed = set(consumed)
topic_text = _walk(tok.tid, tokens, children, temp_consumed, [], root_id=tok.tid,
sentence_idx=sentence_idx, antecedent_text_map=antecedent_text_map)
pred_text = _walk(head_tok.tid, tokens, children, temp_consumed, [], root_id=head_tok.tid,
sentence_idx=sentence_idx, antecedent_text_map=antecedent_text_map)
if topic_text and pred_text:
comprehensive_relations.append((topic_text, "هو", pred_text))
entities.append(topic_text)
entities.append(pred_text)
continue
# ٠جـ. فعل تسمية مبني للمجهول ("سميت"، "تسمى") - فاعله النحوي
# الاسم الجديد المُسنَد، لا الموضوع نفسه. الموضوع ضمني: فاعل الفعل
# الحاكم إن اتصلت بفعل ("سميت: العضلات المخططة")، أو الاسم الحاكم
# نفسه إن اتصلت باسم مباشرة ("خلايا ... تسمى: العصي والمخاريط").
_NAMING_VERB_FORMS = {
"سمي", "سميت", "سميا", "سموا", "يسمى", "تسمى",
"دعي", "دعيت", "دعيا", "دعوا", "يدعى", "تدعى",
}
if (tok.cpos in ("VRB", "VRB-PASS")
and dediac_ar(_clean_form(tok.form)) in _NAMING_VERB_FORMS
and tok.deprel == "MOD"):
# الاسم المُسنَد يتصل عبر SBJ ("سميت: العضلات المخططة") أو
# MOD ("خلايا ... تسمى: العصي والمخاريط") - نمطان بنيويان.
name_id = next((c for c in children.get(tok.tid, []) if tokens[c].deprel == "SBJ"), None)
if name_id is None:
name_id = next(
(c for c in children.get(tok.tid, [])
if tokens.get(c) and tokens[c].deprel == "MOD"
and _is_nominal(tokens[c]) and not _is_pronoun(tokens[c])),
None,
)
# (ب) أولاً: يتعامل تلقائياً مع اتصال مباشر باسم (يُرجعه فوراً)
# أو عبر "الذي/التي" موصولة صريحة.
topic_id = _find_implicit_subject(tok.tid, tokens, children)
if topic_id is None and tok.head in tokens and tokens[tok.head].cpos in ("VRB", "VRB-PASS"):
# (أ): الفعل الحاكم فعل آخر لا اسم (تنسيق/تتابع جملتين) -
# الموضوع الحقيقي فاعل ذلك الفعل الحاكم نفسه.
root_verb = tokens[tok.head]
topic_id = next((c for c in children.get(root_verb.tid, []) if tokens[c].deprel == "SBJ"), None)
if name_id and topic_id:
temp_consumed = set(consumed)
name_pairs = _walk_entities(name_id, tokens, children, temp_consumed, [],
sentence_idx=sentence_idx, antecedent_text_map=antecedent_text_map)
topic_pairs = _walk_entities(topic_id, tokens, children, temp_consumed, [],
sentence_idx=sentence_idx, antecedent_text_map=antecedent_text_map)
name_texts = [t for _, t in name_pairs]
topic_texts = [t for _, t in topic_pairs]
if name_texts and topic_texts:
verb_text = dediac_ar(_clean_form(tok.form))
for topic_text in topic_texts:
for name_text in name_texts:
comprehensive_relations.append((topic_text, verb_text, name_text))
entities.extend(topic_texts)
entities.extend(name_texts)
consumed |= temp_consumed
continue
# ٠ب. بناء وجودي: حرف جر يحمل فاعلاً (SBJ) مباشرة بلا فعل ظاهر
# ("وفي داخله هيولى ونوى" = "وفي داخله [توجد] هيولى ونوى").
if tok.cpos == "PRT" and not _is_stopword(tok) and not _is_conjunction(tok):
exist_sbj_id = next((c for c in children.get(tok.tid, []) if tokens[c].deprel == "SBJ"), None)
exist_obj_id = next(
(c for c in children.get(tok.tid, []) if tokens.get(c) and _looks_like_pp_object(tokens[c])),
None,
)
if exist_sbj_id and exist_obj_id:
temp_consumed = set(consumed)
exist_sbj_pairs = _walk_entities(exist_sbj_id, tokens, children, temp_consumed, [],
sentence_idx=sentence_idx, antecedent_text_map=antecedent_text_map)
exist_obj_pairs = _walk_entities(exist_obj_id, tokens, children, temp_consumed, [],
sentence_idx=sentence_idx, antecedent_text_map=antecedent_text_map)
exist_sbj_texts = [t for _, t in exist_sbj_pairs]
exist_obj_texts = [t for _, t in exist_obj_pairs]
if exist_sbj_texts and exist_obj_texts:
prep_text = dediac_ar(_clean_form(tok.form))
# متمّم حرف الجر اسم ظرفي شبه-حرف-جر بذاته ("داخل" في
# "في داخله") - يُسقَط من نصه الخاص (كلمة توقف صلدة)
# فيبقى الهدف نظيفاً ("الليف العضلي المخطط")، ويُضاف
# صراحة إلى تسمية العلاقة نفسها لتصبح "في داخل" الأدق
# دلالياً بدل "في" وحدها.
exist_obj_tok = tokens.get(exist_obj_id)
if (exist_obj_tok and _is_nominal(exist_obj_tok)
and _is_hard_stopword(exist_obj_tok)):
prep_text = f"{prep_text} {dediac_ar(_clean_form(exist_obj_tok.form))}"
for s in exist_sbj_texts:
for o in exist_obj_texts:
comprehensive_relations.append((s, prep_text, o))
entities.extend(exist_sbj_texts)
entities.extend(exist_obj_texts)
continue
# 1. الأفعال المبنية للمعلوم (Active Verbs)
if tok.cpos == "VRB":
sbj_id = next((c for c in children.get(tok.tid, []) if tokens[c].deprel == "SBJ"), None)
# أفعال جملة الصلة (غشاء ... يحمي الخلية، الميتوكوندريا التي
# تعد مصدر الطاقة) لا فاعل صريح لها في الشجرة - فاعلها الحقيقي
# هو الاسم الذي يتصل به الفعل مباشرة كصفة/صلة، أو عبر "الذي/
# التي" الموصولة. الفاعل الصريح دائماً أولى إن وُجد.
implicit_sbj_id = sbj_id or _find_implicit_subject(tok.tid, tokens, children)
# نجمع كل أبناء OBJ (لا الأول فقط عبر next()) لأن فعلاً واحداً
# قد يحمل مفعولاً به ومفعولاً مطلقاً معاً، كلاهما بعلاقة OBJ.
# نتحقق أيضاً من _is_nominal لأن أدوات مصدرية مثل "أن" (اعتقد
# أن...) تحمل نفس علاقة OBJ رغم كونها حرفاً لا اسماً.
obj_ids = [c for c in children.get(tok.tid, [])
if tokens[c].deprel == "OBJ" and _is_nominal(tokens[c])
and not _is_pronoun(tokens[c])]
obj_id = obj_ids[0] if obj_ids else None
extra_obj_ids = obj_ids[1:]
# حرف الجر الحقيقي (PRT)، أو اسم ظرفي شبه-حرف-جر (داخل/خارج/
# فوق/تحت...) متصل بالفعل عبر MOD ("تصطف ... داخل الليف").
def _is_bi_haythu(prt_id: int) -> bool:
""""بحيث" أداة ربط خطابية مركَّبة، لا متمّم حرف جر حقيقي -
يجب استبعادها من ترشّح prep_id. الفحص عام (كلمة توقف لا
محتوى تحتها) لا تطابق نصي لـ"حيث" وحدها، لاختلاف شكلها
السطحي المحتمل بعد ترقيع NOAN.
"""
prt = tokens.get(prt_id)
if prt is None or dediac_ar(_clean_form(prt.form)) != "ب":
return False
obj_child = next((c for c in children.get(prt_id, []) if tokens.get(c)), None)
if obj_child is None:
return False
obj_tok = tokens[obj_child]
obj_surface = dediac_ar(_clean_form(obj_tok.form))
if obj_surface == "NOAN":
# فشل ترقيع NOAN نفسه (بقي المُعرِّف الداخلي كما هو) -
# "بـ" + رمز غير محلول لا يصلح متمّماً حقيقياً بأي حال.
return True
if not _is_stopword(obj_tok):
return False
probe_text = _walk(obj_child, tokens, children, set(), [], root_id=obj_child)
return not probe_text or probe_text == obj_surface
prep_id = next(
(c for c in children.get(tok.tid, [])
if tokens.get(c) and _is_particle(tokens[c]) and not _is_bi_haythu(c)),
None,
)
prep_is_nominal_relator = False
if prep_id is None:
nominal_prep_id = next(
(c for c in children.get(tok.tid, [])
if tokens.get(c) and tokens[c].deprel == "MOD"
and _is_nominal(tokens[c]) and _is_hard_stopword(tokens[c])),
None,
)
if nominal_prep_id is not None:
prep_id = nominal_prep_id
prep_is_nominal_relator = True
# [كذا]") - المفعول الحقيقي الوحيد ضمير، فيُستبعَد obj_ids
# أعلاه عمداً؛ هذا يستعيده كنص جاهز حين تتوفر خريطة إحالة.
resolved_pronoun_obj = None
if not obj_id and antecedent_text_map:
pronoun_obj_id = next(
(c for c in children.get(tok.tid, [])
if tokens.get(c) and tokens[c].deprel == "OBJ" and _is_pronoun(tokens[c])),
None,
)
if pronoun_obj_id is not None:
resolved_pronoun_obj = antecedent_text_map.get((sentence_idx, pronoun_obj_id))
# أ- فعل متعدٍّ (فاعل + فعل + مفعول به)، بالفاعل الضمني إن غاب
# الصريح. _walk_entities بدل _walk: فاعل/مفعول معطوف يُصدر
# علاقة مستقلة لكل عضو (جداء ديكارتي)، لا دمجاً في كيان واحد.
if implicit_sbj_id and (obj_id or resolved_pronoun_obj):
temp_consumed = set(consumed)
sbj_pairs = _walk_entities(implicit_sbj_id, tokens, children, temp_consumed, [],
sentence_idx=sentence_idx, antecedent_text_map=antecedent_text_map)
if obj_id:
obj_pairs = _walk_entities(obj_id, tokens, children, temp_consumed, [],
sentence_idx=sentence_idx, antecedent_text_map=antecedent_text_map)
for extra_id in extra_obj_ids:
obj_pairs.extend(
_walk_entities(extra_id, tokens, children, temp_consumed, [],
sentence_idx=sentence_idx, antecedent_text_map=antecedent_text_map)
)
obj_texts = [t for _, t in obj_pairs]
else:
obj_texts = [resolved_pronoun_obj]
sbj_texts = [t for _, t in sbj_pairs]
if sbj_texts and obj_texts and not _is_stopword(tok):
verb_text = dediac_ar(_clean_form(tok.form))
for s in sbj_texts:
for o in obj_texts:
comprehensive_relations.append((s, verb_text, o))
entities.extend(sbj_texts)
entities.extend(obj_texts)
# ب- فعل لازم مع حرف جر (فاعل + فعل + حرف جر + اسم مجرور، أو
# حرف جر بمتمّم ضميري مُحال - "يحيط به غشاء الليف")
elif implicit_sbj_id and prep_id:
if prep_is_nominal_relator:
# الاسم الظرفي يأخذ متمِّمه عبر IDF (مضاف إليه)، لا
# OBJ كحرف الجر الحقيقي - "داخل الليف العضلي" إضافة
# نحوية عادية.
prep_obj_id = next(
(c for c in children.get(prep_id, [])
if tokens.get(c) and tokens[c].deprel == "IDF"
and _is_nominal(tokens[c]) and not _is_pronoun(tokens[c])),
None,
)
else:
prep_obj_id = next((c for c in children.get(prep_id, []) if tokens.get(c) and _looks_like_pp_object(tokens[c])), None)
prep_obj_texts: List[str] = []
if prep_obj_id:
temp_consumed = set(consumed)
sbj_pairs = _walk_entities(implicit_sbj_id, tokens, children, temp_consumed, [],
sentence_idx=sentence_idx, antecedent_text_map=antecedent_text_map)
prep_obj_pairs = _walk_entities(prep_obj_id, tokens, children, temp_consumed, [],
sentence_idx=sentence_idx, antecedent_text_map=antecedent_text_map)
sbj_texts = [t for _, t in sbj_pairs]
prep_obj_texts = [t for _, t in prep_obj_pairs]
elif antecedent_text_map:
# لا متمّم اسمي مباشر - جرّب ضميراً متصلاً بحرف الجر
pronoun_id = next(
(c for c in children.get(prep_id, []) if tokens.get(c) and _is_pronoun(tokens[c])),
None,
)
resolved = antecedent_text_map.get((sentence_idx, pronoun_id)) if pronoun_id is not None else None
if resolved:
temp_consumed = set(consumed)
sbj_pairs = _walk_entities(implicit_sbj_id, tokens, children, temp_consumed, [],
sentence_idx=sentence_idx, antecedent_text_map=antecedent_text_map)
sbj_texts = [t for _, t in sbj_pairs]
prep_obj_texts = [resolved]
else:
sbj_texts = []
if prep_obj_texts and sbj_texts and not _is_stopword(tok):
verb_text = dediac_ar(_clean_form(tok.form))
prep_text = dediac_ar(_clean_form(tokens[prep_id].form))
relation_label = f"{verb_text} {prep_text}"
for s in sbj_texts:
for o in prep_obj_texts:
comprehensive_relations.append((s, relation_label, o))
entities.extend(sbj_texts)
entities.extend(prep_obj_texts)
# بب- فعل تبادلي بفاعل معطوف فقط، بلا مفعول ولا حرف جر
# ("تتحاذى الأقراص العاتمة وكذلك الأقراص النيرة") - العلاقة
# هنا بين أطراف الفاعل المعطوف نفسها.
elif implicit_sbj_id and not obj_id and not resolved_pronoun_obj and not prep_id and not _is_stopword(tok):
temp_consumed = set(consumed)
sbj_pairs = _walk_entities(implicit_sbj_id, tokens, children, temp_consumed, [],
sentence_idx=sentence_idx, antecedent_text_map=antecedent_text_map)
if len(sbj_pairs) >= 2:
sbj_texts = [t for _, t in sbj_pairs]
verb_text = dediac_ar(_clean_form(tok.form))
first_text = sbj_texts[0]
for other_text in sbj_texts[1:]:
comprehensive_relations.append((first_text, f"{verb_text} مع", other_text))
entities.extend(sbj_texts)
# جـ- فعل بلا فاعل حتى ضمني - لا نبني علاقة كاملة موثوقة، لكن
# مفعوله وأشباه جمله تستحق الظهور ككيانات مستقلة بدل الاختفاء.
elif not implicit_sbj_id and not _is_stopword(tok):
temp_consumed = set(consumed)
if obj_id:
obj_pairs = _walk_entities(obj_id, tokens, children, temp_consumed, [],
sentence_idx=sentence_idx, antecedent_text_map=antecedent_text_map)
for extra_id in extra_obj_ids:
obj_pairs.extend(_walk_entities(extra_id, tokens, children, temp_consumed, [],
sentence_idx=sentence_idx, antecedent_text_map=antecedent_text_map))
entities.extend(t for _, t in obj_pairs)
all_prep_ids = [c for c in children.get(tok.tid, [])
if tokens.get(c) and _is_particle(tokens[c])
and not _is_conjunction(tokens[c])]
for pid in all_prep_ids:
prep_obj_id = next((c for c in children.get(pid, [])
if tokens.get(c) and _looks_like_pp_object(tokens[c])), None)
if prep_obj_id:
prep_obj_pairs = _walk_entities(prep_obj_id, tokens, children, temp_consumed, [],
sentence_idx=sentence_idx, antecedent_text_map=antecedent_text_map)
entities.extend(t for _, t in prep_obj_pairs)
elif tok.cpos == "VRB-PASS":
sbj_id = next((c for c in children.get(tok.tid, []) if tokens[c].deprel == "SBJ"), None) # نائب الفاعل
prep_id = next((c for c in children.get(tok.tid, []) if tokens.get(c) and _is_particle(tokens[c])), None)
if sbj_id and prep_id:
prep_obj_id = next((c for c in children.get(prep_id, []) if tokens.get(c) and _looks_like_pp_object(tokens[c])), None)
if prep_obj_id:
temp_consumed = set(consumed)
sbj_pairs = _walk_entities(sbj_id, tokens, children, temp_consumed, [],
sentence_idx=sentence_idx, antecedent_text_map=antecedent_text_map)
prep_obj_pairs = _walk_entities(prep_obj_id, tokens, children, temp_consumed, [],
sentence_idx=sentence_idx, antecedent_text_map=antecedent_text_map)
sbj_texts = [t for _, t in sbj_pairs]
prep_obj_texts = [t for _, t in prep_obj_pairs]
if sbj_texts and prep_obj_texts:
verb_text = dediac_ar(_clean_form(tok.form))
prep_text = dediac_ar(_clean_form(tokens[prep_id].form))
relation_label = f"{verb_text} {prep_text}"
for s in sbj_texts:
for o in prep_obj_texts:
comprehensive_relations.append((s, relation_label, o))
entities.extend(sbj_texts)
entities.extend(prep_obj_texts)
# 3. الجمل الاسمية (المبتدأ والخبر)
elif _is_nominal(tok):
prd_id = next((c for c in children.get(tok.tid, []) if tokens[c].deprel == "PRD"), None)
if prd_id:
temp_consumed = set(consumed)
subj_pairs = _walk_entities(tok.tid, tokens, children, temp_consumed, [],
sentence_idx=sentence_idx, antecedent_text_map=antecedent_text_map)
prd_pairs = _walk_entities(prd_id, tokens, children, temp_consumed, [],
sentence_idx=sentence_idx, antecedent_text_map=antecedent_text_map)
subj_texts = [t for _, t in subj_pairs]
prd_texts = [t for _, t in prd_pairs]
if subj_texts and prd_texts:
for s in subj_texts:
for p in prd_texts:
comprehensive_relations.append((s, "يتصف بـ/يعتبر", p))
entities.extend(subj_texts)
entities.extend(prd_texts)
# نستهلك الخبر هنا حتى لا يُعامل ككيان منفصل مقطوع السياق
consumed.add(prd_id)
# 4. جملة اسمية "مبتدأ ... خبر" حيث رأس الشجرة الخبر (اسم لا
# فعل)، والمبتدأ متصل به عبر TPC. الخبر يُمشى عبر _walk
# المفردة عمداً لا _walk_entities: معطوف يتصل بالرأس هنا غالباً
# جملة اسمية مستقلة أُلحقت بنيوياً، لا معطوفاً حقيقياً.
tpc_id = next((c for c in children.get(tok.tid, []) if tokens[c].deprel == "TPC"), None)
if tpc_id and not _is_stopword(tok):
temp_consumed = set(consumed)
tpc_pairs = _walk_entities(tpc_id, tokens, children, temp_consumed, [],
sentence_idx=sentence_idx, antecedent_text_map=antecedent_text_map)
tpc_texts = [t for _, t in tpc_pairs]
pred_text = _walk(tok.tid, tokens, children, temp_consumed, [], root_id=tok.tid,
sentence_idx=sentence_idx, antecedent_text_map=antecedent_text_map)
if tpc_texts and pred_text:
for s in tpc_texts:
comprehensive_relations.append((s, "هو", pred_text))
entities.extend(tpc_texts)
entities.append(pred_text)
# -----------------------------------------------------------------------
# المرحلة الثانية: استخراج الكيانات المتبقية وحروف الجر (الآلية الأصلية)
# -----------------------------------------------------------------------
for tok in sentence:
if tok.tid in consumed:
continue
if _is_pronoun(tok):
continue
if not _is_nominal(tok):
continue
if tok.deprel == "PRD":
consumed.add(tok.tid)
continue
# نعت حال متصل بفعل مباشرة ("تصطف اللييفات ... متوازية") - CATiB
# لا يملك وسماً مستقلاً للحال (كلاهما MOD)، لكن اسماً مرتبطاً بفعل
# مباشرة لا يكون نعتاً حقيقياً لاسم (الأفعال لا تُنعَت)، فهذا كافٍ
# للتمييز. تُستثنى كلمات التوقف ("إضافة"، "وجود") لأنها تُعالَج
# عبر آلية _walk الخاصة بها.
if (tok.deprel == "MOD" and tok.head in tokens
and tokens[tok.head].cpos in ("VRB", "VRB-PASS")
and not _is_stopword(tok)):
consumed.add(tok.tid)
continue
# كلمة توقف متصلة بفعل مباشرة ("إضافة إلى وجود ليفيات" متصلة بـ
# "يبدو") - نربط أي علاقة حقيقية تحتها بفاعل ذلك الفعل بدل تركها
# معلَّقة بمعرّف كلمة لا نص لها.
if (tok.deprel == "MOD" and tok.head in tokens
and tokens[tok.head].cpos in ("VRB", "VRB-PASS")
and _is_stopword(tok)):
consumed.add(tok.tid)
verb_tok = tokens[tok.head]
anchor_id = next((c for c in children.get(verb_tok.tid, []) if tokens[c].deprel == "SBJ"), None)
anchor_id = anchor_id or _find_implicit_subject(verb_tok.tid, tokens, children)
if anchor_id is not None:
temp_consumed = set(consumed)
anchor_pairs = _walk_entities(anchor_id, tokens, children, temp_consumed, [],
sentence_idx=sentence_idx, antecedent_text_map=antecedent_text_map)
anchor_texts = [t for _, t in anchor_pairs]
for child_id in sorted(children.get(tok.tid, [])):
child = tokens.get(child_id)
if child is None or child_id in consumed or not _is_particle(child) or _is_conjunction(child):
continue
anchor_relations: List[Tuple[int, str, str]] = []
_emit_pp_edge(child_id, tokens, children, temp_consumed, anchor_relations,
root_id=-1, depth=0, sentence_idx=sentence_idx,
antecedent_text_map=antecedent_text_map)
# "وجود X" محمول وجودي ("توجد X")، لا حرف جر بمعناه
# الحرفي، فتسمية العلاقة الأدق "يحوي" لا حرف الجر
# السطحي الذي ربطها نحوياً فقط.
passes_through_wujud = any(
tokens.get(c) and dediac_ar(_clean_form(tokens[c].form)) == "وجود"
for c in children.get(child_id, [])
)
for _rid, rel_label, rel_target in anchor_relations:
effective_label = "يحوي" if passes_through_wujud else rel_label
for anchor_text in anchor_texts:
comprehensive_relations.append((anchor_text, effective_label, rel_target))
entities.append(rel_target)
entities.extend(anchor_texts)
consumed |= temp_consumed
continue
# حرف الجر "ب" المرتبط بظرف حال ("تمكنوا بنجاح"، "سير العمليات
# بكفاءة") - نعتمد على شرط "لا محتوى حقيقي تحته" فقط، إذ CATiB6
# لا يميّز المصدر عن الاسم العادي، وهذا الشرط يحمي أسماء المحتوى
# الحقيقي ("الفيزياء" في "دراسة الفيزياء بعمق") من الإسقاط.
governor = tokens.get(tok.head)
if (governor and _is_particle(governor)
and dediac_ar(_clean_form(governor.form)) == "ب"):
grandgovernor = tokens.get(governor.head)
if grandgovernor and (_is_nominal(grandgovernor)
or grandgovernor.cpos in {"VRB", "VRB-PASS"}):
probe_consumed: Set[int] = set(consumed)
probe_relations: List[Tuple[int, str, str]] = []
probe_text = _walk(tok.tid, tokens, children, probe_consumed,
probe_relations, root_id=tok.tid,
sentence_idx=sentence_idx, antecedent_text_map=antecedent_text_map)
probe_text = re.sub(r"\s+", " ", probe_text).strip()
if probe_text == dediac_ar(_clean_form(tok.form)):
continue
# استخدام _walk_entities: كيان مستقل معطوف ("الماء والضوء والهواء"
# حين لا يلتقطه أي فعل) يجب أن يصبح ثلاثة كيانات منفصلة، لا كياناً
# واحداً مدمجاً ولا فقداً لكل ما بعد الأول.
start_idx = len(raw_relations)
pairs = _walk_entities(tok.tid, tokens, children, consumed, raw_relations,
sentence_idx=sentence_idx, antecedent_text_map=antecedent_text_map)
text_by_id = {cid: t for cid, t in pairs}
for i in range(start_idx, len(raw_relations)):
src_id, rel, tgt = raw_relations[i]
if src_id in text_by_id:
raw_relations[i] = (text_by_id[src_id], rel, tgt) # type: ignore
for cid, text in pairs:
src_tok = tokens[cid]
bare_stopword = _is_stopword(src_tok) and text == dediac_ar(_clean_form(src_tok.form))
if bare_stopword:
continue
entities.append(text)
# -----------------------------------------------------------------------
# المرحلة الثالثة: دمج العلاقات وتصفية NOAN والضمائر نهائياً
# -----------------------------------------------------------------------
relations = [(s, r, t) for s, r, t in raw_relations if isinstance(s, str)]
relations.extend(comprehensive_relations)
# فلتر صارم: منع NOAN، منع الكيانات القصيرة جداً (الضمائر المتسربة)، ومنع الجمل الكاملة (أكثر من 6 كلمات)
clean_entities = []
for e in entities:
words_count = len(e.split())
if "NOAN" not in e.upper() and len(e) > 2 and words_count <= 4:
if e not in _PRONOUN_FORMS: # تأكيد إضافي لمنع الضمائر
clean_entities.append(e)
clean_relations = []
for s, r, t in relations:
s_words, t_words = len(s.split()), len(t.split())
# تطبيق نفس الفلتر على أطراف العلاقات
if ("NOAN" not in s.upper() and "NOAN" not in t.upper() and
len(s) > 2 and len(t) > 2 and s_words <= 4 and t_words <= 4 and
s not in _PRONOUN_FORMS and t not in _PRONOUN_FORMS):
clean_relations.append((s, r, t))
return ExtractionResult(
entities=_dedupe_preserve_order(clean_entities),
relations=clean_relations
)
def _dedupe_preserve_order(items: List[str]) -> List[str]:
seen: Set[str] = set()
out: List[str] = []
for item in items:
if item not in seen:
seen.add(item)
out.append(item)
return out
def _normalize_conll_input(conll_text) -> str:
if isinstance(conll_text, (list, tuple)):
chunks = [str(c) for c in conll_text]
if any("\n" in c for c in chunks):
return "\n\n".join(chunks)
return "\n".join(chunks)
return conll_text
def debug_print_parsed_tree(conll_text) -> None:
for s_idx, sentence in enumerate(parse_conllx(_normalize_conll_input(conll_text)), 1):
tokens = _by_id(sentence)
print(f"--- sentence {s_idx} ---")
for tok in sentence:
governor = tokens.get(tok.head)
gov_text = governor.form if governor else "ROOT"
print(f" {tok.tid:>3} {tok.form:<15} pos={tok.cpos:<6} "
f"--[{tok.deprel}]--> {gov_text} (#{tok.head})")
print()
PROTECTED_GLUE_FRAGMENTS = frozenset(dediac_ar(w) for w in {
# نهايات كلمات معرَّبة/علمية شائعة يُخطئ المحلل الصرفي فيها فيفصل أول
# حرف (ب/ل/ك/و/ف) بصفته حرف جر أو عطف حقيقياً، تاركاً الباقي توكناً
# غريباً مستقلاً ("بلازمي" -> "ب" + "لازمي"، "كلور" -> "ك" + "لور").
# وسّع هذه القائمة بمصطلحات مجالك عند ظهور حالات جديدة.
"لازمي", "لازما", "لازم", # بلازمي / بلازما
"روتين", "روتينات", # بروتين / بروتينات
"كتيريا", # بكتيريا
"يولوجيا", "يولوجي", # بيولوجيا / بيولوجي
"وتاسيوم", # بوتاسيوم
"يبسين", # بيبسين
"لور", # كلور
"يتامين", # فيتامين
"يوكيميائي", "يوكيميائية", # بيوكيميائي(ة)
})
def _repair_split_loanwords(sentence: Sentence) -> None:
"""تصلح خطأ تحليل صرفي شائع مع الكلمات المعرَّبة: المحلل يفصل أول
حرف (ب/ل/ك/و/ف) بصفته حرف جر حقيقياً. حين يطابق الباقي كلمة معرَّبة
معروفة، نُعيد لصق الحرف وإعادة توجيه أي أحفاد له، ثم نحذف توكنه.
"""
tokens = _by_id(sentence)
to_remove: List[int] = []
for tok in sentence:
if tok.cpos != "PRT" or len(_clean_form(tok.form)) != 1:
continue
letter = _clean_form(tok.form)
if letter not in ("ب", "ل", "ك", "و", "ف"):
continue
kids = [t for t in sentence if t.head == tok.tid]
if len(kids) != 1:
continue
child = kids[0]
remainder = dediac_ar(_clean_form(child.form))
if remainder not in PROTECTED_GLUE_FRAGMENTS:
continue
child.form = letter + child.form
child.head = tok.head
child.deprel = tok.deprel
to_remove.append(tok.tid)
if to_remove:
drop = set(to_remove)
sentence[:] = [t for t in sentence if t.tid not in drop]
def _group_into_surface_words(sentence: Sentence) -> List[List[int]]:
"""تجمع توكنات الجملة إلى "كلمات سطحية" باستخدام علامة "+" التي
يتركها المحلل على الزوائد الملتصقة - محاذاة بنيوية دقيقة مع
raw_words، بخلاف المطابقة النصية التقريبية القابلة للانزلاق.
"""
toks = [t for t in sentence if t.cpos != "PNX"]
groups: List[List[int]] = []
current: List[int] = []
n = len(toks)
for idx, tok in enumerate(toks):
current.append(tok.tid)
ends_proclitic = tok.form.endswith("+") and tok.form != "+"
next_is_enclitic = idx + 1 < n and toks[idx + 1].form.startswith("+")
if ends_proclitic or next_is_enclitic:
continue
groups.append(current)
current = []
if current:
groups.append(current)
return groups
def _patch_noan_tokens(sentences: List[Sentence], original_text: str) -> None:
"""ترقيع قائم على المحاذاة البنيوية (clitic-boundary) لا مطابقة نصية
تقريبية (تتحقق بسهولة زائدة، وأي تطابق زائف يُزيح كل ما بعده). يتم
فقط حين تتطابق أعداد الكلمات السطحية والأصلية تماماً، وNOAN كلمة
سطحية كاملة لم تنقسم مع زوائد؛ خارج ذلك تبقى NOAN كما هي للفلترة.
"""
if not original_text:
return
raw_words = re.findall(r"[\w]+", dediac_ar(original_text))
for sentence in sentences:
groups = _group_into_surface_words(sentence)
if len(groups) != len(raw_words):
continue
tokens = _by_id(sentence)
for group, word in zip(groups, raw_words):
if len(group) == 1 and "NOAN" in tokens[group[0]].form:
tokens[group[0]].form = word
def extract_kg_entities(conll_text, original_text: str = "") -> ExtractionResult:
conll_text = _normalize_conll_input(conll_text)
merged = ExtractionResult()
sentences = parse_conllx(conll_text)
for sentence in sentences:
_repair_split_loanwords(sentence)
if original_text:
_patch_noan_tokens(sentences, original_text)
# يُحسَب مرة واحدة لكل المستند (لا لكل جملة) لأن الإحالة قد تعبر حدود
# الجملة الواحدة - أُلحقت بمحاولة try/except دفاعية: خريطة إحالة فارغة
# (لا إثراء) أفضل من توقف الاستخراج بالكامل بسبب استثناء غير متوقَّع
# في وحدة إضافية اختيارية.
try:
antecedent_text_map = build_antecedent_text_map(sentences)
except Exception:
antecedent_text_map = {}
for sentence_idx, sentence in enumerate(sentences):
result = extract_entities_and_relations(sentence, sentence_idx, antecedent_text_map)
merged.entities.extend(result.entities)
merged.relations.extend(result.relations)
# إضافة قاعدة الفلترة الصارمة هنا:
# نستبعد أي كيان يحوي NOAN
merged.entities = [e for e in _dedupe_preserve_order(merged.entities) if "NOAN" not in e.upper()]
# نستبعد أي علاقة يكون طرفاها (المصدر أو الهدف) يحويان NOAN
merged.relations = [(s, r, t) for s, r, t in merged.relations
if "NOAN" not in s.upper() and "NOAN" not in t.upper()]
return merged
# --------------------------------------------------------------------------- #
# 5. CamelParser2.0 integration point
# --------------------------------------------------------------------------- #
def run_camel_parser(sentences: List[str], camel_parser_repo_path: str) -> str:
import sys
from pathlib import Path
if isinstance(sentences, str):
sentences = [sentences]
sys.path.insert(0, camel_parser_repo_path)
from camel_tools.utils.charmap import CharMapper # type: ignore
from pandas import read_csv # type: ignore
from src.classes import TextParams # type: ignore
from src.initialize_disambiguator.disambiguator_interface import get_disambiguator # type: ignore
from src.data_preparation import get_tagset, parse_text # type: ignore
from src.utils.model_downloader import get_model_name # type: ignore
from src.conll_output import text_tuples_to_string # type: ignore
root_dir = Path(camel_parser_repo_path)
model_path = root_dir / "models"
parse_model = "catib"
arclean = CharMapper.builtin_mapper("arclean")
clitic_feats_df = read_csv(root_dir / "data" / "clitic_feats.csv")
clitic_feats_df = clitic_feats_df.astype(str).astype(object)
model_name = get_model_name(parse_model, model_path=model_path)
tagset = get_tagset(parse_model)
disambiguator = get_disambiguator("bert", "r13")
file_type_params = TextParams(
sentences, model_path / model_name, arclean, disambiguator,
clitic_feats_df, tagset, "",
)
parsed = parse_text("text", file_type_params)
return text_tuples_to_string(parsed, file_type="text", sentences=sentences)
# =========================================================================== #
# قسم استخراج العلاقات عبر النموذج الهجين (Relation Extraction via mT5)
# =========================================================================== #
def _find_entity_spans(
entity_text: str, sentence: Sentence, max_gap: int = 3
) -> List[List[int]]:
"""تبحث عن كل الذِّكرات الفعلية للكيان في الجملة، لا أي توكن تصادف
وجوده ضمن كلمات نص الكيان (مطابقة "كيس كلمات" تُفسد حساب المسافة
النحوية بين الكيانات وتُنتج أزواجاً وهمية للنموذج). يشترط ظهور كلمات
الكيان بالترتيب الصحيح ومتقاربة (بفارق لا يتجاوز max_gap)، ويسمح
بتخطي كلمة سقطت أثناء _walk. تُرجع كل الذِّكرات لا أولها فقط.
"""
target_words = dediac_ar(entity_text).split()
if not target_words:
return []
content_toks = [t for t in sentence if t.cpos != "PNX"]
forms = [dediac_ar(_clean_form(t.form)) for t in content_toks]
n = len(forms)
used = [False] * n
spans: List[List[int]] = []
start = 0
while start < n:
pos = start
matched_ids: List[int] = []
ok = True
for word in target_words:
found = None
for look in range(pos, min(pos + max_gap + 1, n)):
if not used[look] and forms[look] == word:
found = look
break
if found is None:
ok = False
break
matched_ids.append(content_toks[found].tid)
pos = found + 1
if ok:
spans.append(matched_ids)
for k in range(start, pos):
used[k] = True
start = pos
else:
start += 1
return spans
def _find_root_of_span(span_ids: List[int], tokens: Dict[int, Token]) -> int:
"""تحدد 'رأس' الكيان: وهو التوكن الذي لا ينتمي أبوه (Head) إلى نفس الكيان."""
if not span_ids:
return -1
roots = [tid for tid in span_ids if tokens[tid].head not in span_ids]
if roots:
return roots[0] # نأخذ الرأس الأول في حال وجود عدة رؤوس
return span_ids[0] # خطة بديلة
def _get_path_to_root(tid: int, tokens: Dict[int, Token]) -> List[int]:
"""ترسم المسار النحوي من التوكن المحدد صعوداً إلى جذر الجملة (0)."""
path = []
curr = tid
visited = set()
while curr != 0 and curr in tokens and curr not in visited:
path.append(curr)
visited.add(curr)
curr = tokens[curr].head
path.append(0) # ROOT
return path
def _get_shortest_path_length(tid1: int, tid2: int, tokens: Dict[int, Token]) -> int:
"""تحسب عدد القفزات (Edges) بين توكنين في شجرة الإعراب."""
if tid1 == -1 or tid2 == -1:
return 999 # مسافة بعيدة جداً إذا لم نتمكن من تحديد الرأس
path1 = _get_path_to_root(tid1, tokens)
path2 = _get_path_to_root(tid2, tokens)
# البحث عن أدنى سلف مشترك (Lowest Common Ancestor)
lca = 0
for node in path1:
if node in path2:
lca = node
break
dist1 = path1.index(lca)
dist2 = path2.index(lca)
return dist1 + dist2
def _best_mention_pair(
spans1: List[List[int]], spans2: List[List[int]], tokens: Dict[int, Token]
) -> Optional[Tuple[List[int], List[int], int]]:
"""حين يتكرر أحد الكيانين (أو كلاهما) بأكثر من ذكرٍ في الجملة، تختار
زوج الذِّكرات الأقرب نحوياً من بين كل التوليفات الممكنة - هذا هو الزوج
الأرجح أن يكون طرفَي علاقة فعلية، لا أي توليفة عشوائية بينهما."""
best = None
best_dist = None
for s1 in spans1:
for s2 in spans2:
r1 = _find_root_of_span(s1, tokens)
r2 = _find_root_of_span(s2, tokens)
d = _get_shortest_path_length(r1, r2, tokens)
if best_dist is None or d < best_dist:
best_dist = d
best = (s1, s2, d)
return best
def _is_nested(e1: str, e2: str) -> bool:
"""فحص الاحتواء على مستوى الكلمات (word-set) لا السلسلة الحرفية
الخام، لتجنّب إيجابيات كاذبة لكيانات قصيرة تصادف كونها سلسلة فرعية
حرفياً من كيان آخر غير مرتبط."""
w1 = set(dediac_ar(e1).split())
w2 = set(dediac_ar(e2).split())
return w1 <= w2 or w2 <= w1
def _dediac_with_offsets(text: str) -> Tuple[str, List[int]]:
"""تُرجع نسخة مُجرَّدة من التشكيل مع خريطة مواضع للنص الأصلي - يتيح
البحث في نسخة مُجرَّدة ثم إسقاط التطابق على النص الأصلي بدقة (بديل
أوثق من إعادة البناء من التوكنات، الذي يُسقط الترقيم والتشكيل).
"""
out_chars = []
offsets = []
for i, ch in enumerate(text):
cleaned = dediac_ar(ch)
if cleaned:
out_chars.append(cleaned)
offsets.append(i)
return "".join(out_chars), offsets
def _tag_nth_occurrence(
original_text: str, entity_surface_dediac: str, n: int, tag: str
) -> Optional[str]:
"""تضع ... حول الظهور رقم n تحديداً لنص الكيان، لا أول
ظهور فقط - مهم حين يتكرر النص نفسه بدلالة مختلفة أكثر من مرة.
"""
dediac_text, offsets = _dediac_with_offsets(original_text)
escaped = re.escape(entity_surface_dediac)
matches = list(re.finditer(rf"\b{escaped}\b", dediac_text))
if n >= len(matches):
return None
m = matches[n]
start_orig = offsets[m.start()]
end_orig = offsets[m.end() - 1] + 1
return (
original_text[:start_orig]
+ f"<{tag}>{original_text[start_orig:end_orig]}"
+ original_text[end_orig:]
)
import itertools
import requests
RE_API_URL = "https://judy4444-text2tale.hf.space/extract_relation"
_INVALID_RELATIONS = {"none", "لا يوجد", "", "noan", "no_relation", "no relation"}
# --------------------------------------------------------------------------- #
# كشف النفي: فعل يحمل أداة نفي مباشرة لا يُستخرَج منه علاقة، لأنها ستعكس
# معنى الجملة. "ما" مستبعدة عمداً لكونها أكثر شيوعاً كاسم موصول.
# --------------------------------------------------------------------------- #
_NEGATION_FORMS = {"لم", "لن", "لا", "ليس", "ليست"}
def _find_negated_verb_ids(sentence: Sentence, tokens: Dict[int, Token],
children: Dict[int, List[int]]) -> Set[int]:
"""تُرجع معرفات كل فعل في الجملة يحمل أداة نفي مباشرة ضمن أبنائه."""
negated: Set[int] = set()
for tok in sentence:
if tok.cpos not in {"VRB", "VRB-PASS"}:
continue
for cid in children.get(tok.tid, []):
child = tokens.get(cid)
if child and dediac_ar(_clean_form(child.form)) in _NEGATION_FORMS:
negated.add(tok.tid)
break
return negated
# --------------------------------------------------------------------------- #
# جمل الصلة ("غشاء ... يحمي الخلية") لا فاعل صريح لها - فاعلها الاسم الذي
# يتصل به الفعل مباشرة (نكرة)، أو عبر "الذي/التي" (معرفة).
# --------------------------------------------------------------------------- #
_RELATIVE_PRONOUNS = {"الذي", "التي", "اللذان", "اللتان", "الذين", "اللاتي", "اللائي"}
def _find_implicit_subject(verb_id: int, tokens: Dict[int, Token],
children: Dict[int, List[int]]) -> Optional[int]:
"""تصعد من رأس الفعل بحثاً عن اسم حقيقي: مباشرة، أو متجاوزة سلسلة
"الذي/التي" موصولة. تتوقف عند أول اسم حقيقي أو بعد ٣ قفزات كحد أقصى
دفاعاً ضد شجرة مشوَّهة."""
current = tokens.get(verb_id)
hops = 0
while current is not None and hops < 3:
governor = tokens.get(current.head)
if governor is None:
return None
if dediac_ar(_clean_form(governor.form)) in _RELATIVE_PRONOUNS:
current = governor
hops += 1
continue
if _is_nominal(governor):
return governor.tid
return None
return None
# --------------------------------------------------------------------------- #
# يمنع تزاوج معطوفَين كطرفَي علاقة - لا علاقة سببية حقيقية بينهما غالباً،
# فسؤال النموذج عنها يُنتج تخميناً عشوائياً.
# --------------------------------------------------------------------------- #
def _are_coordinate_siblings(root1: int, root2: int, tokens: Dict[int, Token],
children: Dict[int, List[int]]) -> bool:
if root1 == -1 or root2 == -1:
return False
siblings = _collect_conjunct_heads(root1, tokens, children, set())
return root2 in siblings
# --------------------------------------------------------------------------- #
# تحقُّق أن العلاقة المُرجَعة من النموذج مرتبطة فعلياً بالجملة: هل تشترك
# كلمات العلاقة معها بكلمة محتوى واحدة على الأقل؟
# --------------------------------------------------------------------------- #
def _relation_is_grounded(predicted_relation: str, tagged_source: str) -> bool:
# حظر قطعي: حرف عطف مجرد ليس علاقة أبداً، بصرف النظر عن ارتباطه بالنص -
# ظهوره دائماً عرَض لخطأ آخر (عطف أُسيء تفسيره كطرفَي علاقة، أو علاقة
# حقيقية فُقد فعلها وبقي معطوفها فقط)، لا حالة صحيحة واحدة له. هذا حظر
# مطلق يسبق فحص الارتباط اللفظي، لأن حرف العطف غالباً موجود حرفياً في
# النص المُرسَل فيجتاز فحص الارتباط اللفظي وحده رغم عدم صلاحيته كعلاقة.
predicted_clean = dediac_ar(predicted_relation).strip()
if predicted_clean in _CONJUNCTIONS or len(predicted_clean) <= 1:
return False
source_words = set(dediac_ar(re.sub(r"", "", tagged_source)).split())
relation_words = set(dediac_ar(predicted_relation).split())
# نتجاهل حروف الجر/الكلمات الوظيفية القصيرة جداً عند حساب التقاطع، حتى
# لا يمر تشابه زائف عبر كلمة وظيفية مشتركة بلا دلالة حقيقية
relation_content_words = {w for w in relation_words if len(w) > 1}
if not relation_content_words:
return False
return bool(relation_content_words & source_words)
def hybrid_relation_extraction_api(
sentence_text: str,
parsed_sentence: "Sentence",
rule_based_entities: List[str],
already_related: Optional[Set[frozenset]] = None,
max_hops: int = 2,
) -> List[Tuple[str, str, str]]:
"""تُرسِل للنموذج فقط أزواج كيانات مرتبطة نحوياً ضمن مسافة معقولة،
لا كل التوليفات الممكنة. already_related تستبعد أزواجاً غطّاها
المسار القاعدي (المصدر الموثوق) بالفعل. max_hops=2 يعني تشارك حوكمة
مباشرة تحت نفس الرأس - أبعد من ذلك سؤال غير محدَّد المعنى للنموذج.
"""
final_relations: List[Tuple[str, str, str]] = []
tokens_dict = _by_id(parsed_sentence)
children_dict = _children_map(parsed_sentence)
already_related = already_related or set()
negated_verb_ids = _find_negated_verb_ids(parsed_sentence, tokens_dict, children_dict)
clean_entities = _dedupe_preserve_order(
[e for e in rule_based_entities if "NOAN" not in e]
)
# نحسب كل ذِكرات كل كيان مرة واحدة، بدل إعادة البحث عنها لكل زوج
entity_spans = {e: _find_entity_spans(e, parsed_sentence) for e in clean_entities}
entity_spans = {e: spans for e, spans in entity_spans.items() if spans}
available = list(entity_spans.keys())
for e1, e2 in itertools.combinations(available, 2):
if _is_nested(e1, e2):
continue
if frozenset([e1, e2]) in already_related:
continue
picked = _best_mention_pair(entity_spans[e1], entity_spans[e2], tokens_dict)
if picked is None:
continue
span1, span2, dist = picked
if dist > max_hops:
continue
root1 = _find_root_of_span(span1, tokens_dict)
root2 = _find_root_of_span(span2, tokens_dict)
# إصلاح النفي: أي طرف يقع تحت فعل منفي فسؤال العلاقة كله غير موثوق
path1 = set(_get_path_to_root(root1, tokens_dict))
path2 = set(_get_path_to_root(root2, tokens_dict))
if (path1 & negated_verb_ids) or (path2 & negated_verb_ids):
continue
# إصلاح المعطوفات: طرفان معطوفان على بعضهما ليسا طرفَي علاقة، بل
# مفعولان/فاعلان متوازيان لشيء ثالث
if _are_coordinate_siblings(root1, root2, tokens_dict, children_dict):
continue
occ1 = entity_spans[e1].index(span1)
occ2 = entity_spans[e2].index(span2)
tagged = _tag_nth_occurrence(sentence_text, dediac_ar(e1), occ1, "e1")
if tagged is None:
continue
tagged = _tag_nth_occurrence(tagged, dediac_ar(e2), occ2, "e2")
if tagged is None or "" not in tagged or "" not in tagged:
continue
try:
response = requests.post(RE_API_URL, json={"text": tagged}, timeout=5.0)
if response.status_code == 200:
data = response.json()
predicted_relation = data.get("relation", "").strip()
if (predicted_relation and predicted_relation.lower() not in _INVALID_RELATIONS
and _relation_is_grounded(predicted_relation, tagged)):
final_relations.append((e1, predicted_relation, e2))
else:
print(f"[خطأ API] فشل استخراج العلاقة للزوج ({e1}, {e2}) - Status Code: {response.status_code}")
except requests.exceptions.RequestException as exc:
print(f"[انقطاع اتصال] تعذر الوصول لخادم mT5: {exc}")
return final_relations
# =========================================================================== #
# خط الأنابيب الكامل لمعالجة المستندات (Full NLP Pipeline for Documents)
# =========================================================================== #
def split_text_into_sentences(full_text: str) -> List[str]:
# إدخال مسافة بعد علامة نهاية الجملة إن لم توجد، لضمان تقسيم صحيح
# لحدود الجمل بدل وصول جملتين متصلتين للمحلل كجملة واحدة.
full_text = re.sub(r'([.?!؛])(?=\S)', r'\1 ', full_text)
# لا نُقسِّم عند "؛" حين تتبعها مباشرة "الذي/التي" - الاسم الموصول
# يحتاج مرجعه في الشجرة نفسها ليعمل حل الفاعل الضمني
# (_find_implicit_subject)؛ نستبدلها بفاصلة عادية فتبقى الجملتان
# متصلتين بنفس الشجرة.
relative_pattern = "|".join(re.escape(w) for w in _RELATIVE_PRONOUNS)
full_text = re.sub(rf'؛(?=\s*(?:{relative_pattern})\b)', ',', full_text)
# تقطيع النص بدقة مع إزالة المسافات الزائدة
sentences = re.split(r'(?<=[.?!؛])\s+', full_text)
return [s.strip() for s in sentences if s.strip()]
def process_full_document_pipeline(full_text: str, camel_parser_path: str) -> List[Tuple[str, str, str]]:
"""محرك المعالجة الكامل: تحليل نحوي، استخراج قاعدي، استدلال هجين،
تنقية نهائية. علاقات المسار القاعدي مصدر أساسي موثوق؛ نموذج mT5
تكميلي فقط لأزواج لم يغطّها المسار القاعدي، ولا يُسمَح له بمعارضة
علاقة قاعدية قائمة لنفس الزوج.
مبنية على ثلاث تمريرات: حل الإحالة المرجعية يحتاج رؤية جمل المستند
كاملة معاً قبل أي استخراج، إذ إحالة "ها" مثلاً قد تعبر حدود الجملة.
"""
all_rule_relations: List[Tuple[str, str, str]] = []
all_hybrid_relations: List[Tuple[str, str, str]] = []
sentences = split_text_into_sentences(full_text)
# التمريرة الأولى: استدعاء واحد للمحلل النحوي لكل جمل المستند معاً
# (توصية CamelParser2.0 الرسمية)، إذ كل استدعاء منفصل يُعيد تحميل
# نموذج BERT للتفكيك الصرفي من الصفر.
per_sentence_parsed: List[Tuple[str, List[Sentence]]] = []
all_parsed_sentences: List[Sentence] = []
raw_conllx = run_camel_parser(sentences, camel_parser_path)
if isinstance(raw_conllx, list):
raw_conllx = "\n".join(raw_conllx)
normalized = _normalize_conll_input(raw_conllx)
batch_parsed = parse_conllx(normalized)
if len(batch_parsed) == len(sentences):
# الحالة المتوقَّعة: مطابقة ١:١ بين جملة الإدخال وجملة الخرج.
for sent_str, sentence in zip(sentences, batch_parsed):
_repair_split_loanwords(sentence)
if sent_str:
_patch_noan_tokens([sentence], sent_str)
per_sentence_parsed.append((sent_str, [sentence]))
all_parsed_sentences.append(sentence)
else:
# احتياط دفاعي: إن انقسمت جملة إدخال إلى أكثر من جملة خرج داخلياً
# (نادر لكن ممكن)، لا نُخاطر بمحاذاة خاطئة بين sent_str والشجرة
# الخطأ - نستدعي المحلل لكل جملة على حدة، أبطأ لكن آمن المحاذاة.
for sent_str in sentences:
raw = run_camel_parser(sent_str, camel_parser_path)
if isinstance(raw, list):
raw = "\n".join(raw)
parsed = parse_conllx(_normalize_conll_input(raw))
for s in parsed:
_repair_split_loanwords(s)
if sent_str:
_patch_noan_tokens(parsed, sent_str)
per_sentence_parsed.append((sent_str, parsed))
all_parsed_sentences.extend(parsed)
# -------------------------------------------------------------------
# التمريرة الثانية: حل الإحالة المرجعية على المستند الكامل دفعة واحدة.
# -------------------------------------------------------------------
try:
antecedent_text_map = build_antecedent_text_map(all_parsed_sentences)
except Exception:
antecedent_text_map = {}
# -------------------------------------------------------------------
# التمريرة الثالثة: استخراج الكيانات/العلاقات لكل جملة (بفهرسها الصحيح
# ضمن المستند الكامل، لا الجملة وحدها)، ثم الاستدلال الهجين كالمعتاد.
# -------------------------------------------------------------------
global_idx = 0
for sent_str, parsed in per_sentence_parsed:
merged = ExtractionResult()
for sentence in parsed:
result = extract_entities_and_relations(sentence, global_idx, antecedent_text_map)
merged.entities.extend(result.entities)
merged.relations.extend(result.relations)
global_idx += 1
merged.entities = [e for e in _dedupe_preserve_order(merged.entities) if "NOAN" not in e.upper()]
merged.relations = [
(s, r, t) for s, r, t in merged.relations
if "NOAN" not in s.upper() and "NOAN" not in t.upper()
]
extracted_res = merged
all_rule_relations.extend(extracted_res.relations)
already_related = {
frozenset([e1, e2]) for e1, _rel, e2 in extracted_res.relations
}
# إرسال البيانات للمعالجة الهجينة إذا كان هناك كيانان أو أكثر -
# فقط للأزواج التي لم تُغطِّها القواعد بعلاقة بالفعل
if parsed and len(extracted_res.entities) >= 2:
parsed_sent = parsed[0]
hybrid_rels = hybrid_relation_extraction_api(
sentence_text=sent_str,
parsed_sentence=parsed_sent,
rule_based_entities=extracted_res.entities,
already_related=already_related,
max_hops=2, # انظر تعليق hybrid_relation_extraction_api
)
all_hybrid_relations.extend(hybrid_rels)
# 4. الدمج: القواعد النحوية أولاً وبثقة كاملة، ثم النموذج فقط لأزواج
# لم تغطِّها القواعد إطلاقاً في أي جملة بالمستند (وليس فقط الجملة
# نفسها) - تحسباً لأن يكون نفس الزوج قد ظهر مرتين بجمل مختلفة.
covered_pairs = {frozenset([e1, e2]) for e1, _rel, e2 in all_rule_relations}
filtered_hybrid = [
(e1, rel, e2) for e1, rel, e2 in all_hybrid_relations
if frozenset([e1, e2]) not in covered_pairs
]
all_document_relations = all_rule_relations + filtered_hybrid
# 5. التنظيف النهائي: إزالة العلاقات المكررة تماماً (بغض النظر عن الترتيب)
cleaned_relations = []
seen_pairs = set()
for e1, rel, e2 in all_document_relations:
# استخدام frozenset يضمن أن (e1, e2, rel) تُعامل تماماً مثل (e2, e1, rel)
pair_signature = frozenset([e1, e2, rel])
if pair_signature not in seen_pairs:
cleaned_relations.append((e1, rel, e2))
seen_pairs.add(pair_signature)
return cleaned_relations
# =========================================================================== #
# تحويل الناتج إلى رسم بياني (nodes/edges JSON) مع تصنيف نوع الكيان
# =========================================================================== #
# العلاقات (edges): لا تصنيف - نص العلاقة الحرفي نفسه هو النوع دائماً.
# الكيانات (nodes): تصنيف واعٍ بمجال النص (تاريخ/جغرافيا/علوم)، إذ الفئات
# المتوقَّعة تختلف كلياً بين المجالات. ما لا يُطابَق يأخذ اسمه نفسه.
# --------------------------------------------------------------------------- #
def _normalize_for_typing(text):
return re.sub(r"[أإآ]", "ا", dediac_ar(text))
def _strip_al_prefix(word):
return word[2:] if word.startswith("ال") else word
def _matches_keyword_set(surface, keywords):
words = surface.split()
stripped_surface = " ".join(_strip_al_prefix(w) for w in words)
if any(kw in surface or kw in stripped_surface for kw in keywords if " " in kw):
return True
for word in words:
bare = _strip_al_prefix(word)
if word in keywords or bare in keywords:
return True
return False
def _first_word_matches_keyword_set(surface, keywords, n=1):
"""كالسابقة، لكن مقصورة على أول ن كلمة - تُستخدَم حين الكلمة الرئيسية
وحدها هي الدليل الموثوق. كلمة مفردة: تطابق بادئة لا احتواء حر
("ليف" بادئة لـ"ليفيات" لكن ليست بادئة لـ"سورية").
"""
words = surface.split()[:n]
head = " ".join(_strip_al_prefix(w) for w in words)
if any((" " in kw and kw in head) for kw in keywords):
return True
return any(_strip_al_prefix(w).startswith(kw) for w in words for kw in keywords if " " not in kw)
# بوابة أسماء أماكن حقيقية (دول/مدن/مناطق/جهات) - لا يمكن اكتشافها عبر
# كلمات مفتاحية عامة (لا تحتوي "مدينة"/"دولة" كجزء من الاسم نفسه)، فتحتاج
# قائمة أعلام صريحة. مشتركة بين تاريخ وجغرافيا لأن نفس الأسماء تتكرر.
_PLACE_GAZETTEER = {dediac_ar(w) for w in {
"مصر", "سورية", "سوريا", "العراق", "الاردن", "لبنان", "فلسطين",
"السعودية", "اليمن", "ليبيا", "الجزائر", "المغرب", "تونس", "تركيا",
"ايران", "افريقيا", "اوروبا", "اسيا", "دمشق", "القاهرة", "بغداد",
"القدس", "مكة", "المدينة", "الكوفة", "البصرة", "طرابلس", "القيروان",
"قرطاجة", "القسطنطينية", "اسطنبول", "حلب", "بيروت", "عمان الاردنية",
"شمال", "جنوب", "شرق", "غرب", "شمالي", "جنوبي", "شرقي", "غربي",
"الوطن العربي", "بلاد الشام", "شبه الجزيرة",
}}
# ============================================================
# تصنيف نوع الكيان الدلالي، واعٍ بمجال النص (تاريخ/جغرافيا/علوم) لأن
# الفئات المتوقَّعة تختلف كلياً بين المجالات. كل فئة تُحدَّد عبر قوائم
# كلمات مفتاحية قابلة للتوسع بحسب المجال، بلا افتراض فئة عامة حين لا
# تُطابَق أي كلمة صراحةً - في تلك الحالة يُستخدَم اسم الكيان نفسه كنوعه.
# ============================================================
# ===================== تاريخ =====================
_PERSON_KEYWORDS_HIST = {dediac_ar(w) for w in {
"معتمد", "ملك", "ملكة", "رئيس", "قائد", "وزير", "سفير", "حاكم",
"خليفة", "سلطان", "امير", "زعيم", "قاضي", "محتل", "غازي", "قنصل",
"حلفاء", "شعب", "قوم", "جيش", "جنود", "سكان", "مندوب", "وفد", "جيوش",
"بيزنطيين", "عثمانيين", "مصريين", "بريطانيين", "فرنسيين",
}}
_EVENT_KEYWORDS_HIST = {dediac_ar(w) for w in {
"معركة", "حرب", "حروب", "ثورة", "غزو", "فتح", "حصار", "انتفاضة",
"مؤتمر", "معاهدة", "انقلاب", "احتلال", "تحرير", "وفاة", "ميلاد",
"تاسيس", "سقوط", "توقيع", "انتخابات", "اجتياح", "تمرد", "استقلال",
"معاهدات",
}}
# "دولة" مُستبعَدة عمداً: ملتبسة بين سلالة حاكمة ومكان (بلد).
_ERA_KEYWORDS = {dediac_ar(w) for w in {
"عصر", "عهد", "حقبة", "فترة", "زمن",
"قرن", "قرون", "عام", "اعوام", "سنة", "سنوات", "عقد", "عقود",
"الفية", "موسم",
}}
_PLACE_KEYWORDS_HIST = {dediac_ar(w) for w in {
"جزيرة", "مدينة", "قرية", "دولة", "بلاد", "اقليم", "منطقة", "عاصمة",
"قارة", "قناة", "طريق", "حدود",
}}
_CONCEPT_KEYWORDS = {dediac_ar(w) for w in {
"مبدأ", "مبادئ", "نظرية", "فكرة", "فلسفة", "عقيدة", "دستور", "قانون",
"ايديولوجيا", "مذهب", "نزعة", "تنوع",
}}
_ACTION_KEYWORDS_HIST = {dediac_ar(w) for w in {
"اجراء", "اجراءات", "تدبير", "تدابير", "خطوة", "خطوات", "قرار",
"قرارات", "سياسة", "سياسات", "مقاومة",
}}
_ABSTRACT_SITUATION_KEYWORDS_HIST = {dediac_ar(w) for w in {
"ازمة", "تدخل", "وضع", "حالة", "ظرف", "متانة",
}}
_SYSTEM_KEYWORDS = {dediac_ar(w) for w in {
"نظام", "دائرة", "قطاع", "شبكة", "بنية", "خدمات", "خدمة", "هيئة",
"مؤسسة", "ادارة", "حرس", "شرطة", "بريد", "اسطول", "وزارة", "حكومة",
"زراعة", "نشاط", "انشطة", "استثمار", "شورى", "الوراثة", "نظم", "غلاف",
}}
# ===================== جغرافيا =====================
_PLACE_KEYWORDS_GEO = {dediac_ar(w) for w in {
"بادية", "جبل", "جبال", "وادي", "اودية", "هضبة", "حوضة", "احواض",
"صحراء", "نهر", "بحر", "بحيرة", "سهل", "ساحل", "حدود", "جزيرة",
"منطقة", "اقليم", "مناجم", "حقول", "سبخة", "حماد", "بيئة", "كتل",
"سواحل", "مسطحات",
}}
_MATERIAL_KEYWORDS_GEO = {dediac_ar(w) for w in {
"مراعي", "ملح", "املاح", "فوسفات", "غاز", "نفط", "معادن", "مياه",
"مياه جوفية", "امطار", "لحوم", "حليب", "جلود", "صوف",
}}
_ORGANISM_KEYWORDS = {dediac_ar(w) for w in {
"حيوان", "حيوانات", "نبات", "نباتات", "انسان", "بشر", "كائن",
"كائنات", "مخلوق", "اغنام", "ابقار", "ماعز", "مواشي", "مزروعات",
"حضارات", "ثروة حيوانية",
}}
_ELEMENT_KEYWORDS = {dediac_ar(w) for w in {
"تربة", "تضاريس", "غطاء نباتي", "مظاهر السطح", "سطح", "طبقات",
}}
_PROCESS_KEYWORDS = {dediac_ar(w) for w in {
"عملية", "عمليات", "استصلاح", "تفاعلات", "غسيل", "تركيب ضوئي",
"هضم", "تنفس", "انقسام", "تكاثر", "تمثيل غذائي", "ترشيح", "افراز",
"تكاثف", "اكسدة", "اختزال", "امتصاص", "دورة", "تخمر",
"احتراق", "تعرق", "اخصاب", "تلقيح", "انبات",
}}
_ABSTRACT_KEYWORDS_GEO = {dediac_ar(w) for w in {
"مناخ", "رياح", "ريح", "درجة", "رطوبة", "جفاف", "امتداد", "تكتل",
"مقومات", "فصل", "فصول", "صيف", "شتاء", "ربيع", "خريف", "موجة",
"موجات", "انجراف", "مدى", "رعي", "ري", "تبخر", "تكتل",
}}
_ACTION_KEYWORDS_GEO = {dediac_ar(w) for w in {
"ادارة خاطئة", "رعي جائر", "احتطاب", "زراعات بعلية غير ناجحة",
"استخدام مفرط", "قطع جائر",
}}
# ===================== علوم =====================
_ORGAN_KEYWORDS = {dediac_ar(w) for w in {
"عضو", "اعضاء", "غشاء", "اغشية", "خلية", "خلايا", "نواة", "عضلة",
"عضلات", "نسيج", "انسجة", "جهاز", "قناة", "قنوات", "انبوب", "باحة",
"باحات", "قشرة", "تلفيف", "حصين", "مشبك", "مشابك", "ليف", "الياف",
"اذن", "عين", "جلد", "بشرة", "رحم", "جنين", "مشيمة", "سن", "اسنان",
"عاج", "لب", "معدة", "مريء", "امعاء", "فم", "مطرقة", "ركاب", "سندان",
"نافذة", "هدب", "اهداب", "ميسم", "بذيرة", "طلع", "جراثيم", "فك",
"طبقة", "طبقات", "كرة مخية", "نصف كرة", "عظم", "عظيمات", "عظام",
"نطفة", "نطف", "مركز", "اثنا عشر", "بلعوم", "رغامى", "قزحية", "حدقة",
"شبكية", "مشيمية", "صيوان", "حلزون", "قوقعة", "فص", "عصب",
"هيولى", "نوى", "وعاء", "اوعية", "لطخة", "نقطة عمياء", "نظارات",
"غطاء", "مجرى",
}}
_MATERIAL_KEYWORDS_SCI = {dediac_ar(w) for w in {
"مادة", "مواد", "شوارد", "عصارة", "عصارات", "سائل", "غاز", "معدن",
"ملح", "دهون", "بروتين", "هرمون", "انزيم", "ناقل", "نواقل", "غذاء",
"فضلات", "دم", "حمض", "لمف",
}}
_NONLIVING_ELEMENT_KEYWORDS = {dediac_ar(w) for w in {"ماء", "هواء", "تربة"}}
_ACTION_KEYWORDS_SCI = {dediac_ar(w) for w in {
"اهمال", "افراط", "تدخين", "سوء استخدام", "عمل شاق", "راحة", "نظر",
"تقريب",
}}
_ABSTRACT_CONDITION_KEYWORDS = {dediac_ar(w) for w in {
"نخر", "التهاب", "تراكم", "تورم", "نزيف", "كسر", "التواء", "ضمور",
"تنكس", "قصور", "فشل", "عدوى", "تسمم", "حساسية", "انسداد", "تشنج",
"الم", "تعب", "اجهاد", "اتعاب", "تخريب", "توقف", "تضخم", "عمى",
}}
_GOAL_KEYWORDS = {dediac_ar(w) for w in {"هدف", "غاية", "غايات", "استدامة", "مقصد"}}
_SCI_CONCEPT_KEYWORDS = {dediac_ar(w) for w in {"تنوع"}}
_ORGANISM_KEYWORDS_SCI = {dediac_ar(w) for w in {"حيوان", "حيوانات", "نبات", "نباتات", "كائن", "كائنات"}}
def classify_node_type(label, subject=""):
surface = _normalize_for_typing(label).strip()
subj = _normalize_for_typing(subject).strip()
# سنة ميلادية/هجرية: رقم مجرد، أو رقم متبوع بحرف ميلادي/هجري
# سنة ميلادية/هجرية: رقم مجرد، أو رقم متبوع بحرف ميلادي/هجري، أو نطاق
# سنوات ("1798-1801")، أو مسبوقة بكلمة سياقية شائعة ("خلال عام
# 1928"، "في أواخر 1920") - نفحص أي رقم ٣-٤ خانات ضمن النص بدل
# اشتراط أن يكون النص كله رقماً مجرداً فقط.
if re.search(r"\d{3,4}\s*(م|هـ|ه)?(\s*[-–]\s*\d{3,4}\s*(م|هـ|ه)?)?", surface) and len(surface) <= 40:
return "زمان"
if subj == "تاريخ":
if _matches_keyword_set(surface, _EVENT_KEYWORDS_HIST):
return "حدث"
# مطابقة العبارة كاملة لا الكلمة الأولى فقط: "منتصف القرن التاسع
# عشر"، "خلال عام 1928" - كلمة السياق (منتصف/أوائل/خلال...) تسبق
# كلمة الزمان الحقيقية، فتُفلت من فحص الكلمة الأولى وحدها.
if _matches_keyword_set(surface, _ERA_KEYWORDS):
return "زمان"
if _matches_keyword_set(surface, _PLACE_GAZETTEER):
return "مكان"
if _matches_keyword_set(surface, _SYSTEM_KEYWORDS):
return "نظام"
if _matches_keyword_set(surface, _PLACE_KEYWORDS_HIST):
return "مكان"
if _matches_keyword_set(surface, _CONCEPT_KEYWORDS):
return "مفهوم"
if _matches_keyword_set(surface, _ACTION_KEYWORDS_HIST):
return "فعل"
if _matches_keyword_set(surface, _ABSTRACT_SITUATION_KEYWORDS_HIST):
return "نوع"
# لا تخمين أخير: لم تُطابَق أي كلمة مفتاحية فعلياً، فنستخدم اسم
# الكيان نفسه بدل افتراض فئة "اشخاص" عامة - تصنيف مفاهيم أو أحداث
# مجردة كأشخاص لمجرد عدم مطابقتها أي فئة أخرى تسمية مضلِّلة.
return label
if subj == "جغرافيا":
if _first_word_matches_keyword_set(surface, {"معدل", "كمية", "نسبة", "درجة", "اعداد"}, n=1):
return "نوع"
if _matches_keyword_set(surface, _ACTION_KEYWORDS_GEO):
return "فعل"
if _first_word_matches_keyword_set(surface, _ABSTRACT_KEYWORDS_GEO, n=1):
return "نوع"
if _first_word_matches_keyword_set(surface, _ELEMENT_KEYWORDS, n=1):
return "عنصر"
if _matches_keyword_set(surface, _PLACE_GAZETTEER):
return "مكان"
if _first_word_matches_keyword_set(surface, _MATERIAL_KEYWORDS_GEO, n=1):
return "مادة"
if _first_word_matches_keyword_set(surface, _PLACE_KEYWORDS_GEO, n=1):
return "مكان"
if _matches_keyword_set(surface, _PROCESS_KEYWORDS):
return "عملية"
if _matches_keyword_set(surface, _ORGANISM_KEYWORDS):
return "كائن"
if _matches_keyword_set(surface, _ELEMENT_KEYWORDS):
return "عنصر"
if _matches_keyword_set(surface, _SYSTEM_KEYWORDS):
return "نظام"
if _matches_keyword_set(surface, _PLACE_KEYWORDS_GEO):
return "مكان"
if _matches_keyword_set(surface, _MATERIAL_KEYWORDS_GEO):
return "مادة"
if _matches_keyword_set(surface, _ABSTRACT_KEYWORDS_GEO):
return "نوع"
# لا تخمين أخير: لم تُطابَق أي كلمة مفتاحية فعلياً، فنستخدم اسم
# الكيان نفسه بدل افتراض فئة "مكان" عامة على كيانات لا علاقة لها
# بمكان إطلاقاً.
return label
if subj in ("علوم", "احياء", "بيولوجيا", "كيمياء", "فيزياء"):
if _matches_keyword_set(surface, _ACTION_KEYWORDS_SCI):
return "فعل"
if _matches_keyword_set(surface, _GOAL_KEYWORDS):
return "هدف"
if _matches_keyword_set(surface, _SCI_CONCEPT_KEYWORDS):
return "مفهوم"
if _first_word_matches_keyword_set(surface, _ABSTRACT_CONDITION_KEYWORDS, n=2):
return "نوع"
if _first_word_matches_keyword_set(surface, _NONLIVING_ELEMENT_KEYWORDS, n=1) and len(surface.split()) <= 2:
return "عنصر غير حي"
if _matches_keyword_set(surface, {"مكونات", "مكون"}):
return "عنصر"
if _matches_keyword_set(surface, _PROCESS_KEYWORDS):
return "عملية"
if _matches_keyword_set(surface, _ORGANISM_KEYWORDS_SCI):
return "كائن"
if _first_word_matches_keyword_set(surface, _MATERIAL_KEYWORDS_SCI, n=2):
return "مادة"
if _first_word_matches_keyword_set(surface, _ORGAN_KEYWORDS, n=2):
return "عضو"
if _matches_keyword_set(surface, _ORGAN_KEYWORDS):
return "عضو"
if _matches_keyword_set(surface, _SYSTEM_KEYWORDS):
return "نظام"
if _matches_keyword_set(surface, _MATERIAL_KEYWORDS_SCI):
return "مادة"
# لا تخمين أخير: لم تُطابَق أي كلمة مفتاحية فعلياً، فنستخدم اسم
# الكيان نفسه بدل افتراض فئة "نوع" عامة - تصنيف قسري لكل ما لا
# تُعرَف فئته يُنتج تسميات مضلِّلة أكثر مما يفيد.
return label
if _matches_keyword_set(surface, _EVENT_KEYWORDS_HIST):
return "حدث"
if _matches_keyword_set(surface, _ORGAN_KEYWORDS):
return "عضو"
if _matches_keyword_set(surface, _MATERIAL_KEYWORDS_SCI) or _matches_keyword_set(surface, _MATERIAL_KEYWORDS_GEO):
return "مادة"
if _matches_keyword_set(surface, _PLACE_KEYWORDS_GEO) or _matches_keyword_set(surface, _PLACE_KEYWORDS_HIST):
return "مكان"
return label
def _merge_entity_specificity_variants(labels: List[str]) -> Dict[str, str]:
"""يدمج صيغاً مختلفة لنفس الكيان على مستويين: (١) فروق تعريف/تنكير
لنفس الكلمات ("خلايا حسية بصرية"/"الخلايا الحسية البصرية")، مفضِّلاً
الأكثر تعريفاً؛ (٢) بادئة كلمات مطابقة تماماً بعد تجريد "ال" ("الليف"
<- "الليف العضلي المخطط"). لا دمج عند تشارك جزئي فقط ("غشاء الليف"
كيان مختلف عن "الليف")، ولا عند تعدد مرشحين أطول بنفس الطول (غموض).
"""
unique_labels = sorted(set(labels))
norm_cache: Dict[str, Tuple[str, ...]] = {
label: tuple(_strip_al_prefix(w) for w in label.split()) for label in unique_labels
}
def definiteness_score(label: str) -> int:
return sum(1 for w in label.split() if w.startswith("ال"))
# الخطوة ١: تجميع الصيغ المتطابقة تماماً بعد التطبيع (فروق تعريف
# بحتة) - نختار الأكثر تعريفاً كصيغة قانونية، وإلا الأول أبجدياً
# لضمان نتيجة حتمية عند التعادل.
by_norm: Dict[Tuple[str, ...], List[str]] = {}
for label in unique_labels:
by_norm.setdefault(norm_cache[label], []).append(label)
canonical: Dict[str, str] = {}
for group in by_norm.values():
group.sort(key=lambda s: (-definiteness_score(s), s))
rep = group[0]
for label in group:
canonical[label] = rep
# الخطوة ٢: دمج بادئة الكلمات (بعد التطبيع) بين الممثلين الناتجين
# عن الخطوة ١ - المساواة في الطول مُستبعَدة عمداً هنا (> لا >=)
# لأنها عولجت بالفعل في الخطوة ١، وتضمن عدم وقوع تعادل دائري بين
# صيغتين تُحيل كل منهما للأخرى.
representatives = sorted(set(canonical.values()))
rep_norm = {r: norm_cache[r] for r in representatives}
for short in representatives:
short_norm = rep_norm[short]
supersets = [
long_ for long_ in representatives
if long_ != short and len(rep_norm[long_]) > len(short_norm)
and rep_norm[long_][:len(short_norm)] == short_norm
]
if not supersets:
continue
max_len = max(len(rep_norm[s]) for s in supersets)
longest = [s for s in supersets if len(rep_norm[s]) == max_len]
if len(longest) == 1:
for label, rep in list(canonical.items()):
if rep == short:
canonical[label] = longest[0]
# سلسلة الدمج قد تحتاج أكثر من خطوة (الليف -> الليف العضلي -> الليف
# العضلي المخطط)، فنُكرر حتى الاستقرار.
for _ in range(5):
changed = False
for label in unique_labels:
nxt = canonical.get(canonical[label], canonical[label])
if nxt != canonical[label]:
canonical[label] = nxt
changed = True
if not changed:
break
return canonical
def build_knowledge_graph_json(
relations: List[Tuple[str, str, str]],
extra_entities: Optional[List[str]] = None,
subject: str = "",
) -> Dict:
"""تبني {nodes, edges} من قائمة الثلاثيات (مصدر، علاقة، هدف). كل كيان
فريد عقدة واحدة بمعرّف مستقر؛ extra_entities تُضاف كعقد معزولة بلا
أضلاع؛ subject يُمرَّر لتصنيف الكيانات (انظر classify_node_type).
"""
all_labels = [t[0] for t in relations] + [t[2] for t in relations] + list(extra_entities or [])
canonical_map = _merge_entity_specificity_variants(all_labels)
node_ids: Dict[str, str] = {}
nodes: List[Dict[str, str]] = []
def get_or_create_node(raw_label: str) -> str:
label = canonical_map.get(raw_label, raw_label)
if label not in node_ids:
node_id = f"عقدة_{len(node_ids) + 1}"
node_ids[label] = node_id
nodes.append({
"id": node_id,
"label": label,
"type": classify_node_type(label, subject),
})
return node_ids[label]
edges: List[Dict[str, str]] = []
seen_edges: Set[Tuple[str, str, str]] = set()
for source, rel, target in relations:
source_id = get_or_create_node(source)
target_id = get_or_create_node(target)
# دمج صيغ المصدر/الهدف قد يُنتج ضلعاً مكرراً حرفياً لم يكن مكرراً
# قبل الدمج (مثال: "الليف يحوي الهيولى" و"الليف العضلي يحوي
# الهيولى" تصبحان نفس الضلع بعد الدمج) - نُزيل التكرار الناتج.
edge_key = (source_id, target_id, rel)
if edge_key in seen_edges:
continue
seen_edges.add(edge_key)
# العلاقة تُترَك بنصها الحرفي دائماً كنوعها - لا تصنيف مطلوب هنا
# خلافاً للكيانات، إذ لا غموض دلالياً يستدعيه.
edges.append({"source": source_id, "target": target_id, "type": rel})
for e in (extra_entities or []):
get_or_create_node(e)
return {"nodes": nodes, "edges": edges}
def process_full_document_pipeline_as_graph(
full_text: str, camel_parser_path: str, subject: str = "",
) -> Dict:
"""مثل process_full_document_pipeline تماماً (تُستدعى داخلياً بلا أي
تغيير)، لكن الناتج بصيغة رسم بياني {nodes, edges} جاهزة لـ
json.dumps() مباشرة، بدل قائمة الثلاثيات المسطَّحة. subject تُمرَّر
لتصنيف الكيانات (تاريخ/جغرافيا/علوم)."""
relations = process_full_document_pipeline(full_text, camel_parser_path)
graph = build_knowledge_graph_json(relations, subject=subject)
if subject:
graph = {"subject": subject, **graph}
return graph
# =========================================================================== #
# Main Execution Block
# =========================================================================== #
if __name__ == "__main__":
# 1. مسار مجلد camel_parser على الجهاز
CAMEL_PARSER_PATH = "/Users/mac/camel_parser"
# 2. مجال النص - يحدد فئات تصنيف الكيانات المستخدَمة (انظر classify_node_type)
SUBJECT = "تاريخ"
sample_text = ("الشبكية: طبقة حساسة للضوء، تحتوي خلايا حسية بصرية تسمى: العصي والمخاريط، ويخرج من الشبكة الياف العصب البصري، الذي يتصل بمركز الرؤية في المخ، وتنعدم الخلايا الحسية البصرية في منطقة خروج العصب البصري؛ لذلك تسمى النقطة العمياء، وفي الشبكية مقابل الحدقة اللطخة الصفراء في مركزها انخفاض بسيط يسمى الحفيرة المركزية تبلغ فيها قوة البصر شدتها. ")
print("بدء تشغيل نظام استخراج المعرفة الهجين (Hybrid Knowledge Graph)...\n")
print(f"النص المدخل:\n{sample_text}\n")
print("-" * 60)
print("جاري التحليل النحوي واستخراج الكيانات، والتواصل مع خادم العلاقات (mT5 API)...")
try:
graph = process_full_document_pipeline_as_graph(
full_text=sample_text,
camel_parser_path=CAMEL_PARSER_PATH,
subject=SUBJECT,
)
print("\n--- الرسم البياني المعرفي الناتج (nodes/edges) ---")
if not graph.get("edges"):
print("تنبيه: لم يتم استخراج أي علاقات.")
print(json.dumps(graph, ensure_ascii=False, indent=2))
except ConnectionError:
print("\nخطأ: تعذّر الاتصال بخادم FastAPI.")
print("تلميح: تأكد من تشغيل ملف `api.py` في نافذة Terminal أخرى قبل تشغيل هذا الكود.")
except Exception as e:
print(f"\nخطأ غير متوقع أثناء التشغيل: {e}")