Create app.py

app.py

@@ -0,0 +1,321 @@
+import re
+import difflib
+import numpy as np
+import torch
+import gradio as gr
+import pyarabic.araby as araby
+
+import stanza
+from transformers import AutoTokenizer, AutoModel
+from transformers import AutoTokenizer as HFTokenizer, AutoModelForSeq2SeqLM
+from sentence_transformers import SentenceTransformer, util
+import arabert.preprocess
+import yake
+from bert_score import score as bertscore
+
+DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
+torch.set_grad_enabled(False)
+
+# ---- نماذج وأدوات ----
+ARAELECTRA_NAME = "aubmindlab/araelectra-base-discriminator"
+SBERT_MODEL      = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
+QG_MODEL         = "Mihakram/AraT5-base-question-generation"
+
+# Stanza (أول تشغيل قد يحمّل حزمة العربية ويكاشها)
+stanza.download("ar", verbose=False)
+nlp = stanza.Pipeline(lang="ar", processors="tokenize,pos,lemma,depparse", tokenize_no_ssplit=False, verbose=False)
+
+# Arabert preprocessor
+arabert_prep = arabert.preprocess.ArabertPreprocessor(ARAELECTRA_NAME)
+
+# AraELECTRA (للأوفست والتمثيلات السياقية)
+tokenizer_electra = AutoTokenizer.from_pretrained(ARAELECTRA_NAME)
+model_electra     = AutoModel.from_pretrained(ARAELECTRA_NAME).to(DEVICE)
+
+# sBERT
+sbert = SentenceTransformer(SBERT_MODEL, device=DEVICE)
+
+# AraT5 (توليد سؤال)
+qg_tokenizer = HFTokenizer.from_pretrained(QG_MODEL)
+qg_model     = AutoModelForSeq2SeqLM.from_pretrained(QG_MODEL).to(DEVICE)
+
+# ---- أدوات مساعدة ----
+def normalize(s: str) -> str:
+    t = araby.strip_tashkeel(s)
+    t = t.replace("آ","ا").replace("أ","ا").replace("إ","ا").replace("ى","ي")
+    t = t.replace("ـ","")
+    t = " ".join(t.split())
+    return t
+
+def build_char_map(src: str, tgt: str):
+    sm = difflib.SequenceMatcher(a=src, b=tgt)
+    src2tgt = [-1] * len(src)
+    for tag, i1, i2, j1, j2 in sm.get_opcodes():
+        if tag == "equal":
+            for k in range(i2 - i1):
+                src2tgt[i1 + k] = j1 + k
+        elif tag in ("replace", "delete"):
+            for k in range(i2 - i1):
+                src2tgt[i1 + k] = j1
+    last = 0
+    for i in range(len(src2tgt)):
+        if src2tgt[i] == -1:
+            src2tgt[i] = last
+        else:
+            last = src2tgt[i]
+    return src2tgt
+
+def map_span_src_to_tgt(src2tgt, start, end, tgt_len):
+    if start >= len(src2tgt): start = max(0, len(src2tgt)-1)
+    if end == 0: end = 1
+    if end-1 >= len(src2tgt): end = len(src2tgt)
+    ts = src2tgt[start]; te = src2tgt[end-1] + 1
+    ts = max(0, min(ts, max(0, tgt_len-1)))
+    te = max(ts+1, min(te, tgt_len))
+    return ts, te
+
+def token_indices_overlapping_span(offsets, span_start, span_end):
+    idxs = []
+    for i, (s, e) in enumerate(offsets):
+        if e > span_start and s < span_end:
+            idxs.append(i)
+    return idxs
+
+def electra_hidden_states(prep_text):
+    encoded = tokenizer_electra(prep_text, return_tensors="pt", return_offsets_mapping=True, padding=False, truncation=True).to(DEVICE)
+    offsets = encoded.pop("offset_mapping")[0].tolist()
+    with torch.no_grad():
+        out = model_electra(**encoded)
+    H = out.last_hidden_state.squeeze(0)
+    return offsets, H
+
+def electra_phrase_vec_via_offsets(span_start, span_end, src2tgt, prep_text, offsets, H):
+    ts, te = map_span_src_to_tgt(src2tgt, span_start, span_end, len(prep_text))
+    tok_ids = token_indices_overlapping_span(offsets, ts, te)
+    if not tok_ids:
+        return None
+    vecs = [H[i] for i in tok_ids]
+    return torch.stack(vecs, dim=0).mean(dim=0)
+
+# استخراج عبارات اسمية
+def build_noun_phrases(doc, text_norm):
+    noun_phrases = []
+    for si, sent in enumerate(doc.sentences):
+        words_info = []
+        for ti, tok in enumerate(sent.tokens):
+            for w in tok.words:
+                words_info.append({
+                    "id": w.id, "text": w.text, "upos": w.upos, "deprel": w.deprel,
+                    "head": w.head, "start": tok.start_char, "end": tok.end_char, "tok_idx": ti
+                })
+        for wi in words_info:
+            if wi["upos"] not in {"NOUN","PROPN"}:  # رؤوس اسمية
+                continue
+            head = wi
+            left_mods, right_mods = [], []
+            for cj in words_info:
+                if cj["head"] == head["id"] and cj["deprel"] in {"amod","compound","nmod"}:
+                    (left_mods if cj["start"] <= head["start"] else right_mods).append(cj)
+            left_mods  = sorted(left_mods,  key=lambda x: x["start"])
+            right_mods = sorted(right_mods, key=lambda x: x["start"])
+            phrase_tokens = left_mods + [head] + right_mods
+            if len(phrase_tokens) < 2 and head["upos"] != "PROPN":  # استثناء الأعلام المفردة
+                continue
+            span_start = min(t["start"] for t in phrase_tokens); span_end = max(t["end"] for t in phrase_tokens)
+            phrase_text = re.sub(r"\s+", " ", text_norm[span_start:span_end].strip())
+            if len(phrase_text) >= 2:
+                noun_phrases.append({"text": phrase_text, "start": span_start, "end": span_end})
+    # تمييز
+    uniq = {}
+    for np_item in noun_phrases:
+        key = np_item["text"]
+        if key not in uniq or (np_item["end"]-np_item["start"]) > (uniq[key]["end"]-uniq[key]["start"]):
+            uniq[key] = np_item
+    return list(uniq.values())
+
+# الترتيب: sBERT + ELECTRA + MMR
+def mmr_select(doc_emb, cand_embs, candidates, k=10, lam=0.7):
+    if not candidates: return []
+    chosen, rest = [], list(range(len(candidates)))
+    sim_doc = util.cos_sim(doc_emb, cand_embs)[0].cpu().numpy()
+    first = int(np.argmax(sim_doc)); chosen.append(first); rest.remove(first)
+    sim_between = util.cos_sim(cand_embs, cand_embs).cpu().numpy()
+    while len(chosen) < min(k, len(candidates)) and rest:
+        best_i, best_score = None, -1e9
+        for i in rest:
+            redundancy = max(sim_between[i, j] for j in chosen) if chosen else 0.0
+            score = 0.7*sim_doc[i] - 0.3*redundancy
+            if score > best_score: best_score, best_i = score, i
+        chosen.append(best_i); rest.remove(best_i)
+    return [candidates[i] for i in chosen]
+
+def rank_keyphrases(text_norm, nps, alpha=0.8):
+    phrases = [p["text"] for p in nps]
+    if not phrases: return [], []
+    text_prep = arabert_prep.preprocess(text_norm)
+    src2tgt = build_char_map(text_norm, text_prep)
+    # sBERT
+    doc_emb = sbert.encode([text_prep], convert_to_tensor=True)
+    phr_embs = sbert.encode(phrases, convert_to_tensor=True)
+    sims_sbert = util.cos_sim(doc_emb, phr_embs).cpu().numpy()[0]
+    # ELECTRA
+    offsets, H = electra_hidden_states(text_prep)
+    doc_vec_electra = H.mean(dim=0)
+    sims_electra = []
+    for p in nps:
+        v = electra_phrase_vec_via_offsets(p["start"], p["end"], src2tgt, text_prep, offsets, H)
+        if v is None: sims_electra.append(0.0)
+        else:
+            num = torch.dot(doc_vec_electra, v).item()
+            den = float(doc_vec_electra.norm().item() * v.norm().item() + 1e-9)
+            sims_electra.append(num/den)
+    sims_electra = np.array(sims_electra)
+    blended = alpha*sims_sbert + (1-alpha)*sims_electra
+    order = np.argsort(-blended)
+    ranked = [(phrases[i], float(blended[i]), float(sims_sbert[i]), float(sims_electra[i])) for i in order]
+    diverse = mmr_select(doc_emb, phr_embs, phrases, k=min(12, len(phrases)), lam=0.7)
+    return ranked, diverse
+
+# YAKE
+def yake_scores_for_phrases(text_norm, phrases, max_ngram_size=5, lan="ar"):
+    kw_extractor = yake.KeywordExtractor(lan=lan, n=max_ngram_size, dedupLim=0.9, top=1000)
+    scored = kw_extractor.extract_keywords(text_norm)
+    norm = lambda s: re.sub(r"\s+"," ", s).strip().lower()
+    scored_norm = {norm(k): v for k, v in scored}
+    res = {}
+    for p in phrases:
+        res[p] = scored_norm.get(norm(p))
+    return res
+
+def invert_and_minmax_yake(score_map):
+    vals = [None if v is None else 1/(1+v) for v in score_map.values()]
+    finite = [x for x in vals if x is not None]
+    if not finite: return {k:0.0 for k in score_map.keys()}
+    vmin, vmax = min(finite), max(finite); rng = (vmax-vmin) if vmax>vmin else 1.0
+    out = {}
+    for (k,_), pos in zip(score_map.items(), vals):
+        out[k] = 0.0 if pos is None else (pos - vmin)/rng
+    return out
+
+def blend_semantic_with_yake(ranked_sem, yake_norm, w_sem=0.7, w_yake=0.3):
+    merged = []
+    for phr, sem_sc, sb, el in ranked_sem:
+        y = yake_norm.get(phr, 0.0)
+        final = w_sem*sem_sc + w_yake*y
+        merged.append((phr, final, sem_sc, y, sb, el))
+    merged.sort(key=lambda x: -x[1])
+    return merged
+
+# تقسيم بالنقطة + اختيار جملة داعمة لكل عبارة
+def split_by_dots(text: str):
+    parts = re.split(r"\.{1,}\s*", text)
+    return [p.strip() for p in parts if p.strip()]
+
+def sentence_kind_from_root(stanza_sentence):
+    root = next((w for w in stanza_sentence.words if w.deprel == "root"), None)
+    if not root: return "unknown"
+    return "verbal" if root.upos == "VERB" else "nominal"
+
+def split_and_tag_nominal_verbal_by_dots(text_norm):
+    sents = split_by_dots(text_norm)
+    tagged = []
+    for s in sents:
+        doc_s = nlp(s)
+        if not doc_s.sentences:
+            tagged.append({"text": s, "kind": "unknown"})
+        else:
+            tagged.append({"text": s, "kind": sentence_kind_from_root(doc_s.sentences[0])})
+    return tagged
+
+def best_support_sentence_by_dots(text_norm, phrase):
+    sentences_tagged = split_and_tag_nominal_verbal_by_dots(text_norm)
+    if not sentences_tagged: return ""
+    sent_texts = [m["text"] for m in sentences_tagged]
+    sent_embs  = sbert.encode(sent_texts, convert_to_tensor=True)
+    p_emb      = sbert.encode([phrase], convert_to_tensor=True)
+    sims = util.cos_sim(p_emb, sent_embs)[0].cpu().numpy()
+    best_idx = int(np.argmax(sims))
+    return sent_texts[best_idx], sentences_tagged[best_idx]["kind"]
+
+# توليد سؤال موحّد (بدون hints)
+def gen_unified_question_freeform(phrases, supports, context_text, max_len=96, num_beams=5):
+    context_short = context_text.strip()[:600]
+    items_block = "\n".join([f"- العبارة: {p}\n  جملة داعمة: {s}" for p, s in zip(phrases, supports)])
+    prompt = (
+        "حوّل العبارات التالية إلى سؤال واحد شامل بالعربية يعتمد على السياق. "
+        "يجب أن يغطي جميع العبارات بشكل موجز وواضح.\n"
+        f"{items_block}\n"
+        f"سياق: {context_short}\n"
+        "السؤال الموحد:"
+    )
+    inputs = qg_tokenizer(prompt, return_tensors="pt", truncation=True).to(DEVICE)
+    outputs = qg_model.generate(
+        **inputs, max_length=max_len, num_beams=num_beams,
+        early_stopping=True, no_repeat_ngram_size=3
+    )
+    q = qg_tokenizer.decode(outputs[0], skip_special_tokens=True).strip()
+    q = q.rstrip("?.؟")
+    if q and not q.endswith("؟"): q += "؟"
+    return q
+
+# الواجهة: خطوة واحدة تنفّذ كل شيء وتعرض النتائج
+def run_pipeline(user_text):
+    if not user_text or len(user_text.strip()) < 5:
+        return "رجاءً أدخل نصًا عربيًا أطول.", "", "", "", ""
+
+    text_norm = normalize(user_text)
+    doc = nlp(text_norm)
+
+    # 1) عبارات اسمية
+    nps = build_noun_phrases(doc, text_norm)
+    if not nps:
+        return "لم تُستخرج عبارات اسمية.", "", "", "", ""
+
+    # 2) ترتيب دلالي
+    ranked_sem, diverse = rank_keyphrases(text_norm, nps, alpha=0.8)
+
+    # 3) YAKE + دمج
+    phrases = [r[0] for r in ranked_sem]
+    yake_raw  = yake_scores_for_phrases(text_norm, phrases, max_ngram_size=5, lan="ar")
+    yake_norm = invert_and_minmax_yake(yake_raw)
+    ranked_blended = blend_semantic_with_yake(ranked_sem, yake_norm, w_sem=0.7, w_yake=0.3)
+
+    # 4) أفضل جملة داعمة لأول 5 عبارات
+    top_n = min(5, len(ranked_blended))
+    top_phrases = [ranked_blended[i][0] for i in range(top_n)]
+    supports = []
+    kinds    = []
+    for p in top_phrases:
+        s, kind = best_support_sentence_by_dots(text_norm, p)
+        supports.append(s); kinds.append(kind)
+
+    # 5) سؤال موحّد من الخمس عبارات
+    unified_q = gen_unified_question_freeform(top_phrases, supports, text_norm)
+
+    # إخراج منسق
+    nps_str = "\n".join(f"- {p['text']}" for p in nps[:20])
+    ranked_str = "\n".join(f"{i+1:>2}. {t[0]} (score={t[1]:.3f})" for i, t in enumerate(ranked_blended[:15]))
+    support_str = "\n".join(f"{i+1:>2}. [{kinds[i]}] {top_phrases[i]}  →  {supports[i]}" for i in range(top_n))
+    diverse_str = "\n".join(f"- {d}" for d in diverse[:10])
+
+    return unified_q, ranked_str, support_str, diverse_str, nps_str
+
+title = "Arabic Main Question Generation (Hybrid Pipeline)"
+desc = "أدخل نصًا عربيًا؛ سنستخرج العبارات الاسمية، نرتّبها (sBERT + ELECTRA + YAKE + MMR)، نختار جملًا داعمة، ونولّد سؤالًا موحّدًا بـ AraT5."
+
+with gr.Blocks(title=title) as demo:
+    gr.Markdown(f"# {title}\n{desc}")
+
+    with gr.Row():
+        inp = gr.Textbox(lines=12, label="النص العربي")
+    btn = gr.Button("تشغيل الـPipeline")
+
+    out_unified = gr.Textbox(label="السؤال الموحد (AraT5)")
+    out_ranked  = gr.Textbox(label="Top Noun Phrases (Blended Ranking)")
+    out_support = gr.Textbox(label="أفضل الجمل الداعمة لأول 5 عبارات")
+    out_diverse = gr.Textbox(label="MMR Diverse Selection")
+    out_nps     = gr.Textbox(label="العبارات الاسمية المستخرجة (أول 20)")
+
+    btn.click(run_pipeline, inputs=inp, outputs=[out_unified, out_ranked, out_support, out_diverse, out_nps])
+
+demo.launch()