datasources.telegram.search_telegram

   1"""
   2Search Telegram via API
   3"""
   4import traceback
   5import hashlib
   6import asyncio
   7import json
   8import ural
   9import time
  10import re
  11
  12from backend.lib.search import Search
  13from common.lib.exceptions import QueryParametersException, ProcessorInterruptedException, ProcessorException, \
  14    QueryNeedsFurtherInputException
  15from common.lib.helpers import convert_to_int, UserInput
  16from common.lib.item_mapping import MappedItem, MissingMappedField
  17
  18from datetime import datetime
  19from telethon import TelegramClient
  20from telethon.errors.rpcerrorlist import UsernameInvalidError, TimeoutError, ChannelPrivateError, BadRequestError, \
  21    FloodWaitError, ApiIdInvalidError, PhoneNumberInvalidError, RPCError
  22from telethon.tl.functions.channels import GetFullChannelRequest
  23from telethon.tl.functions.users import GetFullUserRequest
  24from telethon.tl.types import MessageEntityMention
  25
  26
  27
  28class SearchTelegram(Search):
  29    """
  30    Search Telegram via API
  31    """
  32    type = "telegram-search"  # job ID
  33    category = "Search"  # category
  34    title = "Telegram API search"  # title displayed in UI
  35    description = "Scrapes messages from open Telegram groups via its API."  # description displayed in UI
  36    extension = "ndjson"  # extension of result file, used internally and in UI
  37    is_local = False  # Whether this datasource is locally scraped
  38    is_static = False  # Whether this datasource is still updated
  39
  40    # cache
  41    details_cache = None
  42    failures_cache = None
  43    eventloop = None
  44    import_issues = 0
  45    end_if_rate_limited = 600  # break if Telegram requires wait time above number of seconds
  46
  47    max_workers = 1
  48    max_retries = 3
  49    flawless = 0
  50
  51    config = {
  52        "telegram-search.can_query_all_messages": {
  53            "type": UserInput.OPTION_TOGGLE,
  54            "help": "Remove message amount limit",
  55            "default": False,
  56            "tooltip": "Allows users to query unlimited messages from Telegram. This can lead to HUGE datasets!"
  57        },
  58        "telegram-search.max_entities": {
  59            "type": UserInput.OPTION_TEXT,
  60            "help": "Max entities to query",
  61            "coerce_type": int,
  62            "min": 0,
  63            "default": 25,
  64            "tooltip": "Amount of entities that can be queried at a time. Entities are groups or channels. 0 to "
  65                       "disable limit."
  66        },
  67        "telegram-search.max_crawl_depth": {
  68            "type": UserInput.OPTION_TEXT,
  69            "help": "Max crawl depth",
  70            "coerce_type": int,
  71            "min": 0,
  72            "default": 0,
  73            "tooltip": "If higher than 0, 4CAT can automatically add new entities to the query based on forwarded "
  74                       "messages. Recommended to leave at 0 for most users since this can exponentially increase "
  75                       "dataset sizes."
  76        }
  77    }
  78
  79    @classmethod
  80    def get_options(cls, parent_dataset=None, config=None):
  81        """
  82        Get processor options
  83
  84        Just updates the description of the entities field based on the
  85        configured max entities.
  86
  87        :param DataSet parent_dataset:  An object representing the dataset that
  88          the processor would be run on
  89        :param ConfigManager|None config:  Configuration reader (context-aware)
  90        """
  91
  92        max_entities = config.get("telegram-search.max_entities", 25)
  93        options = {
  94            "intro": {
  95                "type": UserInput.OPTION_INFO,
  96                "help": "Messages are scraped in reverse chronological order: the most recent message for a given entity "
  97                        "(e.g. a group) will be scraped first.\n\nTo query the Telegram API, you need to supply your [API "
  98                        "credentials](https://my.telegram.org/apps). 4CAT at this time does not support two-factor "
  99                        "authentication for Telegram."
 100            },
 101            "api_id": {
 102                "type": UserInput.OPTION_TEXT,
 103                "help": "API ID",
 104                "cache": True,
 105            },
 106            "api_hash": {
 107                "type": UserInput.OPTION_TEXT,
 108                "help": "API Hash",
 109                "cache": True,
 110            },
 111            "api_phone": {
 112                "type": UserInput.OPTION_TEXT,
 113                "help": "Phone number",
 114                "cache": True,
 115                "default": "+xxxxxxxxxx"
 116            },
 117            "divider": {
 118                "type": UserInput.OPTION_DIVIDER
 119            },
 120            "query-intro": {
 121                "type": UserInput.OPTION_INFO,
 122                "help": "Separate with commas or line breaks."
 123            },
 124            "query": {
 125                "type": UserInput.OPTION_TEXT_LARGE,
 126                "help": "Entities to scrape",
 127                "tooltip": "Separate with commas or line breaks."
 128            },
 129            "max_posts": {
 130                "type": UserInput.OPTION_TEXT,
 131                "help": "Messages per group",
 132                "min": 1,
 133                "max": 50000,
 134                "default": 10
 135            },
 136            "daterange": {
 137                "type": UserInput.OPTION_DATERANGE,
 138                "help": "Date range"
 139            },
 140            "divider-2": {
 141                "type": UserInput.OPTION_DIVIDER
 142            },
 143            "info-sensitive": {
 144                "type": UserInput.OPTION_INFO,
 145                "help": "Your API credentials and phone number **will be sent to the 4CAT server** and will be stored "
 146                        "there while data is fetched. After the dataset has been created your credentials will be "
 147                        "deleted from the server, unless you enable the option below. If you want to download images "
 148                        "attached to the messages in your collected data, you need to enable this option. Your "
 149                        "credentials will never be visible to other users and can be erased later via the result page."
 150            },
 151            "save-session": {
 152                "type": UserInput.OPTION_TOGGLE,
 153                "help": "Save session:",
 154                "default": False
 155            },
 156            "resolve-entities-intro": {
 157                "type": UserInput.OPTION_INFO,
 158                "help": "4CAT can resolve the references to channels and user and replace the numeric ID with the full "
 159                        "user, channel or group metadata. Doing so allows one to discover e.g. new relevant groups and "
 160                        "figure out where or who a message was forwarded from.\n\nHowever, this increases query time and "
 161                        "for large datasets, increases the chance you will be rate-limited and your dataset isn't able "
 162                        "to finish capturing. It will also dramatically increase the disk space needed to store the "
 163                        "data, so only enable this if you really need it!"
 164            },
 165            "resolve-entities": {
 166                "type": UserInput.OPTION_TOGGLE,
 167                "help": "Resolve references",
 168                "default": False,
 169            }
 170        }
 171
 172        if max_entities:
 173            options["query-intro"]["help"] = (f"You can collect messages from up to **{max_entities:,}** entities "
 174                                              f"(channels or groups) at a time. Separate with line breaks or commas.")
 175
 176        all_messages = config.get("telegram-search.can_query_all_messages", False)
 177        if all_messages:
 178            if "max" in options["max_posts"]:
 179                del options["max_posts"]["max"]
 180        else:
 181            options["max_posts"]["help"] = (f"Messages to collect per entity. You can query up to "
 182                                             f"{options['max_posts']['max']:,} messages per entity.")
 183
 184        if config.get("telegram-search.max_crawl_depth", 0) > 0:
 185            options["crawl_intro"] = {
 186                "type": UserInput.OPTION_INFO,
 187                "help": "Optionally, 4CAT can 'discover' new entities via forwarded messages; for example, if a "
 188                        "channel X you are collecting data for contains a message forwarded from channel Y, 4CAT can "
 189                        "collect messages from both channel X and Y. **Use this feature with caution**, as datasets can "
 190                        "rapidly grow when adding newly discovered entities to the query this way. Note that dataset "
 191                        "progress cannot be accurately tracked when you use this feature."
 192            }
 193            options["crawl-depth"] = {
 194                "type": UserInput.OPTION_TEXT,
 195                "coerce_type": int,
 196                "min": 0,
 197                "max": config.get("telegram-search.max_crawl_depth"),
 198                "default": 0,
 199                "help": "Crawl depth",
 200                "tooltip": "How many 'hops' to make when crawling messages. This is the distance from an initial "
 201                           "query, i.e. at most this many hops can be needed to reach the entity from one of the "
 202                           "starting entities."
 203            }
 204            options["crawl-threshold"] = {
 205                "type": UserInput.OPTION_TEXT,
 206                "coerce_type": int,
 207                "min": 0,
 208                "default": 5,
 209                "help": "Crawl threshold",
 210                "tooltip": "Entities need to be references at least this many times to be added to the query. Only "
 211                           "references discovered below the max crawl depth are taken into account."
 212            }
 213            options["crawl-via-links"] = {
 214                "type": UserInput.OPTION_TOGGLE,
 215                "default": False,
 216                "help": "Extract new groups from links",
 217                "tooltip": "Look for references to other groups in message content via t.me links and @references. "
 218                           "This is more error-prone than crawling only via forwards, but can be a way to discover "
 219                           "links that would otherwise remain undetected."
 220            }
 221
 222        return options
 223
 224
 225    def get_items(self, query):
 226        """
 227        Execute a query; get messages for given parameters
 228
 229        Basically a wrapper around execute_queries() to call it with asyncio.
 230
 231        :param dict query:  Query parameters, as part of the DataSet object
 232        :return list:  Posts, sorted by thread and post ID, in ascending order
 233        """
 234        if "api_phone" not in query or "api_hash" not in query or "api_id" not in query:
 235            self.dataset.update_status("Could not create dataset since the Telegram API Hash and ID are missing. Try "
 236                                       "creating it again from scratch.", is_final=True)
 237            return None
 238
 239        self.details_cache = {}
 240        self.failures_cache = set()
 241        #TODO: This ought to yield as we're holding everything in memory; async generator? execute_queries() also needs to be modified for this
 242        results = asyncio.run(self.execute_queries())
 243
 244        if not query.get("save-session"):
 245            self.dataset.delete_parameter("api_hash", instant=True)
 246            self.dataset.delete_parameter("api_phone", instant=True)
 247            self.dataset.delete_parameter("api_id", instant=True)
 248
 249        if self.flawless:
 250            self.dataset.update_status(f"Dataset completed, but {self.flawless} requested entities were unavailable (they may have "
 251                                       "been private). View the log file for details.", is_final=True)
 252
 253        return results
 254
 255    async def execute_queries(self):
 256        """
 257        Get messages for queries
 258
 259        This is basically what would be done in get_items(), except due to
 260        Telethon's architecture this needs to be called in an async method,
 261        which is this one.
 262
 263        :return list:  Collected messages
 264        """
 265        # session file has been created earlier, and we can re-use it here in
 266        # order to avoid having to re-enter the security code
 267        query = self.parameters
 268
 269        session_id = SearchTelegram.create_session_id(query["api_phone"].strip(),
 270                                                      query["api_id"].strip(),
 271                                                      query["api_hash"].strip())
 272        self.dataset.log(f'Telegram session id: {session_id}')
 273        session_path = self.config.get("PATH_SESSIONS").joinpath(session_id + ".session")
 274
 275        client = None
 276
 277        try:
 278            client = TelegramClient(str(session_path), int(query.get("api_id")), query.get("api_hash"),
 279                                    loop=self.eventloop)
 280            await client.start(phone=SearchTelegram.cancel_start)
 281        except RuntimeError:
 282            # session is no longer useable, delete file so user will be asked
 283            # for security code again. The RuntimeError is raised by
 284            # `cancel_start()`
 285            self.dataset.update_status(
 286                "Session is not authenticated: login security code may have expired. You need to re-enter the security code.",
 287                is_final=True)
 288
 289            if client and hasattr(client, "disconnect"):
 290                await client.disconnect()
 291
 292            if session_path.exists():
 293                session_path.unlink()
 294
 295            return []
 296        except Exception as e:
 297            # not sure what exception specifically is triggered here, but it
 298            # always means the connection failed
 299            self.log.error(f"Telegram: {e}\n{traceback.format_exc()}")
 300            self.dataset.update_status("Error connecting to the Telegram API with provided credentials.", is_final=True)
 301            if client and hasattr(client, "disconnect"):
 302                await client.disconnect()
 303            return []
 304
 305        # ready our parameters
 306        parameters = self.dataset.get_parameters()
 307        queries = [query.strip() for query in parameters.get("query", "").split(",")]
 308        max_items = convert_to_int(parameters.get("items", 10), 10)
 309
 310        # Telethon requires the offset date to be a datetime date
 311        max_date = parameters.get("max_date")
 312        if max_date:
 313            try:
 314                max_date = datetime.fromtimestamp(int(max_date))
 315            except ValueError:
 316                max_date = None
 317
 318        # min_date can remain an integer
 319        min_date = parameters.get("min_date")
 320        if min_date:
 321            try:
 322                min_date = int(min_date)
 323            except ValueError:
 324                min_date = None
 325
 326        posts = []
 327        try:
 328            async for post in self.gather_posts(client, queries, max_items, min_date, max_date):
 329                posts.append(post)
 330            return posts
 331        except ProcessorInterruptedException as e:
 332            raise e
 333        except Exception:
 334            # catch-all so we can disconnect properly
 335            # ...should we?
 336            self.dataset.update_status("Error scraping posts from Telegram; halting collection.")
 337            self.log.error(f"Telegram scraping error (dataset {self.dataset.key}): {traceback.format_exc()}")
 338            # May as well return what was captured, yes?
 339            return posts
 340        finally:
 341            await client.disconnect()
 342
 343    async def gather_posts(self, client, queries, max_items, min_date, max_date):
 344        """
 345        Gather messages for each entity for which messages are requested
 346
 347        :param TelegramClient client:  Telegram Client
 348        :param list queries:  List of entities to query (as string)
 349        :param int max_items:  Messages to scrape per entity
 350        :param int min_date:  Datetime date to get posts after
 351        :param int max_date:  Datetime date to get posts before
 352        :return list:  List of messages, each message a dictionary.
 353        """
 354        resolve_refs = self.parameters.get("resolve-entities")
 355
 356        # Adding flag to stop; using for rate limits
 357        no_additional_queries = False
 358
 359        # This is used for the 'crawl' feature so we know at which depth a
 360        # given entity was discovered
 361        depth_map = {
 362            entity: 0 for entity in queries
 363        }
 364
 365        crawl_max_depth = self.parameters.get("crawl-depth", 0)
 366        crawl_msg_threshold = self.parameters.get("crawl-threshold", 10)
 367        crawl_via_links = self.parameters.get("crawl-via-links", False)
 368
 369        self.dataset.log(f"Max crawl depth: {crawl_max_depth}")
 370        self.dataset.log(f"Crawl threshold: {crawl_msg_threshold}")
 371
 372        # this keeps track of how often an entity not in the original query
 373        # has been mentioned. When crawling is enabled and this exceeds the
 374        # given threshold, the entity is added to the query
 375        crawl_references = {}
 376        full_query = set(queries)
 377        num_queries = len(queries)
 378
 379        # we may not always know the 'entity username' for an entity ID, so
 380        # keep a reference map as we go
 381        entity_id_map = {}
 382
 383        # Collect queries
 384        # Use while instead of for so we can change queries during iteration
 385        # this is needed for the 'crawl' feature which can discover new
 386        # entities during crawl
 387        processed = 0
 388        total_messages = 0
 389        while queries:
 390            query = queries.pop(0)
 391
 392            delay = 10
 393            retries = 0
 394            processed += 1
 395            self.dataset.update_progress(processed / num_queries)
 396
 397            if no_additional_queries:
 398                # Note that we are not completing this query
 399                self.dataset.update_status(f"Rate-limited by Telegram; not executing query {entity_id_map.get(query, query)}")
 400                continue
 401
 402            while True:
 403                self.dataset.update_status(f"Retrieving messages for entity '{entity_id_map.get(query, query)}'")
 404                entity_posts = 0
 405                discovered = 0
 406                try:
 407                    async for message in client.iter_messages(entity=query, offset_date=max_date):
 408                        entity_posts += 1
 409                        total_messages += 1
 410                        if self.interrupted:
 411                            raise ProcessorInterruptedException(
 412                                "Interrupted while fetching message data from the Telegram API")
 413
 414                        if entity_posts % 100 == 0:
 415                            self.dataset.update_status(
 416                                f"Retrieved {entity_posts:,} posts for entity '{entity_id_map.get(query, query)}' ({total_messages:,} total)")
 417
 418                        if message.action is not None:
 419                            # e.g. someone joins the channel - not an actual message
 420                            continue
 421
 422                        # todo: possibly enrich object with e.g. the name of
 423                        # the channel a message was forwarded from (but that
 424                        # needs extra API requests...)
 425                        serialized_message = SearchTelegram.serialize_obj(message)
 426                        if "_chat" in serialized_message:
 427                            # Add query ID to check if queries have been crawled previously
 428                            full_query.add(serialized_message["_chat"]["id"])
 429                            if query not in entity_id_map and serialized_message["_chat"]["id"] == query:
 430                                # once we know what a channel ID resolves to, use the username instead so it is easier to
 431                                # understand for the user
 432                                entity_id_map[query] = serialized_message["_chat"]["username"]
 433                                self.dataset.update_status(f"Fetching messages for entity '{entity_id_map[query]}' (channel ID {query})")
 434
 435                        if resolve_refs:
 436                            serialized_message = await self.resolve_groups(client, serialized_message)
 437
 438                        # Stop if we're below the min date
 439                        if min_date and serialized_message.get("date") < min_date:
 440                            break
 441
 442                        # if crawling is enabled, see if we found something to add to the query
 443                        linked_entities = set()
 444                        if crawl_max_depth and (depth_map.get(query) < crawl_max_depth):
 445                            message_fwd = serialized_message.get("fwd_from")
 446                            fwd_from = None
 447                            fwd_source_type = None
 448                            if message_fwd and message_fwd.get("from_id"):
 449                                if message_fwd["from_id"].get("_type") == "PeerChannel":
 450                                    # Legacy(?) data structure (pre 2024/7/22)
 451                                    # even if we haven't resolved the ID, we can feed the numeric ID
 452                                    # to Telethon! this is nice because it means we don't have to
 453                                    # resolve entities to crawl iteratively
 454                                    fwd_from = int(message_fwd["from_id"]["channel_id"])
 455                                    fwd_source_type = "channel"
 456                                elif message_fwd and message_fwd.get("from_id", {}).get('full_chat',{}):
 457                                    # TODO: do we need a check here to only follow certain types of messages? this is similar to resolving, but the types do not appear the same to me
 458                                    # Note: message_fwd["from_id"]["channel_id"] == message_fwd["from_id"]["full_chat"]["id"] in test cases so far
 459                                    fwd_from = int(message_fwd["from_id"]["full_chat"]["id"])
 460                                    fwd_source_type = "channel"
 461                                elif message_fwd and (message_fwd.get("from_id", {}).get('full_user',{}) or message_fwd.get("from_id", {}).get("_type") == "PeerUser"):
 462                                    # forwards can also come from users
 463                                    # these can never be followed, so don't add these to the crawl, but do document them
 464                                    fwd_source_type = "user"
 465                                else:
 466                                    print(json.dumps(message_fwd))
 467                                    self.log.warning(f"Telegram (dataset {self.dataset.key}): Unknown fwd_from data structure; unable to crawl")
 468                                    fwd_source_type = "unknown"
 469
 470                                if fwd_from:
 471                                    linked_entities.add(fwd_from)
 472
 473
 474                            if crawl_via_links:
 475                                # t.me links
 476                                all_links = ural.urls_from_text(serialized_message["message"])
 477                                all_links = [link.split("t.me/")[1] for link in all_links if ural.get_hostname(link) == "t.me" and len(link.split("t.me/")) > 1]
 478                                for link in all_links:
 479                                    if link.startswith("+"):
 480                                        # invite links
 481                                        continue
 482
 483                                    entity_name = link.split("/")[0].split("?")[0].split("#")[0]
 484                                    linked_entities.add(entity_name)
 485
 486                                # @references
 487                                references = [r for t, r in message.get_entities_text() if type(t) is MessageEntityMention]
 488                                for reference in references:
 489                                    if reference.startswith("@"):
 490                                        reference = reference[1:]
 491
 492                                    reference = reference.split("/")[0]
 493
 494                                    linked_entities.add(reference)
 495
 496                            # Check if fwd_from or the resolved entity ID is already queued or has been queried
 497                            for link in linked_entities:
 498                                if link not in full_query and link not in queries and fwd_source_type not in ("user",):
 499                                    # new entity discovered!
 500                                    # might be discovered (before collection) multiple times, so retain lowest depth
 501                                    # print(f"Potentially crawling {link}")
 502                                    depth_map[link] = min(depth_map.get(link, crawl_max_depth), depth_map[query] + 1)
 503                                    if link not in crawl_references:
 504                                        crawl_references[link] = 0
 505                                    crawl_references[link] += 1
 506
 507                                    # Add to queries if it has been referenced enough times
 508                                    if crawl_references[link] >= crawl_msg_threshold:
 509                                        queries.append(link)
 510                                        full_query.add(link)
 511                                        num_queries += 1
 512                                        discovered += 1
 513                                        self.dataset.update_status(f"Discovered new entity {entity_id_map.get(link, link)} in {entity_id_map.get(query, query)} at crawl depth {depth_map[query]}, adding to query")
 514
 515
 516
 517                        serialized_message["4CAT_metadata"] = {
 518                            "collected_at": datetime.now().isoformat(), # this is relevant for rather long crawls
 519                            "query": query, # possibly redundant, but we are adding non-user defined queries by crawling and may be useful to know exactly what query was used to collect an entity
 520                            "query_depth": depth_map.get(query, 0)
 521                        }
 522                        yield serialized_message
 523
 524                        if entity_posts >= max_items:
 525                            break
 526
 527                except ChannelPrivateError:
 528                    self.dataset.update_status(f"Entity {entity_id_map.get(query, query)} is private, skipping")
 529                    self.flawless += 1
 530
 531                except (UsernameInvalidError,):
 532                    self.dataset.update_status(f"Could not scrape entity '{entity_id_map.get(query, query)}', does not seem to exist, skipping")
 533                    self.flawless += 1
 534
 535                except FloodWaitError as e:
 536                    self.dataset.update_status(f"Rate-limited by Telegram: {e}; waiting")
 537                    if e.seconds < self.end_if_rate_limited:
 538                        time.sleep(e.seconds)
 539                        continue
 540                    else:
 541                        self.flawless += 1
 542                        no_additional_queries = True
 543                        self.dataset.update_status(
 544                            f"Telegram wait grown larger than {int(e.seconds / 60)} minutes, ending")
 545                        break
 546
 547                except BadRequestError as e:
 548                    self.dataset.update_status(
 549                        f"Error '{e.__class__.__name__}' while collecting entity {entity_id_map.get(query, query)}, skipping")
 550                    self.flawless += 1
 551
 552                except ValueError as e:
 553                    self.dataset.update_status(f"Error '{e}' while collecting entity {entity_id_map.get(query, query)}, skipping")
 554                    self.flawless += 1
 555
 556                except ChannelPrivateError as e:
 557                    self.dataset.update_status(
 558                        f"QUERY '{entity_id_map.get(query, query)}' unable to complete due to error {e}. Skipping.")
 559                    break
 560
 561                except TimeoutError:
 562                    if retries < 3:
 563                        self.dataset.update_status(
 564                            f"Tried to fetch messages for entity '{entity_id_map.get(query, query)}' but timed out {retries:,} times. Skipping.")
 565                        self.flawless += 1
 566                        break
 567
 568                    self.dataset.update_status(
 569                        f"Got a timeout from Telegram while fetching messages for entity '{entity_id_map.get(query, query)}'. Trying again in {delay:,} seconds.")
 570                    time.sleep(delay)
 571                    delay *= 2
 572                    continue
 573
 574                self.dataset.log(f"Completed {entity_id_map.get(query, query)} with {entity_posts} messages (discovered {discovered} new entities)")
 575                break
 576
 577    async def resolve_groups(self, client, message):
 578        """
 579        Recursively resolve references to groups and users
 580
 581        :param client:  Telethon client instance
 582        :param dict message:  Message, as already mapped by serialize_obj
 583        :return:  Resolved dictionary
 584        """
 585        resolved_message = message.copy()
 586        for key, value in message.items():
 587            try:
 588                if type(value) is not dict:
 589                    # if it's not a dict, we never have to resolve it, as it
 590                    # does not represent an entity
 591                    continue
 592
 593                elif "_type" in value and value["_type"] in ("InputPeerChannel", "PeerChannel"):
 594                    # forwarded from a channel!
 595                    if value["channel_id"] in self.failures_cache:
 596                        continue
 597
 598                    if value["channel_id"] not in self.details_cache:
 599                        channel = await client(GetFullChannelRequest(value["channel_id"]))
 600                        self.details_cache[value["channel_id"]] = SearchTelegram.serialize_obj(channel)
 601
 602                    resolved_message[key] = self.details_cache[value["channel_id"]]
 603                    resolved_message[key]["channel_id"] = value["channel_id"]
 604
 605                elif "_type" in value and value["_type"] == "PeerUser":
 606                    # a user!
 607                    if value["user_id"] in self.failures_cache:
 608                        continue
 609
 610                    if value["user_id"] not in self.details_cache:
 611                        user = await client(GetFullUserRequest(value["user_id"]))
 612                        self.details_cache[value["user_id"]] = SearchTelegram.serialize_obj(user)
 613
 614                    resolved_message[key] = self.details_cache[value["user_id"]]
 615                    resolved_message[key]["user_id"] = value["user_id"]
 616                else:
 617                    resolved_message[key] = await self.resolve_groups(client, value)
 618
 619            except (TypeError, ChannelPrivateError, UsernameInvalidError) as e:
 620                self.failures_cache.add(value.get("channel_id", value.get("user_id")))
 621                if type(e) in (ChannelPrivateError, UsernameInvalidError):
 622                    self.dataset.log(f"Cannot resolve entity with ID {value.get('channel_id', value.get('user_id'))} of type {value['_type']} ({e.__class__.__name__}), leaving as-is")
 623                else:
 624                    self.dataset.log(f"Cannot resolve entity with ID {value.get('channel_id', value.get('user_id'))} of type {value['_type']}, leaving as-is")
 625
 626        return resolved_message
 627
 628    @staticmethod
 629    def cancel_start():
 630        """
 631        Replace interactive phone number input in Telethon
 632
 633        By default, if Telethon cannot use the given session file to
 634        authenticate, it will interactively prompt the user for a phone
 635        number on the command line. That is not useful here, so instead
 636        raise a RuntimeError. This will be caught below and the user will
 637        be told they need to re-authenticate via 4CAT.
 638        """
 639        raise RuntimeError("Connection cancelled")
 640
 641    @staticmethod
 642    def map_item(message):
 643        """
 644        Convert Message object to 4CAT-ready data object
 645
 646        :param Message message:  Message to parse
 647        :return dict:  4CAT-compatible item object
 648        """
 649        if message["_chat"]["username"]:
 650            # chats can apparently not have usernames???
 651            # truly telegram objects are way too lenient for their own good
 652            thread = message["_chat"]["username"]
 653        elif message["_chat"]["title"]:
 654            thread = re.sub(r"\s", "", message["_chat"]["title"])
 655        else:
 656            # just give up
 657            thread = "unknown"
 658
 659        # determine username
 660        # API responses only include the user *ID*, not the username, and to
 661        # complicate things further not everyone is a user and not everyone
 662        # has a username. If no username is available, try the first and
 663        # last name someone has supplied
 664        fullname = ""
 665        username = ""
 666        user_id = message["_sender"]["id"] if message.get("_sender") else ""
 667        user_is_bot = message["_sender"].get("bot", False) if message.get("_sender") else ""
 668
 669        if message.get("_sender") and message["_sender"].get("username"):
 670            username = message["_sender"]["username"]
 671
 672        if message.get("_sender") and message["_sender"].get("first_name"):
 673            fullname += message["_sender"]["first_name"]
 674
 675        if message.get("_sender") and message["_sender"].get("last_name"):
 676            fullname += " " + message["_sender"]["last_name"]
 677
 678        fullname = fullname.strip()
 679
 680        # determine media type
 681        # these store some extra information of the attachment in
 682        # attachment_data. Since the final result will be serialised as a csv
 683        # file, we can only store text content. As such some media data is
 684        # serialised as JSON.
 685        attachment_type = SearchTelegram.get_media_type(message["media"])
 686        attachment_filename = ""
 687
 688        if attachment_type == "contact":
 689            contact_data = ["phone_number", "first_name", "last_name", "vcard", "user_id"]
 690            if message["media"].get('contact', False):
 691                # Old datastructure
 692                attachment = message["media"]["contact"]
 693            elif all([property in message["media"].keys() for property in contact_data]):
 694                # New datastructure 2022/7/25
 695                attachment = message["media"]
 696            else:
 697                raise ProcessorException('Cannot find contact data; Telegram datastructure may have changed')
 698            attachment_data = json.dumps({property: attachment.get(property) for property in contact_data})
 699
 700        elif attachment_type == "document":
 701            # videos, etc
 702            # This could add a separate routine for videos to make them a
 703            # separate type, which could then be scraped later, etc
 704            attachment_type = message["media"]["document"]["mime_type"].split("/")[0]
 705            if attachment_type == "video":
 706                attachment = message["media"]["document"]
 707                attachment_data = json.dumps({
 708                    "id": attachment["id"],
 709                    "dc_id": attachment["dc_id"],
 710                    "file_reference": attachment["file_reference"],
 711                })
 712            else:
 713                attachment_data = ""
 714
 715        # elif attachment_type in ("geo", "geo_live"):
 716        # untested whether geo_live is significantly different from geo
 717        #    attachment_data = "%s %s" % (message["geo"]["lat"], message["geo"]["long"])
 718
 719        elif attachment_type == "photo" or attachment_type == "url" and message["media"]["webpage"].get("photo"):
 720            # we don't actually store any metadata about the photo, since very
 721            # little of the metadata attached is of interest. Instead, the
 722            # actual photos may be downloaded via a processor that is run on the
 723            # search results
 724            attachment = message["media"]["photo"] if attachment_type == "photo" else message["media"]["webpage"]["photo"]
 725            attachment_data = json.dumps({
 726                "id": attachment["id"],
 727                "dc_id": attachment["dc_id"],
 728                "file_reference": attachment["file_reference"],
 729            })
 730            attachment_filename = thread + "-" + str(message["id"]) + ".jpeg"
 731
 732        elif attachment_type == "poll":
 733            # unfortunately poll results are only available when someone has
 734            # actually voted on the poll - that will usually not be the case,
 735            # so we store -1 as the vote count
 736            attachment = message["media"]
 737            options = {option["option"]: option["text"] for option in attachment["poll"]["answers"]}
 738            attachment_data = json.dumps({
 739                "question": attachment["poll"]["question"],
 740                "voters": attachment["results"]["total_voters"],
 741                "answers": [{
 742                    "answer": options[answer["option"]],
 743                    "votes": answer["voters"]
 744                } for answer in attachment["results"]["results"]] if attachment["results"]["results"] else [{
 745                    "answer": options[option],
 746                    "votes": -1
 747                } for option in options]
 748            })
 749
 750        else:
 751            attachment_data = ""
 752
 753        # was the message forwarded from somewhere and if so when?
 754        forwarded_timestamp = ""
 755        forwarded_name = ""
 756        forwarded_id = ""
 757        forwarded_username = ""
 758        if message.get("fwd_from") and "from_id" in message["fwd_from"] and type(message["fwd_from"]["from_id"]) is not int:
 759            # forward information is spread out over a lot of places
 760            # we can identify, in order of usefulness: username, full name,
 761            # and ID. But not all of these are always available, and not
 762            # always in the same place either
 763            forwarded_timestamp = int(message["fwd_from"]["date"])
 764            from_data = message["fwd_from"]["from_id"]
 765
 766            if from_data:
 767                forwarded_id = from_data.get("channel_id", from_data.get("user_id", ""))
 768
 769            if message["fwd_from"].get("from_name"):
 770                forwarded_name = message["fwd_from"].get("from_name")
 771
 772            if from_data and from_data.get("from_name"):
 773                forwarded_name = message["fwd_from"]["from_name"]
 774
 775            if from_data and from_data.get("users") and len(from_data["users"]) > 0 and "user" not in from_data:
 776                from_data["user"] = from_data["users"][0]
 777
 778            if from_data and ("user" in from_data or "chats" in from_data):
 779                # 'resolve entities' was enabled for this dataset
 780                if "user" in from_data:
 781                    if from_data["user"].get("username"):
 782                        forwarded_username = from_data["user"]["username"]
 783
 784                    if from_data["user"].get("first_name"):
 785                        forwarded_name = from_data["user"]["first_name"]
 786                    if message["fwd_from"].get("last_name"):
 787                        forwarded_name += "  " + from_data["user"]["last_name"]
 788
 789                    forwarded_name = forwarded_name.strip()
 790
 791                elif "chats" in from_data:
 792                    channel_id = from_data.get("channel_id")
 793                    for chat in from_data["chats"]:
 794                        if chat["id"] == channel_id or channel_id is None:
 795                            forwarded_username = chat["username"]
 796
 797            elif message.get("_forward") and message["_forward"].get("_chat"):
 798                if message["_forward"]["_chat"].get("username"):
 799                    forwarded_username = message["_forward"]["_chat"]["username"]
 800
 801                if message["_forward"]["_chat"].get("title"):
 802                    forwarded_name = message["_forward"]["_chat"]["title"]
 803
 804        link_title = ""
 805        link_attached = ""
 806        link_description = ""
 807        reactions = ""
 808
 809        if message.get("media") and message["media"].get("webpage"):
 810            link_title = message["media"]["webpage"].get("title")
 811            link_attached = message["media"]["webpage"].get("url")
 812            link_description = message["media"]["webpage"].get("description")
 813
 814        if message.get("reactions") and message["reactions"].get("results"):
 815            for reaction in message["reactions"]["results"]:
 816                if type(reaction["reaction"]) is dict and "emoticon" in reaction["reaction"]:
 817                    # Updated to support new reaction datastructure
 818                    reactions += reaction["reaction"]["emoticon"] * reaction["count"]
 819                elif type(reaction["reaction"]) is str and "count" in reaction:
 820                    reactions += reaction["reaction"] * reaction["count"]
 821                else:
 822                    # Failsafe; can be updated to support formatting of new datastructures in the future
 823                    reactions += f"{reaction}, "
 824
 825        is_reply = False
 826        reply_to = ""
 827        if message.get("reply_to"):
 828            is_reply = True
 829            reply_to = message["reply_to"].get("reply_to_msg_id", "")
 830
 831        # t.me links
 832        linked_entities = set()
 833        all_links = ural.urls_from_text(message["message"])
 834        all_links = [link.split("t.me/")[1] for link in all_links if
 835                     ural.get_hostname(link) == "t.me" and len(link.split("t.me/")) > 1]
 836
 837        for link in all_links:
 838            if link.startswith("+"):
 839                # invite links
 840                continue
 841
 842            entity_name = link.split("/")[0].split("?")[0].split("#")[0]
 843            linked_entities.add(entity_name)
 844
 845        # @references
 846        # in execute_queries we use MessageEntityMention to get these
 847        # however, after serializing these objects we only have the offsets of
 848        # the mentioned username, and telegram does weird unicode things to its
 849        # offsets meaning we can't just substring the message. So use a regex
 850        # as a 'good enough' solution
 851        all_mentions = set(re.findall(r"@([^\s\W]+)", message["message"]))
 852
 853        # make this case-insensitive since people may use different casing in
 854        # messages than the 'official' username for example
 855        all_connections = set([v for v in [forwarded_username, *linked_entities, *all_mentions] if v])
 856        all_ci_connections = set()
 857        seen = set()
 858        for connection in all_connections:
 859            if connection.lower() not in seen:
 860                all_ci_connections.add(connection)
 861                seen.add(connection.lower())
 862
 863        return MappedItem({
 864            "id": f"{message['_chat']['username']}-{message['id']}",
 865            "thread_id": thread,
 866            "chat": message["_chat"]["username"],
 867            "author": user_id,
 868            "author_username": username,
 869            "author_name": fullname,
 870            "author_is_bot": "yes" if user_is_bot else "no",
 871            "body": message["message"],
 872            "body_markdown": message.get("message_markdown", MissingMappedField("")),
 873            "is_reply": is_reply,
 874            "reply_to": reply_to,
 875            "views": message["views"] if message["views"] else "",
 876            # "forwards": message.get("forwards", MissingMappedField(0)),
 877            "reactions": reactions,
 878            "timestamp": datetime.fromtimestamp(message["date"]).strftime("%Y-%m-%d %H:%M:%S"),
 879            "unix_timestamp": int(message["date"]),
 880            "timestamp_edited": datetime.fromtimestamp(message["edit_date"]).strftime("%Y-%m-%d %H:%M:%S") if message[
 881                "edit_date"] else "",
 882            "unix_timestamp_edited": int(message["edit_date"]) if message["edit_date"] else "",
 883            "author_forwarded_from_name": forwarded_name,
 884            "author_forwarded_from_username": forwarded_username,
 885            "author_forwarded_from_id": forwarded_id,
 886            "entities_linked": ",".join(linked_entities),
 887            "entities_mentioned": ",".join(all_mentions),
 888            "all_connections": ",".join(all_ci_connections),
 889            "timestamp_forwarded_from": datetime.fromtimestamp(forwarded_timestamp).strftime(
 890                "%Y-%m-%d %H:%M:%S") if forwarded_timestamp else "",
 891            "unix_timestamp_forwarded_from": forwarded_timestamp,
 892            "link_title": link_title,
 893            "link_description": link_description,
 894            "link_attached": link_attached,
 895            "attachment_type": attachment_type,
 896            "attachment_data": attachment_data,
 897            "attachment_filename": attachment_filename
 898        })
 899
 900    @staticmethod
 901    def get_media_type(media):
 902        """
 903        Get media type for a Telegram attachment
 904
 905        :param media:  Media object
 906        :return str:  Textual identifier of the media type
 907        """
 908        try:
 909            return {
 910                "NoneType": "",
 911                "MessageMediaContact": "contact",
 912                "MessageMediaDocument": "document",
 913                "MessageMediaEmpty": "",
 914                "MessageMediaGame": "game",
 915                "MessageMediaGeo": "geo",
 916                "MessageMediaGeoLive": "geo_live",
 917                "MessageMediaInvoice": "invoice",
 918                "MessageMediaPhoto": "photo",
 919                "MessageMediaPoll": "poll",
 920                "MessageMediaUnsupported": "unsupported",
 921                "MessageMediaVenue": "venue",
 922                "MessageMediaWebPage": "url"
 923            }[media.get("_type", None)]
 924        except (AttributeError, KeyError):
 925            return ""
 926
 927    @staticmethod
 928    def serialize_obj(input_obj):
 929        """
 930        Serialize an object as a dictionary
 931
 932        Telethon message objects are not serializable by themselves, but most
 933        relevant attributes are simply struct classes. This function replaces
 934        those that are not with placeholders and then returns a dictionary that
 935        can be serialized as JSON.
 936
 937        :param obj:  Object to serialize
 938        :return:  Serialized object
 939        """
 940        scalars = (int, str, float, list, tuple, set, bool)
 941
 942        if type(input_obj) in scalars or input_obj is None:
 943            return input_obj
 944
 945        if type(input_obj) is not dict:
 946            obj = input_obj.__dict__
 947        else:
 948            obj = input_obj.copy()
 949
 950        mapped_obj = {}
 951        for item, value in obj.items():
 952            if type(value) is datetime:
 953                mapped_obj[item] = value.timestamp()
 954            elif type(value).__module__ in ("telethon.tl.types", "telethon.tl.custom.forward"):
 955                mapped_obj[item] = SearchTelegram.serialize_obj(value)
 956            elif type(value) is list:
 957                mapped_obj[item] = [SearchTelegram.serialize_obj(item) for item in value]
 958            elif type(value) is bytes:
 959                mapped_obj[item] = value.hex()
 960            elif type(value) not in scalars and value is not None:
 961                # type we can't make sense of here
 962                continue
 963            else:
 964                mapped_obj[item] = value
 965
 966        # Add the _type if the original object was a telethon type
 967        if type(input_obj).__module__ in ("telethon.tl.types", "telethon.tl.custom.forward"):
 968            mapped_obj["_type"] = type(input_obj).__name__
 969
 970        # Store the markdown-formatted text
 971        if type(input_obj).__name__ == "Message":
 972            mapped_obj["message_markdown"] = input_obj.text
 973
 974        return mapped_obj
 975
 976    @staticmethod
 977    def validate_query(query, request, config):
 978        """
 979        Validate Telegram query
 980
 981        :param config:
 982        :param dict query:  Query parameters, from client-side.
 983        :param request:  Flask request
 984        :param User user:  User object of user who has submitted the query
 985        :param ConfigManager config:  Configuration reader (context-aware)
 986        :return dict:  Safe query parameters
 987        """
 988        # no query 4 u
 989        if not query.get("query", "").strip():
 990            raise QueryParametersException("You must provide a search query.")
 991
 992        if not query.get("api_id", None) or not query.get("api_hash", None) or not query.get("api_phone", None):
 993            raise QueryParametersException("You need to provide valid Telegram API credentials first.")
 994
 995        all_posts = config.get("telegram-search.can_query_all_messages", False)
 996        max_entities = config.get("telegram-search.max_entities", 25)
 997
 998        num_items = query.get("max_posts") if all_posts else min(query.get("max_posts"), SearchTelegram.get_options(
 999            config=config)["max_posts"]["max"])
