Spaces:
Sleeping
Sleeping
File size: 3,191 Bytes
dbe2c62 |
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 |
from collections import OrderedDict
from copy import deepcopy
class ChunkBuilder:
def readInput(self, RawLvlsDict=None, RawDataDict=None):
# Đọc dữ liệu
self.struct_spec = RawLvlsDict[0]
self.paragraphs = sorted(
RawDataDict.get("paragraphs", []),
key=lambda x: x.get("Paragraph", 0)
)
# Chuẩn bị cấu trúc
self.ordered_fields = list(self.struct_spec.keys())
self.last_field = self.ordered_fields[-1]
self.level_fields = self.ordered_fields[:-1]
# Tập marker cho từng field
self.marker_dict = {}
for fld in self.ordered_fields:
vals = self.struct_spec.get(fld, [])
self.marker_dict[fld] = set(vals) if isinstance(vals, list) else set()
# Biến tạm
self.StructDict = []
self.index_counter = 1
# ===== Các hàm tiện ích =====
def _new_temp(self):
return {fld: "" for fld in self.level_fields} | {self.last_field: []}
def _temp_has_data(self, temp):
return any(temp[f].strip() for f in self.level_fields) or bool(temp[self.last_field])
def _reset_deeper(self, temp, touched_field):
idx = self.level_fields.index(touched_field)
for f in self.level_fields[idx+1:]:
temp[f] = ""
temp[self.last_field] = []
def _has_data_from_level(self, temp, fld):
"""Kiểm tra từ level fld trở xuống có dữ liệu không"""
if fld not in self.level_fields:
return False
idx = self.level_fields.index(fld)
for f in self.level_fields[idx:]:
if temp[f].strip():
return True
if temp[self.last_field]:
return True
return False
def _with_index(self, temp, idx):
"""Tạo OrderedDict với Index đứng đầu"""
od = OrderedDict()
od["Index"] = idx
for f in self.level_fields:
od[f] = temp[f]
od[self.last_field] = temp[self.last_field]
return od
# ===== Hàm chính =====
def build(self, RawLvlsDict=None, RawDataDict=None):
self.readInput(RawLvlsDict, RawDataDict)
temp = self._new_temp()
for p in self.paragraphs:
text = p.get("Text") or ""
marker = p.get("MarkerType", None) or "none"
matched_field = None
for fld in self.level_fields:
if marker in self.marker_dict.get(fld, set()):
matched_field = fld
break
if matched_field is not None:
if self._has_data_from_level(temp, matched_field):
self.StructDict.append(self._with_index(deepcopy(temp), self.index_counter))
self.index_counter += 1
temp[matched_field] = text
self._reset_deeper(temp, matched_field)
else:
temp[self.last_field].append(text)
if self._temp_has_data(temp):
self.StructDict.append(self._with_index(deepcopy(temp), self.index_counter))
self.index_counter += 1
return self.StructDict |