src/database_manager.py - master - IvanMysin/Topics

src/database_manager.py
221 строка6 KB
ivan
Working on downloading scripts
17 фев 2026, 15:19
17 фев 2026, 15:192db3ea5
#!/usr/bin/env python
"""
Shared module for database operations across all loading scripts.
"""

import sqlite3
import hashlib
from pathlib import Path
from typing import Tuple, Dict, Any, Optional

from config import DB_CONFIG


def get_articles_config() -> Dict[str, str]:
    """
    Возвращает конфигурацию таблицы Articles.

    Returns:
        Словарь с полями таблицы Articles
    """
    return DB_CONFIG["Articles"]


def create_articles_table(db_path: str) -> None:
    """
    Creates the Articles table with all fields from DB_CONFIG.

    Args:
        db_path: Path to the database file
    """
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()

    articles_config = get_articles_config()

    cursor.execute(f'''
    CREATE TABLE IF NOT EXISTS Articles (
        {articles_config["id_column"]} INTEGER PRIMARY KEY,
        {articles_config["doi_column"]} TEXT UNIQUE,
        {articles_config["title_column"]} TEXT,
        {articles_config["abstract_column"]} TEXT,
        {articles_config["date_column"]} TEXT,
        {articles_config["text_column"]} TEXT,
        {articles_config["journal_column"]} TEXT,
        {articles_config["authors_column"]} TEXT,
        {articles_config["volume_column"]} TEXT,
        {articles_config["issue_column"]} TEXT,
        {articles_config["pages_column"]} TEXT,
        {articles_config["pmid_column"]} TEXT,
        {articles_config["filepath_column"]} TEXT
    )
    ''')

    conn.commit()
    conn.close()


def document_exists(cursor: sqlite3.Cursor, doi: Optional[str], title: str) -> bool:
    """
    Checks if a document already exists in the database.

    Args:
        cursor: Database cursor
        doi: DOI of the document
        title: Title of the document

    Returns:
        True if the document exists
    """
    articles_config = get_articles_config()

    if doi:
        cursor.execute(f"SELECT 1 FROM Articles WHERE {articles_config['doi_column']} = ?", (doi,))
        if cursor.fetchone():
            return True

    cursor.execute(f"SELECT 1 FROM Articles WHERE {articles_config['title_column']} = ?", (title,))
    return cursor.fetchone() is not None


def validate_doi(doi: str) -> Tuple[bool, str]:
    """
    Validates DOI.

    Args:
        doi: DOI to validate

    Returns:
        Tuple(is_valid, message) - whether DOI is valid and error message
    """
    if not doi:
        return False, "DOI отсутствует"

    doi = doi.strip()
    if len(doi) < 10:
        return False, f"DOI слишком короткий: {doi}"

    if doi.startswith("PMID:") or doi.startswith("GEN:"):
        return True, ""

    if "/" not in doi:
        return False, f"DOI содержит некорректный формат: {doi}"

    return True, ""


def validate_title(title: str) -> Tuple[bool, str]:
    """
    Validates article title.

    Args:
        title: Title to validate

    Returns:
        Tuple(is_valid, message) - whether title is valid and error message
    """
    if not title:
        return False, "Название отсутствует"

    title = title.strip()
    if len(title) < 20:
        return False, f"Название слишком короткое ({len(title)} символов): {title}"

    if len(title) > 1000:
        return False, f"Название слишком длинное ({len(title)} символов): {title[:100]}..."

    return True, ""


def validate_abstract(abstract: str) -> Tuple[bool, str]:
    """
    Validates article abstract.

    Args:
        abstract: Abstract to validate

    Returns:
        Tuple(is_valid, message) - whether abstract is valid and error message
    """
    if not abstract:
        return False, "Абстракт отсутствует"

    abstract = abstract.strip()
    if len(abstract) < 50:
        return False, f"Абстракт слишком короткий ({len(abstract)} символов): {abstract}"

    return True, ""


def generate_doi_from_title_abstract(title: str, abstract: str) -> str:
    """
    Generates a DOI hash from title and abstract.

    Args:
        title: Article title
        abstract: Article abstract

    Returns:
        Generated DOI as hash
    """
    combined = f"{title}|{abstract}"
    return "GEN:" + hashlib.md5(combined.encode('utf-8')).hexdigest()


def insert_article(cursor: sqlite3.Cursor, article_data: Dict[str, Any]) -> None:
    """
    Inserts an article into the database.

    Args:
        cursor: Database cursor
        article_data: Dictionary with article fields
    """
    articles_config = get_articles_config()

    cursor.execute(f'''
    INSERT INTO Articles (
        {articles_config["doi_column"]},
        {articles_config["title_column"]},
        {articles_config["abstract_column"]},
        {articles_config["date_column"]},
        {articles_config["text_column"]},
        {articles_config["journal_column"]},
        {articles_config["authors_column"]},
        {articles_config["volume_column"]},
        {articles_config["issue_column"]},
        {articles_config["pages_column"]},
        {articles_config["pmid_column"]},
        {articles_config["filepath_column"]}
    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
    ''', (
        article_data.get('doi'),
        article_data.get('title'),
        article_data.get('abstract'),
        article_data.get('date'),
        article_data.get('full_text'),
        article_data.get('journal'),
        article_data.get('authors'),
        article_data.get('volume'),
        article_data.get('issue'),
        article_data.get('pages'),
        article_data.get('pmid'),
        article_data.get('filepath', '')
    ))


def update_article_filepath(cursor: sqlite3.Cursor, article_id: int, filepath: str) -> None:
    """
    Обновляет путь к файлу для статьи.

    Args:
        cursor: Database cursor
        article_id: ID статьи
        filepath: Путь к файлу
    """
    articles_config = get_articles_config()

    cursor.execute(f'''
    UPDATE Articles
    SET {articles_config["filepath_column"]} = ?
    WHERE {articles_config["id_column"]} = ?
    ''', (filepath, article_id))