testiki-vnmu/_parse_docx.py at main · reddev111/testiki-vnmu · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
#!/usr/bin/env python3
"""
Parse the local DOCX into structured gynecology/obstetrics JSON.

Why this parser exists:
- the source document contains merged answer paragraphs
- some paragraphs contain both the tail of one answer and the next numbered question
- correct answers are marked either by leading stars or text highlighting
"""

from __future__ import annotations

import json
import re
from dataclasses import dataclass
from pathlib import Path

from docx import Document
from docx.oxml.ns import qn


ROOT = Path(__file__).parent
DOCX = max(
    (path for path in ROOT.glob("*.docx") if not path.name.startswith("~$")),
    key=lambda path: path.stat().st_mtime,
)
OUT_GYN = ROOT / "questions" / "gynecology" / "gynecology.json"
OUT_OBS = ROOT / "questions" / "gynecology" / "obstetrics.json"
OUT_CHECK = ROOT / "_result_check.txt"


LETTER_CHARS = "A-JАВСДЕЄЖІЇҐ"
CYR_TO_LAT = str.maketrans({
    "А": "A",
    "В": "B",
    "С": "C",
    "Д": "D",
    "Е": "E",
    "Ж": "J",
})

ANSWER_SEG_RE = re.compile(
    rf"^\s*(?P<stars>\*{{0,3}})\s*(?P<letter>[{LETTER_CHARS}])\.\s*(?P<body>.*)$",
    re.DOTALL | re.IGNORECASE,
)
QUESTION_SEG_RE = re.compile(r"^\s*(?P<num>\d{1,3})\.\s*(?P<body>.*)$", re.DOTALL)

START_QUESTION_RE = re.compile(r"^(?P<num>\d{1,3})\.\s*(?=\S)")
START_ANSWER_RE = re.compile(
    rf"^\s*(?P<stars>\*{{0,3}})\s*(?P<letter>[{LETTER_CHARS}])\.\s*(?=\S)",
    re.IGNORECASE,
)
INNER_QUESTION_RE = re.compile(r"(?<=\s)(?P<num>\d{1,3})\.\s+(?=\S)")
INNER_ANSWER_RE = re.compile(
    rf"(?<=\s)(?P<stars>\*{{0,3}})\s*(?P<letter>[{LETTER_CHARS}])\.\s+(?=\S)",
    re.IGNORECASE,
)


@dataclass
class ParagraphChunk:
    kind: str
    text: str
    highlighted: bool


def run_is_highlighted(run) -> bool:
    rpr = run._r.find(qn("w:rPr"))
    if rpr is None:
        return False
    hl = rpr.find(qn("w:highlight"))
    return hl is not None and hl.get(qn("w:val")) in ("yellow", "cyan")


def paragraph_text_and_mask(paragraph) -> tuple[str, list[bool]]:
    pieces: list[str] = []
    mask: list[bool] = []

    for run in paragraph.runs:
        text = run.text
        if not text:
            continue
        pieces.append(text)
        mask.extend([run_is_highlighted(run)] * len(text))

    if not pieces:
        text = paragraph.text
        return text.strip(), [False] * len(text.strip())

    text = "".join(pieces)
    start = 0
    end = len(text)
    while start < end and text[start].isspace():
        start += 1
    while end > start and text[end - 1].isspace():
        end -= 1

    return text[start:end], mask[start:end]


def normalize_text(text: str) -> str:
    text = text.replace("\r", "")
    text = text.replace("\xa0", " ")
    text = re.sub(r"\s+", " ", text).strip()
    return text


def tokenize_paragraph(text: str, mask: list[bool]) -> list[ParagraphChunk]:
    if not text:
        return []

    starts: list[tuple[int, str]] = []

    if START_QUESTION_RE.match(text):
        starts.append((0, "question"))
    if START_ANSWER_RE.match(text):
        starts.append((0, "answer"))

    starts.extend((m.start(), "question") for m in INNER_QUESTION_RE.finditer(text))
    starts.extend((m.start(), "answer") for m in INNER_ANSWER_RE.finditer(text))
    starts.sort(key=lambda item: (item[0], 0 if item[1] == "question" else 1))

    deduped: list[tuple[int, str]] = []
    seen_positions: set[int] = set()
    for pos, kind in starts:
        if pos in seen_positions:
            continue
        deduped.append((pos, kind))
        seen_positions.add(pos)

    if not deduped:
        return [ParagraphChunk("text", normalize_text(text), any(mask))]

    chunks: list[ParagraphChunk] = []

    if deduped[0][0] > 0:
        prefix = normalize_text(text[: deduped[0][0]])
        if prefix:
            chunks.append(ParagraphChunk("text", prefix, any(mask[: deduped[0][0]])))

    for idx, (start, kind) in enumerate(deduped):
        end = deduped[idx + 1][0] if idx + 1 < len(deduped) else len(text)
        seg_text = normalize_text(text[start:end])
        if not seg_text:
            continue
        seg_mask = mask[start:end] if end <= len(mask) else mask[start:]
        chunks.append(ParagraphChunk(kind, seg_text, any(seg_mask)))

    return chunks


def normalize_letter(letter: str) -> str:
    return letter.upper().translate(CYR_TO_LAT)


