view_model.py 7.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288
  1. """
  2. Implementation of the syndication taxonomy in architecture.md
  3. """
  4. from dataclasses import dataclass, asdict, replace
  5. from typing import List, Dict, Optional, Tuple, Union
  6. import re
  7. @dataclass
  8. class ContentId:
  9. prefix: str
  10. id: str
  11. def __str__ (self):
  12. return self.prefix + self.id
  13. @staticmethod
  14. def parse (content_id, prefix, id_pattern = '(.+)'):
  15. raw_id = content_id[len(prefix):]
  16. id_matches = re.fullmatch(id_pattern, raw_id)
  17. if id_matches:
  18. raw_id = id_matches[0]
  19. return ContentId(prefix=prefix, id=raw_id)
  20. @dataclass
  21. class PublicMetrics:
  22. reply_count: Optional[int] = None
  23. quote_count: Optional[int] = None
  24. retweet_count: Optional[int] = None
  25. like_count: Optional[int] = None
  26. impression_count: Optional[int] = None
  27. bookmark_count: Optional[int] = None
  28. # may be video only
  29. view_count: Optional[int] = None
  30. def items (self):
  31. return asdict(self).items()
  32. @dataclass
  33. class NonPublicMetrics:
  34. impression_count: Optional[int] = None
  35. user_profile_clicks: Optional[int] = None
  36. url_link_clicks: Optional[int] = None
  37. def items (self):
  38. return asdict(self).items()
  39. @dataclass
  40. class MediaItem:
  41. type: str
  42. preview_image_url: str
  43. media_key: Optional[str] = None
  44. image_url: Optional[str] = None
  45. url: Optional[str] = None
  46. content_type: Optional[str] = None
  47. duration_ms: Optional[int] = None
  48. height: Optional[int] = None
  49. width: Optional[int] = None
  50. size: Optional[int] = None
  51. public_metrics: Optional[PublicMetrics] = None
  52. @dataclass
  53. class Card:
  54. display_url: Optional[str] = None
  55. source_url: Optional[str] = None
  56. content: Optional[str] = None
  57. title: Optional[str] = None
  58. preview_image_url: Optional[str] = None
  59. image_url: Optional[str] = None
  60. @dataclass
  61. class FeedItemAction:
  62. route: str
  63. route_params: Dict
  64. # intended as external resources,
  65. # not first class display.
  66. # PDF, chat, embedded image (iframe)
  67. @dataclass
  68. class FeedItemAttachment:
  69. name: str
  70. content_type: str
  71. url: Optional[str] = None
  72. content: Optional[object] = None
  73. size: Optional[int] = None
  74. @dataclass
  75. class UnrepliedSection:
  76. feed_item_id: Optional[str] = None
  77. description: Optional[str] = None
  78. text: Optional[str] = None
  79. span: Optional[Tuple[int, int]] = None
  80. def text_from (self, feed_item: 'FeedItem') -> str:
  81. if not self.span or not feed_item.text:
  82. return
  83. return feed_item.text[self.span[0]:self.span[1]]
  84. ReplyingToSection = UnrepliedSection
  85. @dataclass
  86. class FeedItem:
  87. id: str
  88. created_at: str
  89. display_name: str
  90. handle: str
  91. published_by: Optional['FeedServiceUser'] = None
  92. title: Optional[str] = None
  93. text: Optional[str] = None
  94. html: Optional[str] = None
  95. author_is_verified: Optional[bool] = None
  96. url: Optional[str] = None
  97. conversation_id: Optional[str] = None
  98. avi_icon_url: Optional[str] = None
  99. author_url: Optional[str] = None
  100. author_id: Optional[str] = None
  101. source_url: Optional[str] = None
  102. source_author_url: Optional[str] = None
  103. reply_depth: Optional[int] = 0
  104. is_marked: Optional[bool] = None
  105. is_bookmarked: Optional[bool] = None
  106. card: Optional[Card] = None
  107. public_metrics: Optional[PublicMetrics] = None
  108. non_public_metrics: Optional[NonPublicMetrics] = None
  109. retweeted_tweet_id: Optional[str] = None
  110. source_retweeted_by_url: Optional[str] = None
  111. retweeted_by: Optional[str] = None
  112. retweeted_by_url: Optional[str] = None
  113. videos: Optional[List[MediaItem]] = None
  114. photos: Optional[List[MediaItem]] = None
  115. quoted_tweet_id: Optional[str] = None
  116. quoted_tweet: Optional['FeedItem'] = None
  117. replied_tweet_id: Optional[str] = None
  118. replied_tweet: Optional['FeedItem'] = None
  119. note: Optional[str] = None
  120. debug_source_data: Optional[Dict] = None
  121. attachments: Optional[List[FeedItemAttachment]] = None
  122. # At some point we may move to feed_item_actions when the set is known
  123. actions: Optional[Dict[str, FeedItemAction]] = None
  124. # This is a TBD concept to highlight unreplied parts of a message.
  125. unreplied: Optional[List[UnrepliedSection]] = None
  126. # This is a TBD concept to highlight parts of a message that are a reply.
  127. replying_to: Optional[List[ReplyingToSection]] = None
  128. is_viewed: Optional[bool] = None
  129. # tm = FeedItem(id="1", text="aa", created_at="fs", display_name="fda", handle="fdsafas")
  130. @dataclass
  131. class ThreadItem:
  132. feed_item: Union[FeedItem,'Collection','RoutedMessage']
  133. children: Optional[List['ThreadItem']] = None
  134. parent: Optional['ThreadItem'] = None
  135. parents: Optional[List['ThreadItem']] = None
  136. actions: Optional[Dict[str, FeedItemAction]] = None
  137. @dataclass
  138. class FeedServiceUser:
  139. id: str
  140. url: str
  141. name: str # display_name
  142. username: str # handle
  143. is_verified: Optional[bool] = None
  144. is_protected: Optional[bool] = None
  145. created_at: Optional[str] = None
  146. description: Optional[str] = None
  147. preview_image_url: Optional[str] = None # deprecated... rename to avatar_image_url
  148. poster_image_url: Optional[str] = None
  149. website: Optional[str] = None
  150. location: Optional[str] = None
  151. actions: Optional[Dict[str, FeedItemAction]] = None
  152. source_url: Optional[str] = None
  153. raw_user: Optional = None
  154. @property
  155. def display_name (self) -> str:
  156. return self.name
  157. @property
  158. def handle (self) -> str:
  159. return self.username
  160. @property
  161. def avatar_image_url (self) -> str:
  162. return self.preview_image_url
  163. @dataclass
  164. class Collection:
  165. """
  166. Works for both collections and lists for now
  167. """
  168. id:str
  169. name: str
  170. description: Optional[str] = None
  171. preview_image_url: Optional[str] = None
  172. created_at: Optional[str] = None
  173. updated_at: Optional[str] = None
  174. owner_id: Optional[str] = None
  175. owner: Optional[FeedServiceUser] = None
  176. url: Optional[str] = None
  177. total_count: Optional[int] = None
  178. current_page: Optional['CollectionPage'] = None
  179. @dataclass
  180. class CollectionItem:
  181. item: List[Union[FeedServiceUser,FeedItem,ThreadItem,Collection]] = None
  182. sort_order: Optional[int] = None
  183. after_id: Optional[str] = None
  184. @dataclass
  185. class CollectionPage:
  186. """
  187. Works for Collections and Lists for now, as well as Threads and nested Collections.
  188. Feed is a Collection.
  189. """
  190. id: str
  191. items: Optional[List[Union[FeedServiceUser,FeedItem,ThreadItem,CollectionItem,Collection]]] = None
  192. next_token: Optional[str] = None
  193. last_dt: Optional[str] = None
  194. total_count: Optional[int] = None
  195. includes: Optional[Dict] = None
  196. def cleandict(d):
  197. if isinstance(d, dict):
  198. return {k: cleandict(v) for k, v in d.items() if v is not None}
  199. elif isinstance(d, list):
  200. return [cleandict(v) for v in d]
  201. else:
  202. return d