# Servidor LOCAL do Sistema de Nomeações da DRH — trabalho em equipe em tempo real, sem internet.
# Roda numa máquina da PM; os outros acessam pelo navegador no endereço mostrado abaixo.
# A base (base-nomeacoes-drh.json) fica NESTA máquina — nada vai para a nuvem.
#
# Como usar: python servidor.py (fique com esta janela aberta)
# Depois abra no navegador o endereço que aparecer, e passe-o para a equipe (mesma rede).
import json, os, socket, threading, http.server, socketserver, base64, time, secrets
from urllib.parse import urlparse, parse_qs, quote
from http.cookies import SimpleCookie
RAIZ = os.path.dirname(os.path.abspath(__file__))
BASE = os.path.join(RAIZ, 'base-nomeacoes-drh.json')
HTML = 'Sistema de Nomeações - DRH.html'
PORTA = 8000
# ---- SENHA de acesso (login do navegador). TROQUE por uma senha forte antes de expor externamente.
# Para desligar a senha (só rede interna), deixe SENHA = ''.
USUARIO = 'drh'
SENHA = 'pmrr@gape'
# PIN curto do APP DO CELULAR pra EDITAR (nomear/trocar). Ver é livre; editar pede este PIN. TROQUE por outro.
PIN_APP = 'pmrr@gape'
# Tela de login (centralizada, com a cara da PMRR) — substitui a caixinha feia do navegador (HTTP Basic).
PAGINA_LOGIN = '''
Entrar · Nomeações DRH/PMRR
'''
_lock = threading.Lock()
_estado = {'versao': 1, 'config': {}, 'nomeacoes': [], 'afastamentos': [], 'disposicoes': [], 'recados': [], 'log': []}
_rev = 0
_sessoes = set() # tokens de sessão válidos (login pela tela bonita, em vez da caixinha do navegador)
def carregar():
global _estado
if os.path.exists(BASE):
try:
_estado = json.load(open(BASE, encoding='utf-8'))
_estado.setdefault('config', {}); _estado.setdefault('nomeacoes', []); _estado.setdefault('afastamentos', []); _estado.setdefault('disposicoes', []); _estado.setdefault('recados', []); _estado.setdefault('log', [])
except Exception as e:
print('!! não consegui ler a base:', e)
def salvar():
tmp = BASE + '.tmp'
json.dump(_estado, open(tmp, 'w', encoding='utf-8'), ensure_ascii=False, indent=1)
os.replace(tmp, BASE) # troca atômica (não corrompe se cair no meio)
def _merge(chave, upserts, deletes):
"""Merge por ID numa lista de _estado (upsert vence delete do mesmo id)."""
_estado.setdefault(chave, [])
up_ids = {n.get('id') for n in (upserts or [])}
if deletes:
rem = set(deletes) - up_ids
_estado[chave] = [n for n in _estado[chave] if n.get('id') not in rem]
por_id = {n['id']: i for i, n in enumerate(_estado[chave])}
for n in (upserts or []):
if n.get('id') in por_id:
_estado[chave][por_id[n['id']]] = n
else:
por_id[n['id']] = len(_estado[chave]); _estado[chave].append(n)
def aplicar(upserts, deletes, config, afs_upserts=None, afs_deletes=None, disp_upserts=None, disp_deletes=None, rec_upserts=None, rec_deletes=None):
"""Aplica as mudanças de um cliente por ID (merge sem sobrescrever o que os outros fizeram)."""
global _rev
with _lock:
_merge('nomeacoes', upserts, deletes)
_merge('afastamentos', afs_upserts, afs_deletes)
_merge('disposicoes', disp_upserts, disp_deletes)
_merge('recados', rec_upserts, rec_deletes)
if config:
_estado['config'].update(config)
_rev += 1
salvar()
return _rev
class Handler(http.server.SimpleHTTPRequestHandler):
def __init__(self, *a, **k):
super().__init__(*a, directory=RAIZ, **k)
def log_message(self, *a):
pass # silencioso
def end_headers(self):
# NUNCA deixa o navegador guardar o HTML/JS em cache: assim, ao dar F5, a equipe sempre pega a versão nova.
if not any(h.lower().startswith(b'cache-control') for h in self._headers_buffer):
self.send_header('Cache-Control', 'no-cache, no-store, must-revalidate')
super().end_headers()
def _cookie_ok(self):
try:
c = SimpleCookie(self.headers.get('Cookie', ''))
tok = c['sessao'].value if 'sessao' in c else ''
return bool(tok) and tok in _sessoes
except Exception:
return False
def _auth_ok(self):
if not SENHA:
return True
if self._cookie_ok(): # entrou pela tela de login bonita (cookie de sessão)
return True
h = self.headers.get('Authorization', '')
if h.startswith('Basic '):
try:
u, p = base64.b64decode(h[6:]).decode('utf-8').split(':', 1)
return (u == USUARIO and p == SENHA) or (PIN_APP and p == PIN_APP) # senha do desktop OU PIN do app
except Exception:
return False
return False
def _pagina_login(self):
corpo = PAGINA_LOGIN.encode('utf-8')
self.send_response(401)
self.send_header('Content-Type', 'text/html; charset=utf-8')
self.send_header('Cache-Control', 'no-store')
self.send_header('Content-Length', str(len(corpo)))
self.end_headers()
self.wfile.write(corpo)
def _json(self, obj, code=200):
corpo = json.dumps(obj, ensure_ascii=False).encode('utf-8')
self.send_response(code)
self.send_header('Content-Type', 'application/json; charset=utf-8')
self.send_header('Cache-Control', 'no-store')
self.send_header('Content-Length', str(len(corpo)))
self.end_headers()
self.wfile.write(corpo)
def _publico(self, path):
# PÚBLICO = só o necessário para DESENHAR a tela de login. Todo o resto (app, desktop, dados, fotos) exige ENTRAR.
return path in ('/icone-192.png', '/apple-touch-icon.png', '/favicon-64.png',
'/Acervo/Modelos/brasao_pmrr_web.png', '/Acervo/Modelos/brasao_roraima.svg',
'/Acervo/Modelos/logo_drh_t.png')
def do_GET(self):
u = urlparse(self.path)
if u.path == '/api/logout': # sair: apaga a sessão e o cookie, volta pra tela de login
try:
c = SimpleCookie(self.headers.get('Cookie', '')); _sessoes.discard(c['sessao'].value if 'sessao' in c else '')
except Exception:
pass
corpo = b'{"ok":true}'
self.send_response(200)
self.send_header('Content-Type', 'application/json; charset=utf-8')
self.send_header('Set-Cookie', 'sessao=; Path=/; Max-Age=0')
self.send_header('Content-Length', str(len(corpo)))
self.end_headers(); self.wfile.write(corpo); return
if not self._publico(u.path) and not self._auth_ok():
return self._pagina_login() # tela de login bonita (centralizada), no lugar da caixinha do navegador
if u.path == '/': # atalho: raiz abre o sistema direto (link curto para a equipe)
ua = self.headers.get('User-Agent', '')
if any(x in ua for x in ('Mobi', 'Android', 'iPhone', 'iPad', 'iPod')): # celular -> app enxuto
self.send_response(302); self.send_header('Location', '/app-celular.html'); self.end_headers(); return
try: v = str(int(os.path.getmtime(os.path.join(RAIZ, HTML)))) # versão = data do arquivo; muda a URL a cada atualização (mata o cache do navegador)
except Exception: v = '0'
self.send_response(302); self.send_header('Location', '/' + quote(HTML) + '?v=' + v); self.end_headers(); return
if u.path == '/api/base':
since = parse_qs(u.query).get('since', [None])[0]
with _lock:
if since is not None and since.isdigit() and int(since) == _rev:
self.send_response(204); self.end_headers(); return
self._json({'rev': _rev, 'versao': _estado.get('versao', 1),
'config': _estado.get('config', {}), 'nomeacoes': _estado['nomeacoes'],
'afastamentos': _estado.get('afastamentos', []), 'disposicoes': _estado.get('disposicoes', []),
'recados': _estado.get('recados', []), 'log': _estado.get('log', [])})
return
return super().do_GET() # arquivos (HTML, fotos, etc.)
def do_POST(self):
global _rev
u = urlparse(self.path)
if u.path == '/api/recado': # PÚBLICO (app do celular, sem senha): recado que chega no desktop
n = int(self.headers.get('Content-Length', 0))
try:
d = json.loads(self.rfile.read(n) or b'{}')
except Exception:
return self._json({'erro': 'json inválido'}, 400)
texto = (d.get('texto') or '').strip()
if not texto:
return self._json({'erro': 'vazio'}, 400)
with _lock:
_estado.setdefault('recados', []).append({
'id': 'r' + str(int(time.time() * 1000)), 'quando': time.strftime('%Y-%m-%dT%H:%M:%S'),
'nome': (d.get('nome') or '').strip()[:120], 'texto': texto[:2000], 'lido': False})
_rev += 1; salvar()
return self._json({'ok': True})
if u.path == '/api/login': # PÚBLICO: recebe a senha da tela bonita e devolve um cookie de sessão
n = int(self.headers.get('Content-Length', 0))
try:
d = json.loads(self.rfile.read(n) or b'{}')
except Exception:
return self._json({'erro': 'json inválido'}, 400)
if not SENHA or (d.get('senha') or '') == SENHA:
tok = secrets.token_urlsafe(24); _sessoes.add(tok)
corpo = json.dumps({'ok': True}).encode('utf-8')
self.send_response(200)
self.send_header('Content-Type', 'application/json; charset=utf-8')
self.send_header('Set-Cookie', 'sessao=%s; Path=/; HttpOnly; SameSite=Lax; Max-Age=%d' % (tok, 60 * 60 * 24 * 30))
self.send_header('Content-Length', str(len(corpo)))
self.end_headers(); self.wfile.write(corpo); return
return self._json({'erro': 'senha incorreta'}, 401)
if not self._auth_ok():
return self._json({'erro': 'nao autenticado'}, 401) # 401 simples (sem WWW-Authenticate, pra não abrir a caixinha do navegador)
u = urlparse(self.path)
if u.path == '/api/sync':
n = int(self.headers.get('Content-Length', 0))
try:
d = json.loads(self.rfile.read(n) or b'{}')
except Exception:
return self._json({'erro': 'json inválido'}, 400)
rev = aplicar(d.get('upserts'), d.get('deletes'), d.get('config'), d.get('afsUpserts'), d.get('afsDeletes'), d.get('dispUpserts'), d.get('dispDeletes'), d.get('recUpserts'), d.get('recDeletes'))
return self._json({'rev': rev})
self._json({'erro': 'rota desconhecida'}, 404)
def ips_locais():
ips = set()
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM); s.connect(('8.8.8.8', 80))
ips.add(s.getsockname()[0]); s.close()
except Exception:
pass
try:
for info in socket.getaddrinfo(socket.gethostname(), None, socket.AF_INET):
ips.add(info[4][0])
except Exception:
pass
return sorted(i for i in ips if not i.startswith('127.'))
class Servidor(socketserver.ThreadingMixIn, http.server.HTTPServer):
daemon_threads = True
allow_reuse_address = True
if __name__ == '__main__':
if not os.path.exists(os.path.join(RAIZ, HTML)):
print('!! Rode este arquivo DENTRO da pasta do sistema (onde está o "%s").' % HTML)
carregar()
srv = Servidor(('0.0.0.0', PORTA), Handler)
print('=' * 64)
print(' SISTEMA DE NOMEAÇÕES DA DRH — SERVIDOR LOCAL (tempo real)')
print('=' * 64)
print(' Base carregada:', len(_estado['nomeacoes']), 'lançamentos')
print()
print(' NESTE computador, abra: http://localhost:%d' % PORTA)
print()
print(' Passe para a EQUIPE (mesma rede) este endereço:')
for ip in ips_locais() or ['']:
print(' http://%s:%d' % (ip, PORTA))
print()
if SENHA:
print(' LOGIN (o navegador vai pedir): usuário: %s senha: %s' % (USUARIO, SENHA))
print()
print(' A base fica SÓ nesta máquina. Deixe esta janela ABERTA enquanto trabalham.')
print(' Para encerrar: feche a janela ou aperte Ctrl+C.')
print('=' * 64)
try:
srv.serve_forever()
except KeyboardInterrupt:
print('\nEncerrado.')