#!/usr/bin/env python3 """ UK Energy Bill Analyzer - Backend Server v4.3.5 Features: - Secure API key storage via .env file - User-friendly error messages - EPC certificate lookup and scraping - Claude AI integration for bill analysis """ from http.server import HTTPServer, SimpleHTTPRequestHandler import json import urllib.request import urllib.error import urllib.parse import base64 import os import sys import traceback import re import socket from html.parser import HTMLParser from pathlib import Path # ============================================================================ # USER-FRIENDLY ERROR MESSAGES # ============================================================================ # Error messages written for reading age 9 (plain English) ERROR_MESSAGES = { # API Key errors 'claude_key_missing': { 'title': 'Claude API Key Missing', 'message': 'We need a Claude API key to analyse your bill. ' 'Please add your key to the .env file.', 'help': 'Get a free key at console.anthropic.com' }, 'claude_key_invalid': { 'title': 'Claude API Key Not Working', 'message': 'The Claude API key does not seem to be correct. ' 'Please check you copied the whole key.', 'help': 'Keys start with sk-ant-api03-' }, 'epc_key_missing': { 'title': 'EPC API Key Missing', 'message': 'We need an EPC API key to look up your property. ' 'Please add your key to the .env file.', 'help': 'Get a free key at epc.opendatacommunities.org' }, 'epc_key_invalid': { 'title': 'EPC API Key Not Working', 'message': 'The EPC API key does not seem to be correct. ' 'Please check you copied the whole key.', 'help': 'Make sure your email matches the one you registered with' }, # Network errors 'network_error': { 'title': 'Connection Problem', 'message': 'We could not connect to the internet. ' 'Please check your connection and try again.', 'help': 'Make sure you are connected to the internet' }, 'timeout': { 'title': 'Taking Too Long', 'message': 'The request is taking too long. ' 'The service might be busy. Please try again.', 'help': 'Wait a moment and try again' }, # Data errors 'postcode_missing': { 'title': 'Postcode Needed', 'message': 'Please enter your postcode so we can find your property.', 'help': 'Enter a UK postcode like SW1A 1AA' }, 'postcode_not_found': { 'title': 'Postcode Not Found', 'message': 'We could not find any properties at this postcode. ' 'Please check the postcode is correct.', 'help': 'Make sure you typed the postcode correctly' }, 'no_certificates': { 'title': 'No Energy Certificates Found', 'message': 'We could not find any energy certificates for this postcode. ' 'Your property might not have one yet.', 'help': 'You can check manually at find-energy-certificate.service.gov.uk' }, # Server errors 'server_error': { 'title': 'Something Went Wrong', 'message': 'Sorry, something went wrong on our end. ' 'Please try again in a moment.', 'help': 'If this keeps happening, check the server logs' }, 'service_unavailable': { 'title': 'Service Not Available', 'message': 'The service we need is not available right now. ' 'Please try again later.', 'help': 'GOV.UK or the API might be having problems' } } def create_error_response(error_key: str, details: str = None) -> dict: """Create a user-friendly error response.""" error_info = ERROR_MESSAGES.get(error_key, ERROR_MESSAGES['server_error']) response = { 'error': True, 'errorKey': error_key, 'title': error_info['title'], 'message': error_info['message'], 'help': error_info['help'] } if details: response['details'] = details return response def classify_error(exception: Exception) -> tuple: """Classify an exception into a user-friendly error type.""" error_str = str(exception).lower() # Network and timeout errors if isinstance(exception, socket.timeout): return 'timeout', str(exception) if isinstance(exception, urllib.error.URLError): if 'timed out' in error_str: return 'timeout', str(exception) if 'connection refused' in error_str or 'network' in error_str: return 'network_error', str(exception) if hasattr(exception, 'code'): code = exception.code if code == 401: return 'epc_key_invalid', f'HTTP {code}' if code == 403: return 'epc_key_invalid', f'HTTP {code}' if code == 404: return 'postcode_not_found', f'HTTP {code}' if code >= 500: return 'service_unavailable', f'HTTP {code}' return 'network_error', str(exception) # HTTP errors from Claude API if isinstance(exception, urllib.error.HTTPError): code = exception.code if code == 401: return 'claude_key_invalid', f'HTTP {code}: Unauthorized' if code == 403: return 'claude_key_invalid', f'HTTP {code}: Forbidden' if code == 429: return 'timeout', 'Rate limit reached - too many requests' if code >= 500: return 'service_unavailable', f'HTTP {code}' # Connection errors if isinstance(exception, ConnectionError): return 'network_error', str(exception) # Generic errors if 'timeout' in error_str: return 'timeout', str(exception) if 'connection' in error_str or 'network' in error_str: return 'network_error', str(exception) return 'server_error', str(exception) # ============================================================================ # CONFIGURATION - Load API keys from .env file # ============================================================================ def load_env_file(): """Load environment variables from .env file""" env_file = Path(__file__).parent / '.env' if env_file.exists(): print('āœ“ Loading API keys from .env file') with open(env_file) as f: for line in f: line = line.strip() if line and not line.startswith('#') and '=' in line: key, value = line.split('=', 1) os.environ[key.strip()] = value.strip() return True return False # Load environment variables at startup ENV_LOADED = load_env_file() # Get API keys from environment (fallback to empty string) CLAUDE_API_KEY = os.environ.get('CLAUDE_API_KEY', '') EPC_EMAIL = os.environ.get('EPC_EMAIL', '') EPC_API_KEY = os.environ.get('EPC_API_KEY', '') # ============================================================================ # HTML PARSERS # ============================================================================ class EPCSearchParser(HTMLParser): """Parse GOV.UK EPC search results to extract certificate numbers WITH addresses""" def __init__(self): super().__init__() self.certificates = [] self.in_cert_link = False self.in_address_line = False self.current_cert = None self.current_data = [] def handle_starttag(self, tag, attrs): attrs_dict = dict(attrs) if tag == 'a' and 'href' in attrs_dict: href = attrs_dict['href'] cert_match = re.search(r'/energy-certificate/([0-9]{4}-[0-9]{4}-[0-9]{4}-[0-9]{4}-[0-9]{4})', href) if cert_match: if self.current_cert and self.current_cert.get('certificateNumber'): self.certificates.append(self.current_cert) self.current_cert = { 'certificateNumber': cert_match.group(1), 'url': f"https://find-energy-certificate.service.gov.uk{href}", 'address': '', 'fullAddress': '' } self.in_cert_link = True self.current_data = [] if tag == 'dd' and attrs_dict.get('class', '').find('govuk-summary-list__value') >= 0: self.in_address_line = True self.current_data = [] def handle_endtag(self, tag): if tag == 'a' and self.in_cert_link and self.current_cert: text = ''.join(self.current_data).strip() if text: self.current_cert['address'] = text self.in_cert_link = False self.current_data = [] if tag == 'dd' and self.in_address_line and self.current_cert: text = ''.join(self.current_data).strip() if text and not self.current_cert.get('fullAddress'): self.current_cert['fullAddress'] = text self.in_address_line = False self.current_data = [] def handle_data(self, data): if self.in_cert_link or self.in_address_line: self.current_data.append(data) def close(self): if self.current_cert and self.current_cert.get('certificateNumber'): self.certificates.append(self.current_cert) super().close() def extract_house_identifier(address): """Extract house number or name from address""" if not address: return None, None addr_lower = address.lower().strip() # Try to match house number (handles both "21 Street" and "21, Street") number_match = re.match(r'^(\d+[a-z]?(?:-\d+[a-z]?)?)[\s,]', addr_lower) if number_match: return 'number', number_match.group(1) # Try to match flat number flat_match = re.match(r'^(?:flat|apartment|unit)\s+(\d+[a-z]?)', addr_lower) if flat_match: return 'flat', flat_match.group(1) # Try to match house name name_match = re.match(r'^([a-z\s\-\']+?)(?:,|\s+[A-Z]{1,2}\d)', addr_lower) if name_match: name = name_match.group(1).strip() if len(name) > 5 and not name.startswith('flat'): return 'name', name return None, None def match_address(user_address, certificates): """Find the best matching certificate based on address matching""" if not user_address or not certificates: return 0, 0, "No address provided - using first certificate" user_addr_lower = user_address.lower().strip() user_id_type, user_id_value = extract_house_identifier(user_address) print(f'\n šŸ” Matching address: "{user_address}"') if user_id_type: print(f' Extracted {user_id_type}: "{user_id_value}"') best_match_index = 0 best_match_score = 0 best_match_reason = "" for idx, cert in enumerate(certificates): cert_addr = cert['address'] cert_full = cert.get('fullAddress', cert_addr) cert_addr_lower = cert_addr.lower().strip() cert_full_lower = cert_full.lower().strip() score = 0 reason = [] # EXACT MATCH if user_addr_lower == cert_addr_lower or user_addr_lower == cert_full_lower: score = 1000 reason.append("EXACT MATCH") # HIGH CONFIDENCE - house identifier matches elif user_id_type and user_id_value: cert_id_type, cert_id_value = extract_house_identifier(cert_addr) if cert_id_type == user_id_type and cert_id_value == user_id_value: score = 500 reason.append(f"House {user_id_type} matches: {user_id_value}") # Bonus for street name overlap user_words = set(w for w in user_addr_lower.split() if len(w) > 3) cert_words = set(w for w in cert_addr_lower.split() if len(w) > 3) common = user_words & cert_words if common: score += len(common) * 10 reason.append(f"Street words match: {', '.join(common)}") # MEDIUM CONFIDENCE - substring matching if score < 500: if user_addr_lower in cert_addr_lower or user_addr_lower in cert_full_lower: score = 300 reason.append("User address contained in certificate") elif cert_addr_lower in user_addr_lower: score = 250 reason.append("Certificate address contained in user input") # LOW CONFIDENCE - word overlap if score < 250: user_words = set(w for w in user_addr_lower.split() if len(w) > 3) cert_words = set(w for w in cert_addr_lower.split() if len(w) > 3) cert_full_words = set(w for w in cert_full_lower.split() if len(w) > 3) all_cert_words = cert_words | cert_full_words common_words = user_words & all_cert_words if user_words and common_words: overlap_ratio = len(common_words) / len(user_words) score = int(overlap_ratio * 100) reason.append(f"Word overlap: {len(common_words)}/{len(user_words)} words") print(f' [{idx+1}] Score: {score:4d} - {cert_addr}') if reason: print(f' Reason: {", ".join(reason)}') if score > best_match_score: best_match_score = score best_match_index = idx best_match_reason = ", ".join(reason) if reason else "Weak match" return best_match_index, best_match_score, best_match_reason class EPCImprovementParser(HTMLParser): """Parse GOV.UK EPC certificate page to extract improvements""" def __init__(self): super().__init__() self.improvements = [] self.current_improvement = None self.current_tag = None self.current_data = [] self.in_improvements_section = False self.in_dt = False self.in_dd = False self.last_dt = None def handle_starttag(self, tag, attrs): if tag == 'h2': self.current_tag = 'h2' self.current_data = [] if tag == 'h3' and self.in_improvements_section: self.current_tag = 'h3' self.current_data = [] if tag == 'dt': self.in_dt = True self.current_data = [] if tag == 'dd': self.in_dd = True self.current_data = [] def handle_endtag(self, tag): if tag == 'h2' and self.current_tag == 'h2': text = ''.join(self.current_data).strip() if 'steps you could take' in text.lower() or 'save energy' in text.lower(): self.in_improvements_section = True self.current_tag = None if tag == 'h3' and self.current_tag == 'h3': text = ''.join(self.current_data).strip() match = re.match(r'Step (\d+):\s*(.+)', text, re.IGNORECASE) if match: self.current_improvement = { 'step': int(match.group(1)), 'description': match.group(2).strip(), 'cost': None, 'saving': None, 'ratingAfter': None, 'scoreAfter': None } self.current_tag = None if tag == 'dt' and self.in_dt: self.last_dt = ''.join(self.current_data).strip().lower() self.in_dt = False if tag == 'dd' and self.in_dd: value = ''.join(self.current_data).strip() if self.current_improvement and self.last_dt: if 'cost' in self.last_dt: self.current_improvement['cost'] = value elif 'saving' in self.last_dt: self.current_improvement['saving'] = value elif 'rating' in self.last_dt or 'potential' in self.last_dt: rating_match = re.search(r'(\d+)\s*\|?\s*([A-G])|([A-G])\s*\((\d+)\)', value) if rating_match: if rating_match.group(1): self.current_improvement['scoreAfter'] = rating_match.group(1) self.current_improvement['ratingAfter'] = rating_match.group(2) else: self.current_improvement['ratingAfter'] = rating_match.group(3) self.current_improvement['scoreAfter'] = rating_match.group(4) if (self.current_improvement.get('cost') and self.current_improvement.get('saving') and self.current_improvement.get('ratingAfter')): self.improvements.append(self.current_improvement) self.current_improvement = None self.in_dd = False def handle_data(self, data): if self.current_tag or self.in_dt or self.in_dd: self.current_data.append(data) # ============================================================================ # HTTP REQUEST HANDLER # ============================================================================ class EnergyAnalyzerHandler(SimpleHTTPRequestHandler): def end_headers(self): self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS') self.send_header('Access-Control-Allow-Headers', 'Content-Type, x-api-key') SimpleHTTPRequestHandler.end_headers(self) def do_OPTIONS(self): self.send_response(200) self.end_headers() def do_GET(self): # Handle /api/config endpoint - tells frontend if API keys are configured if self.path == '/api/config': self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() # Check which API keys are configured (don't expose the actual keys!) config = { 'claudeConfigured': bool(CLAUDE_API_KEY), 'epcConfigured': bool(EPC_API_KEY and EPC_EMAIL), 'message': 'API keys loaded from .env file' if CLAUDE_API_KEY else 'Claude API key not found in .env file' } print(f'āœ“ Config check: Claude={"Yes" if config["claudeConfigured"] else "No"}, EPC={"Yes" if config["epcConfigured"] else "No"}') self.wfile.write(json.dumps(config).encode('utf-8')) return # For all other GET requests, serve static files as normal return SimpleHTTPRequestHandler.do_GET(self) def do_POST(self): print(f'\n=== POST {self.path} ===') if self.path == '/api/claude': try: content_length = int(self.headers.get('Content-Length', 0)) post_data = self.rfile.read(content_length) request_data = json.loads(post_data.decode('utf-8')) # Use API key from .env file or request (fallback) api_key = CLAUDE_API_KEY or request_data.get('apiKey', '') if not api_key: self.send_response(400) self.send_header('Content-Type', 'application/json') self.end_headers() error_response = create_error_response('claude_key_missing') self.wfile.write(json.dumps(error_response).encode('utf-8')) print('āœ— Claude API key missing') return print(f'āœ“ Using Claude API key ({"from .env" if CLAUDE_API_KEY else "from request"})') req = urllib.request.Request( 'https://api.anthropic.com/v1/messages', data=json.dumps({ 'model': request_data.get('model', 'claude-sonnet-4-5-20250929'), 'max_tokens': request_data.get('max_tokens', 4096), 'messages': request_data.get('messages', []) }).encode('utf-8'), headers={ 'Content-Type': 'application/json', 'x-api-key': api_key, 'anthropic-version': '2023-06-01' } ) with urllib.request.urlopen(req, timeout=120) as response: self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(response.read()) print('āœ“ Claude response sent') except urllib.error.HTTPError as e: error_key, details = classify_error(e) # Override for Claude-specific errors if e.code == 401 or e.code == 403: error_key = 'claude_key_invalid' print(f'āœ— Claude API error: {e.code} - {details}') self.send_response(e.code) self.send_header('Content-Type', 'application/json') self.end_headers() error_response = create_error_response(error_key, details) self.wfile.write(json.dumps(error_response).encode('utf-8')) except Exception as e: error_key, details = classify_error(e) print(f'āœ— Error: {details}') self.send_response(500) self.send_header('Content-Type', 'application/json') self.end_headers() error_response = create_error_response(error_key, details) self.wfile.write(json.dumps(error_response).encode('utf-8')) elif self.path == '/api/epc': try: content_length = int(self.headers.get('Content-Length', 0)) post_data = self.rfile.read(content_length) request_data = json.loads(post_data.decode('utf-8')) # Use credentials from .env file or request (fallback) postcode = request_data.get('postcode', '') email = EPC_EMAIL or request_data.get('email', '') api_key = EPC_API_KEY or request_data.get('apiKey', '') if not postcode: self.send_response(400) self.send_header('Content-Type', 'application/json') self.end_headers() error_response = create_error_response('postcode_missing') self.wfile.write(json.dumps(error_response).encode('utf-8')) print('āœ— Postcode missing') return if not api_key: self.send_response(400) self.send_header('Content-Type', 'application/json') self.end_headers() error_response = create_error_response('epc_key_missing') self.wfile.write(json.dumps(error_response).encode('utf-8')) print('āœ— EPC API key missing') return print(f'āœ“ Using EPC credentials ({"from .env" if EPC_API_KEY else "from request"})') req = urllib.request.Request( f'https://epc.opendatacommunities.org/api/v1/domestic/search?postcode={urllib.parse.quote(postcode)}', headers={ 'Accept': 'application/json', 'Authorization': f'Basic {base64.b64encode(f"{email}:{api_key}".encode()).decode()}' } ) with urllib.request.urlopen(req, timeout=30) as response: self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(response.read()) print(f'āœ“ EPC data retrieved for {postcode}') except urllib.error.HTTPError as e: error_key, details = classify_error(e) # Override for EPC-specific errors if e.code == 401 or e.code == 403: error_key = 'epc_key_invalid' elif e.code == 404: error_key = 'postcode_not_found' print(f'āœ— EPC API error: {e.code} - {details}') self.send_response(e.code) self.send_header('Content-Type', 'application/json') self.end_headers() error_response = create_error_response(error_key, details) self.wfile.write(json.dumps(error_response).encode('utf-8')) except Exception as e: error_key, details = classify_error(e) print(f'āœ— EPC Error: {details}') self.send_response(500) self.send_header('Content-Type', 'application/json') self.end_headers() error_response = create_error_response(error_key, details) self.wfile.write(json.dumps(error_response).encode('utf-8')) elif self.path == '/api/scrape-epc': try: content_length = int(self.headers.get('Content-Length', 0)) post_data = self.rfile.read(content_length) request_data = json.loads(post_data.decode('utf-8')) postcode = request_data.get('postcode', '') address = request_data.get('address', '') if not postcode: self.send_response(400) self.send_header('Content-Type', 'application/json') self.end_headers() error_response = create_error_response('postcode_missing') self.wfile.write(json.dumps(error_response).encode('utf-8')) print('āœ— Postcode missing for scrape') return print(f'šŸ“ Postcode: {postcode}') print(f'šŸ  User Address: {address}') search_url = f'https://find-energy-certificate.service.gov.uk/find-a-certificate/search-by-postcode?postcode={urllib.parse.quote(postcode)}' print(f'šŸ”Ž Searching: {search_url}') req = urllib.request.Request( search_url, headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'} ) with urllib.request.urlopen(req, timeout=30) as response: search_html = response.read().decode('utf-8') print(f'āœ“ Retrieved {len(search_html)} bytes') parser = EPCSearchParser() parser.feed(search_html) parser.close() if not parser.certificates: self.send_response(404) self.send_header('Content-Type', 'application/json') self.end_headers() error_response = create_error_response('no_certificates', f'Postcode: {postcode}') self.wfile.write(json.dumps(error_response).encode('utf-8')) print(f'āœ— No certificates found for {postcode}') return print(f'āœ“ Found {len(parser.certificates)} certificates:') for i, cert in enumerate(parser.certificates, 1): print(f' {i}. {cert["address"]} [{cert["certificateNumber"]}]') match_idx, match_score, match_reason = match_address(address, parser.certificates) selected = parser.certificates[match_idx] print(f'\nāœ… SELECTED #{match_idx + 1} (score: {match_score})') print(f' Address: {selected["address"]}') print(f' Reason: {match_reason}') print(f' Certificate: {selected["certificateNumber"]}') cert_url = selected['url'] print(f'\nšŸ”— Fetching: {cert_url}') req = urllib.request.Request( cert_url, headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'} ) with urllib.request.urlopen(req, timeout=30) as response: cert_html = response.read().decode('utf-8') imp_parser = EPCImprovementParser() imp_parser.feed(cert_html) print(f'āœ“ Parsed {len(imp_parser.improvements)} improvements') self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({ 'improvements': imp_parser.improvements, 'certificateNumber': selected['certificateNumber'], 'certificateUrl': cert_url, 'matchedAddress': selected['address'], 'matchScore': match_score, 'matchReason': match_reason }).encode('utf-8')) except urllib.error.HTTPError as e: error_key, details = classify_error(e) print(f'āœ— GOV.UK scrape error: {e.code} - {details}') traceback.print_exc() self.send_response(e.code) self.send_header('Content-Type', 'application/json') self.end_headers() error_response = create_error_response(error_key, details) self.wfile.write(json.dumps(error_response).encode('utf-8')) except Exception as e: error_key, details = classify_error(e) print(f'āœ— Scrape Error: {details}') traceback.print_exc() self.send_response(500) self.send_header('Content-Type', 'application/json') self.end_headers() error_response = create_error_response(error_key, details) self.wfile.write(json.dumps(error_response).encode('utf-8')) elif self.path == '/api/british-gas': # British Gas tariff comparison endpoint try: content_length = int(self.headers.get('Content-Length', 0)) post_data = self.rfile.read(content_length) request_data = json.loads(post_data.decode('utf-8')) postcode = request_data.get('postcode', '') annual_electricity = request_data.get('annual_electricity_kwh', 2900) annual_gas = request_data.get('annual_gas_kwh', 12000) current_cost = request_data.get('current_annual_cost', 0) has_ev = request_data.get('has_ev', False) off_peak_percentage = request_data.get('off_peak_percentage', 0.5) print(f'āš” British Gas comparison for {postcode}' + (' (with EV)' if has_ev else '')) # Try to import the British Gas API module try: from scraper.british_gas_api import ( get_british_gas_tariffs, compare_tariffs, get_data_freshness, get_ev_tariff, compare_ev_tariff ) freshness = get_data_freshness() if not freshness.get('available'): self.send_response(503) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({ 'error': 'British Gas data not available', 'message': 'Run the scraper to fetch tariff data: python scraper/british_gas_scraper.py --all', 'available': False }).encode('utf-8')) return if current_cost > 0: # Compare against current cost result = compare_tariffs( current_annual_cost=current_cost, postcode=postcode, annual_electricity_kwh=annual_electricity, annual_gas_kwh=annual_gas ) else: # Just return available tariffs tariffs = get_british_gas_tariffs(postcode=postcode) result = { 'available': True, 'tariffs': tariffs, 'freshness': freshness } # Always include EV tariff data when available ev_tariff = get_ev_tariff(postcode=postcode) if ev_tariff: result['ev_tariff'] = ev_tariff # If user has an EV and current cost provided, include EV comparison if has_ev and current_cost > 0: ev_comparison = compare_ev_tariff( current_annual_cost=current_cost, postcode=postcode, annual_electricity_kwh=annual_electricity, annual_gas_kwh=annual_gas, off_peak_percentage=off_peak_percentage ) result['ev_comparison'] = ev_comparison self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps(result).encode('utf-8')) print(f'āœ“ British Gas comparison complete' + (' (with EV)' if ev_tariff else '')) except ImportError: self.send_response(503) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({ 'error': 'British Gas scraper not installed', 'message': 'The British Gas scraper module is not available', 'available': False }).encode('utf-8')) except Exception as e: print(f'āœ— British Gas Error: {str(e)}') traceback.print_exc() self.send_response(500) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({ 'error': str(e), 'available': False }).encode('utf-8')) else: self.send_response(404) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({'error': 'Not found'}).encode('utf-8')) def log_message(self, format, *args): if not self.path.startswith('/api/'): sys.stdout.write(f'[Static] {self.command} {self.path}\n') # ============================================================================ # SERVER STARTUP # ============================================================================ def run_server(port=8080): httpd = HTTPServer(('', port), EnergyAnalyzerHandler) print('=' * 70) print(' UK Energy Bill Analyzer v4.3.5') print(' WITH: Secure API Key Storage') print('=' * 70) if ENV_LOADED: print('\nšŸ” API Keys loaded from .env file') if CLAUDE_API_KEY: print(' āœ“ Claude API key configured') if EPC_API_KEY: print(' āœ“ EPC API key configured') print(' ā†’ Keys will NOT be visible in browser') else: print('\nāš  No .env file found - will use keys from browser form') print(' ā†’ Create .env file to hide keys from users') print(f'\nāœ“ Server ready: http://localhost:{port}') print(f'ā†’ Open: http://localhost:{port}/energy-analyzer-octopus.html') print('\n' + '=' * 70) print(' Server running - Press Ctrl+C to stop') print('=' * 70 + '\n') try: httpd.serve_forever() except KeyboardInterrupt: print('\nāœ“ Server stopped') if __name__ == '__main__': port = 8080 if len(sys.argv) > 1: try: port = int(sys.argv[1]) except: pass if sys.version_info < (3, 6): print('ERROR: Python 3.6+ required') sys.exit(1) run_server(port)