datasources.vk.search_vk

  1"""
  2VK keyword search
  3"""
  4import datetime
  5
  6import vk_api
  7
  8from backend.lib.search import Search
  9from common.lib.exceptions import QueryParametersException, ProcessorInterruptedException, ProcessorException
 10from common.lib.helpers import UserInput
 11from common.lib.item_mapping import MappedItem
 12
 13
 14class SearchVK(Search):
 15    """
 16    Get posts via the VK API
 17    """
 18    type = "vk-search"  # job ID
 19    title = "VK"
 20    extension = "ndjson"
 21    is_local = False    # Whether this datasource is locally scraped
 22    is_static = False   # Whether this datasource is still updated
 23
 24    previous_request = 0
 25    import_issues = True
 26
 27    references = [
 28        "[VK API documentation](https://vk.com/dev/first_guide)",
 29        "[Python API wrapper](https://github.com/python273/vk_api)"
 30    ]
 31    
 32    expanded_profile_fields = "id,screen_name,first_name,last_name,name,deactivated,is_closed,is_admin,sex,city,country,photo_200,photo_100,photo_50,followers_count,members_count"  # https://vk.com/dev/objects/user & https://vk.com/dev/objects/group
 33
 34    @classmethod
 35    def get_options(cls, parent_dataset=None, config=None):
 36        """
 37        Get VK data source options
 38
 39        :param config:
 40        :param parent_dataset:  Should always be None
 41        :return dict:  Data source options
 42        """
 43
 44        intro_text = ("This data source uses VK's [API](https://vk.com/dev/first_guide) and a python "
 45                      "[wrapper](https://github.com/python273/vk_api) to request information from VK using your "
 46                      "username and password.")
 47
 48        options = {
 49            "intro-1": {
 50                "type": UserInput.OPTION_INFO,
 51                "help": intro_text
 52            },
 53            "query_type": {
 54                "type": UserInput.OPTION_CHOICE,
 55                "help": "Query Type",
 56                "options": {
 57                    "newsfeed": "News Feed search",
 58                },
 59                "default": "newsfeed"
 60            },
 61            "intro-2": {
 62                "type": UserInput.OPTION_INFO,
 63                "help": "Your username and password will be deleted after your query is complete."
 64            },
 65            "username": {
 66                "type": UserInput.OPTION_TEXT,
 67                "sensitive": True,
 68                "cache": True,
 69                "help": "VK Username"
 70            },
 71            "password": {
 72                "type": UserInput.OPTION_TEXT,
 73                "sensitive": True,
 74                "cache": True,
 75                "help": "VK Password"
 76            },
 77            "intro-3": {
 78                "type": UserInput.OPTION_INFO,
 79                "help": "Enter the text to search for below."
 80            },
 81            "query": {
 82                "type": UserInput.OPTION_TEXT_LARGE,
 83                "help": "Query"
 84            },
 85            "amount": {
 86                "type": UserInput.OPTION_TEXT,
 87                "help": "Max items to retrieve",
 88                "min": 0,
 89                "max": 1000,
 90                "default": 100
 91            },
 92            "include_comments": {
 93                "type": UserInput.OPTION_TOGGLE,
 94                "help": "Include post comments",
 95                "default": False,
 96                "tooltip": ""
 97            },
 98            "divider-2": {
 99                "type": UserInput.OPTION_DIVIDER
100            },
101            "daterange-info": {
102                "type": UserInput.OPTION_INFO,
103                "help": "VK daterange defaults vary by type of query. For the News Feed, posts are returned starting "
104                        "with the most recent and working backwards."
105            },
106            "daterange": {
107                "type": UserInput.OPTION_DATERANGE,
108                "help": "Date range"
109            },
110        }
111
112        return options
113
114    def get_items(self, query):
115        """
116        Use the VK API
117
118        :param query:
119        :return:
120        """
121        if self.parameters.get("username") is None or self.parameters.get("password") is None:
122            self.dataset.update_status(
123                "VK query failed or was interrupted; please create new query in order to provide username and password again.",
124                is_final=True)
125            return []
126
127        self.dataset.update_status("Logging in to VK")
128        try:
129            vk_session = self.login(self.parameters.get("username"), self.parameters.get("password"))
130        except vk_api.exceptions.AuthError as e:
131            self.log.warning(f"VK Auth Issues: {e}")
132            self.dataset.update_status(f"VK unable to authorize user: {e}", is_final=True)
133            return []
134
135        query_type = self.parameters.get("query_type")
136        query = self.parameters.get("query")
137        include_comments = self.parameters.get("include_comments", False)
138
139        if query_type == "newsfeed":
140            query_parameters = {"query": query,
141                                "max_amount": self.parameters.get("amount")}
142
143            # Add start and end dates if provided
144            if self.parameters.get("min_date"):
145                query_parameters['start_time'] = self.parameters.get("min_date")
146            if self.parameters.get("max_date"):
147                query_parameters['end_time'] = self.parameters.get("max_date")
148
149            vk_helper = vk_session.get_api()
150
151            # Collect Newsfeed results
152            num_results = 0
153            self.dataset.update_status("Submitting query...")
154            for i, result_batch in enumerate(self.search_newsfeed(vk_helper, **query_parameters)):
155                if self.interrupted:
156                    raise ProcessorInterruptedException("Interrupted while fetching newsfeed data from the VK API")
157
158                self.dataset.update_status(f"Processing results batch {i+1}")
159                for result in result_batch:
160                    result.update({'4cat_item_type': 'post'})
161                    yield result
162                    num_results += 1
163
164                    if include_comments:
165                        for comment in self.collect_all_comments(vk_helper, owner_id=result.get("owner_id"), post_id=result.get("id")):
166                            comment.update({'4cat_item_type': 'comment'})
167                            yield comment
168
169                    self.dataset.update_status(f"Received {num_results} results of max {self.parameters.get('amount')} from the VK API")
170                    self.dataset.update_progress(num_results / self.parameters.get('amount'))
171
172    def login(self, username, password):
173        """
174        Login and authenticate user
175        """
176        vk_session = vk_api.VkApi(username,
177                                  password,
178                                  config_filename=self.config.get("PATH_SESSIONS").joinpath(username+"-vk_config.json"))
179        vk_session.auth()
180
181        return vk_session
182
183    def search_newsfeed(self, vk_helper, query, max_amount, num_collected=0, start_time=None, end_time=None, **kwargs):
184        """
185        Collects all newsfeed posts
186
187        :param Object vk_helper:    Authorized vk_api.VkApi
188        :param str query:           String representing the search query
189        :param int max_amount:      Max number of posts to collect
190        :param int num_collected:   Number of previously collected results
191        :param int start_time:      Timestamp for earliest post
192        :param int end_time:        Timestamp for latest post
193        :return generator:          Yields groups of posts
194        """
195        remaining = max_amount - num_collected
196        parameters = {
197            "q": query,
198            "extended": 1,
199            "count": remaining if remaining < 200 else 200,
200            "fields": self.expanded_profile_fields,
201        }
202        if start_time:
203            parameters["start_time"] = start_time
204        if end_time:
205            parameters["end_time"] = end_time
206
207        response = vk_helper.newsfeed.search(**parameters)
208        news_feed_results = response.get("items", [])
209        num_collected = num_collected + len(news_feed_results)
210
211        # Flesh out profiles and groups
212        author_profiles = self.expand_profile_fields({"profiles": response.get("profiles", []), "groups": response.get("groups", [])})
213        [result.update({"author_profile": author_profiles.get(result.get("from_id"), {})}) for result in news_feed_results]
214
215        yield news_feed_results
216
217        # Collect additional results
218        if response.get("next_from") and num_collected < max_amount:
219            parameters.update({"start_from": response.get("next_from")})
220            for additional_results in self.search_newsfeed(vk_helper, query, max_amount, num_collected=num_collected, **parameters):
221                yield additional_results
222
223    def collect_all_comments(self, vk_helper, owner_id, post_id):
224        """
225        Collects all comments and replies to a VK post
226
227        :param Object vk_helper:           Authorized vk_api.VkApi
228        :param int owner_id:            Owner ID provided by post/comment/etc
229        :param int post_id:             ID of post from which to collect comments
230        :return generator:              Yields comments and replies
231        """
232        # Collect top level comments from post
233        comments = self.get_comments(vk_helper, owner_id, post_id=post_id)
234
235        # Extract replies and collect more if needed
236        for comment in comments:
237            yield comment
238
239            reply_count = comment.get("thread", {}).get("count", 0)
240            replies = comment.get("thread", {}).get("items", [])
241            if reply_count > 10 and len(replies) == 10:
242                # Collect additional replies
243                replies += self.get_comments(vk_helper, owner_id, comment_id=comment.get("id"), last_collected_id=replies[-1].get("id"))[1:]
244
245            for reply in replies:
246                yield reply
247                if reply.get("thread"):
248                    self.log.warning("VK Datasource issue with replies: additional depth needs to be handled; contact 4CAT devs")
249                    # TODO: this will need modification if reply threads gain depth
250
251    def get_comments(self, vk_helper, owner_id, post_id=None, comment_id=None, last_collected_id=None, **kwargs):
252        """
253        Collect comments from either a post or another comment (i.e., replies to another comment). Must provide either
254        post_id or comment_id, but not both.
255
256        More information can be found here:
257        https://vk.com/dev/wall.getComments
258
259        :param Object vk_helper:       Authorized vk_api.VkApi
260        :param int owner_id:            Owner ID provided by post/comment/etc
261        :param int post_id:             ID of post from which to collect comments
262        :param int comment_id:          ID of comment from which to collect comments
263        :param int last_collected_id:   ID of the last comment to collected; used as start to continue collecting comments
264        :return list:                   List of comments
265        """
266        if self.interrupted:
267            raise ProcessorInterruptedException("Interrupted while fetching comments from the VK API")
268
269        if post_id is None and comment_id is None:
270            raise ProcessorException("Must provide either post_id or comment_id to collect comments from VK")
271
272        parameters = {
273            "owner_id": owner_id,
274            "need_likes": 1,
275            "preview_length": 0,
276            "extended": 1,
277            "count": 100,
278            "thread_items_count": 10,
279            "fields": self.expanded_profile_fields,
280        }
281        if post_id:
282            parameters.update({"post_id": post_id})
283        if comment_id:
284            parameters.update({"comment_id": comment_id})
285        if last_collected_id:
286            parameters.update({"start_comment_id": last_collected_id})
287
288        # Collect comments from VK
289        try:
290            response = vk_helper.wall.getComments(**parameters)
291        except vk_api.exceptions.ApiError as e:
292            self.dataset.log(f"Unable to collect comments for owner_id {owner_id} and {'post_id' if post_id is not None else 'comment_id'} {post_id if post_id is not None else comment_id}: {e}")
293            return []
294        comments = response.get("items", [])
295
296        # Flesh out profiles and groups
297        author_profiles = self.expand_profile_fields({"profiles": response.get("profiles", []), "groups": response.get("groups", [])})
298        [comment.update({"author_profile": author_profiles.get(comment.get("from_id"), {})}) for comment in comments]
299        # Also expand replies
300        [reply.update({"author_profile": author_profiles.get(reply.get("from_id"), {})}) for replies in [comment.get("thread", {}).get("items", []) for comment in comments if comment.get("thread")] for reply in replies]
301
302        # Check if there are potentially additional comments
303        if response.get("count") > 100 and len(comments) == 100:
304            # Update params with last collected comment
305            parameters.update({"start_comment_id": comments[-1].get("id")})
306            # Collect additional comments from VK and remove first comment (which is duplicate)
307            comments += self.get_comments(vk_helper=vk_helper, **parameters)[1:]
308
309        return comments
310
311    @ staticmethod
312    def expand_profile_fields(dict_of_profile_types):
313        """
314        Combine various VK profile and group author information for easy lookup. Add 4CAT_author_profile_type field to
315        differentiate source of data later.
316        """
317        author_types = {}
318        for profile_type, profiles in dict_of_profile_types.items():
319            for profile in profiles:
320                if "id" not in profile:
321                    raise ProcessorException("Profile missing id field; VK data format incorrect/changed")
322                elif profile.get("id") in author_types:
323                    raise ProcessorException("Profile id duplicated across profile types; unable to combine profiles")
324                profile.update({"4CAT_author_profile_type": profile_type})
325                author_types[profile.get("id")] = profile
326        return author_types
327
328    @staticmethod
329    def validate_query(query, request, config):
330        """
331        Validate input for a dataset query on the VK data source.
332
333        Will raise a QueryParametersException if invalid parameters are
334        encountered. Parameters are additionally sanitised.
335
336        :param dict query:  Query parameters, from client-side.
337        :param request:  Flask request
338        :param ConfigManager|None config:  Configuration reader (context-aware)
339        :return dict:  Safe query parameters
340        """
341        # Please provide something...
342        if not query.get("query", None):
343            raise QueryParametersException("Please provide a query.")
344
345        # the dates need to make sense as a range to search within
346        # but, on VK, you can also specify before *or* after only
347        after, before = query.get("daterange")
348        if before and after and before < after:
349            raise QueryParametersException("Date range must start before it ends")
350
351        # TODO: test username and password?
352
353        # if we made it this far, the query can be executed
354        params = {
355            "query":  query.get("query"),
356            "query_type": query.get("query_type"),
357            "amount": query.get("amount"),
358            "include_comments": query.get("include_comments"),
359            "min_date": after,
360            "max_date": before,
361            "username": query.get("username"),
362            "password": query.get("password"),
363        }
364
365        return params
366
367    @staticmethod
368    def map_item(item):
369        """
370        Map a nested VK object to a flat dictionary
371
372        :param item:  VK object as originally returned by the VK API
373        :return dict:  Dictionary in the format expected by 4CAT
374        """
375        vk_item_time = datetime.datetime.fromtimestamp(item.get('date'))
376
377        # Process attachments
378        photos = []
379        videos = []
380        audio = []
381        links = []
382        docs = []
383        for attachment in item.get("attachments", []):
384            attachment_type = attachment.get("type")
385            attachment = attachment.get(attachment_type)
386            if attachment_type == "photo":
387                if attachment.get("sizes"):
388                    photos.append(sorted(attachment.get("sizes"), key=lambda d: d['width'], reverse=True)[0].get('url'))
389                else:
390                    photos.append(str(attachment))
391            elif attachment_type == "video":
392                # TODO: can I get the actual URL? Does not seem like it...
393                videos.append(f"https://vk.com/video{attachment.get('owner_id')}_{attachment.get('id')}")
394            elif attachment_type == "audio":
395                # TODO: Seem unable to create the URL with provided information...
396                audio.append(f"{attachment.get('artist')} - {attachment.get('title')}")
397            elif attachment_type == "link":
398                links.append(attachment.get('url', str(attachment)))
399            elif attachment_type == "doc":
400                docs.append(attachment.get('url', str(attachment)))
401
402        # Use 4cat_item_type to populate different fields
403        tread_id = ""
404        in_reply_to_user = ""
405        in_reply_to_comment_id = ""
406        if item.get("4cat_item_type") == "post":
407            tread_id = item.get("id")
408        elif item.get("4cat_item_type") == "comment":
409            tread_id = item.get("post_id")
410            in_reply_to_user = item.get("reply_to_user")
411            in_reply_to_comment_id = item.get("reply_to_comment")
412
413        author_profile = item.get("author_profile", {})
414        profile_source = "user" if author_profile.get("4CAT_author_profile_type") == "profile" else "community" if author_profile.get("4CAT_author_profile_type") == "group" else "N/A"
415        # Use source of author profile if "type" not present (e.g., in users profiles do not seem to have type)
416        author_type = author_profile.get("type", profile_source)
417
418        return MappedItem({
419            "id": item.get("id"),
420            "thread_id": tread_id,
421            "timestamp": vk_item_time.strftime("%Y-%m-%d %H:%M:%S"),
422            "unix_timestamp": int(vk_item_time.timestamp()),
423            "link": f"https://vk.com/wall{item.get('owner_id')}_{item.get('id')}",
424            "item_type": item.get("4cat_item_type"),
425            "body": item.get("text"),
426            "author_id": item.get("from_id"),
427            "author_type": author_type,
428            "author_screen_name": author_profile.get("screen_name"),
429            "author_name": author_profile.get("name", " ".join([author_profile.get("first_name", ""), author_profile.get("last_name", "")])),
430            "author_sex": "F" if author_profile.get("sex") == 1 else "M" if author_profile.get("sex") == 2 else "Not Specified" if author_profile.get("sex") == 0 else author_profile.get("sex", "N/A"),
431            "author_city": author_profile.get("city", {}).get("title", ""),
432            "author_country": author_profile.get("country", {}).get("title", ""),
433            "author_photo": author_profile.get("photo_200",
434                                               author_profile.get("photo_100", author_profile.get("photo_50", ""))),
435            "author_is_admin": True if author_profile.get("is_admin") == 1 else False if author_profile.get("is_admin") == 0 else author_profile.get("is_admin", "N/A"),
436            "author_is_advertiser": True if author_profile.get("is_advertiser") == 1 else False if author_profile.get(
437                "is_advertiser") == 0 else author_profile.get("is_advertiser", "N/A"),
438            "author_deactivated": author_profile.get("is_deactivated", False),
439            "author_privacy_is_closed": 'closed' if author_profile.get("is_closed") == 1 else 'open' if author_profile.get("is_closed") == 0 else 'private' if author_profile.get("is_closed") == 2 else author_profile.get("is_closed", "N/A"),
440            "author_followers": author_profile.get("followers_count", author_profile.get("members_count", "N/A")),
441            "in_reply_to_user": in_reply_to_user,
442            "in_reply_to_comment_id": in_reply_to_comment_id,
443            "source": item.get("post_source", {}).get("type"),
444            "views": item.get("views", {}).get("count"),
445            "likes": item.get("likes", {}).get("count"),
446            "post_comments": item.get("comments", {}).get("count"),
447            "edited": datetime.datetime.fromtimestamp(item.get("edited")).strftime("%Y-%m-%d %H:%M:%S") if item.get("edited", False) else False,
448            "photos": ", ".join(photos),
449            "videos": ", ".join(videos),
450            "audio": ", ".join(audio),
451            "links": ", ".join(links),
452            "docs": ", ".join(docs),
453            "subject": "",
454        })