On my 4G sticks, I’d like to have a way to know how much data is used each day and what the remaining balance is, and have that sent to me on a regular basis (and logged to the victoriametrics time-series database). Can use data to do that or SMS. But I need a way to send and parse multi-step USSD codes as well as send SMS messages and emails. I asked claude.ai to create a python script that does all that for me. After some debugging it came up with the following script. The .conf file has all the adjustable things like app-specific password for gmail, USSD codes specific to my carrier, and the phone number I want to text with weekly summaries. The schedule is managed by service files and timers.<\/p>\n\n\n\n
change permissions to protect the app-specific password<\/p>\n\n\n\n The cron jobs look like<\/p>\n\n\n\n Script itself is here:<\/p>\n\n\n\n Install at \/usr\/local\/bin\/ and change permissions to allow execution.<\/p>\n","protected":false},"excerpt":{"rendered":" On my 4G sticks, I’d like to have a way to know how much data is used each day and what the remaining balance is, and have that sent to me on a regular basis (and logged to the victoriametrics time-series database). Can use data to do that or SMS. But I need a way…<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[],"class_list":["post-363","post","type-post","status-publish","format-standard","hentry","category-uncategorized"],"_links":{"self":[{"href":"http:\/\/blog.ltzs.us\/index.php?rest_route=\/wp\/v2\/posts\/363","targetHints":{"allow":["GET"]}}],"collection":[{"href":"http:\/\/blog.ltzs.us\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"http:\/\/blog.ltzs.us\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"http:\/\/blog.ltzs.us\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"http:\/\/blog.ltzs.us\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=363"}],"version-history":[{"count":5,"href":"http:\/\/blog.ltzs.us\/index.php?rest_route=\/wp\/v2\/posts\/363\/revisions"}],"predecessor-version":[{"id":377,"href":"http:\/\/blog.ltzs.us\/index.php?rest_route=\/wp\/v2\/posts\/363\/revisions\/377"}],"wp:attachment":[{"href":"http:\/\/blog.ltzs.us\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=363"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"http:\/\/blog.ltzs.us\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=363"},{"taxonomy":"post_tag","embeddable":true,"href":"http:\/\/blog.ltzs.us\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=363"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}sudo nano \/etc\/sim-balance-monitor.conf<\/code><\/mark><\/p>\n\n\n\n# SIM Balance Monitor Configuration\n# File: \/etc\/sim-balance-monitor.conf\n\n[General]\n# Modem index (find with: mmcli -L)\nmodem_index = 0\n\n# Data storage paths\ndata_file = \/var\/log\/sim-balance\/balance-data.csv\npending_file = \/var\/log\/sim-balance\/pending-report.txt\nlog_file = \/var\/log\/sim-balance\/monitor.log\n\n[USSD]\n# Data balance USSD codes\ndata_init = #1234#\ndata_responses = 3,1\n\n# SMS balance USSD codes\nsms_init = #123#\nsms_responses = 0,3\n\n[Email]\n# Email recipient\nto = address@gmail.com\n\n# Email sender (optional, will use hostname if not specified)\nfrom = address@gmail.com\n\n# SMTP server settings\nsmtp_host = smtp.gmail.com\nsmtp_port = 587\nsmtp_user = address@gmail.com\nsmtp_password = <app-specific password>\n\n[SMS]\n# SMS recipient for weekly reports\nto = <phone number><\/code><\/pre>\n\n\n\nsudo chmod 600 \/etc\/sim-balance-monitor.conf<\/code><\/mark><\/p>\n\n\n\nsudo crontab -e<\/code><\/mark><\/p>\n\n\n\n0 9 * * * \/usr\/local\/bin\/check_data_balance.sh daily\n0 10 * * 1 \/usr\/local\/bin\/check_data_balance.sh weekly<\/code><\/pre>\n\n\n\n#!\/usr\/bin\/env python3\n"""\nSIM Balance Monitor\nChecks SIM card data balance daily and sends weekly email\/SMS reports\n"""\n\nimport subprocess\nimport re\nimport csv\nimport time\nimport smtplib\nimport socket\nimport os\nfrom datetime import datetime, timedelta\nfrom pathlib import Path\nfrom email.mime.text import MIMEText\nfrom email.mime.multipart import MIMEMultipart\nimport sys\nimport argparse\nimport logging\nimport configparser\n\n# Default configuration file path\nDEFAULT_CONFIG_FILE = '\/etc\/sim-balance-monitor.conf'\n\n# Default values if not specified in config\nDEFAULT_CONFIG = {\n 'modem_index': '0',\n 'data_file': '\/var\/log\/sim-balance\/balance-data.csv',\n 'pending_file': '\/var\/log\/sim-balance\/pending-report.txt',\n 'log_file': '\/var\/log\/sim-balance\/monitor.log',\n 'ussd_init': '#1234#',\n 'ussd_responses': '3,1',\n 'sms_ussd_init': '#123#',\n 'sms_ussd_responses': '0,3',\n}\n\ndef load_config(config_file=DEFAULT_CONFIG_FILE):\n """Load configuration from file"""\n config = configparser.ConfigParser()\n \n if not Path(config_file).exists():\n print(f"ERROR: Configuration file not found: {config_file}", file=sys.stderr)\n print(f"Please create the configuration file at {config_file}", file=sys.stderr)\n print(f"See the installation guide for the configuration template.", file=sys.stderr)\n sys.exit(1)\n \n try:\n config.read(config_file)\n except Exception as e:\n print(f"ERROR: Failed to parse configuration file: {e}", file=sys.stderr)\n sys.exit(1)\n \n # Build configuration dictionary\n cfg = {}\n \n # General settings\n cfg['modem_index'] = int(config.get('General', 'modem_index', fallback=DEFAULT_CONFIG['modem_index']))\n cfg['data_file'] = config.get('General', 'data_file', fallback=DEFAULT_CONFIG['data_file'])\n cfg['pending_file'] = config.get('General', 'pending_file', fallback=DEFAULT_CONFIG['pending_file'])\n cfg['log_file'] = config.get('General', 'log_file', fallback=DEFAULT_CONFIG['log_file'])\n \n # USSD codes\n cfg['ussd_init'] = config.get('USSD', 'data_init', fallback=DEFAULT_CONFIG['ussd_init'])\n cfg['ussd_responses'] = config.get('USSD', 'data_responses', fallback=DEFAULT_CONFIG['ussd_responses']).split(',')\n cfg['sms_ussd_init'] = config.get('USSD', 'sms_init', fallback=DEFAULT_CONFIG['sms_ussd_init'])\n cfg['sms_ussd_responses'] = config.get('USSD', 'sms_responses', fallback=DEFAULT_CONFIG['sms_ussd_responses']).split(',')\n \n # Email settings\n cfg['email_to'] = config.get('Email', 'to', fallback='')\n cfg['email_from'] = config.get('Email', 'from', fallback=f'sim-monitor@{socket.gethostname()}')\n cfg['smtp_host'] = config.get('Email', 'smtp_host', fallback='smtp.gmail.com')\n cfg['smtp_port'] = int(config.get('Email', 'smtp_port', fallback='587'))\n cfg['smtp_user'] = config.get('Email', 'smtp_user', fallback='')\n cfg['smtp_password'] = config.get('Email', 'smtp_password', fallback='')\n \n # SMS settings\n cfg['sms_to'] = config.get('SMS', 'to', fallback='')\n \n return cfg\n\n# Global config variable\nCONFIG = None\n\n# Setup logging\ndef setup_logging():\n log_dir = Path(CONFIG['log_file']).parent\n log_dir.mkdir(parents=True, exist_ok=True)\n \n # Set log level based on environment variable\n log_level = logging.DEBUG if os.getenv('DEBUG') else logging.INFO\n \n logging.basicConfig(\n level=log_level,\n format='[%(asctime)s] %(levelname)s: %(message)s',\n datefmt='%Y-%m-%d %H:%M:%S',\n handlers=[\n logging.FileHandler(CONFIG['log_file']),\n logging.StreamHandler()\n ]\n )\n\nclass USSDSession:\n """Manages USSD communication with the modem"""\n \n def __init__(self, modem_index):\n self.modem_index = modem_index\n \n def _run_mmcli(self, args, timeout=30):\n """Run mmcli command and return output"""\n cmd = ['mmcli', '-m', str(self.modem_index)] + args\n try:\n result = subprocess.run(\n cmd,\n capture_output=True,\n text=True,\n timeout=timeout\n )\n return result.stdout, result.stderr, result.returncode\n except subprocess.TimeoutExpired:\n logging.error(f"Command timeout: {' '.join(cmd)}")\n return None, "Timeout", 1\n \n def cancel(self):\n """Cancel any active USSD session"""\n self._run_mmcli(['--3gpp-ussd-cancel'])\n \n def initiate(self, code):\n """Initiate USSD session"""\n logging.info(f"Initiating USSD: {code}")\n stdout, stderr, rc = self._run_mmcli(['--3gpp-ussd-initiate', code])\n \n # Log full output for debugging\n if stdout:\n logging.debug(f"STDOUT: {stdout}")\n if stderr:\n logging.debug(f"STDERR: {stderr}")\n \n if rc != 0:\n logging.error(f"USSD initiate failed: {stderr}")\n return None\n \n response = self._extract_response(stdout)\n if not response:\n # Sometimes response comes later, try checking status\n logging.info("No immediate response, checking USSD status...")\n time.sleep(2)\n stdout, stderr, rc = self._run_mmcli(['--3gpp-ussd-status'])\n if stdout:\n logging.debug(f"Status output: {stdout}")\n response = self._extract_response(stdout)\n \n return response\n \n def respond(self, response):\n """Respond to USSD prompt"""\n logging.info(f"Responding: {response}")\n stdout, stderr, rc = self._run_mmcli(['--3gpp-ussd-respond', response])\n \n # Log full output for debugging\n if stdout:\n logging.debug(f"STDOUT: {stdout}")\n if stderr:\n logging.debug(f"STDERR: {stderr}")\n \n if rc != 0:\n logging.error(f"USSD respond failed: {stderr}")\n return None\n \n result = self._extract_response(stdout)\n if not result:\n # Try checking status\n logging.info("No immediate response, checking USSD status...")\n time.sleep(2)\n stdout, stderr, rc = self._run_mmcli(['--3gpp-ussd-status'])\n if stdout:\n logging.debug(f"Status output: {stdout}")\n result = self._extract_response(stdout)\n \n return result\n \n def _extract_response(self, output):\n """Extract response text from mmcli output"""\n if not output:\n logging.warning("No output from mmcli command")\n return None\n \n # Log the raw output for debugging\n logging.debug(f"Raw mmcli output: {output}")\n \n # First try to extract from 'network request:' (multi-line format from --3gpp-ussd-status)\n if 'network request:' in output:\n lines = output.split('\\n')\n response_lines = []\n capture = False\n \n for line in lines:\n if 'network request:' in line:\n # Start capturing from this line\n capture = True\n # Get the text after "network request:"\n parts = line.split('network request:', 1)\n if len(parts) > 1 and parts[1].strip():\n response_lines.append(parts[1].strip())\n elif capture:\n # Continue capturing indented lines\n if line.strip() and (line.startswith(' |') or line.startswith(' |')):\n # Remove the leading pipe and whitespace\n text = line.split('|', 1)\n if len(text) > 1:\n response_lines.append(text[1].strip())\n elif 'network notification:' in line or not line.strip().startswith('|'):\n # Stop capturing when we hit another field or unindented line\n break\n \n if response_lines:\n response = ' '.join(response_lines)\n logging.info(f"Extracted response from 'network request': {response[:150]}...")\n return response\n \n # Try to extract from 'new reply from network:' (single-line format from initiate\/respond)\n match = re.search(r"new reply from network: '(.+?)'", output, re.MULTILINE | re.DOTALL)\n if match:\n response = match.group(1).strip()\n # Clean up the response (remove extra whitespace, newlines)\n response = ' '.join(response.split())\n logging.info(f"Extracted response from 'new reply': {response[:150]}...")\n return response\n \n # Try other legacy patterns\n patterns = [\n r"response: '(.+?)'",\n r'response: "(.+?)"',\n ]\n \n for pattern in patterns:\n match = re.search(pattern, output, re.MULTILINE | re.DOTALL)\n if match:\n response = match.group(1).strip()\n response = ' '.join(response.split())\n logging.info(f"Extracted response: {response[:150]}...")\n return response\n \n # If no pattern matches, log the full output\n logging.warning(f"Could not extract response from output: {output[:300]}")\n return None\n\nclass BalanceChecker:\n """Handles balance checking and data storage"""\n \n def __init__(self):\n self.data_file = Path(CONFIG['data_file'])\n self.data_file.parent.mkdir(parents=True, exist_ok=True)\n self.ussd = USSDSession(CONFIG['modem_index'])\n \n def check_data_balance(self):\n """Perform USSD sequence to check data balance"""\n logging.info("Starting data balance check")\n \n # Cancel any existing session\n self.ussd.cancel()\n time.sleep(2)\n \n # Initial USSD\n response = self.ussd.initiate(CONFIG['ussd_init'])\n if not response:\n logging.error("No response from initial USSD")\n return None\n logging.info(f"Data Response 1: {response}")\n \n # Follow-up responses\n for i, resp_code in enumerate(CONFIG['ussd_responses'], 2):\n time.sleep(3)\n response = self.ussd.respond(resp_code)\n if not response:\n logging.error(f"No response from data step {i}")\n self.ussd.cancel()\n return None\n logging.info(f"Data Response {i}: {response}")\n \n self.ussd.cancel()\n return response\n \n def check_sms_balance(self):\n """Perform USSD sequence to check SMS balance"""\n logging.info("Starting SMS balance check")\n \n # Cancel any existing session\n self.ussd.cancel()\n time.sleep(2)\n \n # Initial USSD\n response = self.ussd.initiate(CONFIG['sms_ussd_init'])\n if not response:\n logging.error("No response from initial SMS USSD")\n return None\n logging.info(f"SMS Response 1: {response}")\n \n # Follow-up responses\n for i, resp_code in enumerate(CONFIG['sms_ussd_responses'], 2):\n time.sleep(3)\n response = self.ussd.respond(resp_code)\n if not response:\n logging.error(f"No response from SMS step {i}")\n self.ussd.cancel()\n return None\n logging.info(f"SMS Response {i}: {response}")\n \n self.ussd.cancel()\n return response\n \n def parse_data_balance(self, response):\n """Parse data balance and expiry from response"""\n # Pattern: "6676 Mo valables jusqu'au 10\/11\/2025 \u00e0 17:51:29"\n pattern = r'(\\d+)\\s*Mo\\s*valables\\s*jusqu\\'au\\s*(\\d{2}\/\\d{2}\/\\d{4})\\s*\u00e0\\s*(\\d{2}:\\d{2}:\\d{2})'\n match = re.search(pattern, response)\n \n if match:\n balance_mb = int(match.group(1))\n balance_gb = round(balance_mb \/ 1024, 2)\n expiry_date = match.group(2)\n expiry_time = match.group(3)\n \n logging.info(f"Data Balance: {balance_mb} MB ({balance_gb} GB), Expires: {expiry_date} {expiry_time}")\n return {\n 'balance_mb': balance_mb,\n 'balance_gb': balance_gb,\n 'expiry_date': expiry_date,\n 'expiry_time': expiry_time\n }\n else:\n logging.error(f"Could not parse data balance from: {response}")\n return None\n \n def parse_sms_balance(self, response):\n """Parse SMS balance and expiry from response"""\n # Pattern: "935 SMS, valable jusqu'au 10\/11\/2025 17:51:29"\n pattern = r'(\\d+)\\s*SMS.*?valable\\s*jusqu\\'au\\s*(\\d{2}\/\\d{2}\/\\d{4})\\s*(\\d{2}:\\d{2}:\\d{2})'\n match = re.search(pattern, response)\n \n if match:\n sms_count = int(match.group(1))\n expiry_date = match.group(2)\n expiry_time = match.group(3)\n \n logging.info(f"SMS Balance: {sms_count} SMS, Expires: {expiry_date} {expiry_time}")\n return {\n 'sms_count': sms_count,\n 'expiry_date': expiry_date,\n 'expiry_time': expiry_time\n }\n else:\n logging.error(f"Could not parse SMS balance from: {response}")\n return None\n \n def save_balance(self, data_info, sms_info):\n """Save balance data to CSV"""\n timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')\n \n with open(self.data_file, 'a', newline='') as f:\n writer = csv.writer(f)\n if data_info and sms_info:\n writer.writerow([\n timestamp,\n data_info['balance_mb'],\n data_info['balance_gb'],\n data_info['expiry_date'],\n data_info['expiry_time'],\n sms_info['sms_count'],\n sms_info['expiry_date'],\n sms_info['expiry_time']\n ])\n elif data_info:\n writer.writerow([\n timestamp,\n data_info['balance_mb'],\n data_info['balance_gb'],\n data_info['expiry_date'],\n data_info['expiry_time'],\n 'ERROR', 'ERROR', 'ERROR'\n ])\n elif sms_info:\n writer.writerow([\n timestamp,\n 'ERROR', 'ERROR', 'ERROR', 'ERROR',\n sms_info['sms_count'],\n sms_info['expiry_date'],\n sms_info['expiry_time']\n ])\n else:\n writer.writerow([timestamp, 'ERROR', 'ERROR', 'ERROR', 'ERROR', 'ERROR', 'ERROR', 'ERROR'])\n \n def run(self):\n """Run complete balance check"""\n data_response = self.check_data_balance()\n data_info = self.parse_data_balance(data_response) if data_response else None\n \n # Wait a bit between checks\n time.sleep(5)\n \n sms_response = self.check_sms_balance()\n sms_info = self.parse_sms_balance(sms_response) if sms_response else None\n \n self.save_balance(data_info, sms_info)\n return data_info is not None or sms_info is not None\n\nclass ReportGenerator:\n """Generates reports from balance data"""\n \n def __init__(self):\n self.data_file = Path(CONFIG['data_file'])\n \n def get_weekly_data(self):\n """Get balance data from last 7 days"""\n if not self.data_file.exists():\n return []\n \n cutoff = datetime.now() - timedelta(days=7)\n cutoff_str = cutoff.strftime('%Y-%m-%d')\n \n data = []\n with open(self.data_file, 'r') as f:\n reader = csv.reader(f)\n for row in reader:\n if row and row[0] >= cutoff_str:\n data.append(row)\n \n return data\n \n def calculate_daily_usage(self, data):\n """Calculate average daily data usage"""\n prev_balance = None\n total_usage = 0\n usage_days = 0\n \n for row in data:\n if len(row) > 1 and row[1] not in ['ERROR', 'N\/A']:\n balance = int(row[1])\n if prev_balance is not None:\n daily_usage = prev_balance - balance\n if daily_usage >= 0:\n total_usage += daily_usage\n usage_days += 1\n prev_balance = balance\n \n return total_usage \/\/ usage_days if usage_days > 0 else 0\n \n def calculate_daily_sms_usage(self, data):\n """Calculate average daily SMS usage"""\n prev_balance = None\n total_usage = 0\n usage_days = 0\n \n for row in data:\n if len(row) > 5 and row[5] not in ['ERROR', 'N\/A']:\n balance = int(row[5])\n if prev_balance is not None:\n daily_usage = prev_balance - balance\n if daily_usage >= 0:\n total_usage += daily_usage\n usage_days += 1\n prev_balance = balance\n \n return total_usage \/\/ usage_days if usage_days > 0 else 0\n \n def generate_email_report(self, data):\n """Generate detailed email report"""\n report = "SIM Card Data Balance Report\\n"\n report += "=" * 80 + "\\n\\n"\n report += "Report Period: Last 7 Days\\n"\n report += f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\\n\\n"\n \n if not data:\n report += "No data collected during this period.\\n"\n return report\n \n report += "Daily Balance History:\\n"\n report += "-" * 80 + "\\n"\n report += f"{'Date':<19} | {'Data Balance':<16} | {'SMS Balance':<12} | {'Expiry'}\\n"\n report += "-" * 80 + "\\n"\n \n for row in data:\n if len(row) >= 8:\n timestamp, balance_mb, balance_gb, data_expiry_date, data_expiry_time, sms_count, sms_expiry_date, sms_expiry_time = row[:8]\n else:\n # Old format without SMS data\n timestamp = row[0] if len(row) > 0 else 'ERROR'\n balance_mb = row[1] if len(row) > 1 else 'ERROR'\n balance_gb = row[2] if len(row) > 2 else 'ERROR'\n data_expiry_date = row[3] if len(row) > 3 else 'ERROR'\n data_expiry_time = row[4] if len(row) > 4 else 'ERROR'\n sms_count = 'N\/A'\n sms_expiry_date = 'N\/A'\n sms_expiry_time = ''\n \n if balance_mb == 'ERROR' and sms_count == 'ERROR':\n report += f"{timestamp:<19} | {'ERROR':<16} | {'ERROR':<12} | Check logs\\n"\n else:\n data_str = f"{balance_mb} MB ({balance_gb}GB)" if balance_mb != 'ERROR' else 'ERROR'\n sms_str = f"{sms_count} SMS" if sms_count not in ['ERROR', 'N\/A'] else sms_count\n expiry_str = f"{data_expiry_date} {data_expiry_time}" if data_expiry_date != 'ERROR' else 'ERROR'\n report += f"{timestamp:<19} | {data_str:<16} | {sms_str:<12} | {expiry_str}\\n"\n \n report += "-" * 80 + "\\n\\n"\n \n # Statistics\n valid_data_balances = [int(row[1]) for row in data if len(row) > 1 and row[1] not in ['ERROR', 'N\/A']]\n valid_sms_balances = [int(row[5]) for row in data if len(row) > 5 and row[5] not in ['ERROR', 'N\/A']]\n \n if valid_data_balances or valid_sms_balances:\n report += "Summary Statistics:\\n"\n \n if valid_data_balances:\n avg_data = sum(valid_data_balances) \/\/ len(valid_data_balances)\n min_data = min(valid_data_balances)\n max_data = max(valid_data_balances)\n avg_usage = self.calculate_daily_usage(data)\n \n report += f" Data - Average Balance: {avg_data} MB\\n"\n report += f" Data - Minimum Balance: {min_data} MB\\n"\n report += f" Data - Maximum Balance: {max_data} MB\\n"\n report += f" Data - Average Daily Usage: {avg_usage} MB\/day\\n"\n \n if valid_sms_balances:\n avg_sms = sum(valid_sms_balances) \/\/ len(valid_sms_balances)\n min_sms = min(valid_sms_balances)\n max_sms = max(valid_sms_balances)\n avg_sms_usage = self.calculate_daily_sms_usage(data)\n \n report += f" SMS - Average Balance: {avg_sms} SMS\\n"\n report += f" SMS - Minimum Balance: {min_sms} SMS\\n"\n report += f" SMS - Maximum Balance: {max_sms} SMS\\n"\n report += f" SMS - Average Daily Usage: {avg_sms_usage} SMS\/day\\n"\n \n return report\n \n def generate_sms_report(self, data):\n """Generate compact SMS report"""\n if not data:\n return None\n \n sms = f"SIM Wk {datetime.now().strftime('%m\/%d')}:"\n \n last_expiry = None\n last_sms_count = None\n for row in data:\n if len(row) >= 8:\n timestamp, balance_mb, _, expiry_date, _, sms_count, _, _ = row[:8]\n else:\n # Old format\n timestamp = row[0] if len(row) > 0 else None\n balance_mb = row[1] if len(row) > 1 else 'ERROR'\n expiry_date = row[3] if len(row) > 3 else None\n sms_count = 'N\/A'\n \n if timestamp and balance_mb not in ['ERROR', 'N\/A']:\n day = timestamp.split()[0].split('-')[2]\n sms += f" {day}:{balance_mb}M"\n last_expiry = expiry_date\n if sms_count not in ['ERROR', 'N\/A']:\n last_sms_count = sms_count\n \n # Add data statistics\n avg_usage = self.calculate_daily_usage(data)\n valid_balances = [int(row[1]) for row in data if len(row) > 1 and row[1] not in ['ERROR', 'N\/A']]\n min_balance = min(valid_balances) if valid_balances else 0\n \n sms += f" Avg:{avg_usage}M\/d Min:{min_balance}M"\n \n # Add SMS info if available\n if last_sms_count:\n sms += f" SMS:{last_sms_count}"\n \n # Add expiry\n if last_expiry:\n expiry_compact = '\/'.join(last_expiry.split('\/')[:2])\n sms += f" Exp:{expiry_compact}"\n \n return sms\n\nclass EmailSender:\n """Handles email sending via SMTP"""\n \n def check_network(self):\n """Check if network is available"""\n for host in ['8.8.8.8', '1.1.1.1']:\n try:\n socket.create_connection((host, 53), timeout=5)\n return True\n except OSError:\n continue\n return False\n \n def send_email(self, subject, body):\n """Send email using SMTP"""\n if not self.check_network():\n logging.warning("No network connectivity")\n return False\n \n try:\n msg = MIMEMultipart()\n msg['From'] = CONFIG['email_from']\n msg['To'] = CONFIG['email_to']\n msg['Subject'] = subject\n msg.attach(MIMEText(body, 'plain'))\n \n with smtplib.SMTP(CONFIG['smtp_host'], CONFIG['smtp_port'], timeout=30) as server:\n server.starttls()\n server.login(CONFIG['smtp_user'], CONFIG['smtp_password'])\n server.send_message(msg)\n \n logging.info("Email sent successfully")\n return True\n except Exception as e:\n logging.error(f"Failed to send email: {e}")\n return False\n\nclass SMSSender:\n """Handles SMS sending via mmcli"""\n \n def __init__(self):\n self.modem_index = CONFIG['modem_index']\n \n def send_sms(self, message):\n """Send SMS using mmcli"""\n logging.info(f"Sending SMS ({len(message)} chars): {message}")\n \n # Step 1: Create the SMS\n # Note: Use proper quoting for text with spaces\n create_cmd = [\n 'mmcli', '-m', str(self.modem_index),\n '--messaging-create-sms',\n f'text="{message}",number="{CONFIG["sms_to"]}"'\n ]\n \n try:\n result = subprocess.run(\n create_cmd,\n capture_output=True,\n text=True,\n timeout=30\n )\n \n if result.returncode != 0:\n logging.error(f"Failed to create SMS: {result.stderr}")\n return False\n \n # Extract SMS index from output\n # Output looks like: "Successfully created new SMS: \/org\/freedesktop\/ModemManager1\/SMS\/0"\n match = re.search(r'\/SMS\/(\\d+)', result.stdout)\n if not match:\n logging.error(f"Could not extract SMS index from output: {result.stdout}")\n return False\n \n sms_index = match.group(1)\n logging.info(f"Created SMS with index: {sms_index}")\n \n # Step 2: Send the SMS\n send_cmd = [\n 'mmcli', '-s', sms_index,\n '--send'\n ]\n \n result = subprocess.run(\n send_cmd,\n capture_output=True,\n text=True,\n timeout=30\n )\n \n if result.returncode == 0:\n logging.info(f"SMS sent successfully to {CONFIG['sms_to']}")\n return True\n else:\n logging.error(f"Failed to send SMS: {result.stderr}")\n return False\n \n except Exception as e:\n logging.error(f"SMS send error: {e}")\n return False\n\nclass WeeklyReporter:\n """Manages weekly email and SMS reports"""\n \n def __init__(self):\n self.report_gen = ReportGenerator()\n self.email_sender = EmailSender()\n self.sms_sender = SMSSender()\n self.pending_file = Path(CONFIG['pending_file'])\n \n def send_reports(self):\n """Send both email and SMS reports"""\n logging.info("Starting weekly reports")\n \n # Get current week data\n current_data = self.report_gen.get_weekly_data()\n \n # Load pending data if exists\n pending_data = []\n if self.pending_file.exists():\n logging.info("Found pending data from previous week(s)")\n with open(self.pending_file, 'r') as f:\n reader = csv.reader(f)\n pending_data = list(reader)\n \n # Combine data\n all_data = pending_data + current_data\n \n if not all_data:\n logging.info("No data to report")\n return\n \n # Generate reports\n email_body = self.report_gen.generate_email_report(all_data)\n sms_text = self.report_gen.generate_sms_report(current_data)\n \n # Send email\n subject = f"Weekly SIM Balance Report - {datetime.now().strftime('%Y-%m-%d')}"\n if self.email_sender.send_email(subject, email_body):\n # Clear pending data on success\n if self.pending_file.exists():\n self.pending_file.unlink()\n \n # Archive old data (keep 60 days)\n self.archive_old_data()\n else:\n # Save for next week\n logging.info("Saving data for next week")\n with open(self.pending_file, 'w', newline='') as f:\n writer = csv.writer(f)\n writer.writerows(all_data)\n \n # Send SMS (fire-and-forget)\n if sms_text:\n self.sms_sender.send_sms(sms_text)\n \n def archive_old_data(self):\n """Keep only last 60 days of data"""\n data_file = Path(CONFIG['data_file'])\n if not data_file.exists():\n return\n \n cutoff = datetime.now() - timedelta(days=60)\n cutoff_str = cutoff.strftime('%Y-%m-%d')\n \n temp_file = data_file.with_suffix('.tmp')\n with open(data_file, 'r') as fin, open(temp_file, 'w', newline='') as fout:\n reader = csv.reader(fin)\n writer = csv.writer(fout)\n for row in reader:\n if row and row[0] >= cutoff_str:\n writer.writerow(row)\n \n temp_file.replace(data_file)\n logging.info("Archived old data")\n\ndef main():\n global CONFIG\n \n parser = argparse.ArgumentParser(description='SIM Balance Monitor')\n parser.add_argument('action', choices=['check', 'report'], \n help='Action to perform: check (daily balance) or report (weekly summary)')\n parser.add_argument('-c', '--config', default=DEFAULT_CONFIG_FILE,\n help=f'Configuration file path (default: {DEFAULT_CONFIG_FILE})')\n args = parser.parse_args()\n \n # Load configuration FIRST\n CONFIG = load_config(args.config)\n \n # THEN setup logging (needs CONFIG to be loaded)\n setup_logging()\n \n try:\n if args.action == 'check':\n checker = BalanceChecker()\n success = checker.run()\n sys.exit(0 if success else 1)\n \n elif args.action == 'report':\n reporter = WeeklyReporter()\n reporter.send_reports()\n sys.exit(0)\n \n except Exception as e:\n logging.error(f"Unexpected error: {e}", exc_info=True)\n sys.exit(1)\n\nif __name__ == '__main__':\n main()\n<\/code><\/pre>\n\n\n\n