Spaces:
Runtime error
Runtime error
File size: 5,905 Bytes
0399cbd 4002809 0399cbd 4002809 0399cbd 4002809 0399cbd 4002809 0399cbd 4002809 0399cbd 4002809 | 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 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 | # sudo mysql -u root -p
# CREATE DATABASE monitor_log;
# USE monitor_log;
# SHOW TABLES;
# DROP TABLE anonymous_no;
# DROP TABLE log;
# SELECT * FROM information_schema.columns WHERE table_name = 'log';
import mysql.connector
from mysql.connector import Error
from datetime import datetime
import calendar
# Replace these values with your own
db_host = "mydb.c3a48iwi2mip.ap-southeast-2.rds.amazonaws.com"
db_name = "monitor_log"
db_user = "admin"
db_password = "0123456789"
class Database:
def __init__(self, host, user, password, database):
"""Initialize the database connection."""
self.host = host
self.user = user
self.password = password
self.database = database
self.connection = None
def connect(self):
"""Establish a connection to the database."""
try:
self.connection = mysql.connector.connect(
host=self.host,
user=self.user,
password=self.password,
database=self.database
)
if self.connection.is_connected():
print("Successfully connected to the database.")
except Error as e:
print(f"Error: {e}")
def disconnect(self):
"""Close the database connection."""
if self.connection.is_connected():
self.connection.close()
print("Database connection closed.")
def execute_query(self, query, values=None):
"""Execute a query that does not return results (INSERT, UPDATE, DELETE)."""
cursor = self.connection.cursor()
try:
if values:
cursor.execute(query, values)
else:
cursor.execute(query)
self.connection.commit()
print("Query executed successfully.")
except Error as e:
print(f"Error: {e}")
finally:
cursor.close()
def fetch_all(self, query, values=None):
"""Execute a SELECT query and return all results."""
cursor = self.connection.cursor()
results = []
try:
if values is None:
cursor.execute(query)
else: cursor.execute(query, values)
results = cursor.fetchall()
return results
except Error as e:
print(f"Error: {e}")
finally:
cursor.close()
def create_table(self, create_table_query):
"""Create a new table in the database."""
cursor = self.connection.cursor()
try:
cursor.execute(create_table_query)
print("Table created successfully.")
except Error as e:
print(f"Error: {e}")
finally:
cursor.close()
def formatdate(start="Oct 2024",end="Nov 2025"):
# Convert the text to a datetime object with the day set to 1
start_date_object = datetime.strptime(start, "%b %Y")
# Format to YYYY-MM-DD
start_formatted_date = start_date_object.strftime("%Y-%m-%d")
# Convert the text to a datetime object
end_date_object = datetime.strptime(end, "%b %Y")
# Get the last day of the month
last_day = calendar.monthrange(end_date_object.year, end_date_object.month)[1]
# Format to YYYY-MM-DD with the last day of the month
end_formatted_date = end_date_object.replace(day=last_day).strftime("%Y-%m-%d")
return (start_formatted_date, end_formatted_date)
def querylist(id):
query = None
if id == 1:
query = """
SELECT name, COUNT(*) AS amount
FROM `log`
WHERE datetime BETWEEN %s AND %s AND scheck = 1
GROUP BY name;
"""
elif id == 2:
query = """
SELECT DATE(datetime) AS date, COUNT(*) AS amount
FROM `log`
WHERE datetime BETWEEN %s AND %s AND scheck = 1
GROUP BY DATE(datetime);
"""
elif id == 3:
query = """
SELECT id, name, CONCAT(name, '/', id) AS path
FROM `log`
WHERE scheck = 1
"""
elif id == 4:
query = """
SELECT
DATE_FORMAT(datetime, '%b %Y') AS month,
COUNT(*) AS total_items
FROM
`log`
WHERE
datetime BETWEEN %s AND %s
GROUP BY
month
ORDER BY
month;
"""
return query
if __name__ == "__main__":
db = Database(
host=db_host,
user=db_user,
password=db_password,
database=db_name
)
db.connect()
# db.create_table("""
# CREATE TABLE anonymous_no (
# number INT AUTO_INCREMENT PRIMARY KEY,
# name VARCHAR(60) NOT NULL,
# description VARCHAR(500)
# );
# """)
# db.create_table("""
# CREATE TABLE log (
# id VARCHAR(255) PRIMARY KEY,
# name VARCHAR(60) NOT NULL,
# datetime DATETIME DEFAULT CURRENT_TIMESTAMP,
# scheck VARCHAR(5) DEFAULT '0'
# );
# """)
#result1 = db.fetch_all(query=querylist(1),values=formatdate())
#result2 = db.fetch_all(query=querylist(2),values=formatdate())
#result3 = db.fetch_all(query=querylist(3))
#print(result1)
#print("---------")
#print(result2)
#print("---------")
#print(result3)
result4 = db.fetch_all(query=querylist(4), values=formatdate('Sep 2025', 'Sep 2026'))
print(result4)
db.disconnect()
"""
SELECT name, COUNT(*) AS amount
FROM `log`
WHERE datetime BETWEEN %s AND %s AND scheck = 1
GROUP BY name;
"""
"""
SELECT DATE(datetime) AS date, COUNT(*) AS amount
FROM `log`
WHERE datetime BETWEEN %s AND %s AND scheck = 1
GROUP BY DATE(datetime);
"""
"""
SELECT id, name, CONCAT(name, '/', id) AS path
FROM `log`
WHERE scheck = 1
""" |