Spaces:
Paused
Paused
File size: 2,762 Bytes
8035330 253bbca 6ac4ade 8035330 419fdd6 3b22eeb 2d83410 becbbb8 2d83410 c892caa c0cfc84 253bbca bc828c5 419fdd6 8035330 bc828c5 6ac4ade 6ca460a 6ac4ade 419fdd6 6ac4ade 419fdd6 8035330 c0cfc84 3b22eeb becbbb8 3b22eeb 75141de 2d83410 75141de 8035330 3b22eeb becbbb8 75141de becbbb8 3b22eeb 75141de 2d83410 8035330 253bbca 8035330 419fdd6 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 |
import contextlib
import json
import os
from contextlib import asynccontextmanager
from datetime import datetime
from pathlib import Path
from typing import Annotated
from dotenv import load_dotenv
from fastapi import BackgroundTasks, FastAPI, Header, HTTPException
from fastapi.responses import JSONResponse
from huggingface_hub import CommitScheduler, HfApi, hf_hub_download, whoami
from huggingface_hub.utils._errors import HTTPError
from pydantic import BaseModel, Field
from starlette.responses import RedirectResponse
load_dotenv()
app = FastAPI()
VOTES_FILE = "votes/votes.jsonl"
HF_TOKEN = os.getenv("HF_TOKEN")
hf_api = HfApi(token=HF_TOKEN)
scheduler = CommitScheduler(
repo_id="davanstrien/votes",
repo_type="dataset",
folder_path="votes",
path_in_repo="data",
every=1,
token=HF_TOKEN,
hf_api=hf_api,
)
@asynccontextmanager
async def lifespan(app: FastAPI):
if not Path("votes").exists():
Path("votes").mkdir()
path = hf_hub_download(
"davanstrien/summary-ratings",
"data/votes.jsonl",
repo_type="dataset",
token=HF_TOKEN,
local_dir="votes",
local_dir_use_symlinks=False,
)
print(f"Downloaded votes to {path}")
yield
# # Configure CORS
# origins = [
# "https://huggingface.co",
# "chrome-extension://ogbhjlfpmjgjbjoiffagjogbhgaipopf", # Replace with your Chrome plugin ID
# ]
# app.add_middleware(
# CORSMiddleware,
# allow_origins=origins,
# allow_credentials=True,
# allow_methods=["POST"],
# allow_headers=["*"],
# )
def save_vote(vote_entry):
with open(VOTES_FILE, "a") as file:
date_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
vote_entry["timestamp"] = date_time
json.dump(vote_entry, file)
file.write("\n")
@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)):
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):
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"})
|