Spaces:
Paused
Paused
File size: 2,490 Bytes
8035330 253bbca 6ac4ade 8035330 3b22eeb 6d8e1bf 3b22eeb becbbb8 3b22eeb 6d8e1bf c892caa c0cfc84 253bbca bc828c5 8035330 bc828c5 6ac4ade 6ca460a 6ac4ade 8035330 3b354b4 8035330 c0cfc84 3b22eeb becbbb8 3b22eeb 8035330 3b22eeb becbbb8 3b22eeb 8035330 253bbca 8035330 becbbb8 3b354b4 becbbb8 |
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 |
import contextlib
import json
import os
from contextlib import asynccontextmanager
from datetime import datetime
from pathlib import Path
from dotenv import load_dotenv
from fastapi import BackgroundTasks, FastAPI, Header
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from huggingface_hub import CommitScheduler, HfApi, whoami
from huggingface_hub.utils._errors import HTTPError
from pydantic import BaseModel, Field
from starlette.responses import RedirectResponse
load_dotenv()
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):
Path("votes").mkdir(exist_ok=True)
yield
app = FastAPI()
VOTES_FILE = "votes/votes.jsonl"
# 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:
# add time stamp to the vote entry
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
@app.post("/vote")
async def receive_vote(
vote: Vote,
background_tasks: BackgroundTasks,
):
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)
response = JSONResponse(content={"message": "Vote submitted successfully"})
response.headers[
"Access-Control-Allow-Origin"
] = "chrome-extension://ogbhjlfpmjgjbjoiffagjogbhgaipopf" # Replace with your Chrome plugin ID
response.headers["Access-Control-Allow-Methods"] = "POST"
response.headers["Access-Control-Allow-Headers"] = "*"
return response
|