def parse_answer(segment: ParagraphChunk) -> tuple[str, str, bool] | None:
    match = ANSWER_SEG_RE.match(segment.text)
    if not match:
        return None

    stars = match.group("stars") or ""
    letter = normalize_letter(match.group("letter"))
    body = match.group("body").strip()
    inline_star = body.startswith("***")
    body = re.sub(r"^\*{1,3}\s*", "", body).strip()
    is_correct = bool(stars) or inline_star or segment.highlighted
    return letter, body, is_correct


def parse_question(segment: ParagraphChunk) -> tuple[int | None, str]:
    match = QUESTION_SEG_RE.match(segment.text)
    if not match:
        return None, segment.text
    number = int(match.group("num"))
    body = match.group("body").strip()
    return number, body


def build_chunks(paragraphs) -> list[ParagraphChunk]:
    chunks: list[ParagraphChunk] = []
    for paragraph in paragraphs:
        text, mask = paragraph_text_and_mask(paragraph)
        if not text:
            continue
        chunks.extend(tokenize_paragraph(text, mask))
    return chunks


def is_comment_text(text: str) -> bool:
    stripped = text.strip()
    return stripped.startswith("(") and stripped.endswith(")")


def looks_like_question(text: str) -> bool:
    stripped = text.strip()
    if len(stripped) < 12:
        return False
    return stripped.endswith("?") or stripped.endswith(":")


def parse_section(chunks: list[ParagraphChunk], numbered_only: bool) -> list[dict]:
    questions: list[dict] = []
    seen_questions: set[str] = set()
    current_question: str | None = None
    answers: list[tuple[str, str, bool]] = []

    def flush() -> None:
        nonlocal current_question, answers
        if current_question and len(answers) >= 2:
            question_text = normalize_text(current_question)
            correct_positions = [idx for idx, answer in enumerate(answers) if answer[2]]
            if correct_positions and question_text not in seen_questions:
                correct = correct_positions[0] if len(correct_positions) == 1 else correct_positions
                questions.append({
                    "question": question_text,
                    "answers": [answer[1] for answer in answers],
                    "correct": correct,
                })
                seen_questions.add(question_text)
        current_question = None
        answers = []

    for chunk in chunks:
        if chunk.kind == "question":
            number, body = parse_question(chunk)
            if number is None or not body:
                continue
            flush()
            current_question = body
            continue

        if chunk.kind == "answer":
            parsed = parse_answer(chunk)
            if not parsed:
                continue
            if current_question is None:
                continue
            answers.append(parsed)
            continue

        if is_comment_text(chunk.text):
            continue

        if current_question is None:
            if numbered_only and not looks_like_question(chunk.text):
                continue
            current_question = chunk.text
            continue

        if not answers:
            current_question = normalize_text(f"{current_question} {chunk.text}")
            continue

        if numbered_only:
            if looks_like_question(chunk.text):
                flush()
                current_question = chunk.text
            continue

        flush()
        current_question = chunk.text

    flush()
    return questions


def split_sections(document: Document) -> tuple[list, list]:
    nonempty = [paragraph for paragraph in document.paragraphs if paragraph.text.strip()]
    split_index = None

    for index, paragraph in enumerate(nonempty):
        text = paragraph.text.strip()
        if index > 50 and re.match(r"^1\.\s+.+", text):
            split_index = index
            break

    if split_index is None:
        return nonempty, []
    return nonempty[:split_index], nonempty[split_index:]


def write_check_file(label: str, data: list[dict], handle) -> None:
    handle.write(f"=== {label} ===\n\n")
    for idx, item in enumerate(data, start=1):
        handle.write(f"Q{idx}: {item['question'][:120]}\n")
        correct = item["correct"]
        for ans_idx, answer in enumerate(item["answers"]):
            is_correct = ans_idx in correct if isinstance(correct, list) else ans_idx == correct
            mark = " <--" if is_correct else ""
            handle.write(f"  {ans_idx}: {answer[:100]}{mark}\n")
        handle.write("\n")


def main() -> None:
    document = Document(str(DOCX))
    gynecology_paragraphs, obstetrics_paragraphs = split_sections(document)

    gynecology_chunks = build_chunks(gynecology_paragraphs)
    obstetrics_chunks = build_chunks(obstetrics_paragraphs)

    gynecology_questions = parse_section(gynecology_chunks, numbered_only=False)
    obstetrics_questions = parse_section(obstetrics_chunks, numbered_only=True)

    OUT_GYN.write_text(json.dumps(gynecology_questions, ensure_ascii=False, indent=2), encoding="utf-8")
    OUT_OBS.write_text(json.dumps(obstetrics_questions, ensure_ascii=False, indent=2), encoding="utf-8")

    with OUT_CHECK.open("w", encoding="utf-8") as handle:
        write_check_file("GYNECOLOGY", gynecology_questions, handle)
        write_check_file("OBSTETRICS", obstetrics_questions, handle)

    print(f"DOCX: {DOCX.name}")
    print(f"Gynecology questions: {len(gynecology_questions)}")
    print(f"Obstetrics questions: {len(obstetrics_questions)}")
    print(f"Saved: {OUT_GYN}")
    print(f"Saved: {OUT_OBS}")


if __name__ == "__main__":
    main()