1000
1001        # reformat queries to be a comma-separated list with no wrapping
1002        # whitespace
1003        whitespace = re.compile(r"\s+")
1004        items = whitespace.sub("", query.get("query").replace("\n", ","))
1005        if max_entities > 0 and len(items.split(",")) > max_entities:
1006            raise QueryParametersException(f"You cannot query more than {max_entities:,} items at a time.")
1007
1008        sanitized_items = []
1009        # handle telegram URLs
1010        for item in items.split(","):
1011            if not item.strip():
1012                continue
1013            item = re.sub(r"^https?://t\.me/", "", item)
1014            item = re.sub(r"^/?s/", "", item)
1015            item = re.sub(r"[/]*$", "", item)
1016            sanitized_items.append(item)
1017
1018        # the dates need to make sense as a range to search within
1019        min_date, max_date = query.get("daterange")
1020
1021        # now check if there is an active API session
1022        if not hasattr(config, "user") or not config.user.is_authenticated or config.user.is_anonymous:
1023            raise QueryParametersException("Telegram scraping is only available to logged-in users with personal "
1024                                           "accounts.")
1025
1026        # check for the information we need
1027        session_id = SearchTelegram.create_session_id(query.get("api_phone"), query.get("api_id"),
1028                                                      query.get("api_hash"))
1029        config.user.set_value("telegram.session", session_id)
1030        session_path = config.get('PATH_SESSIONS').joinpath(session_id + ".session")
1031
1032        client = None
1033
1034        # API ID is always a number, if it's not, we can immediately fail
1035        try:
1036            api_id = int(query.get("api_id"))
1037        except ValueError:
1038            raise QueryParametersException("Invalid API ID.")
1039
1040        # maybe we've entered a code already and submitted it with the request
1041        if "option-security-code" in request.form and request.form.get("option-security-code").strip():
1042            code_callback = lambda: request.form.get("option-security-code")  # noqa: E731
1043            max_attempts = 1
1044        else:
1045            code_callback = lambda: -1  # noqa: E731
1046            # max_attempts = 0 because authing will always fail: we can't wait for
1047            # the code to be entered interactively, we'll need to do a new request
1048            # but we can't just immediately return, we still need to call start()
1049            # to get telegram to send us a code
1050            max_attempts = 0
1051
1052        # now try authenticating
1053        needs_code = False
1054        try:
1055            loop = asyncio.new_event_loop()
1056            asyncio.set_event_loop(loop)
1057            client = TelegramClient(str(session_path), api_id, query.get("api_hash"), loop=loop)
1058
1059            try:
1060                client.start(max_attempts=max_attempts, phone=query.get("api_phone"), code_callback=code_callback)
1061
1062            except ValueError as e:
1063                # this happens if 2FA is required
1064                raise QueryParametersException("Your account requires two-factor authentication. 4CAT at this time "
1065                                               f"does not support this authentication mode for Telegram. ({e})")
1066            except RuntimeError:
1067                # A code was sent to the given phone number
1068                needs_code = True
1069        except FloodWaitError as e:
1070            # uh oh, we got rate-limited
1071            raise QueryParametersException("You were rate-limited and should wait a while before trying again. " +
1072                                           str(e).split("(")[0] + ".")
1073        except ApiIdInvalidError:
1074            # wrong credentials
1075            raise QueryParametersException("Your API credentials are invalid.")
1076        except PhoneNumberInvalidError:
1077            # wrong phone number
1078            raise QueryParametersException(
1079                "The phone number provided is not a valid phone number for these credentials.")
1080        except RPCError as e:
1081            # only seen this with an 'UPDATE_APP_TO_LOGIN' status
1082            raise QueryParametersException(f"Could not verify your authentication. You may need to update your "
1083                                           f"Telegram app(s) to the latest version to proceed ({e}).")
1084        except Exception as e:
1085            # ?
1086            raise QueryParametersException(
1087                f"An unexpected error ({e}) occurred and your authentication could not be verified.")
1088        finally:
1089            if client:
1090                client.disconnect()
1091
1092        if needs_code:
1093            raise QueryNeedsFurtherInputException(config={
1094                "code-info": {
1095                    "type": UserInput.OPTION_INFO,
1096                    "help": "Please enter the security code that was sent to your Telegram app to continue."
1097                },
1098                "security-code": {
1099                    "type": UserInput.OPTION_TEXT,
1100                    "help": "Security code",
1101                    "sensitive": True
1102                }})
1103
1104        # simple!
1105        return {
1106            "items": num_items,
1107            "query": ",".join(sanitized_items),
1108            "api_id": query.get("api_id"),
1109            "api_hash": query.get("api_hash"),
1110            "api_phone": query.get("api_phone"),
1111            "save-session": query.get("save-session"),
1112            "resolve-entities": query.get("resolve-entities"),
1113            "min_date": min_date,
1114            "max_date": max_date,
1115            "crawl-depth": query.get("crawl-depth"),
1116            "crawl-threshold": query.get("crawl-threshold"),
1117            "crawl-via-links": query.get("crawl-via-links")
1118        }
1119
1120    @staticmethod
1121    def create_session_id(api_phone, api_id, api_hash):
1122        """
1123        Generate a filename for the session file
1124
1125        This is a combination of phone number and API credentials, but hashed
1126        so that one cannot actually derive someone's phone number from it.
1127
1128        :param str api_phone:  Phone number for API ID
1129        :param int api_id:  Telegram API ID
1130        :param str api_hash:  Telegram API Hash
1131        :return str: A hash value derived from the input
1132        """
1133        hash_base = api_phone.strip().replace("+", "") + str(api_id).strip() + api_hash.strip()
1134        return hashlib.blake2b(hash_base.encode("ascii")).hexdigest()