Repository: pR0Ps/slack-to-discord (language: python)

Issue:
Shard ID None heartbeat blocked
Thanks for your work on this! I've tried a few slack to discord tools and this is the only one I could actually get working.

I got part way through an import, then it stalled and I started getting this (repeats every 10 seconds):

`2022-10-08 09:31:45 WARNING  discord.gateway Shard ID None heartbeat blocked for more than 35830 seconds.
Loop thread traceback (most recent call last):
  File "/Users/benishs/opt/anaconda3/bin/slack-to-discord", line 8, in <module>
    sys.exit(main())
  File "/Users/benishs/opt/anaconda3/lib/python3.8/site-packages/slack_to_discord.py", line 584, in main
    run_import(
  File "/Users/benishs/opt/anaconda3/lib/python3.8/site-packages/slack_to_discord.py", line 563, in run_import
    client.run(token, reconnect=False, log_handler=None)
  File "/Users/benishs/opt/anaconda3/lib/python3.8/site-packages/discord/client.py", line 828, in run
    asyncio.run(runner())
  File "/Users/benishs/opt/anaconda3/lib/python3.8/asyncio/runners.py", line 43, in run
    return loop.run_until_complete(main)
  File "/Users/benishs/opt/anaconda3/lib/python3.8/asyncio/base_events.py", line 603, in run_until_complete
    self.run_forever()
  File "/Users/benishs/opt/anaconda3/lib/python3.8/asyncio/base_events.py", line 570, in run_forever
    self._run_once()
  File "/Users/benishs/opt/anaconda3/lib/python3.8/asyncio/base_events.py", line 1859, in _run_once
    handle._run()
  File "/Users/benishs/opt/anaconda3/lib/python3.8/asyncio/events.py", line 81, in _run
    self._context.run(self._callback, *self._args)
  File "/Users/benishs/opt/anaconda3/lib/python3.8/site-packages/discord/client.py", line 409, in _run_event
    await coro(*args, **kwargs)
  File "/Users/benishs/opt/anaconda3/lib/python3.8/site-packages/slack_to_discord.py", line 420, in on_ready
    await self._run_import(g)
  File "/Users/benishs/opt/anaconda3/lib/python3.8/site-packages/slack_to_discord.py", line 518, in _run_import
    sent = await self._send_slack_msg(ch_send, msg)
  File "/Users/benishs/opt/anaconda3/lib/python3.8/site-packages/slack_to_discord.py", line 442, in _send_slack_msg
    for attempt in file_upload_attempts(data):
  File "/Users/benishs/opt/anaconda3/lib/python3.8/site-packages/slack_to_discord.py", line 368, in file_upload_attempts
    fp=io.BytesIO(urllib.request.urlopen(url).read()),
  File "/Users/benishs/opt/anaconda3/lib/python3.8/urllib/request.py", line 222, in urlopen
    return opener.open(url, data, timeout)
  File "/Users/benishs/opt/anaconda3/lib/python3.8/urllib/request.py", line 525, in open
    response = self._open(req, data)
  File "/Users/benishs/opt/anaconda3/lib/python3.8/urllib/request.py", line 542, in _open
    result = self._call_chain(self.handle_open, protocol, protocol +
  File "/Users/benishs/opt/anaconda3/lib/python3.8/urllib/request.py", line 502, in _call_chain
    result = func(*args)
  File "/Users/benishs/opt/anaconda3/lib/python3.8/urllib/request.py", line 1393, in https_open
    return self.do_open(http.client.HTTPSConnection, req,
  File "/Users/benishs/opt/anaconda3/lib/python3.8/urllib/request.py", line 1354, in do_open
    r = h.getresponse()
  File "/Users/benishs/opt/anaconda3/lib/python3.8/http/client.py", line 1332, in getresponse
    response.begin()
  File "/Users/benishs/opt/anaconda3/lib/python3.8/http/client.py", line 303, in begin
    version, status, reason = self._read_status()
  File "/Users/benishs/opt/anaconda3/lib/python3.8/http/client.py", line 264, in _read_status
    line = str(self.fp.readline(_MAXLINE + 1), "iso-8859-1")
  File "/Users/benishs/opt/anaconda3/lib/python3.8/socket.py", line 669, in readinto
    return self._sock.recv_into(b)
  File "/Users/benishs/opt/anaconda3/lib/python3.8/ssl.py", line 1241, in recv_into
    return self.read(nbytes, buffer)
  File "/Users/benishs/opt/anaconda3/lib/python3.8/ssl.py", line 1099, in read
    return self._sslobj.read(len, buffer)`

