import socket
from io import BytesIO
from datetime import datetime
STREAM_URL = "us4.internet-radio.com"
STREAM_PATH = "/proxy/radioestiloleblon?mp=/stream"
BUFFER_SIZE = 1024
PORT = 443
def stream_reader(sock):
sock.recv(BUFFER_SIZE) # http response
while data := sock.recv(BUFFER_SIZE):
print(f"[{datetime.now()}] Received data: {data[:10]}")
yield data
def get_socket():
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect((STREAM_URL, PORT))
s.send(bytes(f"GET {STREAM_PATH}; HTTP/1.1
Host:{STREAM_URL}
", encoding="UTF-8"))
return s
def write_buffer(filename, bytesio: BytesIO):
buffer = bytesio.getbuffer()
with open(filename, "wb") as file:
file.write(buffer)
print(f"[{datetime.now()}] Wrote buffer. Size: {buffer.nbytes} bytes")
def main():
s = get_socket()
with BytesIO() as bIO:
try:
for data in stream_reader(s):
bIO.write(data)
except KeyboardInterrupt: pass
finally: write_buffer("stream.mp3", bytesio=bIO)
if __name__ == "__main__":
main()