This repository has been archived on 2024-10-10. You can view files and clone it, but cannot push or open issues or pull requests.
TelegramBot/classes/importer/csv.py
2023-11-05 13:20:01 +00:00

65 lines
2.1 KiB
Python

from codecs import decode
from csv import DictReader
from datetime import datetime
from typing import Any, Dict, List, Union
from bson import ObjectId
from classes.importer.abstract import Importer
from modules.database_api import col_entries
class ImporterCSV(Importer):
"""
The ImporterCSV class represents the object with
functionality to import/export garbage collection
records and convert them to other object types
from CSV files.
"""
def __init__(self):
super(Importer, self).__init__()
async def import_data(self, data: bytes) -> List[ObjectId]:
entries: List[Dict[str, Any]] = list(
DictReader(decode(data).split("\n"), delimiter=";")
)
for entry in entries:
entry["locations"] = (
[int(entry["locations"])]
if "," not in entry["locations"]
else [int(id) for id in entry["locations"].split(",")]
)
entry["garbage_type"] = int(entry["garbage_type"])
for key in ("locations", "garbage_type", "date"):
if (
key not in entry
or (key == "garbage_type" and not isinstance(entry[key], int))
or (key == "locations" and not isinstance(entry[key], list))
):
raise ValueError
if key == "date":
try:
datetime.fromisoformat(str(entry[key]))
except (ValueError, TypeError) as exc:
raise ValueError from exc
entries_clean: List[Dict[str, Union[str, int, datetime]]] = [
{
"locations": entry["locations"],
"garbage_type": entry["garbage_type"],
"date": datetime.fromisoformat(str(entry["date"])),
}
for entry in entries
]
inserted = await col_entries.insert_many(entries_clean)
return [] if inserted is None else inserted.inserted_ids
async def export_data(self, data: Any) -> Any:
return None