Is there anything I can do to make it recover and continue the import? (It's still going, at the moment.)

If I re-run slack-to-discord will it ignore the channels / messages it has already imported, or do I need to start over? When I first tried to run slack-to-discord I attempted to remove some unneeded channel directories from the Slack export archive (unzipped, removed some directories, re-zipped), but slack-to-discord wouldn't start – an error about not finding the Users file. It would only work with the original exported .zip from Slack. So removing the already imported channels from that export doesn't seem to be an option.

Anything I can do to prevent this from happening again next time I run it?

Thanks,

Scott 


Discussion / hints:
Ah, this is happening because the program blocks while it's downloading a file from Slack so it can upload it to Discord. This causes the Discord session to emit warnings because it can't properly keep the session alive.

I've already been working on a fix for this that streams the data from Slack to Discord instead of fully downloading it before starting the upload. Should be able to push something soon.

If you run it again, it will start from the beginning so you'll want to delete all the channels it did import from Discord before you run it again.

It should also work to delete the channel data from the zip file but might require tweaking some of the other files so they don't reference the channel anymore - I don't remember offhand. I typically just deleted everything and ran the full import again if I hit issues.
Thanks for the quick reply!

I again tried deleting some channels from the archive, because it took awhile to get through the stuff it has already done, but am getting the same error as before – in case that's useful:

`2022-10-08 10:30:52 INFO     slack_to_discord Logging the bot into Discord
Traceback (most recent call last):
  File "/Users/benishs/opt/anaconda3/bin/slack-to-discord", line 8, in <module>
    sys.exit(main())
  File "/Users/benishs/opt/anaconda3/lib/python3.8/site-packages/slack_to_discord.py", line 584, in main
    run_import(
  File "/Users/benishs/opt/anaconda3/lib/python3.8/site-packages/slack_to_discord.py", line 562, in run_import
    client = SlackImportClient(data_dir=t, **kwargs)
  File "/Users/benishs/opt/anaconda3/lib/python3.8/site-packages/slack_to_discord.py", line 398, in __init__
    self._users = slack_usermap(data_dir, real_names=real_names)
  File "/Users/benishs/opt/anaconda3/lib/python3.8/site-packages/slack_to_discord.py", line 102, in slack_usermap
    with open(os.path.join(d, "users.json"), 'rb') as fp:
FileNotFoundError: [Errno 2] No such file or directory: '/var/folders/q8/n_7347m91r9dhc1m_lh271180000gn/T/tmpnr4ij0hy/users.json'`

It seems to think `users.json` doesn't exist? But it definitely exists in that .zip, and I didn't touch that `users.json` file when I removed some of the channels. Even just removing 1 channel (that was only posted to by a bot) gives this error. ¯\_(ツ)_/¯

If the easiest solution is to just start over, I can do that.  

Thank you!
So, I edited the contents of the .zip without extracting it, using an old app called [Zipster](https://www.howtogeek.com/308468/how-to-open-and-browse-zip-files-on-macos-without-unarchiving-them/), and it seems to be working now with the truncated data.

Not surprisingly, it gave a bunch of errors about the missing channels when it started, e.g.
`Data for channel '#general' not found in export` 
but then proceeded to start on the channels that still exist in the archive.

So, something about unzipping and then re-zipping the archive was causing the issue? That makes no sense to me, but I know very little about this stuff. FWIW, I was previously just using the built in unzip / zip tools in the Finder (macOS Monterey 12.6). Anyway, maybe that helps someone else out someday!

Fix patch:
diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..a732f71
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,27 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+build/
+develop-eggs/
+dist/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# virtualenv
+.venv/
+venv/
diff --git a/setup.py b/setup.py
index 0a0e613..08c3f6c 100644
--- a/setup.py
+++ b/setup.py
@@ -33,12 +33,13 @@
         "License :: OSI Approved :: GNU General Public License v3 (GPLv3)"
     ],
     install_requires = [
-        "discord.py>=2.0.0,<3.0.0"
+        "discord.py>=2.0.0,<3.0.0",
+        "urllib3>=1.26.12,<2.0.0",
     ],
-    py_modules=["slack_to_discord"],
+    packages=["slack_to_discord"],
     entry_points={
         "console_scripts": [
-            "slack-to-discord=slack_to_discord:main"
+            "slack-to-discord=slack_to_discord.__main__:main"
         ]
     },
 )
