from __future__ import annotations
import json
import os
import secrets
import socket
import subprocess
import sys
import threading
import time
from pathlib import Path
from urllib.parse import quote_plus
from cryptography.fernet import Fernet
from fastapi import FastAPI, Form, Request
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
from sqlalchemy import create_engine, text
ROOT = Path(__file__).resolve().parent
ENV_FILE = ROOT / '.env'
INSTALL_LOCK = ROOT / 'storage' / 'installed.json'
LOG_DIR = ROOT / 'storage' / 'logs'
LOG_DIR.mkdir(parents=True, exist_ok=True)
app = FastAPI(title='Instalador Chemitec Enterprise Data Platform', version='0.28.0')
CSS = """
:root{--bg:#07111f;--card:#0e1d31;--line:#253954;--text:#edf6ff;--muted:#a9bad0;--brand:#36c2a0;--danger:#ff7070}
*{box-sizing:border-box}body{margin:0;background:linear-gradient(135deg,#07111f,#102541);font-family:Inter,Arial,sans-serif;color:var(--text)}
main{max-width:980px;margin:32px auto;padding:20px}.hero{padding:24px 0}.hero h1{font-size:32px;margin:0 0 8px}.hero p{color:var(--muted)}
.card{background:rgba(14,29,49,.96);border:1px solid var(--line);border-radius:16px;padding:22px;margin:18px 0;box-shadow:0 16px 50px #0004}
.grid{display:grid;grid-template-columns:repeat(2,minmax(0,1fr));gap:14px}.full{grid-column:1/-1}label{display:block;font-weight:700;margin-bottom:6px}
input,select{width:100%;padding:12px;border:1px solid var(--line);background:#091728;color:var(--text);border-radius:10px}button{padding:12px 18px;border:0;border-radius:10px;font-weight:800;cursor:pointer;background:var(--brand);color:#042018}.secondary{background:#294360;color:white}.danger{background:var(--danger)}
.ok{color:#6ee7b7}.bad{color:#fca5a5}.muted{color:var(--muted)}.row{display:flex;gap:10px;align-items:center;flex-wrap:wrap}.status{padding:12px;border-radius:10px;background:#071523;border:1px solid var(--line);white-space:pre-wrap}code{color:#8ee8d3}@media(max-width:700px){.grid{grid-template-columns:1fr}.full{grid-column:auto}}
"""
def page(body: str, title: str = 'Instalador') -> HTMLResponse:
return HTMLResponse(f"{title}{body}")
def installed() -> bool:
return INSTALL_LOCK.exists() and ENV_FILE.exists()
def requirements() -> list[dict]:
checks = []
checks.append({'name': 'Python 3.11+', 'ok': sys.version_info >= (3, 11), 'value': sys.version.split()[0]})
checks.append({'name': 'Escrita na pasta do sistema', 'ok': os.access(ROOT, os.W_OK), 'value': str(ROOT)})
checks.append({'name': 'Porta 8080 disponível', 'ok': port_available(8080), 'value': '8080'})
checks.append({'name': 'Espaço em storage', 'ok': True, 'value': str(ROOT / 'storage')})
return checks
def port_available(port: int) -> bool:
sock = socket.socket()
try:
sock.bind(('127.0.0.1', port))
return True
except OSError:
return False
finally:
sock.close()
def build_database_url(engine: str, host: str, port: int, database: str, username: str, password: str, sqlite_path: str) -> str:
if engine == 'sqlite':
path = Path(sqlite_path or 'storage/metadata.db')
if not path.is_absolute():
path = ROOT / path
path.parent.mkdir(parents=True, exist_ok=True)
return f"sqlite:///{path.as_posix()}"
user = quote_plus(username)
pwd = quote_plus(password)
db = quote_plus(database)
if engine == 'postgresql':
return f'postgresql+psycopg://{user}:{pwd}@{host}:{port or 5432}/{db}'
if engine == 'mysql':
return f'mysql+pymysql://{user}:{pwd}@{host}:{port or 3306}/{db}?charset=utf8mb4'
raise ValueError('Banco de metadados não suportado')
def test_db(url: str) -> tuple[bool, str]:
try:
engine = create_engine(url, pool_pre_ping=True, connect_args={'timeout': 10} if url.startswith('sqlite') else {})
with engine.connect() as conn:
conn.execute(text('SELECT 1'))
engine.dispose()
return True, 'Conexão realizada com sucesso.'
except Exception as exc:
return False, f'Falha na conexão: {type(exc).__name__}: {exc}'
def env_escape(value: str) -> str:
if any(ch in value for ch in [' ', '#', '"', "'"]):
return json.dumps(value)
return value
def write_env(values: dict[str, str]) -> None:
lines = [f'{key}={env_escape(str(value))}' for key, value in values.items()]
tmp = ENV_FILE.with_suffix('.env.tmp')
tmp.write_text('\n'.join(lines) + '\n', encoding='utf-8')
os.chmod(tmp, 0o600)
tmp.replace(ENV_FILE)
def initialize_database(database_url: str, admin_email: str, admin_password: str, admin_name: str) -> None:
os.environ['METADATA_DATABASE_URL'] = database_url
os.environ['DEFAULT_ADMIN_EMAIL'] = admin_email
os.environ['DEFAULT_ADMIN_PASSWORD'] = admin_password
# Import only after environment is ready.
from app.core.database import Base, SessionLocal, engine
from passlib.context import CryptContext
hash_password = CryptContext(schemes=['bcrypt'], deprecated='auto').hash
from app.models.entities import Role, User
from sqlalchemy import select
Base.metadata.create_all(bind=engine)
db = SessionLocal()
try:
admin = db.scalar(select(User).where(User.email == admin_email.lower().strip()))
if not admin:
db.add(User(email=admin_email.lower().strip(), password_hash=hash_password(admin_password), full_name=admin_name, role=Role.admin, is_active=True))
db.commit()
finally:
db.close()
@app.get('/', response_class=HTMLResponse)
def home():
if installed():
return page("Plataforma já instalada
O arquivo de bloqueio de instalação foi encontrado.
Para reinstalar, remova manualmente storage/installed.json e revise o arquivo .env.
{'✓' if c['ok'] else '✗'}{c['name']}{c['value']}
" for c in requirements())
body = f"""
Instalação da Chemitec Enterprise Data Platform
Instalação nativa para servidor Linux ou Windows, sem Docker.
1. Verificação do servidor
{items}
"""
return page(body)
@app.post('/test-database')
def test_database(db_engine: str = Form(...), db_host: str = Form('localhost'), db_port: int = Form(0), db_name: str = Form(''), db_user: str = Form(''), db_password: str = Form(''), sqlite_path: str = Form('storage/metadata.db')):
try:
url = build_database_url(db_engine, db_host, db_port, db_name, db_user, db_password, sqlite_path)
ok, message = test_db(url)
return JSONResponse({'ok': ok, 'message': message})
except Exception as exc:
return JSONResponse({'ok': False, 'message': str(exc)}, status_code=400)
@app.post('/install', response_class=HTMLResponse)
def install(
db_engine: str = Form(...), db_host: str = Form('localhost'), db_port: int = Form(0), db_name: str = Form(''), db_user: str = Form(''), db_password: str = Form(''), sqlite_path: str = Form('storage/metadata.db'),
admin_name: str = Form(...), admin_email: str = Form(...), admin_password: str = Form(...), admin_password_confirm: str = Form(...),
app_port: int = Form(8080), storage_path: str = Form('storage/data'), public_url: str = Form('http://localhost:8080'), app_env: str = Form('development')
):
if installed():
return RedirectResponse('/', status_code=303)
errors = []
if admin_password != admin_password_confirm: errors.append('As senhas não coincidem.')
if len(admin_password) < 12: errors.append('A senha deve possuir ao menos 12 caracteres.')
try:
database_url = build_database_url(db_engine, db_host, db_port, db_name, db_user, db_password, sqlite_path)
except Exception as exc:
errors.append(str(exc)); database_url = ''
if database_url:
ok, message = test_db(database_url)
if not ok: errors.append(message)
if errors:
return page("Instalação não concluída
", 'Erro de instalação')
storage = Path(storage_path)
if not storage.is_absolute(): storage = ROOT / storage
storage.mkdir(parents=True, exist_ok=True)
values = {
'APP_ENV': app_env,
'API_HOST': '0.0.0.0',
'API_PORT': str(app_port),
'PUBLIC_URL': public_url,
'JWT_SECRET': secrets.token_urlsafe(64),
'JWT_ALGORITHM': 'HS256',
'ACCESS_TOKEN_MINUTES': '30',
'FERNET_KEY': Fernet.generate_key().decode(),
'METADATA_DATABASE_URL': database_url,
'REDIS_URL': 'redis://localhost:6379/0',
'REDIS_OPTIONAL': 'true',
'S3_ENDPOINT': '',
'S3_ACCESS_KEY': '',
'S3_SECRET_KEY': '',
'S3_BUCKET': 'chemitec-datalake',
'S3_SECURE': 'false',
'LOCAL_STORAGE_PATH': str(storage),
'STORAGE_BACKEND': 'local',
'DEFAULT_ADMIN_EMAIL': admin_email.lower().strip(),
'DEFAULT_ADMIN_PASSWORD': admin_password,
'DEFAULT_ADMIN_NAME': admin_name,
'MAX_EXTRACT_ROWS': '5000000',
'QUERY_TIMEOUT_SECONDS': '1800',
'NATIVE_INSTALL': 'true',
}
try:
write_env(values)
initialize_database(database_url, admin_email, admin_password, admin_name)
INSTALL_LOCK.parent.mkdir(parents=True, exist_ok=True)
INSTALL_LOCK.write_text(json.dumps({'version':'V028','installed_at':time.strftime('%Y-%m-%dT%H:%M:%S%z'),'database_engine':db_engine,'public_url':public_url}, indent=2), encoding='utf-8')
except Exception as exc:
if ENV_FILE.exists(): ENV_FILE.unlink(missing_ok=True)
return page(f"Falha ao instalar
{type(exc).__name__}: {exc}
Instalação concluída
A plataforma foi configurada com sucesso.
Próximo passo
Encerre o instalador e execute:
python start_native.py
Acesse {public_url} e entre com {admin_email}.
Por segurança, o instalador foi bloqueado pelo arquivo storage/installed.json.