#!/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 threading import time import datetime import urllib.request import urllib.parse 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') LIMITE_HORAS_MINUTOS = 9 * 60 + 30 # 9h30 = 570 minutos 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 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.""" t = threading.Thread(target=send_telegram, args=(token, chat_id, message), daemon=True) t.start() _ausencia_verificada = set() # datas já verificadas (evita enviar mais de uma vez) def check_ausencias_loop(): """Roda em background e envia alerta de ausência no horário limite.""" while True: time.sleep(30) now = datetime.datetime.now() weekday = now.weekday() # 0=seg, 1=ter, 2=qua, 3=qui, 4=sex, 5=sab, 6=dom if weekday > 4: # fim de semana, ignora continue hora_limite = datetime.time(7, 10) if weekday < 4 else datetime.time(6, 10) hoje_str = now.strftime('%Y-%m-%d') if hoje_str in _ausencia_verificada: continue # já verificou hoje if now.time() < hora_limite: continue # ainda não chegou no horário _ausencia_verificada.add(hoje_str) cfg = read_json(CONFIG_FILE, {}) 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_entrada = any( r.get('nome') == nome and r.get('data') == hoje_str and r.get('tipo') == 'entrada' for r in registros ) if not teve_entrada: ausentes.append(nome) if ausentes: hora_str = hora_limite.strftime('%H:%M') nomes = '\n'.join(f'• {n}' for n in ausentes) msg = ( f"⏰ *{empresa}* — Ausência\n" f"Sem registro de entrada até {hora_str}:\n{nomes}" ) send_telegram_async(token, chat_id, msg) print(f'[Ausência] Alerta enviado para {len(ausentes)} funcionária(s).') def calc_horas_dia(registros, nome, data): """Retorna total de minutos trabalhados por uma pessoa em um dia.""" 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') == 'entrada' and regs[i+1].get('tipo') == 'saida': 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 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 if self.path == '/api/config': write_json(CONFIG_FILE, data) self._json({'ok': True}) elif self.path == '/api/registros': write_json(REGISTROS_FILE, data) self._json({'ok': True}) elif self.path == '/api/registrar': # Registra um único ponto e envia notificação Telegram 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 = 'Entrada ✅' if data.get('tipo') == 'entrada' else 'Saída 🔴' msg = f"📋 *{empresa}*\n{data.get('nome')} — {tipo_label} às {data.get('hora')}" send_telegram_async(token, chat_id, msg) # Verifica limite de 9h30 apenas quando registra saída if data.get('tipo') == 'saida': 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) # Só alerta na primeira vez que cruza o limite if total_novo >= LIMITE_HORAS_MINUTOS and total_prev < LIMITE_HORAS_MINUTOS: h = total_novo // 60 m = total_novo % 60 msg_alerta = ( f"⚠️ *{empresa}* — Horas extras\n" f"{nome} trabalhou *{h}h{m:02d}min* hoje " f"(limite: 9h30)" ) send_telegram_async(token, chat_id, msg_alerta) self._json({'ok': True, 'registros': registros}) 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 if __name__ == '__main__': os.chdir(BASE_DIR) ip = get_local_ip() # Inicia verificação de ausências em background threading.Thread(target=check_ausencias_loop, daemon=True).start() server = http.server.HTTPServer(('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.')