# -*- 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 = [0, 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]): entries = list(data.values()) with path.open("w", encoding="utf-8") as f: json.dump(entries, f, ensure_ascii=False, indent=2) log_ops.info("save_cache: gravou %d entradas em %s", len(entries), path.name) # --------------------------------------------------------------------------- # 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: log_err.error( "fetch_reservations: falhou página %d (Type=%d Status=%d)", page, t, st ) break data = resp.json() log_api.info( "fetch_reservations: Type=%d Status=%d page %d → %d registros", t, st, page, len(data) ) if not data: break out.extend(data) if len(data) < PAGE_SIZE: break page += 1 log_api.info( "fetch_reservations: total único de headers API (Type=%d Status=%d) → %d", t, st, len(out) ) 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: log_upd.info("upsert_reservations: nenhum registro novo/atualizado encontrado") return 0 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": parse_iso(r.get("checkIn")), "checkOut": parse_iso(r.get("checkOut")), "dateAdd": parse_iso(r.get("dateAdd")), "dateUp": parse_iso(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() log_ins.info("upsert_reservations: inseridos/atualizados %d registros", len(rows)) # <<< LOGGING return len(rows) 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: log_upd.info("upsert_reservation_details: nenhum detalhe novo/atualizado a processar") return 0 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" : parse_iso(det.get("dateAdd")), "dateUp" : parse_iso(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() log_ins.info("upsert_reservation_details: processados %d detalhes", len(details)) # <<< LOGGING return len(details) 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 if not FLAG_FILE.exists(): truncate_tables() FLAG_FILE.touch() last_dt = parse_iso(FULL_SYNC_START) else: old_cache = load_cache(path_res) last_dt = max( (parse_iso(r["dateUp"]) for r in old_cache.values()), default=parse_iso(FULL_SYNC_START) ) # 2) Carrega cache em memória old_res = load_cache(path_res) old_det = load_cache(path_det) while True: changed = False # 3) Define 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) # 4) Busca token token = get_token() if not token: time.sleep(POLL_INTERVAL) continue # 5) Busca cabeçalhos na API 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: batch = fetch_reservations(token, win_start, win_end, t, st) for r in batch: k = (r["idReservation"], r["idEntity"]) dt = parse_iso(r["dateUp"]) if k not in all_api or dt > parse_iso(all_api[k]["dateUp"]): all_api[k] = r res_last_dt[r["idReservation"]] = max(res_last_dt[r["idReservation"]], dt) log_ops.info( "main: total de cabeçalhos únicos da API neste ciclo → %d", len(all_api) ) # 6) Determina quais detalhes buscar 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 (res_id, _), v in old_res.items() if res_id == rid ) } log_ops.info( "main: detalhes a buscar (novos/atualizados) → %d", len(to_fetch) ) # 7) Faz fetch dos detalhes 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 det: new_details.append(det) old_det[(det["idReservation"], det["idEntityCompany"])] = det changed = True else: log_err.error("main: fetch_detail falhou para reservation %s", rid) time.sleep(1.0) log_ops.info( "main: detalhes efetivamente recebidos → %d", len(new_details) ) # 8) Upserts no banco n_res = upsert_reservations(list(all_api.values())) n_det = upsert_reservation_details(new_details) log_ops.info( "main: upsert_reservations → %d, upsert_reservation_details → %d", n_res, n_det ) # 9) Mescla incremental no cache de reservas 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 # 10) Grava JSON de cache se houve mudança if changed: save_cache(path_res, old_res) save_cache(path_det, old_det) # 11) Avança janela para próximo ciclo if res_last_dt: last_dt = max(res_last_dt.values()) 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)