diff --git a/slack_to_discord/__init__.py b/slack_to_discord/__init__.py
new file mode 100644
index 0000000..21d17fa
--- /dev/null
+++ b/slack_to_discord/__init__.py
@@ -0,0 +1,4 @@
+#!/usr/bin/env python
+
+from slack_to_discord.importer import run_import
+from slack_to_discord.http_stream import SeekableHTTPStream
diff --git a/slack_to_discord/__main__.py b/slack_to_discord/__main__.py
new file mode 100644
index 0000000..c616c14
--- /dev/null
+++ b/slack_to_discord/__main__.py
@@ -0,0 +1,37 @@
+#!/usr/bin/env python
+
+import argparse
+import logging
+from discord.utils import setup_logging
+
+from slack_to_discord import run_import
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Import Slack chat history into Discord"
+    )
+    parser.add_argument("-z", "--zipfile", help="The Slack export zip file", required=True)
+    parser.add_argument("-g", "--guild", help="The Discord Guild to import history into", required=True)
+    parser.add_argument("-t", "--token", help="The Discord bot token", required=True)
+    parser.add_argument("-s", "--start", help="The date to start importing from", required=False, default=None)
+    parser.add_argument("-e", "--end", help="The date to end importing at", required=False, default=None)
+    parser.add_argument("-p", "--all-private", help="Import all channels as private channels in Discord", action="store_true", default=False)
+    parser.add_argument("-r", "--real-names", help="Use real names from Slack instead of usernames", action="store_true", default=False)
+    parser.add_argument("-v", "--verbose", help="Show more verbose logs", action="store_true")
+    args = parser.parse_args()
+
+    setup_logging(level=logging.DEBUG if args.verbose else logging.INFO)
+
+    run_import(
+        zipfile=args.zipfile,
+        token=args.token,
+        guild_name=args.guild,
+        all_private=args.all_private,
+        real_names=args.real_names,
+        start=args.start,
+        end=args.end
+    )
+
+if __name__ == "__main__":
+    main()
diff --git a/slack_to_discord/http_stream.py b/slack_to_discord/http_stream.py
new file mode 100644
index 0000000..a17ff5d
--- /dev/null
+++ b/slack_to_discord/http_stream.py
@@ -0,0 +1,232 @@
+#!/usr/bin/env python
+
+import io
+
+import urllib3
+
+
+class IterBuffer:
+    """Provides a buffered readable interface for an iterator"""
+
+    def __init__(self, iterator=None):
+        self.reset(iterator)
+
+    def __len__(self):
+        """How much data is available to be read from the buffer"""
+        return len(self._data) - self._pos
+
+    def _buff_read(self, size=-1):
+        l = len(self)
+        if size < 0 or size > l:
+            size = l
+        if size < 1:
+            return b""
+
+        ret = bytes(self._data if size == l else self._data[self._pos:self._pos+size])
+        self._pos += size
+        return ret
+
+    def _buff_fill(self, size, read1=False):
+        """Ensure enough data is in the buffer"""
+        if size == 0 or self._iter is None:
+            return
+
+        cur_size = len(self)
+        if size > 0 and cur_size >= size:
+            # have the data, don't need to read anything
+            return
+
+        if self._pos > 0:
+            self._data = self._data[self._pos:]
+            self._pos = 0
+
+        for x in self._iter:
+            self._data += x
+            cur_size += len(x)
+            if read1 or (size > 0 and cur_size >= size):
+                break
+        else:
+            self._iter = None
+
+    def relseek(self, num):
+        """Relative seek within the buffer
+
+        Given the number of bytes to seek within the buffer (-/+), get as close
+        as possible and return the actual seek amount
+        """
+        actual = max(-self._pos, min(num, len(self)))
+        self._pos += actual
+        return actual
+
+    def read(self, size=-1):
+        """Read and return up to `size` bytes.
+
+        If the argument is omitted or negative, data is read and returned until
+        EOF is reached. An empty bytes object is returned if the stream is
+        already at EOF
+        """
+        self._buff_fill(size)
+        return self._buff_read(size)
+
+    def read1(self, size=-1):
+        """Read and return up to `size` bytes with at most one call to the iterator"""
+        self._buff_fill(size, read1=True)
+        return self._buff_read(size)
+
+    def reset(self, iterator=None):
+        """Zero out the current buffer and assign a new iterator to buffer bytes from"""
+        self._iter = iterator
+        self._data = bytearray()
+        self._pos = 0
+
+
+class SeekableHTTPStream(io.BufferedIOBase):
+    """Make the contents at a URL addressable via a seekable file-like object.
+
+    Seeking to arbitrary offsets is handled using HTTP range requests.
+
+    Notes:
+     - The server must respond with an `Accept-Ranges: bytes` header for seeking to be supported.
+     - Using `__len__` or `io.SEEK_END` to seek requires the server to have sent a valid `Content-Length` header.
+    """
+
+    def __init__(self, url, chunk_size=64*1024):
+        self._pos = 0
+        self._url = url
+        self._pool = urllib3.PoolManager()
+        self._resp = None
+        self._buff = IterBuffer()
+
+        self._chunk_size = chunk_size
+        self._do_request()
+
+        try:
+            self._content_length = int(self._resp.getheader("Content-Length"))
+        except TypeError:
+            self._content_length = None
+
+        self._seekable = self._resp.getheader("Accept-Ranges", "").lower() == "bytes"
+
+    def _reset(self):
+        if self._resp:
+            # release the connection back into the pool
+            self._resp.release_conn()
+        self._resp = None
+        self._buff.reset()
+
+    def _do_request(self, start=0):
+        self._reset()
+
+        headers = {}
+        if start > 0:
+            headers["Range"] = "bytes={}-".format(start)
+
+        resp = self._pool.request(
+            "GET",
+            self._url,
+            headers=headers,
+            preload_content=False,
+        )
+        if start > 0 and self._content_length is None and resp.status == 416:
+            # Hit the end of the file - no more data
+            self._reset()
+        elif resp.status not in (200, 206):
+            self._reset()
+            raise urllib3.exceptions.HTTPError(
+                "Bad status code: {}".format(resp.status)
+            )
+        else:
+            self._resp = resp
+            self._buff.reset(resp.stream(amt=self._chunk_size, decode_content=True))
+
+    def __len__(self):
+        if self._content_length is None:
+            raise TypeError("The length of this {} is unknown".format(self.__class__.__name__))
+        return self._content_length
+
+    def close(self):
+        self._reset()
+        super().close()
+
+    def writable(self):
+        return False
+
+    def readable(self):
+        return True
+
+    def seekable(self):
+        return self._seekable
+
+    def tell(self):
+        return self._pos
+
+    def detach(self):
+        raise io.UnsupportedOperation()
+
+    def read(self, size=-1):
+        ret = self._buff.read(size)
+        self._pos += len(ret)
+        return ret
+
+    def read1(self, size=-1):
+        ret = self._buff.read1(size)
+        self._pos += len(ret)
+        return ret
+
+    def readinto(self, b):
+        data = self.read(len(b))
+        l = len(data)
+        b[:l] = data
+        return l
+
+    def readinto1(self, b):
+        data = self.read1(len(b))
+        l = len(data)
+        b[:l] = data
+        return l
+
+    def seek(self, offset, whence=io.SEEK_SET):
+        if not self.seekable():
+            raise io.UnsupportedOperation(f"URL '{self._url}' does not support range requests")
+
+        if whence == io.SEEK_SET:
+            new_pos = offset
+        elif whence == io.SEEK_CUR:
+            new_pos = self._pos + offset
+        elif whence == io.SEEK_END:
+            if self._content_length is None:
+                raise io.UnsupportedOperation("can't do end-relative seeks without knowing the length")
+            new_pos = len(self) + offset
+        else:
+            raise ValueError("Invalid whence: {}".format(whence))
+
+        if self._content_length is not None:
+            new_pos = min(new_pos, self._content_length)
+        new_pos = max(0, new_pos)
+
+        if new_pos == self._content_length:
+            # seeking to end - no more data
+            self._reset()
+        else:
+            # Figure out how far off we are and seek within the buffered data
+            # to get as close as possible to the target offset
+            pos_diff = new_pos - self._pos
+            pos_diff -= self._buff.relseek(pos_diff)
+
+            if pos_diff == 0:
+                # seeked to an already-buffered offset - nothing else to do
+                pass
+            elif 0 < pos_diff < self._chunk_size * 2:
+                # seeking forwards and we're close enough (within 2 iterations of
+                # the current request) that it makes sense to avoid doing another
+                # fresh HTTP request - read the data until we get to the target
+                # offset
+                self.read(pos_diff)
+            else:
+                # seekable stream and we're before the current position or far
+                # enough ahead that we don't want to read everything up to it - do
+                # a range request starting at the requested offset
+                self._do_request(new_pos)
+
+        self._pos = new_pos
+        return self._pos
diff --git a/slack_to_discord.py b/slack_to_discord/importer.py
similarity index 93%
rename from slack_to_discord.py
rename to slack_to_discord/importer.py
index 0141810..968e0d4 100644
--- a/slack_to_discord.py
+++ b/slack_to_discord/importer.py
@@ -1,18 +1,15 @@
-#1/usr/bin/env python
+#!/usr/bin/env python
 
