Spaces:
Sleeping
Sleeping
| # -*- coding: utf-8 -*- | |
| """ | |
| مستخرج الكيانات والعلاقات من نصوص عربية لبناء رسم بياني معرفي (KG) | |
| =================================================================== | |
| نظرة عامة | |
| --------- | |
| يحوّل هذا الملف نصاً عربياً خاماً إلى قائمتَي (كيانات، علاقات) جاهزتين | |
| لبناء رسم بياني معرفي (Knowledge Graph)، عبر خط أنابيب من مرحلتين: | |
| ١. **استخراج قاعدي (Rule-Based)** — `extract_entities_and_relations`: | |
| يعتمد على شجرة الاعتماد النحوي (Dependency Tree) الناتجة عن | |
| CamelParser2.0 بصيغة CATiB، ويمشي عليها لبناء الكيانات (نعت/إضافة | |
| مدمَجة) والعلاقات (فاعل-فعل-مفعول، حرف جر، جملة اسمية...). هذا | |
| المسار هو المصدر **الموثوق أساساً** - حتمي، قابل للتفسير والتصحيح. | |
| ٢. **استخراج هجين تكميلي (Hybrid/ML)** — `hybrid_relation_extraction_api`: | |
| لأزواج الكيانات التي لم يغطِّها المسار القاعدي، يُستشار نموذج mT5 | |
| مُدرَّب محلياً (عبر خادم FastAPI) لاقتراح علاقة إضافية، مع فلترة | |
| صارمة (استبعاد النفي، المعطوفات، العلاقات غير المرتبطة بالنص). | |
| يجمعهما `process_full_document_pipeline` في محرك واحد يعالج مستنداً كاملاً | |
| (متعدد الجمل)، مُرجعاً قائمة علاقات نهائية بعد إزالة التكرار. | |
| الاعتماديات الخارجية | |
| --------------------- | |
| - CAMeL Tools (التطبيع + التحليل الصرفي وحل الغموض عبر CAMeLBERT) | |
| - CamelParser2.0 (github.com/CAMeL-Lab/camel_parser) - التحليل النحوي، | |
| يتطلب Python 3.11.13 ونماذج SuPar محمَّلة مسبقاً | |
| - خادم FastAPI محلي يخدم نموذج mT5 المُدرَّب على المسار الثاني (اختياري؛ | |
| يعمل المسار القاعدي بمفرده بلا هذا الخادم) | |
| حدود معروفة (لم تُعالَج بعد، أو تحتاج بيانات أكبر لتتحسن) | |
| ----------------------------------------------------------- | |
| - تسميات علاقة من حروف جر مجردة ("من"، "في" وحدها) حين لا يوجد فعل صريح | |
| يربط الطرفين - مقبولة كحل احتياطي، ليست مثالية لغوياً. | |
| - نموذج mT5 الهجين ما زال يحتاج بيانات تدريب أكبر (انظر dataset.json) | |
| ليتحسن تعميمه على أزواج كيانات لم يرَ مثيلاً لها. | |
| - كلمات ذات لمة معجمية مشتركة مع كلمة توقف لكن بمعنى مختلف كلياً (مثال | |
| حقيقي عولج: "أنفس" التي لمتها "نفس") قد توجد حالات أخرى مشابهة لم | |
| تُكتشف بعد - إن فُقد كيان حقيقي رغم أنه ليس فارغ المعنى، هذا أول ما | |
| يستحق الفحص. | |
| - تنسيق الجمل المتعاطفة (جملتان اسميتان بواو العطف تتشاركان فعلاً/رابطاً | |
| محذوفاً) قد يخلط أحياناً مبتدأ أحدهما بخبر الأخرى. | |
| """ | |
| from __future__ import annotations | |
| import os | |
| import re | |
| import json | |
| from dataclasses import dataclass, field | |
| from typing import Dict, List, Optional, Set, Tuple | |
| # --------------------------------------------------------------------------- # | |
| # 0. Dediac helper | |
| # --------------------------------------------------------------------------- # | |
| 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 | |
| # --------------------------------------------------------------------------- # | |
| 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 | |
| # بعض التوكنات (غالباً أعداد مكتوبة كـ"الثلاثة") تخرج من المحلل بوسم | |
| # فارغ تماماً بدل NOM. توكن بلا وسم إطلاقاً أقرب لكونه اسماً غير | |
| # مصنَّف من كونه أداة أو علامة ترقيم (هذه توسَم صراحة دائماً)، فنعامله | |
| # كاسم بشكل احتياطي بدل إسقاطه بصمت. | |
| 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"} | |
| # ملاحظة تصميم: لا يوجد أي تحقق قائم على الوسم (pos/cpos) لتمييز الصفة عن | |
| # الاسم في هذا الملف عمداً - صيغة CATiB6 لا تملك فئة نحوية مستقلة للصفة | |
| # (الصفة تُوسم NOM تماماً كالاسم)، فأي محاولة كهذه صامتة الفشل دون أي خطأ | |
| # ظاهر. الدليل الموثوق الوحيد المستخدَم في كل هذا الملف هو العلاقة النحوية | |
| # (deprel): "MOD" = نعت/تخصيص، "IDF" = مضاف إليه، وهذا ما يبني عليه _walk. | |
| _PRONOUN_FORMS = { | |
| "هو", "هي", "هما", "هم", "هن", "أنا", "انا", "نحن", | |
| "أنت", "انت", "أنتما", "انتما", "أنتم", "انتم", "أنتن", "انتن", | |
| "ها", "نا", "ك", "كم", "هما", "هم" | |
| } | |
| def _is_pronoun(tok: Token) -> bool: | |
| """التحقق مما إذا كان التوكن ضميراً لكي نتجاهله تماماً. | |
| ملاحظة مهمة: وسم CATiB6 الخشن لا يملك فئة مستقلة للضمير - حروف الجر | |
| الحقيقية (في/من/على...) وحروف العطف تُوسم PRT أيضاً (مؤكَّد من تشغيل | |
| فعلي على شجرة تحليل حقيقية). لذلك "PRT" وحدها ليست دليلاً على أنه ضمير | |
| - الاعتماد عليها هنا كان يجعل كل حرف جر يبدو ضميراً ويُستهلك بصمت قبل | |
| أن يصل إلى _emit_pp_edge، وهو ما يفسر غياب العلاقات (Relations) بالكامل | |
| في أي تشغيل فعلي. القائمة المغلقة _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) | |
| # --------------------------------------------------------------------------- # | |
| # مُقتبَس ومُكيَّف من منطق "المصافي المتعددة" (Multi-Pass Sieve) لحل | |
| # الإحالة، مع تكييف جوهري لبنيتنا: النسخة الأصلية تعمل على تحليل صرفي | |
| # منفصل (MLEDisambiguator) وتوكنة خطية بسيطة حيث يبقى الضمير المتصل جزءاً | |
| # من كلمته المضيفة (يُستخرَج عبر حقل enc0). عندنا الضمائر المتصلة توكنات | |
| # مستقلة أصلاً بفعل تجزئة CamelParser2.0 (+ها، +ه...)، فلا حاجة لتحليل | |
| # enc0 - نكتفي بجدول جنس/عدد ثابت لكل صيغة ضمير، ونستخرج جنس/عدد المرشحين | |
| # من عمود FEATS مباشرة (متوفر أصلاً في مخرجات CamelParser2.0 الحقيقية). | |
| 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"), | |
| } | |
| # أعداد مكتوبة - لا تصلح أبداً كمُحال إليه (هي نعت لشيء آخر دائماً، لا | |
| # كيان بحد ذاتها)، رغم كونها قد تحمل صفات جنس/عدد نحوية تطابق ضميراً | |
| # بالمصادفة (مثال واجهناه فعلياً: "الثلاثة" طابقت "ها" قبل "المقابر" | |
| # الأبعد، فسبقتها في البحث للخلف رغم كونها النعت لا الاسم المقصود). | |
| _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 لكل جملة على حدة، إذ يحتاجها فحص التعريف | |
| عبر الإضافة (_is_definite_token) للوصول إلى أبناء أي مرشح داخل جملته | |
| هو تحديداً، لا التيار المسطَّح ذاته.""" | |
| 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 resolve_pronoun_antecedents( | |
| sentences: List[Sentence], | |
| ) -> Dict[Tuple[int, int], Tuple[int, Token]]: | |
| """ | |
| تُرجع خريطة: (رقم جملة الضمير، معرّفه) -> (رقم جملة المُحال إليه، توكنه). | |
| رقما الجملة قد يختلفان (إحالة عبر جمل)، لذا نُرجع رقم جملة المُحال | |
| إليه صراحة - معرّف التوكن وحده (tid) يُعاد استخدامه بين الجمل فلا يكفي | |
| لتحديد أي جملة يخصّها التوكن. | |
| تبحث لكل ضمير غائب متصل (ه/ها/هما/هم/هن) عن أقرب اسم معرَّف سابق في | |
| المستند (عبر حدود الجمل) يطابقه جنساً وعدداً - التعريف هنا يشمل التعريف | |
| الصريح ("ال") والمكتسَب عبر الإضافة معاً - مع تفعيل قاعدة "جمع غير | |
| العاقل يُعامَل معاملة المفرد المؤنث" النحوية العربية. | |
| """ | |
| 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] | |
| for j in range(i - 1, -1, -1): | |
| cand_s_idx, cand = stream[j] | |
| if not _is_nominal(cand) or _is_pronoun(cand): | |
| 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 surface == "ها": | |
| if nh_plural_match: | |
| resolution[(s_idx, tok.tid)] = (cand_s_idx, cand) | |
| break | |
| if direct_match and (s_idx, tok.tid) not in resolution: | |
| resolution[(s_idx, tok.tid)] = (cand_s_idx, cand) | |
| continue | |
| if direct_match or nh_plural_match: | |
| resolution[(s_idx, tok.tid)] = (cand_s_idx, cand) | |
| break | |
| return resolution | |
| def build_antecedent_text_map(sentences: List[Sentence]) -> Dict[Tuple[int, int], str]: | |
| """تبني خريطة (رقم جملة الضمير، معرّفه) -> النص الكامل للمُحال إليه | |
| (لا التوكن وحده، بل عبارته الاسمية الكاملة كما تستخرجها _walk ضمن | |
| جملته هو - نعوته وإضافاته). هذه هي الخريطة التي تُستهلَك مباشرة داخل | |
| extract_entities_and_relations لإثراء الكيانات التي تفقد ضميرها. | |
| المشي هنا معزول تماماً (consumed مستقلة) فلا يمسّ استهلاك أي جملة أخرى. | |
| """ | |
| 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. Stop-lemma list | |
| # --------------------------------------------------------------------------- # | |
| # نوعان مختلفان تماماً من كلمات التوقف - خلطهما في قائمة واحدة هو سبب أن كل | |
| # إضافة جديدة كانت تُصلح حالة وتكسر أخرى: | |
| # | |
| # (١) HARD - أدوات/محددات/كمّيات لا معنى لها أبداً حتى مع محدِّد حقيقي، وكذلك | |
| # أسماء الظرف المكاني والزماني (فوق، تحت، بين...) التي تعمل نحوياً مثل | |
| # حرف جر لا كاسم. تُسقط الكلمة نفسها دائماً، لكن ما تحتها من محتوى حقيقي | |
| # يبقى ككيان مستقل. مثال: "بعض الأنظمة" -> "الأنظمة" (وليس "عض الأنظمة"). | |
| # | |
| # (٢) WEAK - أسماء محتوى حقيقية، لكنها عامة جداً لتكون كياناً مفيداً بمفردها. | |
| # فارغة المعنى فقط حين تأتي عارية بلا أي تخصيص؛ إن أُلحق بها نعت أو | |
| # إضافة حقيقية، يُبقى الرأس ضمن الكيان الكامل لأن المركّب أصبح محدداً. | |
| # مثال: "سياق متصل" تبقى كاملة، بينما "سياق" وحدها تُسقط بالكامل. | |
| # | |
| # راجع هذا التصنيف على نصوصك الفعلية وانقل الكلمات بين القائمتين حسب ما | |
| # تراه - المعيار الحاسم: هل الكلمة عديمة المعنى حتى بعد تخصيصها بنعت أو | |
| # إضافة؟ إن كانت الإجابة نعم دائماً -> HARD. إن كانت الإجابة "فقط حين تكون | |
| # عارية" -> WEAK. | |
| # --------------------------------------------------------------------------- # | |
| _HARD_STOP_FORMS = { | |
| # كمّيات ومحددات لا معنى لها أبداً | |
| "بعض", "عض", "كل", "جميع", "كافة", "كافه", "عدة", "عده", "عديد", | |
| "نوع", "نوعين", "كثير", "كثيره", "كثيرة", "قليل", "قليلة", | |
| "مختلف", "مختلفه", "مختلفة", "غير", "سوى", "نفس", "ذات", "اخر", | |
| # أسماء موصولة وإشارة - وصل نحوي بحت، ليست محتوى أبداً | |
| "الذي", "التي", "والذي", "والتي", "اللذان", "اللتان", "الذين", "اللاتي", | |
| "هذا", "هذه", "ذلك", "تلك", "هؤلاء", "أولئك", | |
| # ظروف مكان/زمان تعمل كحرف جر نحوياً - العلاقة لا الكلمة هي المهمة | |
| "بعد", "قبل", "وسط", "حول", "عند", "بين", "خلال", "داخل", "خارج", | |
| "فوق", "تحت", "حيث", "هنا", "هناك", "قرب", "لدى", | |
| # أدوات خطابية/مقارنة لا تحمل معنى كياني أبداً بذاتها | |
| "اما", "أما", "مثل", "كمثل", | |
| # أسماء تأطير خطابي (discourse framing) - "في سياق"، "في إطار"، "على | |
| # صعيد"، "في نطاق" هي روابط نصية بحتة (تشير إلى الخطاب نفسه لا إلى | |
| # كيان في العالم)، لا تصلح رأساً لكيان حتى بعد وصفها بنعت ("سياق | |
| # متصل" ليست كياناً أكثر من "سياق" وحدها - النعت لا يمنحها مرجعاً | |
| # واقعياً). نُقلت "سياق" هنا من WEAK لأن قاعدة WEAK ("أبقِ الرأس إن | |
| # وُصف") غير صحيحة لهذه الفئة تحديداً؛ رصدنا هذا فعلياً في مخرجات | |
| # اختبار حقيقية أنتجت الكيان الزائف "[سياق متصل]" كطرف علاقة. | |
| "سياق", "اطار", "إطار", "نطاق", "صعيد", | |
| # ظروف حال/تكرار جامدة | |
| "كذلك", "ايضا", "موجود", "موجوده", "موجودة", "يوجد", | |
| "جدا", "حقا", "تماما", "نسبيا", "غالبا", "فقط", "دائما", "احيانا", | |
| "مختلف", "مختلفه", "مختلفة", "غير", "سوى", "نفس", "ذات", "اخر", | |
| "سيما", "لا سيما", "خاصة", "بخاصة", | |
| } | |
| _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 فقط حين يكون النص المُعاد هو شكل الكلمة-المتوقَّفة نفسه بلا أي | |
| إضافة - أي أن _walk لم تُنقذ أي محتوى حقيقي من تحتها. تُستخدم بدل | |
| "not _is_stopword(tok)" الخام في قرار قبول/رفض علاقة أو كيان: تلك | |
| الصيغة الخام كانت تُسقط علاقات صحيحة كلياً كل مرة يكون فيها الرأس نفسه | |
| كلمة توقف لكن نُقل تحته محتوى حقيقي (مثال: "كل عنصر" -> "عنصر" - "كل" | |
| كلمة توقف، لكن النص الناتج "عنصر" ليس فارغاً، فيجب ألا تُرفض).""" | |
| 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. The actual tree walk | |
| # --------------------------------------------------------------------------- # | |
| 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, | |
| ) -> 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) | |
| continue | |
| if _is_particle(child): | |
| if _is_conjunction(child): | |
| continue | |
| _emit_pp_edge(child_id, tokens, children, consumed, relations, root_id, _depth) | |
| 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) | |
| 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) | |
| 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) | |
| 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, | |
| ) -> 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: | |
| return | |
| # استخدام _walk_entities بدل _walk المباشرة: لو كان متمّم حرف الجر | |
| # معطوفاً ("...في التربة والهواء") تُصدر علاقة مستقلة لكل معطوف بدل | |
| # دمجها في نص واحد أو فقدان ما بعد الأول (انظر _collect_conjunct_heads). | |
| for cid, target_text in _walk_entities(obj_id, tokens, children, consumed, | |
| relations, _depth=depth + 1): | |
| 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, | |
| ) -> 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): | |
| 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) فيفقد كل | |
| عناصر التعداد بعد الأول ("الفم والبلعوم والمريء..." تُختزل إلى "الفم" | |
| فقط). تتتبّع هذه الدالة السلسلة عبر أي عمق من حروف العطف المتتالية، | |
| وتستهلك توكن حرف العطف نفسه (لا كيان له) دون استهلاك رؤوس المعطوفات - | |
| تُترك هذه لتُعالجها _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, | |
| ) -> List[Tuple[int, str]]: | |
| """مثل _walk تماماً، لكنها تُرجع **قائمة** (رأس المعطوف، نصه) بدل نص | |
| واحد مُدمَج. تُستخدَم في كل نقاط إصدار الكيانات/العلاقات النهائية | |
| (فاعل، مفعول، متمِّم حرف جر، خبر، كيان مستقل) كي تُفكِّك أي تعداد | |
| معطوف إلى علاقات/كيانات مستقلة بدل دمجها في نص واحد أو فقدان أغلبها | |
| (انظر شرح _collect_conjunct_heads). _walk نفسها تبقى كما هي للاستخدام | |
| الداخلي البحت (بناء نص التركيب الاسمي الواحد من نعوته وإضافاته).""" | |
| 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) | |
| 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 اختياريان تماماً (القيم الافتراضية | |
| تُبقي كل استدعاء قائم يعمل بلا أي تغيير في السلوك). حين يُمرَّران | |
| (كما تفعل extract_kg_entities تلقائياً)، يُستخدَمان لإثراء كيانات | |
| فقدت ضميرها المتصل بنص المُحال إليه المُحلَّل مسبقاً عبر | |
| 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] = [] | |
| def _enrich_with_antecedent(text: str, head_tid: int) -> str: | |
| """إن كان لهذا الرأس ضمير غائب متصل مباشرة (IDF غالباً، كـ"ها" في | |
| "أكبرها") وحُلَّت إحالته مسبقاً، تُلحَق عبارة المُحال إليه الكاملة | |
| بنص الكيان بدل تركه ناقصاً بعد إسقاط الضمير بصمت كالمعتاد.""" | |
| if not antecedent_text_map: | |
| return text | |
| for cid in children.get(head_tid, []): | |
| child = tokens.get(cid) | |
| if child and _is_pronoun(child): | |
| resolved = antecedent_text_map.get((sentence_idx, child.tid)) | |
| if resolved and resolved not in text: | |
| return f"{text} {resolved}" | |
| return text | |
| # يُحسَب مرة واحدة لكل الجملة: أي فعل يحمل أداة نفي مباشرة (لم/لن/لا/ | |
| # ليس) لا يُصدر أي علاقة إطلاقاً، لأن أي علاقة مستخرَجة منه ستكون | |
| # معكوسة تماماً لما تقوله الجملة فعلاً. | |
| 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) | |
| topic_text = _enrich_with_antecedent(topic_text, tok.tid) | |
| pred_text = _walk(head_tok.tid, tokens, children, temp_consumed, [], root_id=head_tok.tid) | |
| if topic_text and pred_text: | |
| comprehensive_relations.append((topic_text, "هو", pred_text)) | |
| entities.append(topic_text) | |
| entities.append(pred_text) | |
| 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:] | |
| prep_id = next((c for c in children.get(tok.tid, []) if tokens.get(c) and _is_particle(tokens[c])), None) | |
| # أ- فعل متعدي (فاعل + فعل + مفعول به) - نستخدم الفاعل الضمني | |
| # إن غاب الصريح، ليشمل جمل الصلة أيضاً لا الجمل ذات الفاعل | |
| # المذكور فقط. | |
| # استخدام _walk_entities بدل _walk: لو كان الفاعل أو المفعول | |
| # معطوفاً ("الفم والبلعوم والمريء...") تُصدر علاقة مستقلة لكل | |
| # معطوف (جداء ديكارتي بين قوائم الفاعل والمفعول) بدل دمج كل | |
| # المعطوفات في كيان واحد أو فقدان ما بعد المعطوف الأول بصمت. | |
| if implicit_sbj_id and obj_id: | |
| temp_consumed = set(consumed) | |
| sbj_pairs = _walk_entities(implicit_sbj_id, tokens, children, temp_consumed, []) | |
| obj_pairs = _walk_entities(obj_id, tokens, children, temp_consumed, []) | |
| for extra_id in extra_obj_ids: | |
| obj_pairs.extend( | |
| _walk_entities(extra_id, tokens, children, temp_consumed, []) | |
| ) | |
| sbj_texts = [t for _, t in sbj_pairs] | |
| obj_texts = [t for _, t in obj_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: | |
| 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(implicit_sbj_id, tokens, children, temp_consumed, []) | |
| prep_obj_pairs = _walk_entities(prep_obj_id, tokens, children, temp_consumed, []) | |
| 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 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 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, []) | |
| for extra_id in extra_obj_ids: | |
| obj_pairs.extend(_walk_entities(extra_id, tokens, children, temp_consumed, [])) | |
| 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, []) | |
| 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, []) | |
| prep_obj_pairs = _walk_entities(prep_obj_id, tokens, children, temp_consumed, []) | |
| 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, []) | |
| prd_pairs = _walk_entities(prd_id, tokens, children, temp_consumed, []) | |
| 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, []) | |
| tpc_texts = [t for _, t in tpc_pairs] | |
| pred_text = _walk(tok.tid, tokens, children, temp_consumed, [], root_id=tok.tid) | |
| pred_text = _enrich_with_antecedent(pred_text, tok.tid) | |
| 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 | |
| # معالجة حرف الجر "ب" المرتبط بظرف حال - سواء تعلّق بفعل مباشرة | |
| # ("تمكنوا بنجاح") أو باسم مصدري ("ضمان سير العمليات بكفاءة"، | |
| # حيث "سير" اسم فعل لا فعلاً بالوسم النحوي الخشن لكنه يحمل نفس | |
| # المعنى الحدثي). لا نميّز بين الاثنين هنا لأن 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) | |
| 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) | |
| 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: | |
| """ | |
| تصلح خطأ تحليل صرفي شائع مع الكلمات المعرَّبة/العلمية: المحلل يفصل أول | |
| حرف من الكلمة (ب/ل/ك/و/ف) بصفته حرف جر أو عطف حقيقياً تاركاً بقية | |
| الكلمة توكناً مستقلاً غريباً. حين يطابق الجزء المتبقي كلمة معرَّبة | |
| معروفة، نُعيد لصق الحرف داخل توكن واحد، ونُعيد توجيه أي أحفاد كان | |
| "الحرف" يحملهم إلى الكلمة المُصلَحة، ثم نحذف توكن الحرف كلياً من | |
| الجملة (لا نتركه معلَّقاً بلا أثر - ذلك كان يُخرب محاذاة الكلمات | |
| السطحية التي تعتمد عليها _patch_noan_tokens). | |
| """ | |
| 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 دقيقة 100% حين تتطابق الأعداد، بعكس المطابقة | |
| التقريبية (substring) التي كانت تنزلق بسهولة وتُلصق كلمة بمكان خاطئ. | |
| علامات الترقيم (PNX) لا تُحسَب، لأنها لا تقابل أي كلمة في 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 grouping) لا على | |
| مطابقة نصية تقريبية. مطابقة تقريبية بسيطة (`k in word or word in k`) | |
| تتحقق بسهولة زائدة لأي شذرة قصيرة، وأي تطابق زائف واحد يُزيح كل ما | |
| بعده (كلمة صحيحة تُلصَق بتوكن NOAN لا يخصها). | |
| الترقيع يتم فقط حين: | |
| (أ) عدد الكلمات السطحية (بعد التجميع البنيوي) = عدد كلمات النص | |
| الأصلي بالضبط - أي لا شيء غامض في المحاذاة، و | |
| (ب) توكن 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 SECTION-------------------- | |
| def _find_entity_spans( | |
| entity_text: str, sentence: Sentence, max_gap: int = 3 | |
| ) -> List[List[int]]: | |
| """ | |
| تبحث عن كل الذِّكرات (mentions) الفعلية للكيان في الجملة، لا عن أي توكن | |
| تصادف أن كلمته ضمن كلمات نص الكيان في أي مكان بالجملة. | |
| هذا يُصلح خللاً جوهرياً: `_get_entity_token_ids` القديمة كانت تجمع | |
| الكيانات بـ"كيس كلمات" (bag-of-words) عبر الجملة بأكملها - فكيان مثل | |
| "نواة مركزية" كان يلتقط *كل* تكرار لكلمة "نواة" في الجملة كلها حتى لو | |
| كانت ذكراً مختلفاً تماماً (مثلاً "النواة" لاحقاً بصيغة معرَّفة منفصلة). | |
| هذا كان يُفسد حساب "الرأس" والمسافة النحوية تماماً، فيجعل كل زوج | |
| كيانات تقريباً يبدو "قريباً" نحوياً بصرف النظر عن علاقته الفعلية - وهو | |
| ما يفسر انفجار عدد الأزواج المُرسَلة للنموذج، والعلاقات المستحيلة | |
| اتجاهياً (نفس الزوج بنفس العلاقة أو بعلاقتين متضاربتين في الاتجاهين). | |
| البحث هنا يشترط أن تظهر كلمات الكيان بالترتيب الصحيح ومتقاربة (بفارق | |
| لا يتجاوز 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]: | |
| """تضع <tag>...</tag> حول الظهور رقم n (0-indexed) تحديداً لنص الكيان | |
| - لا أول ظهور نصي عارض كما كانت تفعل النسخة السابقة (count=1)، والتي | |
| كانت تُخطئ الذكرة الصحيحة كلما تكرر النص بدلالة مختلفة سابقاً في | |
| الجملة. تُعيد None إن تعذّر العثور على العدد الكافي من الظهورات (نص | |
| الكيان المُنظَّف لا يطابق النص الخام حرفياً بعد تجريد التشكيل).""" | |
| 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]}</{tag}>" | |
| + 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 | |
| # --------------------------------------------------------------------------- # | |
| # إصلاح: منع تزاوج المعطوفات مع بعضها كطرفَي علاقة. "حماية أنفسهم وأموالهم" | |
| # أنتجت "[حماية] --(و)--> [أموال]" حرفياً - سؤال النموذج عن العلاقة بين | |
| # معطوفَين (مفعولان متوازيان لشيء ثالث) سؤال لا معنى له، والنموذج يهلوس | |
| # حرف العطف نفسه كأنه علاقة. نعيد استخدام _collect_conjunct_heads الموجودة | |
| # فعلاً (بُنيت أصلاً لتوسيع الكيانات) للتحقق من هذا بلا أي منطق جديد مكرَّر. | |
| # --------------------------------------------------------------------------- # | |
| 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 | |
| # --------------------------------------------------------------------------- # | |
| # إصلاح: لا تحقق سابقاً من أن العلاقة التي يُرجعها النموذج مرتبطة فعلياً | |
| # بالنص المُرسَل - _INVALID_RELATIONS تتحقق فقط من قيم فارغة/"none" صريحة. | |
| # مثال هلوسة حقيقي: "ينظم" ظهرت كعلاقة لجملة لا تحوي هذا الفعل إطلاقاً. | |
| # فحص بسيط لكنه فعّال: هل تُشارك كلمات العلاقة المُتوقَّعة (أفعال غالباً) | |
| # النص المُرسَل بكلمة محتوى واحدة على الأقل؟ علاقة صحيحة من هذه الجملة | |
| # تحديداً يجب أن تُبنى من مفرداتها، لا من فراغ. | |
| # --------------------------------------------------------------------------- # | |
| 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"</?e[12]>", "", 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]]: | |
| """ | |
| تفلتر الكيانات نحوياً ومنطقياً، تضع الوسوم بدقة حول الذِّكرة الصحيحة، | |
| وتتواصل مع خادم API - لكل زوج كيانات مرتبط نحوياً ضمن مسافة معقولة | |
| فقط، لا كل التوليفات الممكنة. | |
| (١) already_related: أي زوج كيانات استخرجت له extract_entities_and_ | |
| relations (المحرك القاعدي/النحوي) علاقة صريحة بالفعل لا يُعاد سؤال | |
| النموذج عنه إطلاقاً. المحرك القاعدي هو المصدر الموثوق (يعرف من فاعل | |
| ومن مفعول عبر شجرة الإعراب)؛ استدعاء النموذج على نفس الزوج قد يُنتج | |
| تسمية مختلفة وخاطئة لعلاقة صحيحة أصلاً. | |
| (٢) 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 "<e1>" not in tagged or "<e2>" 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) | |
| # تقطيع النص بدقة مع إزالة المسافات الزائدة | |
| 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]]: | |
| """ | |
| محرك الـ Pipeline الشامل الجاهز للإنتاج (Production). | |
| يدمج التحليل النحوي، استخراج الكيانات، الاستنتاج عبر API، والتنقية النهائية. | |
| علاقات extract_kg_entities (المحرك القاعدي) تُعتمَد كمصدر أساسي وموثوق | |
| لأنها مبنية على شجرة إعراب فعلية لا تخمين نموذج. نموذج mT5 يُستخدَم | |
| فقط كطبقة تكميلية لسد الثغرات - أزواج كيانات لم يستخرج لها المحرك | |
| القاعدي أي علاقة أصلاً - وحتى عندئذ، أي علاقة من النموذج لزوج غطّته | |
| القواعد بالفعل تُستبعَد نهائياً في الدمج الأخير حتى لو اختلفت التسمية، | |
| كي لا يتناقض ناتجان لنفس الزوج في نفس الغراف. | |
| مبنية على ثلاث تمريرات (لا تمريرة واحدة كسابقاً) لسبب جوهري: حل | |
| الإحالة المرجعية (مثال حقيقي: "ها" في جملة تُحيل إلى اسم في جملة | |
| سابقة) يحتاج رؤية جمل المستند **كاملة معاً** قبل أي استخراج - تحليل كل | |
| جملة بمعزل تام (كما كان يحدث سابقاً، جملة فجملة داخل حلقة واحدة) يمنع | |
| هذا كلياً مهما كانت خريطة الإحالة نفسها صحيحة. | |
| """ | |
| all_rule_relations: List[Tuple[str, str, str]] = [] | |
| all_hybrid_relations: List[Tuple[str, str, str]] = [] | |
| sentences = split_text_into_sentences(full_text) | |
| # ------------------------------------------------------------------- | |
| # التمريرة الأولى: تحليل كل جملة نحوياً (كالسابق تماماً، جملة فجملة | |
| # عبر خدمة CAMeL)، لكن دون استخراج كيانات بعد - فقط نجمع كل أشجار | |
| # المستند معاً أولاً. | |
| # ------------------------------------------------------------------- | |
| per_sentence_parsed: List[Tuple[str, List[Sentence]]] = [] | |
| all_parsed_sentences: List[Sentence] = [] | |
| for sent_str in sentences: | |
| raw_conllx = run_camel_parser(sent_str, camel_parser_path) | |
| if isinstance(raw_conllx, list): | |
| raw_conllx = "\n".join(raw_conllx) | |
| normalized = _normalize_conll_input(raw_conllx) | |
| parsed = parse_conllx(normalized) | |
| 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): **لا تصنيف إطلاقاً** - نص العلاقة الحرفي نفسه هو الـ | |
| # type دائماً، كما تُظهر كل الأمثلة الحقيقية المرفقة (لا توجد علاقة واحدة | |
| # فيها بقيمة type مصنَّفة، كلها نص العلاقة كما استُخرج بالحرف). | |
| # | |
| # للكيانات (nodes): التصنيف واعٍ بمجال النص (subject: تاريخ/جغرافيا/علوم) | |
| # لأن الفئات المتوقَّعة تختلف كلياً بين المجالات - "نوع" في نص علمي تعني | |
| # حالة/عملية/ظاهرة مجردة، بينما نص تاريخي يحتاج حدث/زمان/اشخاص/مفهوم/ | |
| # مكان/فعل، ونص جغرافي يحتاج مكان/مادة بالدرجة الأولى. ما لا ينتمي لأي | |
| # فئة معروفة يأخذ اسمه نفسه (نفس السلوك الاحتياطي المطلوب أصلاً). | |
| # --------------------------------------------------------------------------- # | |
| def _normalize_for_typing(text: str) -> str: | |
| """تطبيع الهمزات (أ/إ/آ -> ا) إضافة لإزالة التشكيل - غيابه كان يمنع | |
| مطابقة كلمات صحيحة تماماً بسبب اختلاف شكل الألف فقط (مثال حقيقي: | |
| "الأذن" لا تطابق "اذن" في قائمة الكلمات المفتاحية بلا هذا التطبيع).""" | |
| return re.sub(r"[أإآ]", "ا", dediac_ar(text)) | |
| def _strip_al_prefix(word: str) -> str: | |
| return word[2:] if word.startswith("ال") else word | |
| def _matches_keyword_set(surface: str, keyword_set: frozenset) -> bool: | |
| """تحقق مطابقة لعبارة كاملة مذكورة **ضمن** النص (لا مساواة تامة - | |
| "عملية التركيب الضوئي" يجب أن تطابق "تركيب ضوئي" رغم كونها أطول)، أو | |
| لأي كلمة مفردة ضمنه (بعد تجريد "ال"). النسخة "منزوعة أل" من كامل | |
| العبارة تُستخدَم أيضاً للعبارات متعددة الكلمات - "نصف الكرة المخية" | |
| يجب أن تطابق "نصف كرة مخية" رغم أن "ال" تتخلل الكلمتين.""" | |
| 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 keyword_set if " " in kw): | |
| return True | |
| for word in words: | |
| bare = _strip_al_prefix(word) | |
| if word in keyword_set or bare in keyword_set: | |
| return True | |
| return False | |
| def _first_word_matches(surface: str, keyword_set: frozenset, n: int = 1) -> bool: | |
| """تتحقق من أول ن كلمة/كلمات فقط - تُستخدَم حين تكون الكلمة الرئيسية | |
| (لا أي معدِّل عميق داخل العبارة) هي الدليل الموثوق. مثال حقيقي دفع هذا: | |
| "نخر السن" (نخر = حالة مجردة، السن = عضو) كانت تُصنَّف "عضو" خطأً | |
| لمجرد احتواء "السن" في مكان ما بالعبارة، رغم أن الكلمة الرئيسية "نخر" | |
| لا علاقة لها بعضو فعلياً.""" | |
| words = surface.split()[:n] | |
| head = " ".join(_strip_al_prefix(w) for w in words) | |
| if any(kw == head or kw in head for kw in keyword_set): | |
| return True | |
| return any(_strip_al_prefix(w) in keyword_set or w in keyword_set for w in words) | |
| # ===== مجال: تاريخ ===== | |
| _PERSON_KEYWORDS_HIST = frozenset(dediac_ar(w) for w in { | |
| "معتمد", "ملك", "ملكة", "رئيس", "قائد", "وزير", "سفير", "حاكم", | |
| "خليفة", "سلطان", "امير", "زعيم", "قاضي", "محتل", "غازي", "قنصل", | |
| "حلفاء", "شعب", "قوم", "جيش", "جنود", "سكان", "مندوب", "وفد", | |
| }) | |
| _EVENT_KEYWORDS_HIST = frozenset(dediac_ar(w) for w in { | |
| "معركة", "حرب", "ثورة", "غزو", "فتح", "حصار", "انتفاضة", "مؤتمر", | |
| "معاهدة", "انقلاب", "احتلال", "تحرير", "وفاة", "ميلاد", "تاسيس", | |
| "سقوط", "توقيع", "انتخابات", "اجتياح", "تمرد", "استقلال", "معاهدات", | |
| }) | |
| _PLACE_KEYWORDS_HIST = frozenset(dediac_ar(w) for w in { | |
| "جزيرة", "مدينة", "قرية", "دولة", "بلاد", "اقليم", "منطقة", "عاصمة", | |
| "قارة", | |
| }) | |
| _CONCEPT_KEYWORDS = frozenset(dediac_ar(w) for w in { | |
| "مبدأ", "مبادئ", "نظرية", "فكرة", "فلسفة", "عقيدة", "دستور", "قانون", | |
| "ايديولوجيا", "مذهب", "نظام سياسي", "نزعة", | |
| }) | |
| _ACTION_KEYWORDS_HIST = frozenset(dediac_ar(w) for w in { | |
| "اجراء", "اجراءات", "عمل", "اعمال", "تدبير", "تدابير", "خطوة", | |
| "خطوات", "قرار", "قرارات", "سياسة", "سياسات", "حملة", | |
| }) | |
| # ===== مجال: جغرافيا ===== | |
| _PLACE_KEYWORDS_GEO = frozenset(dediac_ar(w) for w in { | |
| "بادية", "جبل", "جبال", "وادي", "اودية", "هضبة", "حوضة", "احواض", | |
| "صحراء", "نهر", "بحر", "بحيرة", "سهل", "ساحل", "حدود", "جزيرة", | |
| "منطقة", "اقليم", "مناجم", "حقول", "سبخة", "حماد", | |
| }) | |
| _MATERIAL_KEYWORDS_GEO = frozenset(dediac_ar(w) for w in { | |
| "مراعي", "ملح", "فوسفات", "غاز", "نفط", "معادن", "تربة", "مياه جوفية", | |
| }) | |
| # ===== مجال: علوم ===== | |
| _ORGAN_KEYWORDS = frozenset(dediac_ar(w) for w in { | |
| "عضو", "اعضاء", "غشاء", "اغشية", "خلية", "خلايا", "نواة", "عضلة", | |
| "عضلات", "نسيج", "انسجة", "جهاز", "قناة", "قنوات", "انبوب", "باحة", | |
| "باحات", "قشرة", "تلفيف", "حصين", "مشبك", "مشابك", "ليف", "الياف", | |
| "اذن", "عين", "جلد", "بشرة", "رحم", "جنين", "مشيمة", "سن", "اسنان", | |
| "عاج", "لب", "معدة", "مريء", "امعاء", "فم", "مطرقة", "ركاب", "سندان", | |
| "نافذة", "هدب", "اهداب", "ميسم", "بذيرة", "طلع", "جراثيم", "فك", | |
| "طبقة", "طبقات", "كرة مخية", "نصف كرة", "عظم", "عظيمات", "عظام", | |
| "نطفة", "نطف", "مركز", "اثنا عشر", | |
| }) | |
| _MATERIAL_KEYWORDS_SCI = frozenset(dediac_ar(w) for w in { | |
| "مادة", "مواد", "شوارد", "عصارة", "عصارات", "سائل", "غاز", "معدن", | |
| "ملح", "دهون", "بروتين", "هرمون", "انزيم", "ناقل", "نواقل", "غذاء", | |
| "فضلات", "دم", | |
| }) | |
| # قائمة ضيقة عمداً، لا قائمة "فعل" التاريخ العامة: "عمل" (ضمن "كمون | |
| # العمل" - ظاهرة كهربائية لا فعلاً بشرياً) كانت تُطابَق خطأً لو استُخدمت | |
| # القائمة العامة نفسها هنا - المجالان يحتاجان قوائم مختلفتين تماماً. | |
| _ACTION_KEYWORDS_SCI = frozenset(dediac_ar(w) for w in {"اهمال", "افراط", "تدخين", "سوء استخدام"}) | |
| # كلمة حالة/عملية مجردة تتصدّر العبارة تفوق أي عضو/مادة يليها - "نخر | |
| # السن" حالة مرضية (نوع)، لا عضواً، رغم احتواء "السن" (عضو حقيقي). | |
| _ABSTRACT_CONDITION_KEYWORDS = frozenset(dediac_ar(w) for w in { | |
| "نخر", "التهاب", "تراكم", "تورم", "نزيف", "كسر", "التواء", "ضمور", | |
| "تنكس", "قصور", "فشل", "عدوى", "تسمم", "حساسية", "انسداد", "تشنج", | |
| }) | |
| _HISTORY_CHECK_ORDER = ( | |
| ("حدث", _EVENT_KEYWORDS_HIST), | |
| ("مكان", _PLACE_KEYWORDS_HIST), | |
| ("مفهوم", _CONCEPT_KEYWORDS), | |
| ("فعل", _ACTION_KEYWORDS_HIST), | |
| ("اشخاص", _PERSON_KEYWORDS_HIST), | |
| ) | |
| def classify_node_type(label: str, subject: str = "") -> str: | |
| """تُصنِّف الكيان حسب مجال النص (subject) إلى فئة دلالية مناسبة، أو | |
| تُرجع اسمه نفسه إن لم ينتمِ لأي فئة معروفة - نفس السلوك الاحتياطي | |
| الذي وصفه المستخدم أصلاً.""" | |
| surface = _normalize_for_typing(label).strip() | |
| subj = _normalize_for_typing(subject).strip() | |
| if re.fullmatch(r"\d{3,4}", surface): | |
| return "زمان" | |
| if subj == "تاريخ": | |
| for type_name, kws in _HISTORY_CHECK_ORDER: | |
| if _matches_keyword_set(surface, kws): | |
| return type_name | |
| # احتياطي: أغلب ما تبقى في نصوص التاريخ بعد استبعاد الفئات | |
| # الأخرى أشخاص (أعلام أو مجموعات بشرية) لا يمكن حصرها بقائمة | |
| # كلمات مفتاحية ثابتة (أسماء أعلام لا حصر لها) | |
| return "اشخاص" | |
| if subj == "جغرافيا": | |
| # أولوية الكلمة الرئيسية: "مناجم الفوسفات" (رئيسية=مناجم) مكان، | |
| # "الملح من سبخة الموح" (رئيسية=الملح) مادة رغم احتواء "سبخة" | |
| # (مكانية) لاحقاً - الكلمة الرئيسية هي الدليل الحاسم لا أي كلمة | |
| # تظهر في أي مكان بالعبارة. | |
| if _first_word_matches(surface, _MATERIAL_KEYWORDS_GEO, n=1): | |
| return "مادة" | |
| if _first_word_matches(surface, _PLACE_KEYWORDS_GEO, n=1): | |
| return "مكان" | |
| if _matches_keyword_set(surface, _PLACE_KEYWORDS_GEO): | |
| return "مكان" | |
| if _matches_keyword_set(surface, _MATERIAL_KEYWORDS_GEO): | |
| return "مادة" | |
| return "مكان" # الغالبية الساحقة في نصوص الجغرافيا أماكن | |
| if subj in ("علوم", "احياء", "بيولوجيا", "كيمياء", "فيزياء"): | |
| if _matches_keyword_set(surface, _ACTION_KEYWORDS_SCI): | |
| return "فعل" | |
| if _first_word_matches(surface, _ABSTRACT_CONDITION_KEYWORDS, n=1): | |
| return "نوع" | |
| if _first_word_matches(surface, _MATERIAL_KEYWORDS_SCI, n=2): | |
| return "مادة" | |
| if _first_word_matches(surface, _ORGAN_KEYWORDS, n=2): | |
| return "عضو" | |
| if _matches_keyword_set(surface, _ORGAN_KEYWORDS): | |
| return "عضو" | |
| if _matches_keyword_set(surface, _MATERIAL_KEYWORDS_SCI): | |
| return "مادة" | |
| return "نوع" # الفئة الافتراضية الشاملة لكل ما هو مجرد | |
| # مجال غير معروف: افحص كل الفئات، وإلا استخدم اسم الكيان نفسه | |
| 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 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. | |
| """ | |
| node_ids: Dict[str, str] = {} | |
| nodes: List[Dict[str, str]] = [] | |
| def get_or_create_node(label: str) -> str: | |
| 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]] = [] | |
| for source, rel, target in relations: | |
| source_id = get_or_create_node(source) | |
| target_id = get_or_create_node(target) | |
| # العلاقة تُترَك بنصها الحرفي دائماً كنوعها - هذا ما تُظهره كل | |
| # الأمثلة الحقيقية المرفقة من زميلة العمل بلا استثناء واحد؛ لا | |
| # تصنيف مطلوب هنا خلافاً للكيانات. | |
| 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 | |
| # الفعلي على جهازك) | |
| BASE_DIR = os.path.dirname(os.path.abspath(__file__)) | |
| CAMEL_PARSER_PATH = os.path.dirname(os.path.abspath(__file__)) | |
| # 2. نص اختباري يحتوي كيانات مترابطة (قصير ومكثف لاختبار السرعة والربط) | |
| sample_text = ( | |
| "قبل آلاف السنين، عاش القدماء المصريون على ضفاف نهر النيل. " | |
| "وكان لديهم حكام يلقبون بالفراعنة. " | |
| "اعتقد هؤلاء الملوك أن هناك حياة ثانية بعد الموت، لذلك قرروا بناء مقابر ضخمة وقوية لحماية أنفسهم وأموالهم. " | |
| "أشهر هذه المقابر هي أهرامات الجيزة الثلاثة، وأكبرها الهرم الأكبر الذي بناه الملك خوفو. " | |
| "لم تكن هناك رافعات أو آلات حديثة في ذلك الوقت. " | |
| "لذلك، تعاون آلاف العمال المصريين بذكاء وصبر شديد. " | |
| "قاموا بقطع الحجارة الضخمة من الجبال، ونقلوها عبر نهر النيل، ثم رفعوها فوق بعضها بدقة عالية باستخدام ممرات طينية مائلة." | |
| ) | |
| print("بدء تشغيل نظام استخراج المعرفة الهجين (Hybrid Knowledge Graph)...\n") | |
| print(f"النص المدخل:\n{sample_text}\n") | |
| print("-" * 60) | |
| print("جاري التحليل النحوي واستخراج الكيانات، والتواصل مع خادم العلاقات (mT5 API)...") | |
| try: | |
| # 3. استدعاء محرك الأنابيب الشامل الذي صممناه | |
| extracted_relations = process_full_document_pipeline( | |
| full_text=sample_text, | |
| camel_parser_path=CAMEL_PARSER_PATH | |
| ) | |
| # 4. طباعة النتائج النهائية بطريقة رسومية واضحة | |
| print("\n--- النتائج النهائية للغراف المعرفي (Knowledge Graph) ---") | |
| if not extracted_relations: | |
| print("تنبيه: لم يتم استخراج أي علاقات. (تأكد من دقة الكيانات المستخرجة أو عتبة المسافة النحوية).") | |
| else: | |
| for i, (e1, relation, e2) in enumerate(extracted_relations, 1): | |
| # طباعة العلاقة بشكل مرئي: [الكيان 1] --(العلاقة)--> [الكيان 2] | |
| print(f"{i}. [{e1}] --({relation})--> [{e2}]") | |
| # 5. صيغة الرسم البياني (nodes/edges JSON) لتسليمها لواجهة العرض | |
| graph = build_knowledge_graph_json(extracted_relations) | |
| print("\n--- صيغة الرسم البياني (nodes/edges) ---") | |
| 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}") |