main.py

import asyncio
import json
import logging
import os
import re
from contextlib import asynccontextmanager
from datetime import datetime
from pathlib import Path
from typing import Annotated, List

from cashews import NOT_NONE, cache
from dotenv import load_dotenv
from fastapi import BackgroundTasks, FastAPI, Header, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from httpx import AsyncClient
from huggingface_hub import CommitScheduler, DatasetCard, HfApi, hf_hub_download, whoami
from huggingface_hub.utils import disable_progress_bars
from huggingface_hub.utils._errors import HTTPError
from langfuse.openai import AsyncOpenAI  # OpenAI integration
from pydantic import BaseModel, Field
from starlette.responses import RedirectResponse
from card_processing import parse_markdown, try_load_text, is_empty_template
import logging
import logging.config
from huggingface_hub import list_datasets
import yaml

disable_progress_bars()
load_dotenv()

# Load the logging configuration from the YAML file
with open("log_conf.yaml", "r") as f:
    log_config = yaml.safe_load(f.read())
    logging.config.dictConfig(log_config)

# Create a logger for your application
logger = logging.getLogger("app")

# Use the logger in your application
logger.info("Application started")

Gb = 1073741824
cache.setup("disk://", size_limit=16 * Gb)  # configure as in-memory cache

VOTES_FILE = "data/votes.jsonl"

HF_TOKEN = os.getenv("HF_TOKEN")

hf_api = HfApi(token=HF_TOKEN)
async_httpx_client = AsyncClient()

scheduler = CommitScheduler(
    repo_id="davanstrien/summary-ratings",
    repo_type="dataset",
    folder_path="data",
    path_in_repo="data",
    every=5,
    token=HF_TOKEN,
    hf_api=hf_api,
)


@asynccontextmanager
async def lifespan(app: FastAPI):
    logger.info("Running startup event")
    if not Path(VOTES_FILE).exists():
        path = hf_hub_download(
            repo_id="davanstrien/summary-ratings",
            filename="data/votes.jsonl",
            repo_type="dataset",
            token=HF_TOKEN,
            local_dir=".",
            local_dir_use_symlinks=False,
        )
        logger.info(f"Downloaded votes.jsonl to {path}")
        # pre-populate the cache
        datasets = list(list_datasets(sort="likes7d", direction=-1, limit=1000))
        dataset_ids = [dataset.id for dataset in datasets]
        # batch into 100s
        for i in range(0, len(dataset_ids), 10):
            # sleep for 1 second to avoid rate limiting
            await asyncio.sleep(1)
            await get_summaries(SummariesRequest(dataset_ids=dataset_ids[i : i + 10]))
    else:
        logger.info("Votes file already exists")
    yield


app = FastAPI(lifespan=lifespan)
# Configure CORS
# origins = [
#     "https://huggingface.co/*",
#     # "chrome-extension://deckahggoiaphiebdipfbiinmaihfpbk",  # Replace with your Chrome plugin ID
# ]


# # Configure CORS settings
# app.add_middleware(
#     CORSMiddleware,``
#     allow_origins=["https://huggingface.co/*"],  # Update with your frontend URL
#     allow_credentials=True,
#     allow_methods=["*"],
#     allow_headers=["*"],
# )


def save_vote(vote_entry):
    with scheduler.lock:
        with open(VOTES_FILE, "a") as file:
            date_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            vote_entry["timestamp"] = date_time
            file.write(json.dumps(vote_entry) + "\n")
        logger.info(f"Vote saved: {vote_entry}")


@app.get("/", include_in_schema=False)
def root():
    return RedirectResponse(url="/docs")


class Vote(BaseModel):
    dataset: str
    description: str
    vote: int = Field(..., ge=-1, le=1)
    userID: str


def validate_token(token: str = Header(None)) -> bool:
    try:
        whoami(token)
        return True
    except HTTPError:
        return False


@app.post("/vote")
async def receive_vote(
    vote: Vote,
    Authorization: Annotated[str, Header()],
    background_tasks: BackgroundTasks,
):
    if not validate_token(Authorization):
        logger.error("Invalid token")
        raise HTTPException(status_code=401, detail="Invalid token")
    vote_entry = {
        "dataset": vote.dataset,
        "vote": vote.vote,
        "description": vote.description,
        "userID": vote.userID,
    }
    # Append the vote entry to the JSONL file
    background_tasks.add_task(save_vote, vote_entry)
    return JSONResponse(content={"message": "Vote submitted successfully"})


