LongArticleTaskServer/app/infra/internal/long_articles.py

from typing import List, Dict

from app.core.database import DatabaseManager


async def get_top_article_title_list(pool: DatabaseManager) -> List[Dict]:
    query = """
        select distinct title, source_id
        from datastat_sort_strategy
        where produce_plan_name = %s and source_id is not null;
    """
    return await pool.async_fetch(query=query, params=("TOP100",))


async def get_hot_titles(
    pool: DatabaseManager, date_string: str, position: int, read_times_threshold: float
) -> List[str]:
    """get titles of hot articles"""
    query = """
        select distinct title
        from datastat_sort_strategy
        where position < %s and read_rate >= %s and date_str >= %s;
    """
    response = await pool.async_fetch(
        query=query, params=(position, read_times_threshold, date_string)
    )
    return [i["title"] for i in response]