src/dataset.py

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)