def format_prompt(card: str) -> str:
    return f"""
Write a tl;dr summary of a dataset based on the dataset card. Focus on the most critical aspects of the dataset. 
The summary should aim to concisely describe the dataset.

CARD: \n\n{card[:6000]}
---

\n\nInstructions:
If the card provides the necessary information, say what the dataset can be used for.
You do not need to mention that the dataset is hosted or available on the Hugging Face Hub.
Do not mention the license of the dataset.
Do not mention the number of examples in the training or test split.
Only mention size if there is extensive discussion of the scale of the dataset in the dataset card.
Do not speculate on anything not explicitly mentioned in the dataset card.
In general avoid references to the quality of the dataset i.e. don't use phrases like 'a high-quality dataset' in the summary.

\n\nOne sentence summary:"""


async def check_when_dataset_last_modified(dataset_id: str) -> datetime | None:
    try:
        response = await async_httpx_client.get(
            f"https://huggingface.co/api/datasets/{dataset_id}"
        )
        if last_modified := response.json().get("lastModified"):
            return datetime.fromisoformat(last_modified)
        return None
    except Exception as e:
        logger.error(e)
        return None


@cache(ttl="48h", condition=NOT_NONE, key="predict:{dataset_id}")
async def predict(card: str, dataset_id: str) -> str | None:
    try:
        prompt = format_prompt(card)
        client = AsyncOpenAI(
            base_url="https://api-inference.huggingface.co/models/mistralai/Mixtral-8x7B-Instruct-v0.1/v1",
            api_key=HF_TOKEN,
        )

        chat_completion = await client.chat.completions.create(
            model="tgi",
            messages=[
                {"role": "user", "content": prompt},
            ],
            stream=False,
            tags=["tldr-summaries"],
        )
        return chat_completion.choices[0].message.content.strip()
    except Exception as e:
        logger.error(e)
        return None


@app.get("/summary")
async def get_summary(dataset_id: str) -> str | None:
    """
    Get a summary for a dataset based on the provided dataset ID.

    Args:
        dataset_id (str): The ID of the dataset to retrieve the summary for.

    Returns:
        str | None: The generated summary for the dataset, or None if no summary is available or an error occurs."""

    try:
        # dataset_id = request.dataset_id
        try:
            resp = await async_httpx_client.get(
                f"https://huggingface.co/datasets/{dataset_id}/raw/main/README.md",
                follow_redirects=True,
            )
            if resp.status_code != 200:
                return None
        except Exception as e:
            logger.error(e)
            return None
        card_text = resp.text
        card = DatasetCard(card_text)
        text = card.text
        parsed_text = parse_markdown(text)
        if is_empty_template(parsed_text):
            return None
        cache_key = f"predict:{dataset_id}"
        cached_data = await cache.get(cache_key)
        if cached_data is not None:
            cached_summary, cached_last_modified_time = cached_data
            # Get the current last modified time of the dataset
            current_last_modified_time = await check_when_dataset_last_modified(
                dataset_id
            )

            if (
                current_last_modified_time is None
                or cached_last_modified_time >= current_last_modified_time
            ):
                # Use the cached summary if the cached last modified time is greater than or equal to the current last modified time
                logger.info("Using cached summary")
                return cached_summary
        summary = await predict(parsed_text, dataset_id)
        current_last_modified_time = await check_when_dataset_last_modified(dataset_id)
        await cache.set(cache_key, (summary, current_last_modified_time))
        return summary
    except Exception as e:
        logger.error(e)
        return None


class SummariesRequest(BaseModel):
    dataset_ids: List[str]


@cache(ttl="1h", condition=NOT_NONE)
@app.post("/summaries")
async def get_summaries(request: SummariesRequest) -> dict:
    """
    Get summaries for a list of datasets based on the provided dataset IDs.

    Args:
        dataset_ids (List[str]): A list of dataset IDs to retrieve the summaries for.

    Returns:
        dict: A dictionary mapping dataset IDs to their corresponding summaries.
    """
    dataset_ids = request.dataset_ids

    async def get_summary_wrapper(dataset_id):
        return dataset_id, await get_summary(dataset_id)

    summary_tasks = [get_summary_wrapper(dataset_id) for dataset_id in dataset_ids]
    summaries = dict(await asyncio.gather(*summary_tasks))
    for dataset_id in dataset_ids:
        if summaries[dataset_id] is None:
            summaries[dataset_id] = "No summary available"
    return summaries