Spaces:
Runtime error
Runtime error
| # 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 | |
| """ |