Spaces:
Paused
Paused
File size: 2,809 Bytes
8035330 253bbca 6ac4ade 8035330 3b22eeb 6d8e1bf 3b22eeb becbbb8 3b22eeb 6d8e1bf c892caa c0cfc84 75141de 253bbca bc828c5 8035330 bc828c5 6ac4ade 6ca460a 6ac4ade 8035330 3b354b4 8035330 c0cfc84 3b22eeb becbbb8 3b22eeb 75141de 8035330 3b22eeb becbbb8 75141de becbbb8 3b22eeb 75141de 8035330 253bbca 8035330 becbbb8 0724f28 becbbb8 75141de 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 100 101 102 103 104 105 106 107 108 109 |
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
from typing import Annotated
from fastapi import FastAPI, Header
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
def validate_token(token: str = Header(None)):
return token == "ABCD"
@app.post("/vote")
async def receive_vote(
vote: Vote,
Authorization: Annotated[str, Header()],
background_tasks: BackgroundTasks,
):
if not validate_token(Authorization):
return JSONResponse(status_code=401, content={"message": "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)
response = JSONResponse(content={"message": "Vote submitted successfully"})
response.headers[
"Access-Control-Allow-Origin"
] = "chrome-extension://ogbhjlfpmjgjbjoiffagjogbhgaipopf" # TODO Replace with your Chrome plugin ID
response.headers["Access-Control-Allow-Methods"] = "POST"
response.headers["Access-Contr`ol-Allow-Headers"] = "*"
return response
|