src/report_generator.py - common_agents_notes - IvanMysin/Topics

src/report_generator.py
511 строк27 KB
ivan
Working on preport scripts
18 фев 2026, 16:26
18 фев 2026, 16:26007c082
import pandas as pd
import numpy as np
import os
from datetime import datetime
from tqdm.auto import tqdm
from config import RESULTS_DIR, DB_CONFIG, SUMMARIZATION_CONFIG, LLM_CONFIG
from typing import Dict, Optional
import pickle
import json
from pathlib import Path
import markdown
from markdown.extensions.tables import TableExtension

# Добавляем импорт для генерации описаний
try:
    from .llm_descriptor import ClusterDescriptor, FallbackDescriptor
except ImportError:
    # Резервный импорт для случая, если модуль не найден
    from llm_descriptor import ClusterDescriptor, FallbackDescriptor


def contains_non_digits(text):
    for char in text:
        if not char.isdigit():
            return True
    return False


class ReportGenerator:
    def __init__(self, keyword_method: str = "tfidf"):
        self.results_dir = RESULTS_DIR
        self.results_dir.mkdir(exist_ok=True)
        self.descriptor = ClusterDescriptor()
        self.fallback_descriptor = FallbackDescriptor()
        self.templates_dir = Path(__file__).parent / "templates"
        self.templates_dir.mkdir(exist_ok=True)


    def generate_all_reports(self, df: pd.DataFrame, topic_words: dict, similarities: dict,
                             method: str = "kmeans", generate_descriptions: bool = True,
                             cluster_descriptions: dict = None, output_format: str = "markdown", keyword_method='tfidf', ollama_model: str = ''):
        """Генерирует все отчеты с возможностью использования готовых описаний"""
        print(f"📊 Генерация отчетов (формат: {output_format})...")

        # Генерация описаний кластеров
        if cluster_descriptions is not None:
            # Используем готовые описания
            print("📝 Используются предоставленные описания кластеров")
        elif generate_descriptions:
            # Генерируем описания с помощью LLM
            clusters_data = self._prepare_clusters_data(df, topic_words)
            cluster_descriptions = self.descriptor.generate_all_descriptions(
                clusters_data, topic_words, method
            )
        else:
            cluster_descriptions = {}

        # Определяем форматы для генерации
        formats_to_generate = []
        if output_format == "both":
            formats_to_generate = ["markdown", "html"]
        else:
            formats_to_generate = [output_format]

        # Генерируем отчеты во всех запрошенных форматах
        for fmt in formats_to_generate:
            # Сводный отчет
            self._generate_summary_report(df, topic_words, method, cluster_descriptions, fmt, keyword_method, ollama_model)

            # Отчеты по кластерам
            unique_clusters = np.unique(df['cluster'])
            unique_clusters = [c for c in unique_clusters if c != -1]

            for cluster_id in tqdm(unique_clusters, desc=f"Создание отчетов кластеров ({fmt})"):
                cluster_df = df[df['cluster'] == cluster_id]
                description = cluster_descriptions.get(cluster_id)
                self._generate_cluster_report(cluster_id, cluster_df, topic_words, method, description, fmt, keyword_method, ollama_model)

            # Отчет по шумовым точкам
            if -1 in df['cluster'].values:
                noise_df = df[df['cluster'] == -1]
                self._generate_noise_report(noise_df, method, fmt)

        print(f"✅ Отчеты сгенерированы в формате: {', '.join(formats_to_generate)}")

    def _prepare_clusters_data(self, df: pd.DataFrame, topic_words: dict) -> Dict:
        """Подготавливает данные кластеров для генерации описаний"""
        clusters_data = {}

        for cluster_id in np.unique(df['cluster']):
            if cluster_id == -1:
                continue

            cluster_df = df[df['cluster'] == cluster_id]
            clusters_data[cluster_id] = cluster_df

        return clusters_data

    def _generate_summary_report(self, df: pd.DataFrame, topic_words: dict,
                                 method: str, descriptions: dict, output_format: str = "markdown", keyword_method='tfidf', ollama_model: str = ''):
        """Генерирует сводный отчет с описаниями кластеров"""
        if output_format == "html":
            self._generate_summary_report_html(df, topic_words, method, descriptions, keyword_method, ollama_model)
        else:
            self._generate_summary_report_markdown(df, topic_words, method, descriptions, keyword_method, ollama_model)

    def _generate_summary_report_markdown(self, df: pd.DataFrame, topic_words: dict,
                                          method: str, descriptions: dict, keyword_method='tfidf', ollama_model: str = ''):
        """Генерирует сводный отчет в формате Markdown"""
        filename = self.results_dir / "00_summary.md"

        cluster_stats = df[df['cluster'] != -1]['cluster'].value_counts().sort_index()
        noise_count = len(df[df['cluster'] == -1])

        from config import KEYWORD_EXTRACTION_METHODS

        with open(filename, 'w', encoding='utf-8') as f:
            f.write("# Сводный отчет по кластеризации научных статей\n\n")
            f.write(f"**Дата анализа:** {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
            f.write(f"**Всего статей:** {len(df)}\n")
            f.write(f"**Метод кластеризации:** {method}\n")
            f.write(f"**Количество кластеров:** {len(cluster_stats)}\n")
            if noise_count > 0:
                f.write(f"**Шумовые точки:** {noise_count}\n")
            f.write(f"**Текст для кластеризации:** {DB_CONFIG['text_column']}\n")
            f.write(f"**Описания кластеров:** {'Сгенерированы' if descriptions else 'Отсутствуют'}\n")
            f.write(f"**Метод извлечения ключевых слов:** {KEYWORD_EXTRACTION_METHODS.get(keyword_method, keyword_method)}\n")
            f.write("\n## Методология\n\n")

            f.write(f"- **Описания кластеров:** Сгенерированы с помощью {ollama_model}\n")
            f.write(f"- **Ключевые слова:** Извлечены с использованием {keyword_method.upper()}\n")
            f.write("- **Схожесть:** Рассчитана как косинусная схожесть с центром кластера\n")

            f.write("\n")

            f.write("## Статистика по кластерам\n\n")
            f.write("| Кластер | Статей | Ключевые слова |\n")
            f.write("|---------|--------|----------|\n")

            for cluster_id in cluster_stats.index:
                count = cluster_stats[cluster_id]
                description_preview = ""
                topic_words_list = topic_words.get(cluster_id, [])

                for word in topic_words_list:
                    description_preview += f"{word[0]} ({word[1]}), "
                description_preview = description_preview[:-2]

                f.write(f"| [{cluster_id}](cluster_{cluster_id:02d}.md) | {count} | {description_preview} |\n")

            if noise_count > 0:
                f.write(f"| [Шум](noise_documents.md) | {noise_count} | - |\n")

    def _generate_cluster_report(self, cluster_id: int, cluster_df: pd.DataFrame,
                                 topic_words: dict, method: str, description: Optional[str] = None,
                                 output_format: str = "markdown", keyword_method='tfidf', ollama_model: str = ''):
        """Генерирует отчет для одного кластера с описанием"""
        if output_format == "html":
            self._generate_cluster_report_html(cluster_id, cluster_df, topic_words, method, description, keyword_method, ollama_model)
        else:
            self._generate_cluster_report_markdown(cluster_id, cluster_df, topic_words, method, description, keyword_method, ollama_model)

    def _generate_cluster_report_markdown(self, cluster_id: int, cluster_df: pd.DataFrame,
                                          topic_words: dict, method: str, description: Optional[str] = None, keyword_method='tfidf', ollama_model: str = ''):
        """Генерирует отчет для одного кластера в формате Markdown"""
        filename = self.results_dir / f"cluster_{cluster_id:02d}.md"

        # Сортируем по схожести
        cluster_df = cluster_df.sort_values('similarity', ascending=False)

        # Если описание не предоставлено, генерируем fallback
        if description is None:
            topic_words_list = topic_words.get(cluster_id, [])
            description = self.fallback_descriptor.generate_description(
                topic_words_list, len(cluster_df)
            )

        with open(filename, 'w', encoding='utf-8') as f:
            f.write(f"# Кластер {cluster_id}\n\n")
            f.write(f"**Метод кластеризации:** {method}\n")
            f.write(f"**Метод извлечения ключевых слов:** {keyword_method.upper()}\n")
            f.write(f"**Количество статей:** {len(cluster_df)}\n")
            f.write(f"**Текст для кластеризации:** {DB_CONFIG['text_column']}\n\n")
            f.write(f"**Модель ИИ для генерации обзора:** {ollama_model} \n\n")

            # Ключевые слова
            if cluster_id in topic_words and topic_words[cluster_id]:
                f.write("## Ключевые слова\n\n")
                words_str = " ".join([f"**{word}** ({count})" for word, count in topic_words[cluster_id][:10]])
                f.write(f"{words_str}\n\n")

            # Описание кластера
            f.write("## Описание кластера\n\n")
            f.write(f"{description}\n\n")
            # Статьи
            f.write("## Статьи\n\n")
            f.write("Статьи отсортированы по релевантности теме кластера:\n\n")

            for idx, (_, row) in enumerate(cluster_df.iterrows(), 1):
                self._write_article_entry(f, idx, row)

    def _write_article_entry(self, file_handle, index: int, row):
        """Записывает информацию об одной статье в файл"""
        title = row[DB_CONFIG['Articles']['title_column']]
        doi = row[DB_CONFIG['Articles']['doi_column']]
        abstract = row[DB_CONFIG['Articles']['abstract_column']]
        similarity = row['similarity']
        date = row[DB_CONFIG['Articles']['date_column']] if pd.notna(row[DB_CONFIG['Articles']['date_column']]) else ""

        # Проверяем наличие колонки AI summary
        ai_summary_col = DB_CONFIG['Articles']["ai_summury_column"]
        if ai_summary_col in row.index:
            ai_summary = row[ai_summary_col]
        else:
            ai_summary = None

        article_id = row['id']

        # Форматируем DOI ссылку
        doi_link = self._format_doi_link(doi)

        # Форматируем дату
        date_str = f" ({date})" if date else ""

        # Заголовок статьи
        file_handle.write(f"### {index}. {title} \n\n")
        file_handle.write(f"**Идентификатор** <a name='{article_id}'> {article_id} </a> \n\n")
        file_handle.write(f"**Дата публикации:** {date_str} \n\n")

        # DOI ссылка
        if 'doi' in doi_link:
            file_handle.write(f"**DOI:** {doi_link}  \n")
        else:
            file_handle.write(f"**PMID:** {doi_link}  \n")

        # Схожесть с темой
        file_handle.write(f"**Схожесть с темой:** {similarity:.3f}  \n\n")

        # Абстракт
        formatted_abstract = self._format_abstract(abstract)
        file_handle.write(f"**Абстракт:** {formatted_abstract}\n\n")

        ai_summary = self._format_abstract(ai_summary)
        if ai_summary != '*Текст отсутствует*':
            file_handle.write(f"**ИИ пересказ:** {ai_summary}\n\n")

        file_handle.write("---\n\n")

    def _format_abstract(self, abstract: str, max_length: int = 15000) -> str:
        """Форматирует абстракт для красивого отображения"""
        if pd.isna(abstract) or abstract == "":
            return "*Текст отсутствует*"

        abstract_clean = str(abstract).strip()

        # Если абстракт слишком длинный, обрезаем и добавляем многоточие
        if len(abstract_clean) > max_length:
            return abstract_clean[:max_length] + "..."

        return abstract_clean

    def _format_doi_link(self, doi: str) -> str:
        """Форматирует DOI в виде активной ссылки Markdown"""
        if pd.isna(doi) or doi == "":
            return ""

        # Очищаем DOI от лишних пробелов
        doi_clean = str(doi).strip()

        if contains_non_digits(doi_clean):
            # Создаем Markdown ссылку
            return f"[{doi_clean}](https://doi.org/{doi_clean})"
        else:
            return f"[{doi_clean}](https://pubmed.ncbi.nlm.nih.gov/{doi_clean}/)"

    def _format_doi_link_html(self, doi: str) -> str:
        """Форматирует DOI в виде активной ссылки HTML"""
        if pd.isna(doi) or doi == "":
            return ""

        # Очищаем DOI от лишних пробелов
        doi_clean = str(doi).strip()

        if contains_non_digits(doi_clean):
            return f'<a href="https://doi.org/{doi_clean}" class="doi-link" target="_blank">{doi_clean}</a>'
        else:
            return f'<a href="https://pubmed.ncbi.nlm.nih.gov/{doi_clean}/" class="doi-link" target="_blank">{doi_clean}</a>'

    def _load_html_template(self) -> str:
        """Загружает HTML шаблон"""
        template_path = self.templates_dir / "html_template.html"
        if template_path.exists():
            with open(template_path, 'r', encoding='utf-8') as f:
                return f.read()
        else:
            return self._get_default_html_template()

    def _get_default_html_template(self) -> str:
        """Возвращает базовый HTML шаблон"""
        return """<!DOCTYPE html>
<html lang="ru">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>{page_title}</title>
    <style>
        body {{ font-family: Arial, sans-serif; line-height: 1.6; color: #333; background: #f5f7fa; padding: 20px; }}
        .container {{ max-width: 1200px; margin: 0 auto; background: white; border-radius: 8px; padding: 30px; }}
        h1 {{ color: #2c3e50; border-bottom: 3px solid #3498db; padding-bottom: 10px; }}
        h2 {{ color: #34495e; margin-top: 30px; }}
        table {{ width: 100%; border-collapse: collapse; margin: 20px 0; }}
        th {{ background: #3498db; color: white; padding: 12px; text-align: left; }}
        td {{ padding: 10px 12px; border-bottom: 1px solid #ecf0f1; }}
        tr:hover {{ background: #f8f9fa; }}
        .article {{ background: #fafafa; border: 1px solid #e0e0e0; border-radius: 6px; padding: 20px; margin-bottom: 20px; }}
        .keyword {{ background: #e8f4f8; padding: 3px 8px; border-radius: 12px; font-size: 0.9em; }}
        .doi-link {{ color: #3498db; text-decoration: none; }}
        .similarity-score {{ background: #d4edda; padding: 2px 8px; border-radius: 4px; }}
        .abstract {{ background: white; padding: 15px; border-left: 3px solid #28a745; }}
        .info-box {{ background: #ecf0f1; border-left: 4px solid #3498db; padding: 15px; margin-bottom: 20px; border-radius: 4px; }}
        .info-label {{ font-weight: bold; color: #2c3e50; }}
    </style>
</head>
<body>
    <div class="container">
        {content}
    </div>
</body>
</html>"""

    def _generate_summary_report_html(self, df: pd.DataFrame, topic_words: dict,
                                        method: str, descriptions: dict, keyword_method: str, ollama_model: str = ''):
        """Генерирует сводный отчет в формате HTML"""
        filename = self.results_dir / "00_summary.html"

        cluster_stats = df[df['cluster'] != -1]['cluster'].value_counts().sort_index()
        noise_count = len(df[df['cluster'] == -1])

        from config import KEYWORD_EXTRACTION_METHODS

        content = "<h1>Сводный отчет по кластеризации научных статей</h1>"
        content += '<div class="info-box">'
        content += f"<p><span class=\"info-label\">Дата анализа:</span> {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>"
        content += f"<p><span class=\"info-label\">Всего статей:</span> {len(df)}</p>"
        content += f"<p><span class=\"info-label\">Метод кластеризации:</span> {method}</p>"
        content += f"<p><span class=\"info-label\">Количество кластеров:</span> {len(cluster_stats)}</p>"
        if noise_count > 0:
            content += f"<p><span class=\"info-label\">Шумовые точки:</span> {noise_count}</p>"
        content += f"<p><span class=\"info-label\">Текст для кластеризации:</span> {DB_CONFIG['text_column']}</p>"
        content += f"<p><span class=\"info-label\">Описания кластеров:</span> {'Сгенерированы' if descriptions else 'Отсутствуют'}</p>"
        content += f"<p><span class=\"info-label\">Метод извлечения ключевых слов:</span> {KEYWORD_EXTRACTION_METHODS.get(keyword_method, keyword_method)}</p>"
        content += "</div>"

        content += "<h2>Методология</h2>"
        content += "<ul>"
        content += f"<li><strong>Описания кластеров:</strong> Сгенерированы с помощью {ollama_model}</li>"
        content += f"<li><strong>Ключевые слова:</strong> Извлечены с использованием {keyword_method.upper()}</li>"
        content += "<li><strong>Схожесть:</strong> Рассчитана как косинусная схожесть с центром кластера</li>"
        content += "</ul>"

        content += "<h2>Статистика по кластерам</h2>"
        content += "<table>"
        content += "<tr><th>Кластер</th><th>Статей</th><th>Ключевые слова</th></tr>"

        for cluster_id in cluster_stats.index:
            count = cluster_stats[cluster_id]
            topic_words_list = topic_words.get(cluster_id, [])
            keywords_html = " ".join([f'<span class="keyword">{word} </span>' for word in topic_words_list])
            content += f"<tr><td><a href=\"cluster_{cluster_id:02d}.html\" class=\"doi-link\" target=\"_blank\">{cluster_id}</a></td><td>{count}</td><td>{keywords_html}</td></tr>"

        if noise_count > 0:
            content += "<tr><td><a href=\"noise_documents.html\" class=\"doi-link\" target=\"_blank\">Шум</a></td>"
            content += f"<td>{noise_count}</td><td>-</td></tr>"

        content += "</table>"

        html_template = self._load_html_template()
        html_output = html_template.replace("{page_title}", "Сводный отчет").replace("{content}", content)

        with open(filename, 'w', encoding='utf-8') as f:
            f.write(html_output)

    def _generate_cluster_report_html(self, cluster_id: int, cluster_df: pd.DataFrame,
                                         topic_words: dict, method: str, description: Optional[str] = None, keyword_method: str = 'tfidf', ollama_model: str = ''):
        """Генерирует HTML отчет для одного кластера"""
        filename = self.results_dir / f"cluster_{cluster_id:02d}.html"

        # Сортируем по схожести
        cluster_df = cluster_df.sort_values('similarity', ascending=False)

        # Если описание не предоставлено, генерируем fallback
        if description is None:
            topic_words_list = topic_words.get(cluster_id, [])
            description = self.fallback_descriptor.generate_description(
                topic_words_list, len(cluster_df)
            )
        else:
            description = markdown.markdown(description, extensions=[TableExtension()])

        content = f"<h1>Кластер {cluster_id}</h1>"
        content += '<div class="info-box">'
        content += f"<p><span class=\"info-label\">Метод кластеризации:</span> {method}</p>"
        content += f"<p><span class=\"info-label\">Метод извлечения ключевых слов:</span> {keyword_method.upper()}</p>"
        content += f"<p><span class=\"info-label\">Количество статей:</span> {len(cluster_df)}</p>"
        content += f"<p><span class=\"info-label\">Текст для кластеризации:</span> {DB_CONFIG['text_column']}</p>"
        content += f"<p><span class=\"info-label\">Модель ИИ для обзора </span> {ollama_model}</p>"
        content += "</div>" 

        content += "<h2>Описание кластера</h2>"
        content += f"<p>{description}</p>"

        if cluster_id in topic_words and topic_words[cluster_id]:
            content += "<h2>Ключевые слова</h2>"
            keywords_html = " ".join([f'<span class="keyword">{word} ({count})</span>' for word, count in topic_words[cluster_id]])
            content += f"<p>{keywords_html}</p>"

        content += "<h2>Статьи</h2>"
        content += "<p>Статьи отсортированы по релевантности теме кластера:</p>"

        for idx, (_, row) in enumerate(cluster_df.iterrows(), 1):
            content += self._write_article_entry_html(idx, row)

        html_template = self._load_html_template()
        html_output = html_template.replace("{page_title}", f"Кластер {cluster_id}").replace("{content}", content)

        with open(filename, 'w', encoding='utf-8') as f:
            f.write(html_output)

    def _write_article_entry_html(self, index: int, row) -> str:
        """Записывает информацию об одной статье в HTML формате"""
        title = row[DB_CONFIG['Articles']['title_column']]
        doi = row[DB_CONFIG['Articles']['doi_column']]
        abstract = row[DB_CONFIG['Articles']['abstract_column']]
        similarity = row['similarity']
        date = row[DB_CONFIG['Articles']['date_column']] if pd.notna(row[DB_CONFIG['Articles']['date_column']]) else ""
        # Проверяем наличие колонки AI summary
        ai_summary_col = SUMMARIZATION_CONFIG["database"]["summary_column"]
        if ai_summary_col in row.index:
            ai_summary = row[ai_summary_col]
        else:
            ai_summary = None

        article_id = row['id']

        doi_link = self._format_doi_link_html(doi)
        date_str = f" ({date})" if date else ""

        content = f'<div class="article">'
        content += f"<h3>{index}. {title} </h3>"
        content += f'<p class="article-meta"><strong>Идентификатор статьи: </strong> <a name={article_id}> {article_id} </a> </span></p>'
        content += f'<p class="article-meta"><strong>Дата публикации </strong> {date_str} </span></p>'

        if doi_link:
            if 'doi.org' in doi_link.lower():
                doi_type = "DOI"
            else:
                doi_type = "PMID"
            content += f'<p class="article-meta"><strong>{doi_type}:</strong> {doi_link}</p>'

        content += f'<p class="article-meta"><strong>Схожесть с темой:</strong> <span class="similarity-score">{similarity:.3f}</span></p>'

        formatted_abstract = self._format_abstract(abstract)
        content += f'<div class="abstract"><strong>Абстракт:</strong> {formatted_abstract}</div>'

        ai_summary = self._format_abstract(ai_summary)
        if ai_summary != '*Текст отсутствует*':
            content += f'<div class="abstract"><strong>ИИ пересказ:</strong> {ai_summary}</div>'

        content += "</div>"
        return content

    def _generate_noise_report(self, noise_df: pd.DataFrame, method: str, output_format: str = "markdown"):
        """Генерирует отчет по шумовым точкам"""
        if output_format == "html":
            self._generate_noise_report_html(noise_df, method)
        else:
            self._generate_noise_report_markdown(noise_df, method)

    def _generate_noise_report_markdown(self, noise_df: pd.DataFrame, method: str):
        """Генерирует отчет по шумовым точкам в формате Markdown"""
        filename = self.results_dir / "noise_documents.md"

        with open(filename, 'w', encoding='utf-8') as f:
            f.write("# Шумовые точки\n\n")
            f.write(f"**Метод кластеризации:** {method}\n")
            f.write(f"**Количество шумовых точек:** {len(noise_df)}\n\n")
            f.write("## Статьи, не вошедшие в кластеры\n\n")

            for idx, (_, row) in enumerate(noise_df.iterrows(), 1):
                self._write_article_entry(f, idx, row)

    def _generate_noise_report_html(self, noise_df: pd.DataFrame, method: str):
        """Генерирует HTML отчет по шумовым точкам"""
        filename = self.results_dir / "noise_documents.html"

        content = "<h1>Шумовые точки</h1>"
        content += '<div class="info-box">'
        content += f"<p><span class=\"info-label\">Метод кластеризации:</span> {method}</p>"
        content += f"<p><span class=\"info-label\">Количество шумовых точек:</span> {len(noise_df)}</p>"
        content += "</div>"

        content += "<h2>Статьи, не вошедшие в кластеры</h2>"

        for idx, (_, row) in enumerate(noise_df.iterrows(), 1):
            content += self._write_article_entry_html(idx, row)

        html_template = self._load_html_template()
        html_output = html_template.replace("{page_title}", "Шумовые точки").replace("{content}", content)

        with open(filename, 'w', encoding='utf-8') as f:
            f.write(html_output)