| """ |
| DefectRadar Annotation Tool |
| Human annotation of definitional defects in Ukrainian legislation. |
| Part of the LegDefQA benchmark construction. |
| """ |
|
|
| import os |
| import json |
| import uuid |
| import datetime |
| import gradio as gr |
| from pathlib import Path |
| from huggingface_hub import HfApi, hf_hub_download, upload_file |
|
|
| HF_TOKEN = os.environ.get("HF_TOKEN") |
| DATASET_REPO = "overthelex/defectradar-annotations" |
| SAMPLE_REPO = "overthelex/defectradar-annotator" |
| SAMPLE_FILE = "annotation_sample.jsonl" |
|
|
| api = HfApi(token=HF_TOKEN) |
|
|
|
|
| def ensure_dataset_repo(): |
| try: |
| api.repo_info(repo_id=DATASET_REPO, repo_type="dataset") |
| except Exception: |
| api.create_repo( |
| repo_id=DATASET_REPO, |
| repo_type="dataset", |
| private=False, |
| ) |
|
|
|
|
| def load_sample(): |
| path = hf_hub_download( |
| repo_id=SAMPLE_REPO, |
| filename=SAMPLE_FILE, |
| repo_type="space", |
| token=HF_TOKEN, |
| ) |
| items = [] |
| with open(path, encoding="utf-8") as f: |
| for line in f: |
| if line.strip(): |
| items.append(json.loads(line)) |
| return items |
|
|
|
|
| def load_existing_annotations(): |
| try: |
| path = hf_hub_download( |
| repo_id=DATASET_REPO, |
| filename="annotations.jsonl", |
| repo_type="dataset", |
| token=HF_TOKEN, |
| ) |
| seen = set() |
| with open(path, encoding="utf-8") as f: |
| for line in f: |
| if line.strip(): |
| row = json.loads(line) |
| key = (row.get("annotator_id", ""), row.get("definition_id")) |
| seen.add(key) |
| return seen |
| except Exception: |
| return set() |
|
|
|
|
| def save_annotation(record: dict): |
| ensure_dataset_repo() |
| line = json.dumps(record, ensure_ascii=False) + "\n" |
| ts = datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%d_%H%M%S") |
| fname = f"raw/{record['annotator_id']}_{record['definition_id']}_{ts}.jsonl" |
| tmp = Path(f"/tmp/{uuid.uuid4().hex}.jsonl") |
| tmp.write_text(line, encoding="utf-8") |
| upload_file( |
| path_or_fileobj=str(tmp), |
| path_in_repo=fname, |
| repo_id=DATASET_REPO, |
| repo_type="dataset", |
| token=HF_TOKEN, |
| ) |
| tmp.unlink(missing_ok=True) |
|
|
|
|
| SAMPLE = load_sample() |
|
|
|
|
| def get_annotator_id(request: gr.Request = None): |
| return str(uuid.uuid4())[:8] |
|
|
|
|
| def format_definition(item): |
| law = item.get("law_title", "") |
| rada = item.get("rada_id", "") |
| definiendum = item.get("definiendum", "") |
| genus = item.get("genus_proximum", "") |
| diff = item.get("differentia_specifica", "") |
| full = item.get("full_definiens", "") |
|
|
| if genus and diff: |
| definiens_display = f"**{genus}**, {diff}" |
| else: |
| definiens_display = full |
|
|
| pipeline_flags = [] |
| if item.get("pipeline_circulus"): |
| lemmas = ", ".join(item.get("overlapping_lemmas", [])) |
| pipeline_flags.append(f"circulus (леми: {lemmas})") |
| if item.get("pipeline_ignotum"): |
| terms = ", ".join(item.get("undefined_terms", [])) |
| pipeline_flags.append(f"ignotum (терміни: {terms})") |
| if not pipeline_flags: |
| pipeline_flags.append("дефектів не виявлено") |
|
|
| flags_str = "; ".join(pipeline_flags) |
|
|
| md = f"""### Закон: {law} |
| **Рада ID:** `{rada}` |
| |
| --- |
| |
| **Термін (definiendum):** {definiendum} |
| |
| **Визначення (definiens):** {definiens_display} |
| |
| --- |
| |
| **Pipeline:** {flags_str} |
| """ |
| return md |
|
|
|
|
| CIRCULUS_CHOICES = [ |
| "TRUE_DEFECT -- визначення нормативно порожнє, не додає розрізнювального критерію", |
| "DOMAIN_REUSE -- корінь повторюється, але визначення змістовне", |
| "FALSE_POSITIVE -- pipeline помилився, повтору немає", |
| "НЕ ВПЕВНЕНИЙ -- потребує додаткового аналізу", |
| ] |
|
|
| IGNOTUM_CHOICES = [ |
| "CRITICAL -- термін не визначений ніде, норма не може бути застосована", |
| "CROSS_REF -- термін визначений в іншому законі, потрібне посилання", |
| "COMMON -- термін загальновживаний, визначення не потрібне", |
| "FALSE_POSITIVE -- pipeline помилився, термін визначений у цьому ж законі", |
| "НЕ ВПЕВНЕНИЙ -- потребує додаткового аналізу", |
| ] |
|
|
|
|
| def build_app(): |
|
|
| with gr.Blocks( |
| title="DefectRadar -- Анотація дефектів визначень", |
| theme=gr.themes.Soft(), |
| css=""" |
| .definition-box { border-left: 4px solid #6366f1; |
| padding: 16px; border-radius: 4px; margin: 8px 0; } |
| .definition-box * { color: var(--body-text-color) !important; } |
| .progress-text { font-size: 1.1em; font-weight: 600; } |
| """, |
| ) as demo: |
|
|
| annotator_state = gr.State("") |
| current_idx = gr.State(0) |
| order_state = gr.State([]) |
|
|
| gr.Markdown(""" |
| # DefectRadar -- Анотація дефектів визначень у законодавстві |
| |
| Інструмент для ручної анотації дефектів у визначеннях українського законодавства. |
| Результати використовуються для побудови gold-standard бенчмарку **LegDefQA**. |
| |
| ### Інструкція |
| 1. Введіть ваше ім'я або ідентифікатор |
| 2. Прочитайте визначення та оцінки pipeline |
| 3. Дайте вашу експертну оцінку для кожного типу дефекту |
| 4. Натисніть «Зберегти та далі» |
| |
| **Критерій для circulus:** Чи додає визначення хоча б один розрізнювальний критерій, якого немає в самому терміні? |
| |
| **Критерій для ignotum:** Чи може юрист застосувати цю норму без звернення до інших законів? |
| """) |
|
|
| with gr.Row(): |
| annotator_input = gr.Textbox( |
| label="Ваше ім'я / ідентифікатор", |
| placeholder="напр. Іван Петренко", |
| scale=3, |
| ) |
| start_btn = gr.Button("Почати анотацію", variant="primary", scale=1) |
|
|
| progress_md = gr.Markdown("", elem_classes=["progress-text"]) |
|
|
| definition_md = gr.Markdown("", elem_classes=["definition-box"]) |
|
|
| with gr.Group(visible=False) as annotation_group: |
| gr.Markdown("### Ваша оцінка") |
|
|
| with gr.Row(): |
| with gr.Column(): |
| circulus_radio = gr.Radio( |
| choices=CIRCULUS_CHOICES, |
| label="Circulus in definiendo", |
| info="Чи є визначення тавтологічним?", |
| ) |
| circulus_comment = gr.Textbox( |
| label="Коментар (circulus)", |
| placeholder="Необов'язково", |
| lines=2, |
| ) |
|
|
| with gr.Column(): |
| ignotum_radio = gr.Radio( |
| choices=IGNOTUM_CHOICES, |
| label="Ignotum per ignotum", |
| info="Чи використовує визначення невизначені терміни?", |
| ) |
| ignotum_comment = gr.Textbox( |
| label="Коментар (ignotum)", |
| placeholder="Необов'язково", |
| lines=2, |
| ) |
|
|
| with gr.Row(): |
| skip_btn = gr.Button("Пропустити", variant="secondary") |
| save_btn = gr.Button("Зберегти та далі", variant="primary") |
|
|
| done_md = gr.Markdown("", visible=False) |
|
|
| def start_annotation(name): |
| if not name.strip(): |
| return ( |
| gr.update(), |
| gr.update(), |
| gr.update(), |
| "**Введіть ваше ім'я для початку.**", |
| "", |
| gr.update(visible=False), |
| gr.update(visible=False), |
| ) |
|
|
| annotator_id = name.strip().replace(" ", "_").lower() |
| done = load_existing_annotations() |
|
|
| order = [] |
| for i, item in enumerate(SAMPLE): |
| key = (annotator_id, item["id"]) |
| if key not in done: |
| order.append(i) |
|
|
| if not order: |
| return ( |
| annotator_id, |
| 0, |
| order, |
| f"**Ви вже проанотували всі {len(SAMPLE)} визначень!**", |
| "", |
| gr.update(visible=False), |
| gr.update(visible=False), |
| ) |
|
|
| item = SAMPLE[order[0]] |
| remaining = len(order) |
| total = len(SAMPLE) |
| annotated = total - remaining |
|
|
| return ( |
| annotator_id, |
| 0, |
| order, |
| f"**Прогрес:** {annotated}/{total} (залишилось {remaining})", |
| format_definition(item), |
| gr.update(visible=True), |
| gr.update(visible=False), |
| ) |
|
|
| start_btn.click( |
| fn=start_annotation, |
| inputs=[annotator_input], |
| outputs=[ |
| annotator_state, |
| current_idx, |
| order_state, |
| progress_md, |
| definition_md, |
| annotation_group, |
| done_md, |
| ], |
| ) |
|
|
| def submit_annotation( |
| annotator_id, idx, order, circ_val, circ_comment, igno_val, igno_comment |
| ): |
| if not order or idx >= len(order): |
| return ( |
| idx, |
| "**Готово! Дякуємо за анотацію.**", |
| "", |
| gr.update(visible=False), |
| gr.update(visible=True, value="## Дякуємо! Всі визначення проанотовано."), |
| None, |
| "", |
| None, |
| "", |
| ) |
|
|
| item = SAMPLE[order[idx]] |
|
|
| circ_label = circ_val.split(" -- ")[0] if circ_val else "SKIPPED" |
| igno_label = igno_val.split(" -- ")[0] if igno_val else "SKIPPED" |
|
|
| record = { |
| "annotator_id": annotator_id, |
| "definition_id": item["id"], |
| "rada_id": item["rada_id"], |
| "definiendum": item["definiendum"], |
| "pipeline_circulus": item.get("pipeline_circulus", False), |
| "pipeline_ignotum": item.get("pipeline_ignotum", False), |
| "human_circulus": circ_label, |
| "human_circulus_comment": circ_comment or "", |
| "human_ignotum": igno_label, |
| "human_ignotum_comment": igno_comment or "", |
| "timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(), |
| "stratum": item.get("stratum", ""), |
| } |
|
|
| try: |
| save_annotation(record) |
| except Exception as e: |
| return ( |
| idx, |
| f"**Помилка збереження:** {e}", |
| format_definition(item), |
| gr.update(visible=True), |
| gr.update(visible=False), |
| gr.update(), |
| gr.update(), |
| gr.update(), |
| gr.update(), |
| ) |
|
|
| next_idx = idx + 1 |
| if next_idx >= len(order): |
| total = len(SAMPLE) |
| return ( |
| next_idx, |
| f"**Прогрес:** {total}/{total}", |
| "", |
| gr.update(visible=False), |
| gr.update( |
| visible=True, |
| value="## Дякуємо! Всі визначення проанотовано.\n\nВаші відповіді збережено у датасет на HuggingFace.", |
| ), |
| None, |
| "", |
| None, |
| "", |
| ) |
|
|
| next_item = SAMPLE[order[next_idx]] |
| remaining = len(order) - next_idx |
| total = len(SAMPLE) |
| annotated = total - remaining |
|
|
| return ( |
| next_idx, |
| f"**Прогрес:** {annotated}/{total} (залишилось {remaining})", |
| format_definition(next_item), |
| gr.update(visible=True), |
| gr.update(visible=False), |
| None, |
| "", |
| None, |
| "", |
| ) |
|
|
| save_btn.click( |
| fn=submit_annotation, |
| inputs=[ |
| annotator_state, |
| current_idx, |
| order_state, |
| circulus_radio, |
| circulus_comment, |
| ignotum_radio, |
| ignotum_comment, |
| ], |
| outputs=[ |
| current_idx, |
| progress_md, |
| definition_md, |
| annotation_group, |
| done_md, |
| circulus_radio, |
| circulus_comment, |
| ignotum_radio, |
| ignotum_comment, |
| ], |
| ) |
|
|
| skip_btn.click( |
| fn=lambda aid, idx, order: submit_annotation( |
| aid, idx, order, "SKIPPED -- пропущено", "", "SKIPPED -- пропущено", "" |
| ), |
| inputs=[annotator_state, current_idx, order_state], |
| outputs=[ |
| current_idx, |
| progress_md, |
| definition_md, |
| annotation_group, |
| done_md, |
| circulus_radio, |
| circulus_comment, |
| ignotum_radio, |
| ignotum_comment, |
| ], |
| ) |
|
|
| return demo |
|
|
|
|
| demo = build_app() |
| demo.launch() |
|
|
