1import os
  2import asyncio
  3import subprocess
  4import tempfile
  5from aiohttp_retry import ExponentialRetry
  6from shazamio import Shazam, Serialize, HTTPClient, SearchParams
  7from yt_dlp import YoutubeDL
  8
  9
 10class Shazamnnn:
 11
 12  # change to relative/absolute path to executable
 13  # if not present on PATH env variable
 14  exec_ffmpeg = 'ffmpeg'
 15
 16  @staticmethod
 17  def is_url(s):
 18    # https://stackoverflow.com/questions/7160737/how-to-validate-a-url-in-python-malformed-or-not
 19    from urllib.parse import urlparse
 20    try:
 21      result = urlparse(s)
 22      return all([result.scheme, result.netloc])
 23    except AttributeError:
 24      return False
 25
 26  @staticmethod
 27  def _sec_to_time_sig(n):
 28    m, s = divmod(n, 60)
 29    h, m = divmod(m, 60)
 30    return f'{h:>02d}:{m:>02d}:{s:>02d}'
 31
 32  @staticmethod
 33  def _time_sig_to_sec(t):
 34    # this does not handle stupid formatted time signature
 35    # beyond its intended purpose like 0:2131231491:69 or 
 36    # foo:!!^*@^:zero or 2mins so be aware
 37    t, _, s = t.rpartition(':')
 38    h, _, m = t.rpartition(':')
 39    s = int(s) if s else 0
 40    m = int(m) if m else 0
 41    h = int(h) if h else 0
 42    m += h * 60
 43    s += m * 60
 44    return s
 45
 46  @classmethod
 47  def _parse_seconds(cls, t):
 48    if isinstance(t, str):
 49      return cls._time_sig_to_sec(t)
 50    return int(t)
 51
 52  @classmethod
 53  def _pack_ffmpeg_default(cls, src, dst, seek, duration, asamplerate=16000, aformat='ogg', acodec='libvorbis'):
 54    seek = cls._parse_seconds(seek) if seek else 0
 55    duration = cls._parse_seconds(duration) if duration else None
 56    ffmpeg_args = list()
 57    ffmpeg_args.extend(['-loglevel', 'warning'])
 58    ffmpeg_args.extend(['-ss', f'{seek}'])
 59    if src:
 60      ffmpeg_args.extend(['-i', src])
 61    if duration:
 62      ffmpeg_args.extend(['-t', f'{duration}'])
 63    if aformat:
 64      ffmpeg_args.extend(['-f', aformat])
 65    if asamplerate:
 66      ffmpeg_args.extend(['-ar', f'{asamplerate}'])
 67    if acodec:
 68      ffmpeg_args.extend(['-c:a', acodec])
 69    if dst:
 70      ffmpeg_args.append(dst if dst != '-' else 'pipe:')
 71    return ffmpeg_args
 72
 73  @classmethod
 74  def ytdl_audio_partial(cls, url, dst=None, seek=None, duration=None, asamplerate=16000, aformat='ogg', acodec='libvorbis'):
 75    ffmpeg_args = cls._pack_ffmpeg_default(None, None, seek, duration, asamplerate, aformat, acodec)
 76    opts = {}
 77    opts['overwrites'] = True
 78    opts['external_downloader'] = {'default': 'ffmpeg'}
 79    opts['external_downloader_args'] = {'ffmpeg': ffmpeg_args}
 80    opts['postprocessors'] = [
 81      {
 82        'key': 'FFmpegExtractAudio',
 83        'nopostoverwrites': False,
 84        'preferredcodec': 'best',
 85        'preferredquality': '5'
 86      }
 87    ]
 88    if dst:
 89      opts['outtmpl'] = {'default': dst}
 90    with YoutubeDL(opts) as ytdl:
 91      return ytdl.download(url)
 92
 93  @classmethod
 94  def ffmpeg_audio_partial(cls, src, dst=None, seek=None, duration=None, asamplerate=16000, aformat='ogg', acodec='libvorbis'): # 22050
 95    cmds = [cls.exec_ffmpeg]
 96    ffmpeg_args = cls._pack_ffmpeg_default(src, dst, seek, duration, asamplerate, aformat, acodec)
 97    cmds.extend(ffmpeg_args)
 98    pipe = subprocess.Popen(cmds, stdout=subprocess.PIPE)
 99    buf = b''
100    size = 2**12
101    while True:
102      b = pipe.stdout.read(size)
103      if not b: break
104      buf += b
105    return buf
106
107async def recognize_from_file(shazam_client, path, seek, duration, asamplerate=16000, aformat='ogg', acodec='libvorbis'):
108  buf = Shazamnnn.ffmpeg_audio_partial(path, '-', seek, duration, asamplerate, aformat, acodec)
109  duration = Shazamnnn._parse_seconds(duration)
110  dat = await shazam_client.recognize(buf, options=SearchParams(segment_duration_seconds=duration))
111  return dat
112
113async def recognize_from_url(shazam_client, url, seek, duration, asamplerate=16000, aformat='ogg', acodec='libvorbis'):
114  # ive tried various methods while avoiding subprocess 
115  # for piping ytdl output into stdout but no success so far,
116  # thus tempfile it is
117  fd, path = tempfile.mkstemp()
118  os.close(fd)
119  Shazamnnn.ytdl_audio_partial(url, path, seek, duration, asamplerate, aformat, acodec)
120  buf = Shazamnnn.ffmpeg_audio_partial(path, '-', 0, duration, asamplerate, aformat, acodec)
121  duration = Shazamnnn._parse_seconds(duration)
122  dat = await shazam_client.recognize(buf, options=SearchParams(segment_duration_seconds=duration))
123  os.remove(path)
124  return dat
125
126if __name__ == '__main__':
127  import json
128  import sys
129  from argparse import ArgumentParser
130
131  parser = ArgumentParser('shazamnnnnn')
132  parser.add_argument('source')
133  parser.add_argument('-s', '--seek', default='0')
134  parser.add_argument('-t', '--duration', default='10')
135
136  args = parser.parse_args()
137
138  client = Shazam(
139    http_client=HTTPClient(
140      retry_options=ExponentialRetry(
141        attempts=12, max_timeout=204.8, statuses={500, 502, 503, 504, 429}
142      )
143    )
144  )
145  
146  coro = recognize_from_url if Shazamnnn.is_url(args.source) else recognize_from_file
147
148  dat = asyncio.run(
149    coro(
150      client,
151      args.source,
152      args.seek,
153      args.duration
154      )
155    )
156
157  if dat: json.dump(dat, sys.stdout, indent=2)