eriks f551cd71d4 Update service/app.py
Signed-off-by: eriks <eriks@git.72.lv>
2024-01-14 11:01:08 +02:00

124 lines
3.6 KiB
Python

import datetime
import json
import uuid
from collections import defaultdict
from io import BytesIO
from typing import Annotated
from fastapi import FastAPI, Form, HTTPException, responses
from icalendar import Alarm, Calendar, Event
from pydantic import BaseModel
from unidecode import unidecode
from uvicorn.workers import UvicornWorker
def generate_ical_for_mapping(cal: dict[datetime.date, list[str]]) -> BytesIO:
ical = Calendar()
ical["VERSION"] = "2.0"
ical["PRODID"] = "NameDays"
for date, names in sorted(cal.items(), key=lambda x: x[0]):
ev = Event()
ev.add("SUMMARY", ", ".join(sorted(names)))
ev.add("DTSTART", date)
ev.add("DTEND", date + datetime.timedelta(days=1))
ev.add("DTSTAMP", datetime.datetime(2000, 1, 1))
ev.add("RRULE", {"FREQ": "YEARLY"})
ev.add("CATEGORY", "Anniversary")
ev.add("UID", uuid.uuid5(uuid.NAMESPACE_OID, date.strftime("%m-%d")))
alert = Alarm()
alert.add("action", "DISPLAY")
alert.add("TRIGGER", datetime.timedelta(hours=9))
alert.add("DESCRIPTION", "Default description")
ev.add_component(alert)
ical.add_component(ev)
return BytesIO(ical.to_ical(True))
def starts_with(string_to_check: str, check_string: str) -> bool:
value = unidecode(string_to_check.lower(), errors="preserve")
query = unidecode(check_string.lower(), errors="preserve")
return value.startswith(query)
with open("service/mapping.json") as f:
MAPPING = json.load(f)
with open("service/vardadienas.json") as f:
NAMEDAYS = json.load(f)
LV_MONTHS = {
1: "jan",
2: "feb",
3: "mar",
4: "apr",
5: "mai",
6: "jūn",
7: "jūl",
8: "aug",
9: "sep",
10: "okt",
11: "nov",
12: "dec",
}
class SearchResult(BaseModel):
text: str
id: str
class SearchResultSection(BaseModel):
text: str
children: list[SearchResult]
class SearchResponse(BaseModel):
results: list[SearchResultSection]
pagination: dict[str, bool] = {"more": False}
app = FastAPI()
@app.get("/", response_class=responses.HTMLResponse)
async def index_html():
with open("assets/index.html") as f:
return responses.HTMLResponse(f.read(), 201)
@app.get("/api/search")
async def search_words(term: str) -> SearchResponse:
result_map = {}
for section, names in MAPPING.items():
result_map[section] = []
for key, value in names.items():
if starts_with(value, term):
result_map[section].append(
SearchResult(id=key, text=f"{value} ({key.split('__')[1]}. {LV_MONTHS[int(key.split('__')[0])]}.)")
)
return SearchResponse(
results=[
SearchResultSection(text=section.title(), children=results)
for section, results in result_map.items()
if results
]
)
@app.post("/api/download", response_class=responses.StreamingResponse)
async def download_ical(words: Annotated[list[str], Form()]):
cal = defaultdict(list)
for selected_name in words:
month, day, name = selected_name.split("__")
vdmd = NAMEDAYS[str(int(month))][str(int(day))]
if name in vdmd["normal"] or name in vdmd["special"]:
date = datetime.date(2000, int(month), int(day))
cal[date].append(name)
if cal:
return responses.StreamingResponse(
content=generate_ical_for_mapping(cal),
media_type="text/calendar",
headers={"Content-Disposition": f'attachment; filename="{uuid.uuid4().hex}.ics"'},
)
raise HTTPException(404, "No names have been found!")