(self, session: AsyncSession, content_item: Dict)
| 134 | await self.add_content(session, content_item) |
| 135 | |
| 136 | async def add_content(self, session: AsyncSession, content_item: Dict): |
| 137 | add_ts = int(get_current_timestamp()) |
| 138 | last_modify_ts = int(get_current_timestamp()) |
| 139 | note = XhsNote( |
| 140 | creator_hash=content_item.get("creator_hash"), |
| 141 | nickname=content_item.get("nickname"), |
| 142 | add_ts=add_ts, |
| 143 | last_modify_ts=last_modify_ts, |
| 144 | note_id=content_item.get("note_id"), |
| 145 | type=content_item.get("type"), |
| 146 | title=content_item.get("title"), |
| 147 | desc=content_item.get("desc"), |
| 148 | video_url=content_item.get("video_url"), |
| 149 | time=content_item.get("time"), |
| 150 | last_update_time=content_item.get("last_update_time"), |
| 151 | liked_count=str(content_item.get("liked_count")), |
| 152 | collected_count=str(content_item.get("collected_count")), |
| 153 | comment_count=str(content_item.get("comment_count")), |
| 154 | share_count=str(content_item.get("share_count")), |
| 155 | image_list=json.dumps(content_item.get("image_list")), |
| 156 | tag_list=json.dumps(content_item.get("tag_list")), |
| 157 | note_url=content_item.get("note_url"), |
| 158 | source_keyword=content_item.get("source_keyword", ""), |
| 159 | xsec_token=content_item.get("xsec_token", "") |
| 160 | ) |
| 161 | session.add(note) |
| 162 | |
| 163 | async def update_content(self, session: AsyncSession, content_item: Dict): |
| 164 | note_id = content_item.get("note_id") |
no test coverage detected