msg.py 1.7 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162
  1. from typing import Optional
  2. from collections import namedtuple
  3. from datetime import datetime
  4. from sql.msg import read_msg_list, get_msg_count, create_msg, read_msg, get_user_msg_count, delete_msg
  5. import object.user
  6. class _Message:
  7. message_tuple = namedtuple("Message", "email content update_time secret")
  8. @staticmethod
  9. def get_message_list(limit: Optional[int] = None, offset: Optional[int] = None, show_secret: bool = False):
  10. ret = []
  11. for i in read_msg_list(limit=limit, offset=offset, show_secret=show_secret):
  12. ret.append(Message(i))
  13. return ret
  14. @staticmethod
  15. def get_msg_count(auth: "object.user.User" = None):
  16. if auth is None:
  17. return get_msg_count()
  18. return get_user_msg_count(auth.id)
  19. @staticmethod
  20. def create(auth: "object.user.User", content, secret: bool = False):
  21. ret = create_msg(auth.id, content, secret)
  22. if ret is not None:
  23. return Message(ret)
  24. return None
  25. class Message(_Message):
  26. def __init__(self, msg_id):
  27. self.id = msg_id
  28. @property
  29. def info(self):
  30. return Message.message_tuple(*read_msg(self.id))
  31. @property
  32. def auth(self):
  33. return object.user.User(self.info.email)
  34. @property
  35. def content(self):
  36. return self.info.content
  37. @property
  38. def update_time(self):
  39. return datetime.utcfromtimestamp(datetime.timestamp(self.info.update_time))
  40. @property
  41. def secret(self):
  42. return self.info.secret
  43. @property
  44. def is_delete(self):
  45. return not self.auth.is_authenticated and len(self.content) != 0
  46. def delete(self):
  47. return delete_msg(self.id)