-
Notifications
You must be signed in to change notification settings - Fork 0
/
app.py
138 lines (116 loc) · 5.42 KB
/
app.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
import os
from urllib.parse import urlparse
from loguru import logger
import dotenv as dotenv
from flask import Flask, request, jsonify
from selenium import webdriver
from selenium.common import TimeoutException, NoSuchElementException
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.support import expected_conditions as ec
import time
from selenium.webdriver.support.wait import WebDriverWait
from selenium_stealth import stealth
# populate the environment variables
dotenv.load_dotenv()
# env vars
HEADLESS = os.getenv('HEADLESS', 'True').lower() == 'true'
DEPLOYMENT = os.getenv('DEPLOYMENT', 'local')
app = Flask(__name__)
def validate_url(music_url):
parsed_url = urlparse(music_url)
return parsed_url.scheme in ['http', 'https'] and parsed_url.netloc
@app.route('/convert', methods=['POST'])
def convert_music_link():
try:
# Get the music service URL from the request
music_url = request.json.get('url')
if not music_url:
return jsonify({"error": "URL not provided"}), 400
if not validate_url(music_url):
return jsonify({"error": "Invalid URL"}), 400
# Use Selenium to automate the interaction with song.link
song_link_url = get_song_link(music_url)
if song_link_url:
return jsonify({"song_link": song_link_url})
else:
return jsonify({"error": "Could not retrieve song.link URL after multiple attempts."}), 504
except Exception as e:
logger.exception("An unexpected error occurred")
return jsonify({"error": str(e)}), 500
def get_song_link(music_url, max_attempts=3):
attempt = 0
while attempt < max_attempts:
attempt += 1
logger.info(f"Attempt {attempt} of {max_attempts} to get song.link URL.")
driver = None # Initialize driver within the loop
try:
# Set up Chrome options
chrome_options = Options()
if HEADLESS:
chrome_options.add_argument("--headless")
chrome_options.add_argument("--no-sandbox")
chrome_options.add_argument("--disable-gpu")
chrome_options.add_argument("--disable-blink-features=AutomationControlled")
chrome_options.add_argument("--disable-dev-shm-usage")
chrome_options.add_argument("--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/91.0.4472.124 Safari/537.36")
if DEPLOYMENT == 'heroku':
# NOTE: this is for chrome-for-testing buildpack!
# **Set the correct Chrome binary location**
chrome_options.binary_location = "/app/.chrome-for-testing/chrome-linux64/chrome"
# **Set the correct Chromedriver location**
chrome_service = Service(executable_path="/app/.chrome-for-testing/chromedriver-linux64/chromedriver")
driver = webdriver.Chrome(service=chrome_service, options=chrome_options)
else:
driver = webdriver.Chrome(options=chrome_options)
# Disable navigator.webdriver detection
driver.execute_script("Object.defineProperty(navigator, 'webdriver', {get: () => undefined})")
# Use stealth to avoid detection
stealth(driver,
languages=["en-US", "en"],
vendor="Google Inc.",
platform="Win32",
webgl_vendor="Intel Inc.",
renderer="Intel Iris OpenGL Engine",
fix_hairline=True)
# Navigate to the song.link homepage
driver.get('https://odesli.co')
# Find the input element by its ID and enter the music service URL
search_input = driver.find_element(By.ID, 'search-page-downshift-input')
search_input.send_keys(music_url)
search_input.send_keys(Keys.ENTER)
# Wait for the page to load
WebDriverWait(driver, 10).until(
ec.presence_of_element_located((By.CSS_SELECTOR, 'img[alt="Album artwork"]'))
)
# Get the current URL after redirection (this will be the universal link)
song_link_url = driver.current_url
# Validate the obtained URL
if validate_url(song_link_url):
logger.info(f"Successfully obtained song.link URL: {song_link_url}")
return song_link_url
else:
logger.warning("Invalid song_link_url obtained.")
time.sleep(1) # Wait before retrying
except (TimeoutException, NoSuchElementException) as e:
logger.warning(f"Attempt {attempt} failed due to: {str(e)}")
time.sleep(1) # Wait before retrying
except Exception as e:
logger.exception(f"An unexpected error occurred on attempt {attempt}: {str(e)}")
time.sleep(1) # Wait before retrying
finally:
if driver:
driver.quit()
# If all attempts failed
logger.error("All attempts to obtain song.link URL have failed.")
return None
def print_log(logs):
print("-"*60)
for entry in logs:
print(entry)
if __name__ == '__main__':
app.run(debug=True)