import http.server
import socketserver
import json
import time
import os
import hashlib
from urllib.parse import urlparse
PORT = 8000
LEDGER_FILE = "ledger.json"
STRIPE_WEBHOOK_SECRET = "whsec_OMU1a5L8rQwJ8peTxla2tXZ4G8VCNAP2"
DEFAULT_STATE = {
"totalGrossRevenue": 0.00,
"totalBooksSold": 0,
"topVideo": "Waiting for traffic...",
"nodes": {"Video": "HIBERNATING", "Delivery": "HIBERNATING", "Sales": "HIBERNATING"},
"log": ["[SYSTEM INIT] Axiom Local Core Production Thread active."]
}
def load_permanent_ledger():
if os.path.exists(LEDGER_FILE):
try:
with open(LEDGER_FILE, 'r') as file:
saved_state = json.load(file)
saved_state["nodes"] = DEFAULT_STATE["nodes"]
saved_state["log"].append(f"[{time.strftime('%H:%M:%S')}] [LEDGER] Core matrix fully restored from local disk registry.")
return saved_state
except Exception: pass
return DEFAULT_STATE.copy()
def save_permanent_ledger(current_state):
try:
state_to_save = current_state.copy()
state_to_save["nodes"] = DEFAULT_STATE["nodes"]
with open(LEDGER_FILE, 'w') as file:
json.dump(state_to_save, file, indent=4)
except Exception as e: print(f"[ERROR] Disk ledger flush failed: {e}")
def execute_jit_watermark(buyer_email, book_id, tx_id):
timestamp = time.strftime("%Y-%m-%d %H:%M:%S UTC")
raw_sig_base = f"{buyer_email}:{tx_id}:SOVEREIGN_CORE_KEY"
crypto_hash = hashlib.sha256(raw_sig_base.encode()).hexdigest()[:16]
watermark_payload = (
f"// --- AXIOM SECURITY MANIFEST BLOCK ---\n"
f"// Licensed exclusively to: {buyer_email}\n"
f"// Transaction ID: {tx_id}\n"
f"// Timestamp: {timestamp}\n"
f"// Cryptographic Anchor: [AX_SIG_{crypto_hash.upper()}]\n"
)
output_filename = f"stamped_{book_id}_{crypto_hash}.txt"
with open(output_filename, 'w') as secure_file:
secure_file.write(watermark_payload)
secure_file.write(f"\n\n[BOOK CONTENT DEPLOYED: {book_id}]\n")
return crypto_hash, output_filename
# --- NEW TASK 1: EMAIL DISPATCHER NODE ---
def dispatch_fulfillment_email(buyer_email, book_id, attachment_filename):
"""
Constructs the final email payload. In production, this fires to SendGrid.
Locally, it saves a mock email file so we can verify the formatting.
"""
# Clean up the internal ID to a readable title for the customer
book_title = "The Midnight Bridge" if "midnight" in book_id else "The Mirror Effect"
email_html = f"""
Your Book Has Arrived!
Hello,
Thank you for your purchase. Your personal, digitally-secured copy of {book_title} is ready for download.
To protect your purchase, this file has been uniquely stamped with your transaction signature. Please keep it secure.
Attached File: {attachment_filename}
Enjoy the read, and thank you for investing in your personal growth.
Best,
The Lover's Mindset Team
"""
receipt_filename = f"outbox_{buyer_email.replace('@', '_at_')}.html"
try:
with open(receipt_filename, 'w') as email_file:
email_file.write(email_html)
return True, receipt_filename
except Exception as e:
print(f"[ERROR] Failed to dispatch email: {e}")
return False, None
engine_state = load_permanent_ledger()
class AxiomCoreHandler(http.server.SimpleHTTPRequestHandler):
def do_GET(self):
parsed_url = urlparse(self.path)
if parsed_url.path == '/api/state':
self.send_response(200)
self.send_header('Content-type', 'application/json')
self.end_headers()
self.wfile.write(json.dumps(engine_state).encode())
for node in engine_state["nodes"]: engine_state["nodes"][node] = "HIBERNATING"
else:
return http.server.SimpleHTTPRequestHandler.do_GET(self)
def do_POST(self):
parsed_url = urlparse(self.path)
if parsed_url.path == '/webhook/stripe':
content_length = int(self.headers['Content-Length'])
raw_post_data = self.rfile.read(content_length)
try:
payload = json.loads(raw_post_data.decode('utf-8'))
session_data = payload["data"]["object"]
actual_amount_dollars = round(session_data["amount_total"] / 100.0, 2)
buyer_email = session_data["customer_details"]["email"]
transaction_id = session_data["id"]
book_id = session_data["metadata"].get("book_id", "unknown_asset")
video_source = session_data["metadata"].get("video_source", "Direct_Traffic")
timestamp = time.strftime("%H:%M:%S")
# 1. Generate the secure watermark asset
sig_hash, file_path = execute_jit_watermark(buyer_email, book_id, transaction_id)
# 2. Fire the Fulfillment Node to "send" the email
dispatch_success, outbox_file = dispatch_fulfillment_email(buyer_email, book_id, file_path)
# 3. Update core memory
engine_state["totalGrossRevenue"] = round(engine_state["totalGrossRevenue"] + actual_amount_dollars, 2)
engine_state["totalBooksSold"] += 1
engine_state["topVideo"] = video_source
engine_state["nodes"]["Video"] = "ACTIVE"
engine_state["nodes"]["Delivery"] = "ACTIVE"
engine_state["nodes"]["Sales"] = "ACTIVE"
engine_state["log"].append(f"[{timestamp}] [WATERMARK] Secured asset generated for {buyer_email}.")
if dispatch_success:
engine_state["log"].append(f"[{timestamp}] [FULFILLMENT] Email dispatched successfully -> {outbox_file}")
save_permanent_ledger(engine_state)
self.send_response(200)
self.send_header('Content-type', 'application/json')
self.end_headers()
self.wfile.write(json.dumps({"status": "webhook_processed"}).encode())
except Exception as e:
self.send_response(400)
self.end_headers()
print(f"[CRITICAL] Webhook parsing fault: {e}")
socketserver.TCPServer.allow_reuse_address = True
with socketserver.TCPServer(("", PORT), AxiomCoreHandler) as runtime_server:
print("=" * 80)
print(f"AXIOM PRODUCTION ENGINE (FULFILLMENT ACTIVE) AT: http://localhost:{PORT}")
print("=" * 80)
try:
runtime_server.serve_forever()
except KeyboardInterrupt:
print("\n[SHUTDOWN] Exiting runtime environment cleanly.")