import glob import json import logging import os import pickle import string from pathlib import Path import lxml import lxml.html import yaml from bs4 import BeautifulSoup, Tag from lxml import etree from progress.bar import Bar from transformers import MarkupLMFeatureExtractor from consts import id2label, label2id from processor import NewsProcessor from utils import TextUtils logging.basicConfig(level=logging.INFO) class NewsDatasetBuilder: __processor: NewsProcessor = None __utils: TextUtils = None def __init__(self): self.__processor = NewsProcessor() self.__utils = TextUtils() logging.debug('NewsDatasetBuilder Sınıfı oluşturuldu') def __get_dom_tree(self, html): """ Verilen HTML içeriğinden bir DOM ağacı oluşturur. Args: html (str): Oluşturulacak DOM ağacının temelini oluşturacak HTML içeriği. Returns: ElementTree: Oluşturulan DOM ağacı. """ html = self.__processor.encode(html) x = lxml.html.fromstring(html) dom_tree = etree.ElementTree(x) return dom_tree @staticmethod def __get_config(config_file_path): """ Belirtilen konfigürasyon dosyasını okuyarak bir konfigürasyon nesnesi döndürür. Args: config_file_path (str): Okunacak konfigürasyon dosyasının yolunu belirtir. Returns: dict: Okunan konfigürasyon verilerini içeren bir sözlük nesnesi. """ with open(config_file_path, "r") as yaml_file: _config = yaml.load(yaml_file, Loader=yaml.FullLoader) return _config def __non_ascii_equal(self, value, node_text): """ Verilen değer ve düğüm metni arasında benzerlik kontrolü yapar. Benzerlik için cosine similarity kullanılır. Eğer benzerlik oranı %70'in üzerinde ise bu iki metin benzer kabul edilir. Args: value (str): Karşılaştırılacak değer. node_text (str): Karşılaştırılacak düğüm metni. Returns: bool: Değer ve düğüm metni arasında belirli bir benzerlik eşiği üzerinde eşleşme durumunda True, aksi halde False. """ value = self.__utils.clean_format_str(value) # value = re.sub(r"[^a-zA-Z0-9.:]", "", value, 0) value_nopunct = "".join([char for char in value if char not in string.punctuation]) node_text = self.__utils.clean_format_str(node_text) # node_text = re.sub(r"[^a-zA-Z0-9.:]", "", node_text, 0) node_text_nopunct = "".join([char for char in node_text if char not in string.punctuation]) sim = self.__utils.cosine(value_nopunct, node_text_nopunct) return sim > 0.7 # value.strip() == node_text.strip() def __get_truth_value(self, site_config, html, label): """ Belirtilen site'ya ait konfigürasyondan label parametresi ile gönderilen tarih, başlık, spot (açıklama) ve içerik alanlarının konfigürasyona göre belirtilen css-query ile bulunup çıkartılır ve döndürülür. Args: site_config (dict): Site konfigürasyon verilerini içeren bir sözlük. html (str): İşlenecek HTML içeriği. label (str): Etiket adı. Returns: list: Etiket adına bağlı doğruluk değerlerini içeren bir liste. """ result = [] tree = BeautifulSoup(html, 'html.parser') qs = site_config["css-queries"][label] for q in qs: found = tree.select(q) if found: el = found[0] for c in el: if type(c) is Tag: c.decompose() if el.name == "meta": text = el.attrs["content"] else: text = el.text if text: text = self.__utils.clean_format_str(text) text = text.strip() result.append(text) return result def __annotation(self, html, site_config, feature_extractor): """ Verilen HTML içeriği, site konfigürasyonu ve özellik çıkarıcısıyla ilişkili bir etiketleme yapar. Bu kısımda sitelerin önceden hazırladığımız css-query leri ile ilgili html bölümlerini bulup, bunu kullanarak otomatik olarak veri işaretlemesi yapılmasını sağlamaktayız. Args: html (str): Etiketleme işlemine tabi tutulacak HTML içeriği. site_config (dict): Site konfigürasyon verilerini içeren bir sözlük. feature_extractor (function): Özellik çıkarıcısı fonksiyonu. Returns: dict or None: Etiketleme sonucunu içeren bir sözlük nesnesi veya None. """ annotations = dict() for _id in id2label: if _id == -100: continue label = id2label[_id] # Önceden belirlediğimiz tarih (date), başlık (title), spot (description) ve içerik (content), # alanlarını site konfigürasyonuna göre çıkartıyoruz annotations[label] = self.__get_truth_value(site_config, html, label) if len(annotations["content"]) == 0: return None # MarkupLMFeatureExtractor ile sayfadaki node text ve xpath'leri çıkarıyoruz. # MarkupLMFeatureExtractor html içeriğindeki head > meta kısımlarını dikkate almaz # sadece body elementinin altındaki node'ları ve xpath'leri çıkarır encoding = feature_extractor(html) labels = [[]] nodes = [[]] xpaths = [[]] # MarkupLMFeatureExtractor tarafından çıkarılan her bir node'u annotations fonksiyonu ile otomatik olarak # bulduğumuz bölümleri node'ların textleri ile karşılaştırıp otomatik olarak veri işaretlemesi yapıyoruz. for idx, node_text in enumerate(encoding['nodes'][0]): xpath = encoding.data["xpaths"][0][idx] match = False for label in annotations: for mark in annotations[label]: if self.__non_ascii_equal(mark, node_text): node_text = self.__utils.clean_format_str(node_text) labels[0].append(label2id[label]) nodes[0].append(node_text) xpaths[0].append(xpath) match = True if not match: labels[0].append(label2id["other"]) nodes[0].append(node_text) xpaths[0].append(xpath) item = {'nodes': nodes, 'xpaths': xpaths, 'node_labels': labels} return item def __transform_file(self, name, file_path, output_path): """ Belirtilen dosyayı dönüştürerek temizlenmiş HTML içeriğini yeni bir dosyaya kaydeder. Args: name (str): Dosyanın adı. file_path (str): Dönüştürülecek dosyanın yolunu belirtir. output_path (str): Temizlenmiş HTML içeriğinin kaydedileceği dizin yolunu belirtir. Returns: None Raises: IOError: Dosya veya dizin oluşturma hatası durumunda fırlatılır. """ with open(file_path, 'r') as html_file: html = html_file.read() clean_html = self.__processor.transform(html) file_dir = f"{output_path}/{name}" file_name = Path(file_path).name if not os.path.exists(file_dir): os.makedirs(file_dir) file_path = f"{file_dir}/{file_name}" with open(file_path, 'w', encoding='utf-8') as output: output.write(clean_html) def __transform(self, name, raw_html_path, output_path, count): """ Belirtilen site için, ham HTML dosyalarının yolunu, çıkış dizin yolunu ve sayımı kullanarak HTML dönüştürme işlemini gerçekleştirir. Args: name (str): İşlem yapılacak site adı. raw_html_path (str): Ham HTML dosyalarının yolunu belirtir. output_path (str): Dönüştürülmüş HTML dosyalarının kaydedileceği dizin yolunu belirtir. count (int): İşlem yapılacak dosya sayısı. Returns: None Raises: IOError: Dosya veya dizin oluşturma hatası durumunda fırlatılır. """ files_path = f"{raw_html_path}/{name}" lfs = glob.glob(f"{files_path}/*.html") _max = count # len(lfs) logging.info(f"{name} html transform started.\n") with Bar(f'{name} Transforming html files', max=_max, suffix='%(percent).1f%% | %(index)d | %(remaining)d | %(max)d | %(eta)ds') as bar: i = 0 for lf in lfs: try: self.__transform_file(name, lf, output_path) bar.next() i = i + 1 if i > count: break except Exception as e: logging.error(f"An exception occurred id: {lf} error: {str(e)}") bar.finish() logging.info(f"{name} html transform completed.\n") def __auto_annotation(self, name, config_path, meta_path, clean_html_path, output_path, count): """ Belirtilen site için, yapılandırma dosyası yolunu, meta dosya yolunu, temizlenmiş HTML dosyalarının yolunu, çıkış dizin yolunu ve işlem yapılacak dosya sayısını kullanarak otomatik etiketleme işlemini gerçekleştirir. Args: name (str): İşlem yapılacak site adı. config_path (str): Yapılandırma dosyasının yolunu belirtir. meta_path (str): Meta dosyasının yolunu belirtir. clean_html_path (str): Temizlenmiş HTML dosyalarının yolunu belirtir. output_path (str): Oluşturulan veri setinin kaydedileceği dizin yolunu belirtir. count (int): İşlem yapılacak dosya sayısı. Returns: None Raises: IOError: Dosya veya dizin oluşturma hatası durumunda fırlatılır. """ config = self.__get_config(config_path) annotation_config = config[name] feature_extractor = MarkupLMFeatureExtractor() dataset = [] with open(f'{meta_path}/{name}.json', 'r') as json_file: links = json.load(json_file) _max = count # len(links) logging.info(f"{name} auto annotation started.\n") with Bar(f'{name} Building DataSet', max=_max, suffix='%(percent).1f%% | %(index)d | %(remaining)d | %(max)d | %(eta)ds') as bar: i = 0 for link in links: try: _id = link["id"] url = link["url"] i = i + 1 html_file_path = f"{clean_html_path}/{name}/{_id}.html" if not os.path.exists(html_file_path): continue with open(html_file_path, 'r') as html_file: html = html_file.read() item = self.__annotation(html, annotation_config, feature_extractor) if item: dataset.append(item) bar.next() if len(dataset) >= _max: break except Exception as e: logging.info(f"An exception occurred id: {url} error: {str(e)}") bar.finish() pickle_file_path = f'{output_path}/{name}.pickle' logging.info(f"Writing the dataset for {name}") with open(pickle_file_path, "wb") as f: pickle.dump(dataset, f) json_file_path = f'{output_path}/{name}.json' with open(json_file_path, 'w', encoding='utf-8') as f: json.dump(dataset, f, ensure_ascii=False, indent=4) def run(self, name, config_path, meta_path, raw_html_path, clean_html_path, dataset_path, count): """ Belirtilen site için, yapılandırma dosyası yolunu, meta dosya yolunu, ham HTML dosyalarının yolunu, temizlenmiş HTML dosyalarının yolunu, veri seti dosyasının yolunu ve işlem yapılacak dosya sayısını kullanarak veri seti oluşturma işlemini gerçekleştirir. Args: name (str): İşlem yapılacak site adı. config_path (str): Yapılandırma dosyasının yolunu belirtir. meta_path (str): Meta dosyasının yolunu belirtir. raw_html_path (str): Ham HTML dosyalarının yolunu belirtir. clean_html_path (str): Temizlenmiş HTML dosyalarının yolunu belirtir. dataset_path (str): Oluşturulan veri setinin kaydedileceği dizin yolunu belirtir. count (int): İşlem yapılacak dosya sayısı. Returns: None """ logging.info(f"{name} build dataset started.") self.__transform(name=name, raw_html_path=raw_html_path, output_path=clean_html_path, count=count) self.__auto_annotation(name=name, config_path=config_path, meta_path=meta_path, clean_html_path=clean_html_path, output_path=dataset_path, count=count) logging.info(f"{name} build dataset completed.") if __name__ == '__main__': # sites = ["aa", "aksam", "cnnturk", "cumhuriyet", "ensonhaber", "haber7", "haberglobal", "haberler", "haberturk", # "hurriyet", "milliyet", "ntv", "trthaber"] sites = ["aa", "aksam", "cnnturk", "cumhuriyet", "ensonhaber", "haber7", "haberglobal", "haberler", "haberturk", "hurriyet"] count_per_site = 10 total = count_per_site * len(sites) builder = NewsDatasetBuilder() _config_path = "../annotation-config.yaml" _meta_path = "../data/meta" _raw_html_path = "../data/html/raw" _clean_html_path = "../data/html/clean" _dataset_path = f"../data/dataset/{total}" for name in sites: builder.run(name=name, config_path=_config_path, meta_path=_meta_path, raw_html_path=_raw_html_path, clean_html_path=_clean_html_path, dataset_path=_dataset_path, count=count_per_site)