# This file is part of torf.
#
# torf is free software: you can redistribute it and/or modify it under the
# terms of the GNU General Public License as published by the Free Software
# Foundation, either version 3 of the License, or (at your option) any later
# version.
#
# torf is distributed in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
# A PARTICULAR PURPOSE.  See the GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with torf.  If not, see <https://www.gnu.org/licenses/>.

import base64
import binascii
import io
import re
import time
import urllib
from collections import abc, defaultdict

from . import _errors as error
from . import _utils as utils


class Magnet:
    """
    BTIH Magnet URI

    :param str xt: eXact Topic: Info hash (Base 16 or 32)
    :param str dn: Display Name: Name of the torrent
    :param int xl: eXact Length: Size in bytes
    :param list tr: TRacker: Iterable of tracker URLs
    :param str xs: eXact Source: Torrent file URL
    :param str as\\_: Acceptable Source: Fallback torrent file URL
    :param list ws: WebSeeds: Iterable of webseed URLs (see BEP19)
    :param list kt: Keyword Topic: List of search keywords

    All keyword arguments that start with ``x_`` go into the :attr:`x`
    dictionary with the part after the underscore as the key.  They appear as
    "x.<name>" in the rendered URI.

    References:
        | https://www.bittorrent.org/beps/bep_0009.html
        | https://en.wikipedia.org/wiki/Magnet_URL
        | http://magnet-uri.sourceforge.net/magnet-draft-overview.txt
        | https://wiki.theory.org/index.php/BitTorrent_Magnet-URI_Webseeding
        | http://shareaza.sourceforge.net/mediawiki/index.php/Magnet_URI_scheme
    """

    _INFOHASH_REGEX = re.compile(r"^[0-9a-f]{40}|[a-z2-7]{32}$", flags=re.IGNORECASE)
    _XT_REGEX = re.compile(r"^urn:btih:([0-9a-f]{40}|[a-z2-7]{32})$", flags=re.IGNORECASE)

    def __init__(
        self,
        xt,
        *,
        dn=None,
        xl=None,
        tr=None,
        xs=None,
        as_=None,
        ws=None,
        kt=None,
        **kwargs,
    ):
        self._tr = utils.MonitoredList(type=utils.URL)
        self._ws = utils.MonitoredList(type=utils.URL)
        self.xt = xt
        self.dn = dn
        self.xl = xl
        self.tr = tr
        self.xs = xs
        self.as_ = as_
        self.ws = ws
        self.kt = kt

        self._x = defaultdict(lambda: None)
        for key in tuple(kwargs):
            if key.startswith("x_"):
                self._x[key[2:]] = kwargs.pop(key)

        if kwargs:
            key, value = next(iter(kwargs.items()))
            raise TypeError(f"Unrecognized argument: {key}={value!r}")

    @property
    def dn(self):
        """Display Name: Name of the torrent or ``None``"""
        return self._dn

    @dn.setter
    def dn(self, value):
        self._dn = str(value).replace("\n", " ") if value is not None else None

    @property
    def xt(self):
        """
        EXact Topic: URN containing the info hash as base 16 or base 32

        Example:
            urn:btih:3bb9561e35b06175bb6d2c2330578dc83846cc5d

        For convenience, this property may be set to the info hash without the
        ``urn:btih`` part.

        :raises MagnetError: if set to an invalid value
        """
        return f"urn:btih:{self._infohash}"

    @xt.setter
    def xt(self, value):
        value = str(value)
        if self._INFOHASH_REGEX.match(value):
            self._infohash = value
        else:
            match = self._XT_REGEX.match(value)
            if match:
                self._infohash = match.group(1)
        if not hasattr(self, "_infohash"):
            raise error.MagnetError(value, 'Invalid exact topic ("xt")')

    @property
    def infohash(self):
        """
        Info hash as base 16 or base 32

        :raises MagnetError: if set to an invalid value
        """
        return self._infohash

    @infohash.setter
    def infohash(self, value):
        value = str(value)
        match = self._INFOHASH_REGEX.match(value)
        if match:
            self._infohash = value
        else:
            raise error.MagnetError(value, "Invalid info hash")

    @property
    def xl(self):
        """
        EXact Length: Size in bytes or ``None``

        :raises MagnetError: if set to an invalid value
        """
        return self._xl

    @xl.setter
    def xl(self, value):
        if value is not None:
            try:
                value = int(value)
            except ValueError:
                raise error.MagnetError(value, 'Invalid exact length ("xl")')
            else:
                if value < 1:
                    raise error.MagnetError(value, "Must be 1 or larger")
                else:
                    self._xl = value
        else:
            self._xl = None

    @property
    def tr(self):
        """
        TRackers: List of tracker URLs, single tracker URL or ``None``

        :raises URLError: if any of the URLs is invalid
        """
        return self._tr

    @tr.setter
    def tr(self, value):
        if value is None:
            self._tr.clear()
        elif isinstance(value, str):
            self._tr.replace((value,))
        else:
            self._tr.replace(value)

    @property
    def xs(self):
        """
        EXact Source: Torrent file URL or ``None``

        :raises URLError: if the URL is invalid
        """
        return self._xs

    @xs.setter
    def xs(self, value):
        self._xs = utils.URL(value) if value is not None else None

    @property
    def as_(self):
        """
        Acceptable Source: Fallback torrent file URL or ``None``

        (The trailing underscore is needed because "as" is a keyword in Python.)

        :raises URLError: if the URL is invalid
        """
        return self._as

    @as_.setter
    def as_(self, value):
        self._as = utils.URL(value) if value is not None else None

    @property
    def ws(self):
        """
        WebSeeds: List of webseed URLs, single webseed URL or ``None``

        See BEP19.

        :raises URLError: if any of the URLs is invalid
        """
        return self._ws

    @ws.setter
    def ws(self, value):
        if value is None:
            self._ws.clear()
        elif isinstance(value, str):
            self._ws.replace((value,))
        else:
            self._ws.replace(value)

    @property
    def kt(self):
        """Keyword Topic: List of search keywords or ``None``"""
        return self._kt

    @kt.setter
    def kt(self, value):
        if value is None:
            self._kt = []
        elif isinstance(value, str):
            self._kt = [value]
        elif isinstance(value, abc.Iterable):
            self._kt = [str(v) for v in value] if value is not None else None
        else:
            raise error.MagnetError(value, 'Invalid keyword topic ("kt")')

    @property
    def x(self):
        """
        Mapping of custom keys to their values

        For example, "x.pe" (a peer address) would be accessed as
        ``magnet.x['pe']``.
        """
        return self._x

    def torrent(self):
        """:class:`Torrent` instance"""
        # Prevent circular import issues
        from ._torrent import Torrent

        torrent = Torrent()
        torrent.name = self.dn
        if self.tr:
            torrent.trackers = self.tr
        if self.ws:
            torrent.webseeds = self.ws
        if self.xl:
            torrent._metainfo["info"]["length"] = self.xl
        if hasattr(self, "_info"):
            torrent.metainfo["info"] = self._info
        elif len(self.infohash) == 40:
            torrent._infohash = self.infohash
        else:
            # Convert base 32 to base 16 (SHA1)
            torrent._infohash = (
                base64.b16encode(base64.b32decode(self.infohash)).decode("utf-8").lower()
            )
        return torrent

    def get_info(self, validate=True, timeout=60, callback=None):
        """
        Download the torrent's "info" section

        Try the following sources in this order: :attr:`xs`, :attr:`as`,
        :attr:`tr`

        :meth:`torrent` can only return a complete torrent if this method is
        called first.

        :param validate: Whether to ensure the downloaded "info" section is
            valid
        :param timeout: Give up after this many seconds
        :type timeout: int, float
        :param callback callable: Callable that is called with a
            :class:`TorfError` instance if a source is specified but fails

        :return: ``True`` if the "info" section was successfully downloaded,
            ``False`` otherwise
        """

        def success():
            return hasattr(self, "_info")

        torrent_urls = []
        if self.xs:
            torrent_urls.append(self.xs)
        if self.as_:
            torrent_urls.append(self.as_)
        torrent_urls.extend(url.rstrip("/") + ".torrent" for url in self.ws)
        # I couldn't find any documentation for the "/file?info_hash=..." GET request, but
        # it seems to work for HTTP trackers.
        # https://stackoverflow.com/a/1019588
        for url in self.tr:
            if url.scheme in ("http", "https"):
                infohash_enc = urllib.parse.quote_from_bytes(binascii.unhexlify(self.infohash))
                torrent_urls.append(f"{url.scheme}://{url.netloc}/file?info_hash={infohash_enc}")

        start = time.monotonic()
        for url in torrent_urls:
            to = timeout - (time.monotonic() - start)
            try:
                torrent = utils.download(url, timeout=to)
            except error.ConnectionError as e:
                if callback:
                    callback(e)
            else:
                self._set_info_from_torrent(torrent, validate, callback)
            if success() or to <= 0:
                break

        return success()

    def _set_info_from_torrent(self, torrent_data, validate=True, callback=False):
        """Extract "info" section from `torrent_data` for :meth:`torrent`"""
        # Prevent circular import issues
        from ._torrent import Torrent

        stream = io.BytesIO(torrent_data)
        try:
            torrent = Torrent.read_stream(stream, validate=validate)
        except error.TorfError as e:
            if callback:
                callback(e)
        else:
            if validate and self.infohash != torrent.infohash:
                raise error.MetainfoError(
                    f"Mismatching info hashes: {self.infohash} != {torrent.infohash}"
                )
            elif torrent.metainfo["info"]:
                self._info = torrent.metainfo["info"]

    _KNOWN_PARAMETERS = ("xt", "dn", "xl", "tr", "xs", "as", "ws", "kt")

    @classmethod
    def from_string(cls, uri):
        """
        Create :class:`Magnet` URI from string

        :raises URLError: if `uri` contains an invalid URL (e.g. :attr:`tr`)
        :raises MagnetError: if `uri` is not a valid magnet URI
        """
        info = urllib.parse.urlparse(uri.strip(), scheme="magnet", allow_fragments=False)
        if not info.scheme == "magnet":
            raise error.MagnetError(uri, "Not a magnet URI")
        else:
            query = urllib.parse.parse_qs(info.query)

        # Check for unknown parameters
        for key in query:
            if key not in cls._KNOWN_PARAMETERS and not key.startswith("x_"):
                raise error.MagnetError(uri, f"{key}: Unknown parameter")

        if "xt" not in query:
            raise error.MagnetError(uri, 'Missing exact topic ("xt")')
        elif len(query["xt"]) > 1:
            raise error.MagnetError(uri, 'Multiple exact topics ("xt")')
        else:
            self = cls(xt=query["xt"][0])

        # Parameters that accept only one value
        for param, attr, name, parse in (
            ("dn", "dn", "display name", lambda v: v),
            ("xl", "xl", "exact length", lambda v: v),
            ("xs", "xs", "exact source", lambda v: v),
            ("as", "as_", "acceptable source", lambda v: v),
            ("kt", "kt", "keyword topic", lambda v: v.split()),
        ):
            if param in query:
                if len(query[param]) > 1:
                    raise error.MagnetError(uri, f'Multiple {name}s ("{param}")')
                else:
                    setattr(self, attr, parse(query[param][0]))

        # Parameters that accept multiple values
        for param, name in (("tr", "tracker"), ("ws", "webseed")):
            if param in query:
                setattr(self, param, query[param])

        return self

    def __str__(self):
        uri = [f"magnet:?xt={self.xt}"]

        for key in ("dn", "xl", "xs", "as_"):
            value = getattr(self, f"{key}")
            if value is not None:
                if isinstance(value, str):
                    uri.append(f"{key}={utils.urlquote(value)}")
                else:
                    uri.append(f"{key}={value}")

        if self.kt:
            uri.append(f"kt={'+'.join(utils.urlquote(k) for k in self.kt)}")

        for key in ("tr", "ws"):
            seq = getattr(self, f"{key}")
            if seq is not None:
                for item in seq:
                    uri.append(f"{key}={utils.urlquote(item)}")

        for key, value in self._x.items():
            uri.append(f"x.{key}={utils.urlquote(value)}")

        return "&".join(uri)

    def __repr__(self):
        clsname = type(self).__name__
        kwargs = {}
        for param in self._KNOWN_PARAMETERS:
            if param == "as":
                param = "as_"
            value = getattr(self, param)
            if value:
                kwargs[param] = value
        for k, v in self.x.items():
            kwargs[f"x_{k}"] = v
        kwargs_str = ", ".join(f"{k}={v!r}" for k, v in kwargs.items())
        return f"{clsname}({kwargs_str})"