# -*- coding: utf-8 -*- """ Integrador HITS-PMS – Reservas & Detalhes ---------------------------------------- Versão: 20/05/2025 Sincronia estrita entre API ↔ JSON ↔ MySQL: • INSERT de reservas novas • UPDATE quando dateUp muda • DELETE quando a reserva some da API Principais recursos • requests.Session() | MySQLConnectionPool | Transações atômicas • Chaves de cache corretas: (idReservation,idEntity) e (idReservation,idEntityCompany) • Logs separados: operations, inserts, updates, deletes, errors, api """ from __future__ import annotations import json import logging import os import sys import time from collections import defaultdict from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Dict, List, Tuple import mysql.connector from mysql.connector import pooling import requests from zoneinfo import ZoneInfo # --------------------------------------------------------------------------- # Configuração geral # --------------------------------------------------------------------------- TZ = ZoneInfo("America/Fortaleza") POLL_INTERVAL = 300 # segundos PAGE_SIZE = 50 STATUSES = [1, 2, 3, 4] # Confirmada, Cancelada, Processada, Bloqueada TYPES = [1, 2] # 1 = inclusão, 2 = atualização (0 = check-in se desejar) FULL_SYNC_START = "2025-03-01T00:00:00Z" BASE_DIR = Path(sys.executable if getattr(sys, "frozen", False) else __file__).resolve().parent DATA_DIR = BASE_DIR / "data" LOG_ROOT = BASE_DIR / "logs" FLAG_FILE = DATA_DIR / "flag_fullsync.ok" for p in (DATA_DIR, LOG_ROOT): p.mkdir(parents=True, exist_ok=True) LOG_DIRS = {n: LOG_ROOT / n for n in ["errors", "updates", "inserts", "deletes", "operations", "api"]} for p in LOG_DIRS.values(): p.mkdir(exist_ok=True) FMT = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s", "%Y-%m-%d %H:%M:%S") def _logger(name: str, path: Path, level=logging.INFO): lg = logging.getLogger(name) lg.propagate = False if lg.handlers: return lg lg.setLevel(level) fh = logging.FileHandler(path, encoding="utf-8") fh.setFormatter(FMT) lg.addHandler(fh) sh = logging.StreamHandler() sh.setFormatter(FMT) lg.addHandler(sh) lg.info("%s logger iniciado (%s)", name, path) return lg log_ops = _logger("operations", LOG_DIRS["operations"] / "operations.log") log_err = _logger("errors", LOG_DIRS["errors"] / "errors.log", level=logging.ERROR) log_upd = _logger("updates", LOG_DIRS["updates"] / "updates.log") log_ins = _logger("inserts", LOG_DIRS["inserts"] / "inserts.log") log_del = _logger("deletes", LOG_DIRS["deletes"] / "deletes.log") log_api = _logger("api", LOG_DIRS["api"] / "api.log") # --------------------------------------------------------------------------- # Pool de conexões MySQL # --------------------------------------------------------------------------- POOL = pooling.MySQLConnectionPool( pool_name = "hits_pool", pool_size = 6, host = os.getenv("DB_HOST", "localhost"), user = os.getenv("DB_USER", "root"), password = os.getenv("DB_PASS", ""), database = os.getenv("DB_NAME", "bihits"), charset = "utf8mb4", autocommit = False, ) def db_conn(): return POOL.get_connection() # --------------------------------------------------------------------------- # Helpers de data / cache # --------------------------------------------------------------------------- def parse_iso(dt: str | None): if not dt: return None if dt.endswith("Z"): dt = dt[:-1] + "+00:00" return datetime.fromisoformat(dt).astimezone(timezone.utc).replace(tzinfo=None) def load_cache(path: Path): if not path.is_file(): return {} with path.open("r", encoding="utf-8") as f: data = json.load(f) return {(i["idReservation"], i.get("idEntity") or i.get("idEntityCompany")): i for i in data} def save_cache(path: Path, data: Dict[Tuple[int, int], dict]): with path.open("w", encoding="utf-8") as f: json.dump(list(data.values()), f, ensure_ascii=False, indent=2) # --------------------------------------------------------------------------- # API helpers # --------------------------------------------------------------------------- SESSION = requests.Session() API_BASE = "https://api.hitspms.net/Datashare/WebCheckinOut" def make_headers(token: str): return { "Accept" : "application/json", "X-API-VERSION" : "1", "X-API-TENANT-NAME" : "thecoralbeachresort", "X-API-PROPERTY-CODE": "1", "X-API-PARTNER-USERID": "0", "X-API-LANGUAGE-CODE": "pt-br", "X-Client-Id" : "THECORALBR", "Authorization" : f"Bearer {token}", } def get_token(): url = "https://bric-investment.com/BI-HITS/dados/Hits/token/token.php" try: r = SESSION.post(url, data={"username": os.getenv("API_USER"), "password": os.getenv("API_PASS")}, timeout=30) r.raise_for_status() txt = r.text.strip() return txt.split(":", 1)[1].strip() if txt.startswith("Token obtido:") else r.json().get("token") except Exception as exc: log_err.error("Erro ao obter token: %s", exc) return None def safe_get(url: str, headers: dict, params: dict | None = None, timeout: int = 30): delay = 5 for _ in range(6): try: r = SESSION.get(url, headers=headers, params=params, timeout=timeout) log_api.info("GET %s → %s", r.url, r.status_code) except requests.RequestException as exc: log_err.error("RequestException: %s", exc) time.sleep(delay); delay = min(delay * 2, 60) continue # 401 – renova token if r.status_code == 401: new_tk = get_token() if new_tk: headers["Authorization"] = f"Bearer {new_tk}" continue return None # 429 – back-off if r.status_code == 429: wait = int(r.headers.get("Retry-After", delay)) time.sleep(wait) continue if r.ok: return r time.sleep(delay) return None # --------------------------------------------------------------------------- # Fetchers # --------------------------------------------------------------------------- def fetch_reservations(token: str, start: str, end: str, t: int, st: int): url = f"{API_BASE}/Reservations" hdr = make_headers(token) out: List[dict] = [] page = 0 while True: resp = safe_get(url, hdr, { "Type" : t, "Status" : st, "InitialDate": start, "FinalDate" : end, "Page" : page, "Size" : PAGE_SIZE, }) if not resp: break data = resp.json() if not data: break out.extend(data) if len(data) < PAGE_SIZE: break page += 1 return out def fetch_detail(token: str, rid: int): resp = safe_get(f"{API_BASE}/Reservation/{rid}", make_headers(token)) return resp.json() if resp else None # --------------------------------------------------------------------------- # SQL statements # --------------------------------------------------------------------------- SQL_INS_RES = ( "INSERT INTO reservations " "(idReservation,idEntity,name,phone,main,mail,zipCode,addressDetails,address,neighborhood,number," "city,country,stateName,stateCode,federalRegistrationNumber,documentType,checkIn,checkOut," "dateAdd,dateUp,status,integrator,channel,raw_json) " "VALUES (%(idReservation)s,%(idEntity)s,%(name)s,%(phone)s,%(main)s,%(mail)s,%(zipCode)s," "%(addressDetails)s,%(address)s,%(neighborhood)s,%(number)s,%(city)s,%(country)s,%(stateName)s," "%(stateCode)s,%(federalRegistrationNumber)s,%(documentType)s,%(checkIn)s,%(checkOut)s," "%(dateAdd)s,%(dateUp)s,%(status)s,%(integrator)s,%(channel)s,%(raw_json)s) " "ON DUPLICATE KEY UPDATE " "name=VALUES(name),phone=VALUES(phone),main=VALUES(main),mail=VALUES(mail),zipCode=VALUES(zipCode)," "addressDetails=VALUES(addressDetails),address=VALUES(address),neighborhood=VALUES(neighborhood)," "number=VALUES(number),city=VALUES(city),country=VALUES(country),stateName=VALUES(stateName)," "stateCode=VALUES(stateCode),federalRegistrationNumber=VALUES(federalRegistrationNumber)," "documentType=VALUES(documentType),checkIn=VALUES(checkIn),checkOut=VALUES(checkOut)," "dateUp=VALUES(dateUp),status=VALUES(status),integrator=VALUES(integrator)," "channel=VALUES(channel),raw_json=VALUES(raw_json)" ) SQL_INS_DET_MAIN = ( "INSERT INTO reservation_details " "(idReservation,idEntityCompany,companyName,idRequesterCompany,requesterCompanyName,groupName," "contactName,contact1,contact2,dateAdd,dateUp,creditState) " "VALUES (%(idReservation)s,%(idEntityCompany)s,%(companyName)s,%(idRequesterCompany)s," "%(requesterCompanyName)s,%(groupName)s,%(contactName)s,%(contact1)s,%(contact2)s," "%(dateAdd)s,%(dateUp)s,%(creditState)s) " "ON DUPLICATE KEY UPDATE " "companyName=VALUES(companyName),idRequesterCompany=VALUES(idRequesterCompany)," "requesterCompanyName=VALUES(requesterCompanyName),groupName=VALUES(groupName)," "contactName=VALUES(contactName),contact1=VALUES(contact1),contact2=VALUES(contact2)," "dateUp=VALUES(dateUp),creditState=VALUES(creditState)" ) SQL_NOTE = "INSERT INTO reservation_notes (idReservation,idEntityCompany,note_index,note) VALUES (%s,%s,%s,%s)" SQL_ROOM = ("INSERT INTO reservation_rooms " "(idReservation,idEntityCompany,room_index,idRoom,code,checkIn,checkOut,idRoomType," "roomTypeName,amount,status,pax,chd1,chd2,chd3," "ratePlanId,ratePlanName,mealPlanDesc) " "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)") SQL_REQ = "INSERT INTO reservation_room_requirements (idReservation,idEntityCompany,room_index,req_index,requirement) VALUES (%s,%s,%s,%s,%s)" SQL_GUEST = ("INSERT INTO reservation_guests " "(idReservation,idEntityCompany,guest_index,idEntity,name,idRoom,contactMail,contactPhone,main," "federalRegistrationNumber,documentType,gender,birthDate,mainDocType,docCpfCnpjPassport," "addressZipCode,addressCountry,addressStateCode,addressStateName,addressCity,addressNeighborhood," "addressAddress,addressDetails,addressNumber) " "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)") SQL_GNOTE = "INSERT INTO reservation_guest_notes (idReservation,idEntityCompany,guest_index,note_index,note_text) VALUES (%s,%s,%s,%s,%s)" SQL_COMM = "INSERT INTO reservation_commissions (idReservation,idEntityCompany,comm_index,idEntity,percentage,value) VALUES (%s,%s,%s,%s,%s,%s)" SQL_REV = ("INSERT INTO reservation_revenue " "(idReservation,idEntityCompany,marketSegmentId,marketSegmentName,idReservationChannel,nameChannel," "purposeStayId,purpose,managerID,managerName) " "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)") # --------------------------------------------------------------------------- # Funções de upsert # --------------------------------------------------------------------------- def get_existing_reservations_map(): conn = db_conn(); cur = conn.cursor() cur.execute("SELECT idReservation,idEntity,dateUp FROM reservations") m = {(r,e): du for r,e,du in cur} cur.close(); conn.close() return m def get_existing_details_map(): conn = db_conn(); cur = conn.cursor() cur.execute("SELECT idReservation,idEntityCompany,dateUp FROM reservation_details") m = {(r,e): du for r,e,du in cur} cur.close(); conn.close() return m def upsert_reservations(recs: List[dict]): if not recs: return existing = get_existing_reservations_map() rows = [] for r in recs: key = (r["idReservation"], r["idEntity"]) if key not in existing or parse_iso(r["dateUp"]) > existing[key]: rows.append({ "idReservation": r["idReservation"], "idEntity": r["idEntity"], "name": r.get("name"), "phone": r.get("phone"), "main": r.get("main"), "mail": r.get("mail"), "zipCode": r.get("zipCode"), "addressDetails":r.get("addressDetails"), "address": r.get("address"), "neighborhood": r.get("neighborhood"), "number": r.get("number"), "city": r.get("city"), "country": r.get("country"), "stateName": r.get("stateName"), "stateCode": r.get("stateCode"), "federalRegistrationNumber": r.get("federalRegistrationNumber"), "documentType": r.get("documentType"), "checkIn": r.get("checkIn"), "checkOut": r.get("checkOut"), "dateAdd": r.get("dateAdd"), "dateUp": r.get("dateUp"), "status": r.get("status"), "integrator": r.get("integrator"), "channel": r.get("channel"), "raw_json": json.dumps(r, ensure_ascii=False), }) (log_ins if key not in existing else log_upd).info("reservations %s", key) if not rows: return conn = db_conn(); cur = conn.cursor() try: cur.executemany(SQL_INS_RES, rows) conn.commit() except Exception: conn.rollback(); log_err.exception("upsert_reservations") raise finally: cur.close(); conn.close() def upsert_reservation_details(details: List[dict]): if not details: return existing = get_existing_details_map() conn = db_conn(); cur = conn.cursor() try: conn.start_transaction() for det in details: key = (det["idReservation"], det["idEntityCompany"]) dt_a = parse_iso(det["dateUp"]) if key in existing and dt_a <= existing[key]: continue # upsert registro principal cur.execute(SQL_INS_DET_MAIN, { "idReservation" : det["idReservation"], "idEntityCompany" : det["idEntityCompany"], "companyName" : det.get("companyName"), "idRequesterCompany" : det.get("idRequesterCompany"), "requesterCompanyName":det.get("requesterCompanyName"), "groupName" : det.get("groupName"), "contactName" : det.get("contactName"), "contact1" : det.get("contact1"), "contact2" : det.get("contact2"), "dateAdd" : det.get("dateAdd"), "dateUp" : det.get("dateUp"), "creditState" : det.get("creditState"), }) # limpa filhas for tbl in ["reservation_notes","reservation_room_requirements", "reservation_rooms","reservation_guest_notes", "reservation_guests","reservation_commissions", "reservation_revenue"]: cur.execute(f"DELETE FROM {tbl} WHERE idReservation=%s AND idEntityCompany=%s", key) # notas cur.executemany(SQL_NOTE, [ (det["idReservation"], det["idEntityCompany"], i, n.get("note")) for i,n in enumerate(det.get("notes", [])) ]) # quartos + reqs room_rows, req_rows = [], [] for r_idx, room in enumerate(det.get("rooms", [])): room_rows.append(( det["idReservation"], det["idEntityCompany"], r_idx, room.get("idRoom"), room.get("code"), room.get("checkIn"), room.get("checkOut"), room.get("idRoomType"), room.get("roomTypeName"), room.get("amount"), room.get("status"), room.get("pax"), room.get("chd1"), room.get("chd2"), room.get("chd3"), room.get("ratePlanId"), room.get("ratePlanName"), room.get("mealPlanDesc"), )) req_rows.extend([ (det["idReservation"], det["idEntityCompany"], r_idx, q_idx, json.dumps(req, ensure_ascii=False)) for q_idx, req in enumerate(room.get("requirementReservation", [])) ]) if room_rows: cur.executemany(SQL_ROOM, room_rows) if req_rows: cur.executemany(SQL_REQ, req_rows) # guests + notes g_rows, gn_rows = [], [] for g_idx, g in enumerate(det.get("guests", [])): g_rows.append(( det["idReservation"], det["idEntityCompany"], g_idx, g.get("idEntity"), g.get("name"), g.get("idRoom"), g.get("contactMail"), g.get("contactPhone"), g.get("main"), g.get("federalRegistrationNumber"), g.get("documentType"), g.get("gender"), g.get("birthDate"), g.get("mainDocType"), g.get("docCpfCnpjPassport"), g.get("addressZipCode"), g.get("addressCountry"), g.get("addressStateCode"), g.get("addressStateName"), g.get("addressCity"), g.get("addressNeighborhood"), g.get("addressAddress"), g.get("addressDetails"), g.get("addressNumber"), )) gn_rows.extend([ (det["idReservation"], det["idEntityCompany"], g_idx, n_idx, n.get("note")) for n_idx, n in enumerate(g.get("notes", [])) ]) if g_rows: cur.executemany(SQL_GUEST, g_rows) if gn_rows: cur.executemany(SQL_GNOTE, gn_rows) # commissions c_rows = [ (det["idReservation"], det["idEntityCompany"], i, c.get("idEntity"), c.get("percentage"), c.get("value")) for i,c in enumerate(det.get("commissions", [])) ] if c_rows: cur.executemany(SQL_COMM, c_rows) # revenue rm = det.get("revenueManagement", {}) if rm: cur.execute(SQL_REV, ( det["idReservation"], det["idEntityCompany"], rm.get("marketSegmentId"), rm.get("marketSegmentName"), rm.get("idReservationChannel"), rm.get("nameChannel"), rm.get("purposeStayId"), rm.get("purpose"), rm.get("managerID"), rm.get("managerName"), )) (log_ins if key not in existing else log_upd).info("reservation_details %s", key) conn.commit() except Exception: conn.rollback(); log_err.exception("upsert_reservation_details") raise finally: cur.close(); conn.close() # --------------------------------------------------------------------------- # Truncamento (full-sync inicial) # --------------------------------------------------------------------------- def truncate_tables(): conn = db_conn(); cur = conn.cursor() cur.execute("SET FOREIGN_KEY_CHECKS=0") for tbl in ["reservation_room_requirements","reservation_rooms", "reservation_guest_notes","reservation_guests", "reservation_commissions","reservation_revenue", "reservation_notes","reservation_details","reservations"]: cur.execute(f"TRUNCATE TABLE {tbl}") cur.execute("SET FOREIGN_KEY_CHECKS=1") conn.commit() cur.close(); conn.close() log_ops.info("Tabelas truncadas – full-sync inicial") # --------------------------------------------------------------------------- # Loop principal # --------------------------------------------------------------------------- def main(): path_res = DATA_DIR / "reservations.json" path_det = DATA_DIR / "reservation_details.json" # --- Full sync inicial apenas trunca tabelas e cria flag, sem salvar JSON vazio --- if not FLAG_FILE.exists(): truncate_tables() FLAG_FILE.touch() last_dt = parse_iso(FULL_SYNC_START) else: # Nas execuções seguintes, vamos inicializar last_dt pelo cache em disco old_res_temp = load_cache(path_res) last_dt = max( (parse_iso(r["dateUp"]) for r in old_res_temp.values()), default=parse_iso(FULL_SYNC_START) ) # Carrega cache em memória ({} na primeira vez, ou conteúdo existente) old_res = load_cache(path_res) old_det = load_cache(path_det) while True: # Janela de busca: de 5 minutos antes do último update até agora win_start = (last_dt - timedelta(minutes=5)).strftime("%Y-%m-%dT%H:%M:%SZ") win_end = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ") log_ops.info("Window %s → %s", win_start, win_end) token = get_token() if not token: time.sleep(POLL_INTERVAL) continue # 1. Busca todas as reservas na janela para cada tipo/status all_api: Dict[Tuple[int,int], dict] = {} res_last_dt: Dict[int, datetime] = defaultdict(lambda: datetime(1970,1,1)) for t in TYPES: for st in STATUSES: for r in fetch_reservations(token, win_start, win_end, t, st): k = (r["idReservation"], r["idEntity"]) dt = parse_iso(r["dateUp"]) # mantém apenas a versão mais recente por key if dt > parse_iso(all_api.get(k, {}).get("dateUp", FULL_SYNC_START)): all_api[k] = r # track do maior dateUp por reservation id res_last_dt[r["idReservation"]] = max(res_last_dt[r["idReservation"]], dt) # 2. Identifica quais reservas precisam de detalhe existing_ids = {k[0] for k in old_res} to_fetch = { rid for rid, dt in res_last_dt.items() if rid not in existing_ids or dt > max( parse_iso(v["dateUp"]) for k,v in old_res.items() if k[0] == rid ) } new_details: List[dict] = [] for idx, rid in enumerate(to_fetch, start=1): log_ops.info("[%d/%d] fetching detail %s", idx, len(to_fetch), rid) det = fetch_detail(token, rid) if not det: continue real_k = (det["idReservation"], det["idEntityCompany"]) # atualiza somente este registro no cache de detalhes old_det[real_k] = det new_details.append(det) # garante que o cache "rasa" old_res fique consistente rasa_k = (det["idReservation"], det.get("idEntity")) if rasa_k in all_api: old_res[rasa_k] = all_api[rasa_k] time.sleep(1.0) # 3. Upserts no MySQL try: upsert_reservations(list(all_api.values())) upsert_reservation_details(new_details) except Exception: log_err.exception("Falha em upserts; ciclo interrompido") break # 4. Deleções estritas no banco e no cache previous_keys = set(old_res) current_keys = set(all_api) missing = previous_keys - current_keys if missing: conn = db_conn() cur = conn.cursor() conn.start_transaction() for k in missing: rid, ent = k cur.execute("DELETE FROM reservation_details WHERE idReservation=%s", (rid,)) cur.execute( "DELETE FROM reservations WHERE idReservation=%s AND idEntity=%s", (rid, ent) ) log_del.info("reservation deleted %s", k) old_res.pop(k, None) # remove também detalhes relacionados desse reservation for dk in [d for d in old_det if d[0] == rid]: old_det.pop(dk, None) conn.commit() cur.close() conn.close() # 5. Atualiza o cache em disco **incremental** (só novos ou modificados) for k, v in all_api.items(): if k not in old_res or parse_iso(v["dateUp"]) > parse_iso(old_res[k]["dateUp"]): old_res[k] = v save_cache(path_res, old_res) save_cache(path_det, old_det) # Prepara last_dt para próxima iteração if res_last_dt: last_dt = max(res_last_dt.values()) log_ops.info( "Ciclo completo — inserts/updates: %d, deletes: %d", len(to_fetch), len(missing) ) time.sleep(POLL_INTERVAL) # ───────────────────────────────────────────────────────────────────────────── # ponto de entrada # ───────────────────────────────────────────────────────────────────────────── if __name__ == "__main__": try: main() except Exception as e: log_err.exception("Erro fatal no main: %s", e)