-import argparse
 import contextlib
 import functools
 import glob
 import html
-import io
 import json
 import logging
 import os
 import re
 import tempfile
 import textwrap
-import urllib
 from zipfile import ZipFile
 from datetime import datetime
 from urllib.parse import urlparse
@@ -20,6 +17,8 @@
 import discord
 from discord.errors import Forbidden
 
+from slack_to_discord.http_stream import SeekableHTTPStream
+
 
 # Discord size limits
 MAX_MESSAGE_SIZE = 2000
@@ -372,7 +371,7 @@ def file_upload_attempts(data):
 
         try:
             f = discord.File(
-                fp=io.BytesIO(urllib.request.urlopen(url).read()),
+                fp=SeekableHTTPStream(url),
                 filename=filename
             )
         except Exception:
@@ -578,33 +577,3 @@ def run_import(*, zipfile, token, **kwargs):
         client.run(token, reconnect=False, log_handler=None)
         if client._exception:
             raise client._exception
-
-
-def main():
-    parser = argparse.ArgumentParser(
-        description="Import Slack chat history into Discord"
-    )
-    parser.add_argument("-z", "--zipfile", help="The Slack export zip file", required=True)
-    parser.add_argument("-g", "--guild", help="The Discord Guild to import history into", required=True)
-    parser.add_argument("-t", "--token", help="The Discord bot token", required=True)
-    parser.add_argument("-s", "--start", help="The date to start importing from", required=False, default=None)
-    parser.add_argument("-e", "--end", help="The date to end importing at", required=False, default=None)
-    parser.add_argument("-p", "--all-private", help="Import all channels as private channels in Discord", action="store_true", default=False)
-    parser.add_argument("-r", "--real-names", help="Use real names from Slack instead of usernames", action="store_true", default=False)
-    parser.add_argument("-v", "--verbose", help="Show more verbose logs", action="store_true")
-    args = parser.parse_args()
-
-    discord.utils.setup_logging(level=logging.DEBUG if args.verbose else logging.INFO)
-
-    run_import(
-        zipfile=args.zipfile,
-        token=args.token,
-        guild_name=args.guild,
-        all_private=args.all_private,
-        real_names=args.real_names,
-        start=args.start,
-        end=args.end
-    )
-
-if __name__ == "__main__":
-    main()


