# -*- 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" # 1) Full‐sync inicial: trunca tabelas e define last_dt if not FLAG_FILE.exists(): truncate_tables() FLAG_FILE.touch() last_dt = parse_iso(FULL_SYNC_START) else: existing = load_cache(path_res) last_dt = max( (parse_iso(r["dateUp"]) for r in existing.values()), default=parse_iso(FULL_SYNC_START) ) # 2) Carrega o cache completo em memória old_res = load_cache(path_res) old_det = load_cache(path_det) while True: changed = False # 3) Janela de busca 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 # 4) Busca apenas as reservas que mudaram nesse intervalo 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"]) if dt > parse_iso(all_api.get(k, {}).get("dateUp", FULL_SYNC_START)): all_api[k] = r res_last_dt[r["idReservation"]] = max(res_last_dt[r["idReservation"]], dt) # 5) Identifica e busca detalhes apenas de registros novos ou atualizados 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"]) old_det[real_k] = det new_details.append(det) changed = True time.sleep(1.0) # 6) Upserts (INSERT/UPDATE) 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 # 7) Deleções DESABILITADAS (para não remover registros que não vieram na janela de updates) # previous_keys = set(old_res) # current_keys = set(all_api) # missing = previous_keys - current_keys # if missing: # ... (não executado) ... # 8) Mescla incremental no cache em memória 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 changed = True # 9) Grava JSON só se houve alteração if changed: save_cache(path_res, old_res) save_cache(path_det, old_det) # 10) Atualiza last_dt para a 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), 0 ) 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)