""" 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(), # annotator_state gr.update(), # current_idx gr.update(), # order_state "**Введіть ваше ім'я для початку.**", "", 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()