-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
154 lines (129 loc) · 4.44 KB
/
main.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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
from flask import Flask, request, redirect, jsonify
import hashlib
import mysql.connector
from urllib.parse import urlparse
import requests
app = Flask(__name__)
# MySQL connection configuration
mysql_config = {
'host': 'prakhardoneria.mysql.pythonanywhere-services.com',
'user': 'prakhardoneria',
'password': '', # Replace with your MySQL password
'database': 'prakhardoneria$TGDB',
}
def connect_to_mysql():
try:
connection = mysql.connector.connect(**mysql_config)
print("Connected to MySQL successfully!")
return connection
except mysql.connector.Error as err:
print(f"Error: {err}")
return None
def create_url_table(connection):
try:
cursor = connection.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS urls (
id INT AUTO_INCREMENT PRIMARY KEY,
original_url VARCHAR(200) UNIQUE NOT NULL,
shortcode VARCHAR(10) UNIQUE NOT NULL
)
""")
connection.commit()
print("URL table created or already exists.")
except mysql.connector.Error as err:
print(f"Error: {err}")
def shorten_url(original_url):
shortcode = hashlib.sha1(original_url.encode()).hexdigest()[:6]
return shortcode
def get_short_url_from_db(original_url):
try:
connection = connect_to_mysql()
if not connection:
return None
cursor = connection.cursor()
cursor.execute("SELECT shortcode FROM urls WHERE original_url = %s", (original_url,))
result = cursor.fetchone()
if result:
return result[0]
else:
return None
except Exception as e:
print(f"Error fetching short URL from database: {e}")
return None
finally:
if connection:
connection.close()
def insert_url_into_db(original_url, shortcode):
try:
connection = connect_to_mysql()
if not connection:
return False
cursor = connection.cursor()
cursor.execute("INSERT INTO urls (original_url, shortcode) VALUES (%s, %s)", (original_url, shortcode))
connection.commit()
return True
except Exception as e:
print(f"Error inserting URL into database: {e}")
return False
finally:
if connection:
connection.close()
def is_valid_url(url):
try:
result = urlparse(url)
return all([result.scheme, result.netloc])
except ValueError:
return False
def is_website_reachable(url):
try:
response = requests.head(url, timeout=5)
return response.status_code == 200
except requests.RequestException:
return False
@app.route('/short', methods=['GET'])
def short():
long_url = request.args.get('long')
if not long_url:
return jsonify({'error': 'Missing long URL parameter'}), 400
if not is_valid_url(long_url):
return jsonify({'error': 'Invalid URL format'}), 400
if "pythonanywhere" in long_url:
return jsonify({'error': 'Invalid URL'}), 400
if not is_website_reachable(long_url):
return jsonify({'error': 'URL is not reachable'}), 400
try:
short_url = get_short_url_from_db(long_url)
if short_url:
return jsonify({'shortened_url': f'{request.host_url}{short_url}'}), 200
shortcode = shorten_url(long_url)
if insert_url_into_db(long_url, shortcode):
return jsonify({'shortened_url': f'{request.host_url}{shortcode}'}), 200
else:
return jsonify({'error': 'Failed to create short URL'}), 500
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/')
def hello_world():
return 'YAY ITS WORKING'
@app.route('/<shortcode>')
def redirect_to_original(shortcode):
try:
connection = connect_to_mysql()
if not connection:
return jsonify({'error': 'Failed to connect to the database.'}), 500
cursor = connection.cursor()
cursor.execute("SELECT original_url FROM urls WHERE shortcode = %s", (shortcode,))
result = cursor.fetchone()
if result:
original_url = result[0]
return redirect(original_url)
else:
return jsonify({'error': 'Shortcode not found'}), 404
except Exception as e:
return jsonify({'error': str(e)}), 500
finally:
if connection:
connection.close()
if __name__ == '__main__':
app.run(port=5003)