Test patch:
diff --git a/tests/test_http_stream.py b/tests/test_http_stream.py
new file mode 100755
index 0000000..fe53ab4
--- /dev/null
+++ b/tests/test_http_stream.py
@@ -0,0 +1,111 @@
+#!/usr/bin/env python
+
+from http import HTTPStatus
+import http.server
+import io
+import threading
+
+import pytest
+
+from slack_to_discord import SeekableHTTPStream
+
+
+RESP_SIZE = 100
+
+
+# TODO: make tests for non-seekable and/or unknown length responses
+
+def gen_bytes(s, e):
+    return b"".join(bytes([x]) for x in range(s, e))
+
+
+class HTTPRangeRequestHandler(http.server.SimpleHTTPRequestHandler):
+    def do_GET(self):
+        range_ = self.headers.get("Range")
+        start, end = 0, RESP_SIZE
+        if range_ is not None:
+            _, r = range_.split("bytes=", 1)
+            s, e = r.split("-", 1)
+            if e:
+                end = int(e)
+            if s:
+                start = int(s)
+
+            if not 0 <= start < end <= RESP_SIZE:
+                self.send_response(HTTPStatus.REQUESTED_RANGE_NOT_SATISFIABLE)
+                self.end_headers()
+                return
+            else:
+                self.send_response(HTTPStatus.PARTIAL_CONTENT)
+                self.send_header("Content-Range", f"bytes {start}-{end-1}/{end}")
+        else:
+            self.send_response(HTTPStatus.OK)
+
+        self.send_header("Accept-Ranges", "bytes")
+        self.send_header("Content-Length", RESP_SIZE)
+        self.end_headers()
+        self.wfile.write(gen_bytes(start, end))
+
+
+@pytest.fixture(scope="module")
+def mockserver():
+    s = http.server.HTTPServer(("127.0.0.1", 0), HTTPRangeRequestHandler)
+    thread = threading.Thread(target=s.serve_forever)
+    thread.start()
+    try:
+        yield f"http://127.0.0.1:{s.server_port}"
+    finally:
+        s.shutdown()
+        thread.join()
+
+
+def test_server(mockserver):
+    s = SeekableHTTPStream(mockserver, chunk_size=10)
+
+    assert len(s) == RESP_SIZE
+
+    # test reading chunks
+    assert s.read(10) == gen_bytes(0, 10)
+    assert s.read(10) == gen_bytes(10, 20)
+    assert s.read(10) == gen_bytes(20, 30)
+
+    # test seeking
+    assert s.seek(10, io.SEEK_CUR) == 40
+    assert s.seek(-10, io.SEEK_CUR) == 30
+    assert s.seek(0, io.SEEK_END) == RESP_SIZE
+    assert s.seek(-10, io.SEEK_END) == RESP_SIZE - 10
+    assert s.seek(5, io.SEEK_SET) == 5
+
+    # test reading across chunks
+    assert s.read(10) == gen_bytes(5, 15)
+    assert s.tell() == 15
+
+    # test seeking bck from current pos and getting data we just got just reads
+    # it from the buffer
+    assert s.seek(-10, io.SEEK_CUR) == 5
+    assert s.read(10) == gen_bytes(5, 15)
+
+    # test seeking past the end stops at the end and returns no data
+    assert s.seek(RESP_SIZE * 2) == RESP_SIZE
+    assert s.read() == b""
+
+    # test read with no args gets everything
+    assert s.seek(50) == 50
+    assert s.read() == gen_bytes(50, 100)
+
+    # test read1 only reads at most 1 chunk from the stream
+    assert s.seek(0) == 0
+    assert s.read(5) == gen_bytes(0, 5)
+    assert s.read1() == gen_bytes(5, 20) # 5 in the buff + 1 read of 10 bytes
+
+    # test readinto
+    b = bytearray(20)
+    assert s.seek(5) == 5
+    assert s.readinto(b) == 20
+    assert s.tell() == 25
+    assert bytes(b) == gen_bytes(5, 25)
+
+    # test readinto1 (partial read)
+    b = bytearray(20)
+    assert s.readinto1(b) == 15
+    assert bytes(b) == gen_bytes(25, 40) + bytes(5)