Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- from pymongo import MongoClient
- from utils import *
- from files.config import *
- import logging
- logging.basicConfig(
- filename="logs/{}.log".format(get_time()),
- filemode="w",
- level="DEBUG",
- format="[%(asctime)s %(levelname)s] %(message)s"
- )
- log = logging.getLogger("VKBot")
- class Database:
- DATE: str = get_time(False)
- __URL: str = "mongodb+srv://test:test123@cluster0-vzzgu.mongodb.net/test?retryWrites=true&w=majority"
- def __init__(self):
- self.con: MongoClient = MongoClient(self.__URL)
- self.db = self.con["support"]
- self.col = self.db["stats"]
- log.info("Connected to Database!")
- self.__load()
- if self.DATE not in stats.keys():
- self.update_date()
- stats[self.DATE]['messages'] += 10
- def __load(self):
- """ @Protected method.
- Loads all data from collection.
- :return: -> None
- """
- for i in self.col.find({}).sort([("_id", -1)]):
- stats.update({i["_id"]: {
- **i
- }})
- stats[i["_id"]].pop("_id")
- def update_stats(self):
- """ Updates collection stats
- every 5 minutes.
- :return:
- """
- self.col.update(
- {"_id": self.DATE},
- {"$set": {**stats[self.DATE]}}
- )
- def update_date(self):
- """ Inserts new date
- every 00:00 (GMT+3)
- :return:
- """
- dialogs.clear()
- self.DATE = get_time(False)
- query = {
- '_id': self.DATE,
- 'messages': 0,
- 'symbols': 0,
- 'dialogs': 0,
- 'photo': 0,
- 'video': 0,
- 'stickers': 0,
- 'buttons': 0,
- 'smiles': 0,
- 'swears': 0
- }
- self.col.insert(query)
- stats.update({self.DATE: {
- **query
- }})
- stats[self.DATE].pop("_id")
- log.info(f"Updated collection. Added date: {self.DATE}")
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement