pSWGn4nU

1import argparse
2import sqlite3
3import time
4import requests
5
6# Arguments
7parser = argparse.ArgumentParser()
8parser.add_argument('url')
9parser.add_argument('-f', required=False, default='myRadio.mp3')
10parser.add_argument('--duration', '-d', type=int, default=20)
11parser.add_argument('--list', '-l', required=False)
12
13
14class Utils:
15    @staticmethod
16    def run_sql_command(db_file, command, args=None):
17        if args is None:
18            args = []
19        db = sqlite3.connect(db_file)
20        cursor = db.cursor()
21        cursor.execute(command, args)
22        db.commit()
23        return cursor.fetchall()
24
25
26class AudioRecorder:
27
28    def __init__(self, db_file) -> None:
29        super().__init__()
30        self.db = db_file
31        Utils.run_sql_command(self.db,
32                              'create table if not exists recordings (name text not null primary key, source text not '
33                              'null, duration integer not null)')
34
35    def save_audio_stream_to_file(self, file_name, url, duration):
36        r = requests.get(url, timeout=duration, stream=True)
37        end_time = time.time() + duration
38        with open(file_name, 'wb') as f:
39            for block in r.iter_content(1024):
40                if time.time() >= end_time:
41                    break
42                f.write(block)
43        Utils.run_sql_command(self.db, 'insert or replace into recordings values (?, ?, ?)', [file_name, url, duration])
44
45    def get_all_recordings(self):
46        res = Utils.run_sql_command(self.db, 'select * from recordings')
47        for row in res:
48            print(row)
49
50
51if __name__ == '__main__':
52    arg = parser.parse_args()
53    ad = AudioRecorder(db_file='audio_db.sqlite')
54    ad.save_audio_stream_to_file(arg.f, arg.url, arg.duration)
55    if arg.list:
56        ad.get_all_recordings()