#!/usr/bin/env python3 """ Servidor local do Sistema de Ponto. Execute este arquivo para iniciar. Depois acesse http://localhost:80 """ import http.server import json import os import socket import socketserver import sys import threading import time import datetime import urllib.request PORT = 80 BASE_DIR = os.path.dirname(os.path.abspath(__file__)) CONFIG_FILE = os.path.join(BASE_DIR, 'dados_config.json') REGISTROS_FILE = os.path.join(BASE_DIR, 'dados_registros.json') # Rastreamento de alertas já enviados (evita repetição) _ausencia_verificada = set() # {data_str} _saida_almoco_alertado = set() # {(nome, data_str)} — por funcionária, persiste no restart via init_alertas() _retorno_alertado = set() # {(nome, data, hora_saida_almoco)} TIPOS_IN = {'entrada', 'retorno_almoco'} TIPOS_OUT = {'saida_almoco', 'saida'} TIPO_LABEL = { 'entrada': 'Entrada ✅', 'saida_almoco': 'Saída Almoço 🍽️', 'retorno_almoco': 'Retorno Almoço 🔄', 'saida': 'Saída 🔴', } def get_local_ip(): try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(('8.8.8.8', 80)) ip = s.getsockname()[0] s.close() return ip except Exception: return 'localhost' def read_json(path, default): try: with open(path, 'r', encoding='utf-8') as f: return json.load(f) except Exception: return default def write_json(path, data): with open(path, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) def get_alertas(cfg): """Retorna configuração de alertas com valores padrão.""" a = cfg.get('alertas', {}) return { 'entrada_seg_qui': a.get('entrada_seg_qui', '07:10'), 'entrada_sex': a.get('entrada_sex', '06:10'), 'saida_almoco': a.get('saida_almoco', '13:30'), 'retorno_almoco_max_min': int(a.get('retorno_almoco_max_min', 70)), 'limite_horas_padrao': int(a.get('limite_horas_padrao', 570)), } def get_limite_horas(cfg, nome): """Retorna o limite de horas diário para uma funcionária (individual ou padrão).""" for f in cfg.get('funcionarias', []): if f.get('nome') == nome: lim = f.get('limite_horas') if lim: try: return int(lim) except Exception: pass return get_alertas(cfg)['limite_horas_padrao'] def send_telegram(token, chat_id, message): """Envia mensagem via Telegram Bot API.""" if not token or not chat_id: return try: url = f"https://api.telegram.org/bot{token}/sendMessage" payload = json.dumps({ 'chat_id': chat_id, 'text': message, 'parse_mode': 'Markdown' }).encode('utf-8') req = urllib.request.Request(url, data=payload, headers={ 'Content-Type': 'application/json', 'User-Agent': 'SistemaPonto/1.0' }) urllib.request.urlopen(req, timeout=10) except Exception as e: print(f'[Telegram] Erro ao enviar: {e}') def send_telegram_async(token, chat_id, message): """Dispara send_telegram em background para não travar o servidor.""" threading.Thread(target=send_telegram, args=(token, chat_id, message), daemon=True).start() def send_document_telegram(token, chat_id, filepath, caption=''): """Envia um arquivo via Telegram Bot API (sendDocument, multipart).""" if not token or not chat_id: return boundary = b'SistemaPontoBoundary' filename = os.path.basename(filepath).encode('utf-8') with open(filepath, 'rb') as f: file_content = f.read() def part(name, value, extra_headers=b''): return ( b'--' + boundary + b'\r\n' + b'Content-Disposition: form-data; name="' + name + b'"' + extra_headers + b'\r\n\r\n' + value + b'\r\n' ) body = ( part(b'chat_id', chat_id.encode()) + part(b'caption', caption.encode('utf-8')) + part(b'parse_mode', b'Markdown') + part(b'document', file_content, b'; filename="' + filename + b'"\r\nContent-Type: application/octet-stream') + b'--' + boundary + b'--\r\n' ) url = f'https://api.telegram.org/bot{token}/sendDocument' req = urllib.request.Request(url, data=body, headers={ 'Content-Type': f'multipart/form-data; boundary={boundary.decode()}', 'Content-Length': str(len(body)), 'User-Agent': 'SistemaPonto/1.0', }) urllib.request.urlopen(req, timeout=30) def backup_loop(): """Envia dados_registros.json via Telegram a cada 3 dias. Persiste a data do último backup em dados_config.json para sobreviver a restarts.""" INTERVALO_DIAS = 3 while True: time.sleep(3600) # verifica a cada hora try: cfg = read_json(CONFIG_FILE, {}) tg = cfg.get('telegram', {}) token = tg.get('token', '') chat_id = tg.get('chat_id', '') if not token or not chat_id: continue hoje_str = datetime.datetime.now().strftime('%Y-%m-%d') ultimo_backup = cfg.get('ultimo_backup', '') # Calcula dias desde o último backup if ultimo_backup: try: delta = (datetime.datetime.strptime(hoje_str, '%Y-%m-%d') - datetime.datetime.strptime(ultimo_backup, '%Y-%m-%d')).days if delta < INTERVALO_DIAS: continue except Exception: pass empresa = cfg.get('empresa', 'Sistema de Ponto') now_str = datetime.datetime.now().strftime('%d/%m/%Y às %H:%M') caption = (f"📦 *{empresa}* — Backup automático\n" f"🗓 {now_str}\n" f"Arquivo: `dados_registros.json`") send_document_telegram(token, chat_id, REGISTROS_FILE, caption) # Persiste data do backup na config cfg['ultimo_backup'] = hoje_str write_json(CONFIG_FILE, cfg) print(f'[Backup] Backup enviado via Telegram em {hoje_str}.') except Exception as e: print(f'[Backup] Erro: {e}') def calc_horas_dia(registros, nome, data): """Retorna total de minutos trabalhados em um dia (exclui intervalo de almoço).""" regs = [r for r in registros if r.get('nome') == nome and r.get('data') == data] regs.sort(key=lambda r: r.get('timestamp', '')) total = 0 i = 0 while i < len(regs) - 1: if regs[i].get('tipo') in TIPOS_IN and regs[i+1].get('tipo') in TIPOS_OUT: try: eh, em = map(int, regs[i]['hora'].split(':')) sh, sm = map(int, regs[i+1]['hora'].split(':')) diff = (sh * 60 + sm) - (eh * 60 + em) if diff > 0: total += diff except Exception: pass i += 2 else: i += 1 return total # ============================================================ # INIT: PRÉ-POPULAR RASTREAMENTO A PARTIR DOS REGISTROS DO DIA # ============================================================ def init_alertas(): """ Chamada uma vez no startup. Lê os registros de hoje e marca como 'já tratado' qualquer funcionária que já tenha registrado saída para almoço (ou retorno/saída), evitando alertas duplicados após restart. """ hoje_str = datetime.datetime.now().strftime('%Y-%m-%d') registros = read_json(REGISTROS_FILE, []) # Agrupa registros de hoje por funcionária (ordenados por timestamp) por_func = {} for r in registros: if r.get('data') != hoje_str: continue nome = r.get('nome', '') por_func.setdefault(nome, []).append(r) for nome, regs in por_func.items(): regs.sort(key=lambda r: r.get('timestamp', '')) tipos = {r.get('tipo', '') for r in regs} # Se funcionária já saiu para almoço (ou retornou/saiu), marcar como tratada if tipos & {'saida_almoco', 'retorno_almoco', 'saida'}: _saida_almoco_alertado.add((nome, hoje_str)) # Para retorno: se já retornou após uma saida_almoco, marcar o par hora_saida_almoco = None for r in regs: if r.get('tipo') == 'saida_almoco': hora_saida_almoco = r.get('hora', '') elif r.get('tipo') in ('retorno_almoco', 'saida') and hora_saida_almoco: _retorno_alertado.add((nome, hoje_str, hora_saida_almoco)) hora_saida_almoco = None # reset para possível 2º ciclo print(f'[Init] Alertas pré-populados: ' f'{len(_saida_almoco_alertado)} saídas almoço, ' f'{len(_retorno_alertado)} retornos já tratados.') # ============================================================ # LOOPS DE ALERTAS EM BACKGROUND # ============================================================ def check_ausencias_loop(): """Alerta se funcionária não registrou entrada até o horário limite.""" while True: time.sleep(30) now = datetime.datetime.now() weekday = now.weekday() if weekday > 4: continue cfg = read_json(CONFIG_FILE, {}) alertas = get_alertas(cfg) lim_str = alertas['entrada_sex'] if weekday == 4 else alertas['entrada_seg_qui'] h, m = map(int, lim_str.split(':')) hora_lim = datetime.time(h, m) hoje_str = now.strftime('%Y-%m-%d') if hoje_str in _ausencia_verificada: continue if now.time() < hora_lim: continue # Verifica se hoje é feriado feriados = cfg.get('feriados', []) if hoje_str in feriados: _ausencia_verificada.add(hoje_str) continue _ausencia_verificada.add(hoje_str) registros = read_json(REGISTROS_FILE, []) tg = cfg.get('telegram', {}) token = tg.get('token', '') chat_id = tg.get('chat_id', '') empresa = cfg.get('empresa', 'Sistema de Ponto') ausentes = [] for func in cfg.get('funcionarias', []): nome = func.get('nome', '') teve = any( r.get('nome') == nome and r.get('data') == hoje_str and r.get('tipo') == 'entrada' for r in registros ) if not teve: ausentes.append(nome) if ausentes: nomes = '\n'.join(f'• {n}' for n in ausentes) msg = ( f"⏰ *{empresa}* — Ausência\n" f"Sem registro de entrada até {lim_str}:\n{nomes}" ) send_telegram_async(token, chat_id, msg) print(f'[Ausência] Alerta enviado para {len(ausentes)} funcionária(s).') def check_saida_almoco_loop(): """Alerta se funcionária não saiu para almoço até o horário limite. Rastreamento por funcionária evita alertas duplicados após restart do servidor.""" while True: time.sleep(30) now = datetime.datetime.now() weekday = now.weekday() if weekday > 4: continue cfg = read_json(CONFIG_FILE, {}) alertas = get_alertas(cfg) lim_str = alertas['saida_almoco'] h, m = map(int, lim_str.split(':')) hora_lim = datetime.time(h, m) if now.time() < hora_lim: continue hoje_str = now.strftime('%Y-%m-%d') registros = read_json(REGISTROS_FILE, []) tg = cfg.get('telegram', {}) token = tg.get('token', '') chat_id = tg.get('chat_id', '') empresa = cfg.get('empresa', 'Sistema de Ponto') sem_almoco = [] for func in cfg.get('funcionarias', []): nome = func.get('nome', '') key = (nome, hoje_str) if key in _saida_almoco_alertado: continue # já alertada ou já saiu regs_hoje = [r for r in registros if r.get('nome') == nome and r.get('data') == hoje_str] regs_hoje.sort(key=lambda r: r.get('timestamp', '')) if not regs_hoje: continue ultimo = regs_hoje[-1].get('tipo', '') # Funcionária já saiu para almoço, retornou ou encerrou o dia → marcar como tratada if ultimo in ('saida_almoco', 'retorno_almoco', 'saida'): _saida_almoco_alertado.add(key) continue # Ainda trabalhando e ainda não saiu para almoço → alertar (uma só vez) if ultimo == 'entrada': _saida_almoco_alertado.add(key) sem_almoco.append(nome) if sem_almoco: nomes = '\n'.join(f'• {n}' for n in sem_almoco) msg = ( f"🍽️ *{empresa}* — Almoço\n" f"Ainda sem saída para almoço às {lim_str}:\n{nomes}" ) send_telegram_async(token, chat_id, msg) print(f'[Almoço] Alerta de saída enviado para {len(sem_almoco)} funcionária(s).') def check_retorno_almoco_loop(): """Alerta se funcionária passou do tempo máximo no almoço sem retornar.""" while True: time.sleep(30) now = datetime.datetime.now() hoje_str = now.strftime('%Y-%m-%d') cfg = read_json(CONFIG_FILE, {}) alertas = get_alertas(cfg) max_min = alertas['retorno_almoco_max_min'] tg = cfg.get('telegram', {}) token = tg.get('token', '') chat_id = tg.get('chat_id', '') empresa = cfg.get('empresa', 'Sistema de Ponto') registros = read_json(REGISTROS_FILE, []) for func in cfg.get('funcionarias', []): nome = func.get('nome', '') regs_hoje = [r for r in registros if r.get('nome') == nome and r.get('data') == hoje_str] regs_hoje.sort(key=lambda r: r.get('timestamp', '')) if not regs_hoje: continue ultimo = regs_hoje[-1] if ultimo.get('tipo') != 'saida_almoco': continue hora_saida = ultimo.get('hora', '') key = (nome, hoje_str, hora_saida) if key in _retorno_alertado: continue try: sh, sm = map(int, hora_saida.split(':')) saida_dt = now.replace(hour=sh, minute=sm, second=0, microsecond=0) diff_min = (now - saida_dt).total_seconds() / 60 if diff_min >= max_min: _retorno_alertado.add(key) h_e = int(diff_min // 60) m_e = int(diff_min % 60) lim_h = max_min // 60 lim_m = max_min % 60 msg = ( f"⏰ *{empresa}* — Almoço longo\n" f"{nome} está no almoço há *{h_e}h{m_e:02d}min* " f"(limite: {lim_h}h{lim_m:02d}min)" ) send_telegram_async(token, chat_id, msg) print(f'[Almoço] Alerta de retorno enviado para {nome}.') except Exception as e: print(f'[Retorno] Erro ao verificar {nome}: {e}') # ============================================================ # SERVIDOR HTTP MULTI-THREAD # ============================================================ class PontoServer(socketserver.ThreadingMixIn, http.server.HTTPServer): """Servidor HTTP que atende cada requisição em thread separada. Evita bloqueio quando múltiplas funcionárias registram ao mesmo tempo.""" daemon_threads = True # threads morrem com o processo principal def handle_error(self, request, client_address): """Suprime erros de conexão encerrada pelo cliente (normais em browsers/mobile).""" exc = sys.exc_info()[0] if exc in (ConnectionResetError, BrokenPipeError, ConnectionAbortedError): return # não logar — não é um problema real super().handle_error(request, client_address) def get_request(self): """Define timeout de 30s na conexão — descarta conexões penduradas automaticamente.""" sock, addr = super().get_request() sock.settimeout(30) return sock, addr # ============================================================ # HANDLER HTTP # ============================================================ class PontoHandler(http.server.SimpleHTTPRequestHandler): def do_OPTIONS(self): self.send_response(200) self._cors() self.end_headers() def do_GET(self): if self.path == '/api/config': self._json(read_json(CONFIG_FILE, {})) elif self.path == '/api/registros': self._json(read_json(REGISTROS_FILE, [])) else: super().do_GET() def do_POST(self): length = int(self.headers.get('Content-Length', 0)) body = self.rfile.read(length) try: data = json.loads(body) except Exception: self.send_response(400) self.end_headers() return # ── Salvar config completa ───────────────────────────── if self.path == '/api/config': write_json(CONFIG_FILE, data) self._json({'ok': True}) # ── Salvar registros completos ───────────────────────── elif self.path == '/api/registros': write_json(REGISTROS_FILE, data) self._json({'ok': True}) # ── Registrar ponto (funcionária) ────────────────────── elif self.path == '/api/registrar': registros = read_json(REGISTROS_FILE, []) registros.append(data) write_json(REGISTROS_FILE, registros) cfg = read_json(CONFIG_FILE, {}) tg = cfg.get('telegram', {}) token = tg.get('token', '') chat_id = tg.get('chat_id', '') empresa = cfg.get('empresa', 'Sistema de Ponto') tipo_label = TIPO_LABEL.get(data.get('tipo', ''), data.get('tipo', '')) nome_reg = data.get('nome', '') dia_reg = data.get('data', '') if data.get('tipo') == 'saida': # Calcula horas do dia total_dia = calc_horas_dia(registros, nome_reg, dia_reg) h_d = total_dia // 60 m_d = total_dia % 60 extra = f"\n📊 Total hoje: *{h_d}h{m_d:02d}min*" # Na sexta, adiciona total da semana try: data_dt = datetime.datetime.strptime(dia_reg, '%Y-%m-%d') if data_dt.weekday() == 4: # sexta segunda = data_dt - datetime.timedelta(days=4) total_sem = 0 d_iter = segunda while d_iter <= data_dt: total_sem += calc_horas_dia(registros, nome_reg, d_iter.strftime('%Y-%m-%d')) d_iter += datetime.timedelta(days=1) h_s = total_sem // 60 m_s = total_sem % 60 extra += f"\n📅 Semana: *{h_s}h{m_s:02d}min*" except Exception: pass msg = f"📋 *{empresa}*\n{nome_reg} — {tipo_label} às {data.get('hora')}{extra}" else: msg = f"📋 *{empresa}*\n{nome_reg} — {tipo_label} às {data.get('hora')}" send_telegram_async(token, chat_id, msg) # Verifica limite de horas ao registrar qualquer saída if data.get('tipo') in TIPOS_OUT: nome = data.get('nome') dia = data.get('data') prev_regs = [r for r in registros if r.get('id') != data.get('id')] total_prev = calc_horas_dia(prev_regs, nome, dia) total_novo = calc_horas_dia(registros, nome, dia) limite = get_limite_horas(cfg, nome) if total_novo >= limite and total_prev < limite: h = total_novo // 60 m = total_novo % 60 h_lim = limite // 60 m_lim = limite % 60 msg_alerta = ( f"⚠️ *{empresa}* — Horas extras\n" f"{nome} trabalhou *{h}h{m:02d}min* hoje " f"(limite: {h_lim}h{m_lim:02d}min)" ) send_telegram_async(token, chat_id, msg_alerta) self._json({'ok': True, 'registros': registros}) # ── Admin registra ponto por funcionária ─────────────── elif self.path == '/api/admin-registrar': registros = read_json(REGISTROS_FILE, []) registros.append(data) registros.sort(key=lambda r: (r.get('data', ''), r.get('hora', ''))) write_json(REGISTROS_FILE, registros) # Atualiza rastreamento de alertas para evitar duplicatas nome = data.get('nome', '') dia = data.get('data', '') tipo = data.get('tipo', '') if tipo in ('saida_almoco', 'retorno_almoco', 'saida'): _saida_almoco_alertado.add((nome, dia)) if tipo in ('retorno_almoco', 'saida'): # Tenta marcar a saida_almoco correspondente no rastreamento de retorno hora_sa = next((r.get('hora') for r in reversed(registros) if r.get('nome') == nome and r.get('data') == dia and r.get('tipo') == 'saida_almoco'), None) if hora_sa: _retorno_alertado.add((nome, dia, hora_sa)) self._json({'ok': True, 'registros': registros}) # ── Editar horário de um registro ────────────────────── elif self.path == '/api/registros/editar': reg_id = data.get('id') nova_hora = data.get('hora') registros = read_json(REGISTROS_FILE, []) for r in registros: if r.get('id') == reg_id: r['hora'] = nova_hora break write_json(REGISTROS_FILE, registros) self._json({'ok': True, 'registros': registros}) # ── Excluir um registro ──────────────────────────────── elif self.path == '/api/registros/excluir': reg_id = data.get('id') registros = read_json(REGISTROS_FILE, []) registros = [r for r in registros if r.get('id') != reg_id] write_json(REGISTROS_FILE, registros) self._json({'ok': True, 'registros': registros}) # ── Testar Telegram ──────────────────────────────────── elif self.path == '/api/testar-telegram': cfg = read_json(CONFIG_FILE, {}) tg = cfg.get('telegram', {}) token = tg.get('token', '') chat_id = tg.get('chat_id', '') empresa = cfg.get('empresa', 'Sistema de Ponto') if not token or not chat_id: self._json({'ok': False, 'erro': 'Token ou Chat ID não configurados.'}) return send_telegram(token, chat_id, f"✅ *{empresa}* — Notificações Telegram funcionando!") self._json({'ok': True}) else: self.send_response(404) self.end_headers() def _json(self, data): body = json.dumps(data, ensure_ascii=False).encode('utf-8') self.send_response(200) self.send_header('Content-Type', 'application/json; charset=utf-8') self.send_header('Content-Length', str(len(body))) self._cors() self.end_headers() self.wfile.write(body) def _cors(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') def log_message(self, format, *args): pass # suprime logs do console if __name__ == '__main__': os.chdir(BASE_DIR) ip = get_local_ip() # Pré-popula rastreamento de alertas a partir dos registros do dia init_alertas() # Inicia todos os loops de alerta e backup em background for loop_fn in [check_ausencias_loop, check_saida_almoco_loop, check_retorno_almoco_loop, backup_loop]: threading.Thread(target=loop_fn, daemon=True).start() server = PontoServer(('0.0.0.0', PORT), PontoHandler) print('=' * 52) print(' SISTEMA DE PONTO — SERVIDOR INICIADO') print('=' * 52) print(f' Neste computador: http://localhost') print(f' Outros computadores: http://{ip}') print() print(' Para parar: feche esta janela ou Ctrl+C') print('=' * 52) try: server.serve_forever() except KeyboardInterrupt: print('\nServidor encerrado.')