import datetime import logging import os import psycopg2 import redis import json from typing import Dict, Any # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') class DailyOperationsReport: """ Generates and saves a daily operations summary report. """ def __init__(self, db_host: str, db_port: int, db_name: str, db_user: str, db_password: str, redis_host: str, redis_port: int, report_dir: str): """ Initializes the DailyOperationsReport class. Args: db_host (str): The hostname of the PostgreSQL database. db_port (int): The port number of the PostgreSQL database. db_name (str): The name of the PostgreSQL database. db_user (str): The username for the PostgreSQL database. db_password (str): The password for the PostgreSQL database. redis_host (str): The hostname of the Redis server. redis_port (int): The port number of the Redis server. report_dir (str): The directory to save the reports to. """ self.db_host = db_host self.db_port = db_port self.db_name = db_name self.db_user = db_user self.db_password = db_password self.redis_host = redis_host self.redis_port = redis_port self.report_dir = report_dir # Ensure the report directory exists os.makedirs(self.report_dir, exist_ok=True) def _get_db_connection(self): """ Establishes a connection to the PostgreSQL database. Returns: psycopg2.extensions.connection: A connection object to the database. """ try: conn = psycopg2.connect( host=self.db_host, port=self.db_port, dbname=self.db_name, user=self.db_user, password=self.db_password ) return conn except psycopg2.Error as e: logging.error(f"Error connecting to PostgreSQL: {e}") raise def _get_redis_connection(self): """ Establishes a connection to the Redis server. Returns: redis.Redis: A connection object to the Redis server. """ try: r = redis.Redis(host=self.redis_host, port=self.redis_port, decode_responses=True) r.ping() # Check the connection return r except redis.exceptions.ConnectionError as e: logging.error(f"Error connecting to Redis: {e}") raise def _get_tasks_completed(self, conn, date: datetime.date) -> int: """ Retrieves the number of tasks completed on a given date from the database. Args: conn (psycopg2.extensions.connection): The database connection. date (datetime.date): The date to query for. Returns: int: The number of tasks completed on the given date. """ try: cursor = conn.cursor() query = "SELECT COUNT(*) FROM tasks WHERE completion_date = %s" cursor.execute(query, (date,)) count = cursor.fetchone()[0] return count except psycopg2.Error as e: logging.error(f"Error querying tasks completed: {e}") raise def _get_costs_incurred(self, redis_conn, date: datetime.date) -> float: """ Retrieves the total costs incurred on a given date from Redis. Args: redis_conn (redis.Redis): The Redis connection. date (datetime.date): The date to query for. Returns: float: The total costs incurred on the given date. """ try: key = f"costs:{date.isoformat()}" costs = redis_conn.get(key) if costs is None: return 0.0 return float(costs) except redis.exceptions.RedisError as e: logging.error(f"Error retrieving costs from Redis: {e}") raise def _get_issues_found(self, conn, date: datetime.date) -> int: """ Retrieves the number of issues found on a given date from the database. Args: conn (psycopg2.extensions.connection): The database connection. date (datetime.date): The date to query for. Returns: int: The number of issues found on the given date. """ try: cursor = conn.cursor() query = "SELECT COUNT(*) FROM issues WHERE date_found = %s" cursor.execute(query, (date,)) count = cursor.fetchone()[0] return count except psycopg2.Error as e: logging.error(f"Error querying issues found: {e}") raise def generate_report(self, date: datetime.date) -> Dict[str, Any]: """ Generates a daily operations report for the given date. Args: date (datetime.date): The date for which to generate the report. Returns: Dict[str, Any]: A dictionary containing the report data. """ try: db_conn = self._get_db_connection() redis_conn = self._get_redis_connection() tasks_completed = self._get_tasks_completed(db_conn, date) costs_incurred = self._get_costs_incurred(redis_conn, date) issues_found = self._get_issues_found(db_conn, date) # Get previous day's data for comparison previous_date = date - datetime.timedelta(days=1) previous_tasks_completed = self._get_tasks_completed(db_conn, previous_date) previous_costs_incurred = self._get_costs_incurred(redis_conn, previous_date) previous_issues_found = self._get_issues_found(db_conn, previous_date) report_data = { "date": date.isoformat(), "tasks_completed": tasks_completed, "costs_incurred": costs_incurred, "issues_found": issues_found, "previous_date": previous_date.isoformat(), "previous_tasks_completed": previous_tasks_completed, "previous_costs_incurred": previous_costs_incurred, "previous_issues_found": previous_issues_found } return report_data except Exception as e: logging.error(f"Error generating report: {e}") raise finally: if 'db_conn' in locals() and db_conn: db_conn.close() def save_report(self, report_data: Dict[str, Any]): """ Saves the report data to a JSON file in the report directory. Args: report_data (Dict[str, Any]): The report data to save. """ date_str = report_data["date"] filename = os.path.join(self.report_dir, f"daily_ops_report_{date_str}.json") try: with open(filename, "w") as f: json.dump(report_data, f, indent=4) logging.info(f"Report saved to {filename}") except IOError as e: logging.error(f"Error saving report to file: {e}") raise def main(): """ Main function to generate and save the daily operations report. """ db_host = "postgresql-genesis-u50607.vm.elestio.app" db_port = 25432 db_name = "genesis" # Replace with your actual database name db_user = "genesis_user" # Replace with your actual database user db_password = "genesis_password" # Replace with your actual database password redis_host = "redis-genesis-u50607.vm.elestio.app" redis_port = 26379 report_dir = "/mnt/e/genesis-system/logs/daily_ops/" report_generator = DailyOperationsReport(db_host, db_port, db_name, db_user, db_password, redis_host, redis_port, report_dir) today = datetime.date.today() try: report_data = report_generator.generate_report(today) report_generator.save_report(report_data) except Exception as e: logging.error(f"Failed to generate and save report: {e}") if __name__ == "__main__": main()