#!/usr/bin/env python3 """Extension Platform API Server. Runs on Coolify. Connects to PostgreSQL via Docker internal network. Builds extensions from Gitea (no local files needed). Pipeline endpoints (called by n8n): POST /publish — Download from Gitea, build ZIP, publish to Composer Registry + Release POST /deploy — Deploy extension to a specific customer via EspoCRM API POST /deploy-all — Deploy to all licensed customers POST /sync — Sync installed extensions from EspoCRM servers with platform DB Dashboard API: GET /api/stats GET /api/extensions GET /api/extensions/:id/versions GET /api/customers POST /api/customers PUT /api/customers/:id GET /api/customers/:id POST /api/customers/:id/extensions DELETE /api/customers/:id/extensions/:ext_id GET /api/deployments """ from http.server import HTTPServer, BaseHTTPRequestHandler import json, os, re, ssl, mimetypes, tempfile, shutil, zipfile, io import base64, urllib.request, urllib.error import psycopg2, psycopg2.extras # --- Config from environment --- DB_HOST = os.environ.get('DB_HOST', 'pss404gosg8oo0o8k40o4wco') DB_PORT = int(os.environ.get('DB_PORT', '5432')) DB_NAME = os.environ.get('DB_NAME', 'extension_platform') DB_USER = os.environ.get('DB_USER', 'postgres') DB_PASS = os.environ.get('DB_PASS', '9cugHhKR8bCFXzm') GITEA_URL = os.environ.get('GITEA_URL', 'https://gitea.dev.marcus-law.co.il') GITEA_TOKEN = os.environ.get('GITEA_TOKEN', '90acb3ebc0f98edfb7a373ef9902bde1cc8eff96') GITEA_ORG = os.environ.get('GITEA_ORG', 'espocrm-extensions') DIST_DIR = os.environ.get('DIST_DIR', '/app/dashboard/dist') # Module name mapping MODULES = { "CrmAssistant": "crm-assistant", "DigitalSignature": "digital-signature", "EmailDefaults": "email-defaults", "GoogleIntegration": "google-integration", "GreenInvoiceBilling": "green-invoice-billing", "HebrewLanguage": "hebrew-language", "KlearBranding": "klear-branding", "LegalCrm": "legal-crm", "LicenseManager": "license-manager", "LocalDocuments": "local-documents", "MpdfEngine": "mpdf-engine", "NetHamishpat": "net-hamishpat", "NetworkStorageIntegration": "network-storage-integration", "NextCloudIntegration": "nextcloud-integration", "OfficeAssistant": "office-assistant", "PhoneIsraelFormat": "phone-israel-format", "SmartAssistant": "smart-assistant", "SmsIntegration": "sms-integration", "WhatsAppIntegration": "whatsapp-integration", "DataMigration": "data-migration", } # Extension dependency tree and install order DEPENDENCIES = { 'HebrewLanguage': [], 'KlearBranding': ['HebrewLanguage'], 'PhoneIsraelFormat': ['HebrewLanguage'], 'EmailDefaults': ['HebrewLanguage'], 'MpdfEngine': ['HebrewLanguage'], 'LicenseManager': ['HebrewLanguage'], 'LocalDocuments': ['HebrewLanguage'], 'GoogleIntegration': ['HebrewLanguage'], 'NextCloudIntegration': ['HebrewLanguage'], 'NetworkStorageIntegration': ['HebrewLanguage'], 'SmsIntegration': ['HebrewLanguage'], 'WhatsAppIntegration': ['HebrewLanguage'], 'GreenInvoiceBilling': ['HebrewLanguage'], 'CrmAssistant': ['HebrewLanguage'], 'OfficeAssistant': ['HebrewLanguage'], 'LegalCrm': ['HebrewLanguage'], 'NetHamishpat': ['HebrewLanguage', 'LegalCrm'], 'DigitalSignature': ['HebrewLanguage', 'LegalCrm'], 'SmartAssistant': ['HebrewLanguage', 'LegalCrm'], 'DataMigration': [], } INSTALL_ORDER = [ 'HebrewLanguage', 'KlearBranding', 'PhoneIsraelFormat', 'EmailDefaults', 'MpdfEngine', 'LicenseManager', 'LocalDocuments', 'GoogleIntegration', 'NextCloudIntegration', 'NetworkStorageIntegration', 'SmsIntegration', 'WhatsAppIntegration', 'GreenInvoiceBilling', 'CrmAssistant', 'OfficeAssistant', 'LegalCrm', 'NetHamishpat', 'DigitalSignature', 'SmartAssistant', 'DataMigration', ] SSL_CTX = ssl.create_default_context() SSL_CTX.check_hostname = False SSL_CTX.verify_mode = ssl.CERT_NONE # --- DB helpers --- def get_db(): return psycopg2.connect(host=DB_HOST, port=DB_PORT, dbname=DB_NAME, user=DB_USER, password=DB_PASS) def query(sql, params=None, fetch='all'): conn = get_db() cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) cur.execute(sql, params) if fetch == 'all': result = [dict(r) for r in cur.fetchall()] elif fetch == 'one': row = cur.fetchone() result = dict(row) if row else None else: result = None conn.commit() cur.close() conn.close() return result # --- Gitea helpers --- def gitea_api(method, path, data=None, raw=False): url = f"{GITEA_URL}/api/v1{path}" headers = {'Authorization': f'token {GITEA_TOKEN}'} if data and not raw: headers['Content-Type'] = 'application/json' data = json.dumps(data).encode() req = urllib.request.Request(url, data=data, headers=headers, method=method) try: resp = urllib.request.urlopen(req, context=SSL_CTX, timeout=60) return json.loads(resp.read()) if not raw else resp.read() except urllib.error.HTTPError as e: body = e.read().decode() return {'error': True, 'code': e.code, 'message': body[:300]} def gitea_upload(url, file_data, filename): """Upload file to Gitea (Composer registry or release asset).""" boundary = '----FormBoundary' + os.urandom(8).hex() body = ( f'--{boundary}\r\n' f'Content-Disposition: form-data; name="package"; filename="{filename}"\r\n' f'Content-Type: application/zip\r\n\r\n' ).encode() + file_data + f'\r\n--{boundary}--\r\n'.encode() headers = { 'Authorization': f'token {GITEA_TOKEN}', 'Content-Type': f'multipart/form-data; boundary={boundary}' } req = urllib.request.Request(url, data=body, headers=headers, method='PUT') try: resp = urllib.request.urlopen(req, context=SSL_CTX, timeout=60) return {'status': 'ok', 'code': resp.getcode()} except urllib.error.HTTPError as e: code = e.code msg = e.read().decode()[:200] if code == 409: return {'status': 'exists', 'code': 409, 'message': 'already exists'} return {'status': 'error', 'code': code, 'message': msg} # --- Build logic (from Gitea, no local files) --- def build_and_publish(module, version): """Download extension from Gitea, inject composer.json, publish to registry + release.""" if module not in MODULES: return {'status': 'error', 'message': f'Unknown module: {module}'} kebab = MODULES[module] tag = f'v{version}' composer_name = f'{GITEA_ORG}/{kebab}' # 1. Download tag archive from Gitea archive_url = f"{GITEA_URL}/api/v1/repos/{GITEA_ORG}/{module}/archive/{tag}.zip" headers = {'Authorization': f'token {GITEA_TOKEN}'} req = urllib.request.Request(archive_url, headers=headers) try: resp = urllib.request.urlopen(req, context=SSL_CTX, timeout=60) archive_data = resp.read() except Exception as e: return {'status': 'error', 'step': 'download', 'message': str(e)} # 2. Extract and read manifest.json with tempfile.TemporaryDirectory() as tmpdir: archive_path = os.path.join(tmpdir, 'archive.zip') with open(archive_path, 'wb') as f: f.write(archive_data) with zipfile.ZipFile(archive_path) as zf: # Gitea archives have a top-level directory top_dirs = set(n.split('/')[0] for n in zf.namelist() if '/' in n) prefix = list(top_dirs)[0] + '/' if len(top_dirs) == 1 else '' # Read manifest manifest_path = f'{prefix}manifest.json' try: manifest = json.loads(zf.read(manifest_path)) except KeyError: return {'status': 'error', 'step': 'manifest', 'message': f'manifest.json not found (tried {manifest_path})'} # Validate version if manifest.get('version') != version: return {'status': 'error', 'step': 'version', 'message': f"Mismatch: tag={version}, manifest={manifest.get('version')}"} # 3. Generate composer.json composer = { "name": composer_name, "description": manifest.get('description', ''), "version": version, "type": "espocrm-extension", "license": "proprietary", "authors": [{"name": manifest.get('author', 'klear')}], "require": {"php": manifest.get('php', ['>=8.1'])[0] if isinstance(manifest.get('php'), list) else '>=8.1'}, "extra": { "espocrm-module": module, "espocrm-min-version": manifest.get('acceptableVersions', ['>=8.0.0'])[0] if isinstance(manifest.get('acceptableVersions'), list) else '>=8.0.0' } } # 4. Build extension ZIP with composer.json at root ext_zip_path = os.path.join(tmpdir, f'{module}-{version}.zip') with zipfile.ZipFile(ext_zip_path, 'w', zipfile.ZIP_DEFLATED) as ext_zip: ext_zip.writestr('composer.json', json.dumps(composer, indent=2)) # Re-read archive and copy contents without the top-level dir with zipfile.ZipFile(archive_path) as src_zip: for item in src_zip.namelist(): if item.endswith('/'): continue # Strip top-level directory rel_path = item[len(prefix):] if prefix and item.startswith(prefix) else item if not rel_path or rel_path.startswith('.git'): continue ext_zip.writestr(rel_path, src_zip.read(item)) with open(ext_zip_path, 'rb') as f: ext_zip_data = f.read() # 5. Publish to Gitea Composer Registry registry_url = f"{GITEA_URL}/api/packages/{GITEA_ORG}/composer" reg_result = gitea_upload(registry_url, ext_zip_data, f'{module}-{version}.zip') reg_status = 'published' if reg_result.get('code') == 201 else 'exists' if reg_result.get('code') == 409 else f"error:{reg_result.get('code')}" # 6. Create Gitea Release + upload asset release_result = gitea_api('POST', f'/repos/{GITEA_ORG}/{module}/releases', { 'tag_name': tag, 'name': tag, 'body': f'Release {version}', 'draft': False, 'prerelease': False }) release_id = release_result.get('id') asset_status = 'skipped' if release_id and not release_result.get('error'): # Upload asset using multipart boundary = '----FormBoundary7MA4YWxkTrZu0gW' body = ( f'--{boundary}\r\n' f'Content-Disposition: form-data; name="attachment"; filename="{module}-{version}.zip"\r\n' f'Content-Type: application/zip\r\n\r\n' ).encode() + ext_zip_data + f'\r\n--{boundary}--\r\n'.encode() asset_url = f"{GITEA_URL}/api/v1/repos/{GITEA_ORG}/{module}/releases/{release_id}/assets?name={module}-{version}.zip" headers = { 'Authorization': f'token {GITEA_TOKEN}', 'Content-Type': f'multipart/form-data; boundary={boundary}' } req = urllib.request.Request(asset_url, data=body, headers=headers, method='POST') try: urllib.request.urlopen(req, context=SSL_CTX, timeout=60) asset_status = 'uploaded' except urllib.error.HTTPError as e: asset_status = f'error:{e.code}' elif release_result.get('error'): asset_status = f'release_failed:{release_result.get("message","")[:100]}' # 7. Update DB db_ok = False try: conn = get_db() cur = conn.cursor() cur.execute("""INSERT INTO extension_versions (extension_id, version, zip_filename, git_tag) SELECT id, %s, %s, %s FROM extensions WHERE name = %s ON CONFLICT (extension_id, version) DO NOTHING""", (version, f'{module}-{version}.zip', tag, module)) cur.execute("UPDATE extensions SET latest_version = %s, updated_at = NOW() WHERE name = %s", (version, module)) conn.commit() cur.close() conn.close() db_ok = True except Exception as e: pass return { 'status': 'ok', 'module': module, 'version': version, 'composer_name': composer_name, 'registry': reg_status, 'release_id': release_id, 'asset': asset_status, 'db_updated': db_ok, 'zip_size': len(ext_zip_data) } # --- Deploy logic --- def deploy_to_customer(customer, module, version): """Deploy extension to a customer via EspoCRM API. Downloads ZIP from Gitea.""" espo_url = customer['espocrm_url'].rstrip('/') api_key = customer['espocrm_api_key'] # Download ZIP from Gitea release tag = f'v{version}' releases = gitea_api('GET', f'/repos/{GITEA_ORG}/{module}/releases/tags/{tag}') if releases.get('error'): return {'status': 'error', 'step': 'find_release', 'message': releases.get('message','')} zip_url = None for asset in releases.get('assets', []): if asset['name'].endswith('.zip'): zip_url = asset['browser_download_url'] break if not zip_url: return {'status': 'error', 'step': 'find_zip', 'message': 'No ZIP asset in release'} # Download ZIP req = urllib.request.Request(zip_url, headers={'Authorization': f'token {GITEA_TOKEN}'}) try: resp = urllib.request.urlopen(req, context=SSL_CTX, timeout=60) zip_data = resp.read() except Exception as e: return {'status': 'error', 'step': 'download_zip', 'message': str(e)} zip_base64 = base64.b64encode(zip_data).decode() headers = {'Espo-Authorization': api_key, 'Content-Type': 'application/json'} # Upload to EspoCRM upload_data = json.dumps({'file': f'data:application/zip;base64,{zip_base64}'}).encode() req = urllib.request.Request(f'{espo_url}/api/v1/Extension/action/upload', data=upload_data, headers=headers, method='POST') try: resp = urllib.request.urlopen(req, context=SSL_CTX, timeout=60) ext_id = json.loads(resp.read()).get('id') except urllib.error.HTTPError as e: return {'status': 'error', 'step': 'upload', 'code': e.code, 'message': e.read().decode()[:200]} except Exception as e: return {'status': 'error', 'step': 'upload', 'message': str(e)} # Install req2 = urllib.request.Request(f'{espo_url}/api/v1/Extension/action/install', data=json.dumps({'id': ext_id}).encode(), headers=headers, method='POST') try: urllib.request.urlopen(req2, context=SSL_CTX, timeout=120) except urllib.error.HTTPError as e: return {'status': 'error', 'step': 'install', 'code': e.code, 'message': e.read().decode()[:200]} except Exception as e: return {'status': 'error', 'step': 'install', 'message': str(e)} # Rebuild req3 = urllib.request.Request(f'{espo_url}/api/v1/Admin/rebuild', data=b'true', headers=headers, method='POST') try: urllib.request.urlopen(req3, context=SSL_CTX, timeout=60) except: pass return {'status': 'ok', 'ext_id': ext_id} # --- Sync logic --- def sync_customer_extensions(customer): """Query a customer's EspoCRM to see what's really installed, compare with platform DB.""" espo_url = customer['espocrm_url'].rstrip('/') api_key = customer['espocrm_api_key'] cid = customer['id'] # 1. Query EspoCRM for installed extensions headers = {'Espo-Authorization': api_key} req = urllib.request.Request(f'{espo_url}/api/v1/Extension?select=name,version&maxSize=200', headers=headers) try: resp = urllib.request.urlopen(req, context=SSL_CTX, timeout=30) data = json.loads(resp.read()) server_exts = {e['name']: e.get('version', '') for e in data.get('list', [])} except Exception as e: return {'customer': customer['name'], 'status': 'error', 'message': f'Cannot reach EspoCRM: {str(e)[:200]}'} # 2. Get what platform thinks is installed platform_exts = query(""" SELECT e.name, ce.installed_version, ce.license_status FROM customer_extensions ce JOIN extensions e ON ce.extension_id = e.id WHERE ce.customer_id = %s AND ce.license_status = 'active'""", (cid,)) platform_map = {r['name']: r['installed_version'] for r in platform_exts} # 3. Compare drifts = [] updated = 0 # Check each platform-tracked extension for ext_name, platform_ver in platform_map.items(): server_ver = server_exts.get(ext_name) if server_ver is None: drifts.append({'extension': ext_name, 'type': 'missing_on_server', 'platform': platform_ver, 'server': None}) if platform_ver is not None: query("""UPDATE customer_extensions SET installed_version = NULL, installed_at = NULL, updated_at = NOW() FROM extensions e WHERE customer_extensions.extension_id = e.id AND customer_extensions.customer_id = %s AND e.name = %s""", (cid, ext_name), fetch=None) updated += 1 elif server_ver != platform_ver: drifts.append({'extension': ext_name, 'type': 'version_mismatch', 'platform': platform_ver, 'server': server_ver}) # Update platform DB to reflect reality query("""UPDATE customer_extensions SET installed_version = %s, updated_at = NOW() FROM extensions e WHERE customer_extensions.extension_id = e.id AND customer_extensions.customer_id = %s AND e.name = %s""", (server_ver, cid, ext_name), fetch=None) updated += 1 # Check for extensions on server not tracked by platform for ext_name, server_ver in server_exts.items(): if ext_name not in platform_map and ext_name in MODULES: drifts.append({'extension': ext_name, 'type': 'extra_on_server', 'platform': None, 'server': server_ver}) return { 'customer': customer['name'], 'customer_id': cid, 'status': 'ok', 'server_count': len(server_exts), 'platform_count': len(platform_map), 'drifts': drifts, 'drift_count': len(drifts), 'updated': updated } def sync_all_customers(): """Run sync for all active customers.""" customers = query("""SELECT id, name, espocrm_url, espocrm_api_key FROM customers WHERE is_active AND espocrm_api_key IS NOT NULL""") results = [] total_drifts = 0 for c in customers: r = sync_customer_extensions(c) results.append(r) total_drifts += r.get('drift_count', 0) return { 'status': 'ok', 'customers_checked': len(results), 'total_drifts': total_drifts, 'results': results } # --- HTTP Handler --- class APIHandler(BaseHTTPRequestHandler): def do_GET(self): path = self.path.split('?')[0] if path == '/api/extensions': self._get_extensions() elif re.match(r'/api/extensions/\d+/versions', path): self._get_extension_versions(int(re.search(r'/api/extensions/(\d+)/versions', path).group(1))) elif path == '/api/customers': self._get_customers() elif re.match(r'/api/customers/\d+$', path): self._get_customer(int(re.search(r'/api/customers/(\d+)', path).group(1))) elif path == '/api/deployments': self._get_deployments() elif path == '/api/stats': self._get_stats() else: self._serve_static(path) def do_POST(self): body = self._read_body() path = self.path.split('?')[0] if path == '/publish': self._handle_publish(body) elif path == '/deploy': self._handle_deploy(body) elif path == '/deploy-all': self._handle_deploy_all(body) elif path == '/api/customers': self._create_customer(body) elif re.match(r'/api/customers/\d+/extensions', path): self._assign_extension(int(re.search(r'/api/customers/(\d+)/extensions', path).group(1)), body) elif path == '/api/deploy': self._handle_deploy(body) elif path == '/sync': self._handle_sync(body) else: self._respond(404, {'error': 'Not found'}) def do_PUT(self): body = self._read_body() path = self.path.split('?')[0] if re.match(r'/api/customers/\d+$', path): self._update_customer(int(re.search(r'/api/customers/(\d+)', path).group(1)), body) else: self._respond(404, {'error': 'Not found'}) def do_DELETE(self): path = self.path.split('?')[0] if re.match(r'/api/customers/\d+/extensions/\d+$', path): m = re.search(r'/api/customers/(\d+)/extensions/(\d+)', path) self._remove_extension(int(m.group(1)), int(m.group(2))) elif re.match(r'/api/customers/\d+$', path): cid = int(re.search(r'/api/customers/(\d+)', path).group(1)) self._delete_customer(cid) else: self._respond(404, {'error': 'Not found'}) def do_OPTIONS(self): self.send_response(204) self._cors_headers() self.end_headers() # --- Dashboard API --- def _get_extensions(self): self._respond(200, query(""" SELECT e.*, (SELECT COUNT(*) FROM extension_versions ev WHERE ev.extension_id = e.id) as version_count, (SELECT COUNT(*) FROM customer_extensions ce WHERE ce.extension_id = e.id AND ce.license_status = 'active') as customer_count FROM extensions e ORDER BY e.name""")) def _get_extension_versions(self, ext_id): self._respond(200, query("SELECT * FROM extension_versions WHERE extension_id = %s ORDER BY released_at DESC", (ext_id,))) def _get_customers(self): rows = query("""SELECT c.id, c.name, c.email, c.espocrm_url, c.is_active, c.notes, c.created_at, c.updated_at, c.coolify_url, c.coolify_app_uuid, (SELECT COUNT(*) FROM customer_extensions ce WHERE ce.customer_id = c.id AND ce.license_status = 'active') as extension_count, (SELECT COUNT(*) FROM customer_extensions ce WHERE ce.customer_id = c.id AND ce.license_status = 'active' AND ce.installed_version IS NOT NULL) as installed_count, (SELECT MAX(dl.started_at) FROM deployment_log dl WHERE dl.customer_id = c.id) as last_deploy FROM customers c ORDER BY c.name""") self._respond(200, rows) def _get_customer(self, cid): customer = query("SELECT id, name, email, espocrm_url, is_active, notes, coolify_url, coolify_app_uuid, created_at, updated_at FROM customers WHERE id = %s", (cid,), fetch='one') if not customer: self._respond(404, {'error': 'Not found'}); return customer['extensions'] = query(""" SELECT e.id, e.name, e.composer_name, e.latest_version, ce.installed_version, ce.license_status, ce.auto_update, ce.installed_at FROM customer_extensions ce JOIN extensions e ON ce.extension_id = e.id WHERE ce.customer_id = %s ORDER BY e.name""", (cid,)) customer['dependencies'] = DEPENDENCIES customer['install_order'] = INSTALL_ORDER self._respond(200, customer) def _create_customer(self, body): for f in ['name', 'espocrm_url', 'espocrm_api_key']: if not body.get(f): self._respond(400, {'error': f'{f} required'}); return row = query("""INSERT INTO customers (name, email, espocrm_url, espocrm_api_key, coolify_url, coolify_token, coolify_app_uuid, notes) VALUES (%s,%s,%s,%s,%s,%s,%s,%s) RETURNING id""", (body['name'], body.get('email'), body['espocrm_url'], body['espocrm_api_key'], body.get('coolify_url'), body.get('coolify_token'), body.get('coolify_app_uuid'), body.get('notes')), fetch='one') cid = row['id'] # Auto-assign all active extensions to new customer all_exts = query("SELECT id FROM extensions WHERE is_active", fetch='all') for ext in all_exts: query("""INSERT INTO customer_extensions (customer_id, extension_id, auto_update, license_status) VALUES (%s, %s, true, 'active') ON CONFLICT DO NOTHING""", (cid, ext['id']), fetch=None) self._respond(201, {'id': cid, 'name': body['name']}) def _update_customer(self, cid, body): fields, values = [], [] for col in ['name','email','espocrm_url','espocrm_api_key','coolify_url','coolify_token','coolify_app_uuid','is_active','notes']: if col in body: fields.append(f"{col} = %s"); values.append(body[col]) if not fields: self._respond(400, {'error': 'No fields'}); return fields.append("updated_at = NOW()"); values.append(cid) query(f"UPDATE customers SET {', '.join(fields)} WHERE id = %s", values, fetch=None) self._respond(200, {'status': 'ok'}) def _assign_extension(self, cid, body): if not body.get('extension_id'): self._respond(400, {'error': 'extension_id required'}); return query("""INSERT INTO customer_extensions (customer_id, extension_id, auto_update, license_status) VALUES (%s, %s, %s, %s) ON CONFLICT (customer_id, extension_id) DO UPDATE SET license_status = EXCLUDED.license_status, auto_update = EXCLUDED.auto_update, updated_at = NOW()""", (cid, body['extension_id'], body.get('auto_update', True), body.get('license_status', 'active')), fetch=None) self._respond(201, {'status': 'ok'}) def _delete_customer(self, cid): query("DELETE FROM customer_extensions WHERE customer_id = %s", (cid,), fetch=None) query("DELETE FROM deployment_log WHERE customer_id = %s", (cid,), fetch=None) query("DELETE FROM customers WHERE id = %s", (cid,), fetch=None) self._respond(200, {'status': 'ok'}) def _remove_extension(self, cid, ext_id): query("DELETE FROM customer_extensions WHERE customer_id = %s AND extension_id = %s", (cid, ext_id), fetch=None) self._respond(200, {'status': 'ok'}) def _get_deployments(self): qs = self.path.split('?')[1] if '?' in self.path else '' params = dict(p.split('=') for p in qs.split('&') if '=' in p) if qs else {} limit = min(int(params.get('limit', 50)), 200) self._respond(200, query("""SELECT dl.*, c.name as customer_name, e.name as extension_name FROM deployment_log dl JOIN customers c ON dl.customer_id = c.id JOIN extensions e ON dl.extension_id = e.id ORDER BY dl.started_at DESC LIMIT %s""", (limit,))) def _get_stats(self): self._respond(200, query("""SELECT (SELECT COUNT(*) FROM extensions WHERE is_active) as total_extensions, (SELECT COUNT(*) FROM customers WHERE is_active) as total_customers, (SELECT COUNT(*) FROM customer_extensions WHERE license_status = 'active') as total_licenses, (SELECT COUNT(*) FROM deployment_log WHERE status = 'ok') as successful_deploys, (SELECT COUNT(*) FROM deployment_log WHERE status != 'ok') as failed_deploys, (SELECT MAX(started_at) FROM deployment_log) as last_deploy""", fetch='one')) # --- Pipeline --- def _handle_publish(self, body): module, version = body.get('module',''), body.get('version','') if not module or not version: self._respond(400, {'error': 'module and version required'}); return result = build_and_publish(module, version) self._respond(200, result) def _handle_deploy(self, body): cid, module, version = body.get('customer_id'), body.get('module',''), body.get('version','') if not all([cid, module, version]): self._respond(400, {'error': 'customer_id, module, version required'}); return customer = query("SELECT id, name, espocrm_url, espocrm_api_key FROM customers WHERE id = %s AND is_active", (cid,), fetch='one') if not customer: self._respond(404, {'error': 'Customer not found'}); return result = deploy_to_customer(customer, module, version) # Log query("""INSERT INTO deployment_log (customer_id, extension_id, version, status, trigger_type, error_message) SELECT %s, e.id, %s, %s, 'manual', %s FROM extensions e WHERE e.name = %s""", (cid, version, result['status'], result.get('message',''), module), fetch=None) if result['status'] == 'ok': query("""UPDATE customer_extensions SET installed_version = %s, installed_at = NOW() FROM extensions e WHERE customer_extensions.extension_id = e.id AND customer_extensions.customer_id = %s AND e.name = %s""", (version, cid, module), fetch=None) self._respond(200, {**result, 'customer': customer['name'], 'module': module, 'version': version}) def _handle_deploy_all(self, body): module, version = body.get('module',''), body.get('version','') if not module or not version: self._respond(400, {'error': 'module and version required'}); return customers = query("""SELECT c.id, c.name, c.espocrm_url, c.espocrm_api_key FROM customers c JOIN customer_extensions ce ON c.id = ce.customer_id JOIN extensions e ON ce.extension_id = e.id WHERE e.name = %s AND ce.auto_update AND ce.license_status = 'active' AND c.is_active""", (module,)) results = [] for c in customers: r = deploy_to_customer(c, module, version) results.append({**r, 'customer': c['name']}) query("""INSERT INTO deployment_log (customer_id, extension_id, version, status, trigger_type, error_message) SELECT %s, e.id, %s, %s, 'auto', %s FROM extensions e WHERE e.name = %s""", (c['id'], version, r['status'], r.get('message',''), module), fetch=None) if r['status'] == 'ok': query("""UPDATE customer_extensions SET installed_version = %s, installed_at = NOW() FROM extensions e WHERE customer_extensions.extension_id = e.id AND customer_extensions.customer_id = %s AND e.name = %s""", (version, c['id'], module), fetch=None) self._respond(200, {'status': 'ok', 'module': module, 'version': version, 'deployed_to': len(results), 'results': results}) # --- Sync --- def _handle_sync(self, body): cid = body.get('customer_id') if cid: customer = query("SELECT id, name, espocrm_url, espocrm_api_key FROM customers WHERE id = %s AND is_active", (cid,), fetch='one') if not customer: self._respond(404, {'error': 'Customer not found'}); return result = sync_customer_extensions(customer) self._respond(200, result) else: result = sync_all_customers() self._respond(200, result) # --- Helpers --- def _read_body(self): length = int(self.headers.get('Content-Length', 0)) return json.loads(self.rfile.read(length)) if length > 0 else {} def _serve_static(self, path): if path == '/': path = '/index.html' file_path = os.path.join(DIST_DIR, path.lstrip('/')) if not os.path.isfile(file_path): # Only fallback to index.html for routes, not for asset files if path.startswith('/assets/') or '.' in path.split('/')[-1]: self._respond(404, {'error': 'Not found'}); return file_path = os.path.join(DIST_DIR, 'index.html') try: with open(file_path, 'rb') as f: content = f.read() self.send_response(200) self.send_header('Content-Type', mimetypes.guess_type(file_path)[0] or 'application/octet-stream') self.send_header('Content-Length', len(content)) self.end_headers() self.wfile.write(content) except: self._respond(404, {'error': 'Not found'}) def _respond(self, code, data): self.send_response(code) self._cors_headers() self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps(data, default=str).encode()) def _cors_headers(self): self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Allow-Methods', 'GET, POST, PUT, DELETE, OPTIONS') self.send_header('Access-Control-Allow-Headers', 'Content-Type') def log_message(self, fmt, *args): print(f"[api] {args[0]}") if __name__ == '__main__': port = int(os.environ.get('PORT', '8080')) server = HTTPServer(('0.0.0.0', port), APIHandler) print(f"Extension Platform API on port {port}") server.serve_forever()