sporttery/pinnacle/libs/syncData.js

import path from "node:path";
import { fileURLToPath } from "node:url";

import { pinnacleGet, getPsteryRelations, updateBaseEvents, notifyException } from "./pinnacleClient.js";
import { parseGame, parseIorDetail } from "./parseGameData.js";
import Logs from "./logs.js";
import { getData, setData } from "./cache.js";
import { GLOBAL_DATA } from "./globalData.js";

const __filename = fileURLToPath(import.meta.url);
const __dirname = path.dirname(__filename);

const gamesMapCacheFile = path.join(__dirname, '../data/gamesCache.json');
const globalDataCacheFile = path.join(__dirname, '../data/globalDataCache.json');
const TP = 'ps_9_9_1';
const IS_DEV = process.env.NODE_ENV == 'development';

/**
 * 重置版本计数
 */
const resetVersionsCount = () => {
  GLOBAL_DATA.straightFixturesVersion = 0;
  GLOBAL_DATA.specialFixturesVersion = 0;
  GLOBAL_DATA.straightOddsVersion = 0;
  GLOBAL_DATA.specialsOddsVersion = 0;

  GLOBAL_DATA.straightFixturesCount = 0;
  GLOBAL_DATA.specialFixturesCount = 0;
}

/**
 * 增加版本计数
 */
const incrementVersionsCount = () => {
  GLOBAL_DATA.straightFixturesCount += 1;
  GLOBAL_DATA.specialFixturesCount += 1;
}


/**
 * 获取指定时区当前日期或时间
 * @param {number} offsetHours - 时区相对 UTC 的偏移（例如：-4 表示 GMT-4）
 * @param {boolean} [withTime=false] - 是否返回完整时间（默认只返回日期）
 * @returns {string} 格式化的日期或时间字符串
 */
const getDateInTimezone = (offsetHours, withTime = false) => {
  const nowUTC = new Date();
  const targetTime = new Date(nowUTC.getTime() + offsetHours * 60 * 60 * 1000);

  const year = targetTime.getUTCFullYear();
  const month = String(targetTime.getUTCMonth() + 1).padStart(2, '0');
  const day = String(targetTime.getUTCDate()).padStart(2, '0');

  if (!withTime) {
    return `${year}-${month}-${day}`;
  }

  const hours = String(targetTime.getUTCHours()).padStart(2, '0');
  const minutes = String(targetTime.getUTCMinutes()).padStart(2, '0');
  const seconds = String(targetTime.getUTCSeconds()).padStart(2, '0');

  return `${year}-${month}-${day} ${hours}:${minutes}:${seconds}`;
}

/**
 * 判断两个数组是否相等
 * @param {*} arr1
 * @param {*} arr2
 * @returns
 */
const isArrayEqualUnordered = (arr1, arr2) => {
  if (arr1.length !== arr2.length) return false;
  const s1 = [...arr1].sort();
  const s2 = [...arr2].sort();
  return s1.every((v, i) => v === s2[i]);
}

/**
 * 获取已过滤比赛列表
 * @returns {Promise<void>}
 */
const getFiltedGames = () => {
  return new Promise(resolve => {
    const updateFiltedGames = () => {
      getPsteryRelations()
      .then(res => {
        if (res.statusCode !== 200) {
          throw new Error(res.message);
        }
        // Logs.outDev('update filted games', res.data);
        const games = res.data.map(game => {
          const { eventId, leagueId } = game?.rel?.ps ?? {};
          return {
            eventId,
            leagueId,
          };
        });

        const newFiltedLeagues = [...new Set(games.map(game => game.leagueId).filter(leagueId => leagueId))];

        if (!isArrayEqualUnordered(newFiltedLeagues, GLOBAL_DATA.filtedLeagues)) {
          Logs.outDev('filted leagues changed', newFiltedLeagues);
          resetVersionsCount();
          GLOBAL_DATA.filtedLeagues = newFiltedLeagues;
        }

        GLOBAL_DATA.filtedGames = games.map(game => game.eventId).filter(eventId => eventId);

        resolve();
        setTimeout(updateFiltedGames, 1000 * 60);
      })
      .catch(err => {
        Logs.err('failed to update filted games:', err.message);
        setTimeout(updateFiltedGames, 1000 * 5);
      });
    }
    updateFiltedGames();
  });
}

/**
 * 获取胜平负盘口数据
 * @returns {Promise<Object>}
 */
const getStraightFixtures = async () => {
  if (!GLOBAL_DATA.filtedLeagues.length) {
    return Promise.reject(new Error('no filted leagues'));
  }
  const leagueIds = GLOBAL_DATA.filtedLeagues.join(',');
  let since = GLOBAL_DATA.straightFixturesVersion;
  if (GLOBAL_DATA.straightFixturesCount >= 12) {
    since = 0;
    GLOBAL_DATA.straightFixturesCount = 0;
  }
  if (since == 0) {
    Logs.outDev('full update straight fixtures');
  }
  return pinnacleGet('/v3/fixtures', { sportId: 29, leagueIds, since })
  .then(data => {
    const { league, last } = data;
    if (!last) {
      return {};
    }
    GLOBAL_DATA.straightFixturesVersion = last;
    const games = league?.map(league => {
      const { id: leagueId, events } = league;
      return events?.map(event => {
        const { starts } = event;
        const timestamp = new Date(starts).getTime();
        return { leagueId, ...event, timestamp };
      });
    })
    .flat() ?? [];
    const update = since == 0 ? 'full' : 'increment';
    return { games, update };
  });
}

/**
 * 更新胜平负盘口数据
 * @returns {Promise<void>}
 */
const updateStraightFixtures = async () => {
  return getStraightFixtures()
  .then(data => {
    const { games, update } = data;
    const { gamesMap } = GLOBAL_DATA;
    if (games?.length) {
      games.forEach(game => {
        const { id } = game;
        if (!gamesMap[id]) {
          gamesMap[id] = game;
        }
        else {
          Object.assign(gamesMap[id], game);
        }
      });
    }
    if (update && update == 'full') {
      const gamesSet = new Set(games.map(game => game.id));
      Object.keys(gamesMap)?.forEach(key => {
        if (!gamesSet.has(+key)) {
          delete gamesMap[key];
        }
      });
    }
  });
}

/**
 * 获取胜平负赔率数据
 * @returns {Promise<Array>}
 */
const getStraightOdds = async () => {
  if (!GLOBAL_DATA.filtedLeagues.length) {
    return Promise.reject(new Error('no filted leagues'));
  }
  const leagueIds = GLOBAL_DATA.filtedLeagues.join(',');
  const since = GLOBAL_DATA.straightOddsVersion;
  // if (GLOBAL_DATA.straightOddsCount >= 27) {
  //   since = 0;
  //   GLOBAL_DATA.straightOddsCount = 3;
  // }
  if (since == 0) {
    Logs.outDev('full update straight odds');
  }
  return pinnacleGet('/v4/odds', { sportId: 29, oddsFormat: 'Decimal', leagueIds, since })
  .then(data => {
    const { leagues, last } = data;
    if (!last) {
      return [];
    }
    GLOBAL_DATA.straightOddsVersion = last;
    const games = leagues?.flatMap(league => league.events);
    // return games;
    return games?.map(item => {
      const { periods, ...rest } = item;
      const straight = periods?.find(period => period.number == 0);
      const straight1st = periods?.find(period => period.number == 1);
      const gameInfo = rest;
      if (straight || straight1st) {
        gameInfo.periods = {};
      }
      if (straight) {
        gameInfo.periods.straight = straight;
      }
      if (straight1st) {
        gameInfo.periods.straight1st = straight1st;
      }
      return gameInfo;
    }) ?? [];
  });
}

/**
 * 更新胜平负赔率数据
 * @returns {Promise<void>}
 */
const updateStraightOdds = async () => {
  return getStraightOdds()
  .then(games => {
    if (games?.length) {
      const { gamesMap={} } = GLOBAL_DATA;
      games.forEach(game => {
        const { id, periods, ...rest } = game;
        const localGame = gamesMap[id];
        if (localGame) {
          Object.assign(localGame, rest);
          if (!localGame.periods && periods) {
            localGame.periods = periods;
          }
          else if (periods) {
            Object.assign(localGame.periods, periods);
          }
        }
      });
    }
  });
}

/**
 * 获取特殊盘口列表数据
 * @returns {Promise<Object>}
 */
const getSpecialFixtures = async () => {
  if (!GLOBAL_DATA.filtedLeagues.length) {
    return Promise.reject(new Error('no filted leagues'));
  }
  const leagueIds = GLOBAL_DATA.filtedLeagues.join(',');
  let since = GLOBAL_DATA.specialFixturesVersion;
  if (GLOBAL_DATA.specialFixturesCount >= 18) {
    since = 0;
    GLOBAL_DATA.specialFixturesCount = 6;
  }
  if (since == 0) {
    Logs.outDev('full update special fixtures');
  }
  return pinnacleGet('/v2/fixtures/special', { sportId: 29, leagueIds, since })
  .then(data => {
    const { leagues, last } = data;
    if (!last) {
      return { specials: [], update: 'increment' };
    }
    GLOBAL_DATA.specialFixturesVersion = last;
    const specials = leagues?.map(league => {
      const { specials } = league;
      return specials?.filter(special => special.event)
      .map(special => {
        const { event: { id: eventId, periodNumber }, ...rest } = special ?? { event: {} };
        return { eventId, periodNumber, ...rest };
      }) ?? [];
    })
    .flat()
    .filter(special => {
      if (special.name.includes('Winning Margin') || special.name.includes('Exact Total Goals')) {
        return true;
      }
      return false;
    }) ?? [];
    const update = since == 0 ? 'full' : 'increment';
    return { specials, update };
  });
}

/**
 * 合并盘口数据
 * @param {*} localContestants
 * @param {*} remoteContestants
 * @returns
 */
const mergeContestants = (localContestants=[], remoteContestants=[]) => {
  const localContestantsMap = new Map(localContestants.map(contestant => [contestant.id, contestant]));
  remoteContestants.forEach(contestant => {
    const localContestant = localContestantsMap.get(contestant.id);
    if (localContestant) {
      Object.assign(localContestant, contestant);
    }
    else {
      localContestants.push(contestant);
    }
  });
  const remoteContestantsMap = new Map(remoteContestants.map(contestant => [contestant.id, contestant]));
  for (let i = localContestants.length - 1; i >= 0; i--) {
    if (!remoteContestantsMap.has(localContestants[i].id)) {
      localContestants.splice(i, 1);
    }
  }
  return localContestants;
}

/**
 * 合并特殊盘口数据
 * @param {*} localSpecials
 * @param {*} remoteSpecials
 * @param {*} specialName
 * @returns
 */
const mergeSpecials = (localSpecials, remoteSpecials, specialName) => {
  if (localSpecials[specialName] && remoteSpecials[specialName]) {
    const { contestants: specialContestants, ...specialRest } = remoteSpecials[specialName];
    Object.assign(localSpecials[specialName], specialRest);
    mergeContestants(localSpecials[specialName].contestants, specialContestants);
  }
  else if (remoteSpecials[specialName]) {
    localSpecials[specialName] = remoteSpecials[specialName];
  }
}

/**
 * 更新特殊盘口列表数据
 * @returns {Promise<void>}
 */
const updateSpecialFixtures = async () => {
  return getSpecialFixtures()
  .then(data => {
    const { specials, update } = data;
    if (specials?.length) {
      const { gamesMap={} } = GLOBAL_DATA;
      const gamesSpecialsMap = {};
      specials.forEach(special => {
        const { eventId } = special;
        if (!gamesSpecialsMap[eventId]) {
          gamesSpecialsMap[eventId] = {};
        }
        if (special.name == 'Winning Margin') {
          gamesSpecialsMap[eventId].winningMargin = special;
        }
        else if (special.name == 'Exact Total Goals') {
          gamesSpecialsMap[eventId].exactTotalGoals = special;
        }
        else if (special.name == 'Winning Margin 1st Half') {
          gamesSpecialsMap[eventId].winningMargin1st = special;
        }
        else if (special.name == 'Exact Total Goals 1st Half') {
          gamesSpecialsMap[eventId].exactTotalGoals1st = special;
        }
      });

      Object.keys(gamesSpecialsMap).forEach(eventId => {
        if (!gamesMap[eventId]) {
          return;
        }

        if (!gamesMap[eventId].specials) {
          gamesMap[eventId].specials = {};
        }

        const localSpecials = gamesMap[eventId].specials;
        const remoteSpecials = gamesSpecialsMap[eventId];

        mergeSpecials(localSpecials, remoteSpecials, 'winningMargin');
        mergeSpecials(localSpecials, remoteSpecials, 'exactTotalGoals');
        mergeSpecials(localSpecials, remoteSpecials, 'winningMargin1st');
        mergeSpecials(localSpecials, remoteSpecials, 'exactTotalGoals1st');

      });
    }
  });
}

/**
 * 获取特殊盘口赔率数据
 * @returns {Promise<Array>}
 */
const getSpecialsOdds = async () => {
  if (!GLOBAL_DATA.filtedLeagues.length) {
    return Promise.reject(new Error('no filted leagues'));
  }
  const leagueIds = GLOBAL_DATA.filtedLeagues.join(',');
  const since = GLOBAL_DATA.specialsOddsVersion;
  // if (GLOBAL_DATA.specialsOddsCount >= 33) {
  //   since = 0;
  //   GLOBAL_DATA.specialsOddsCount = 9;
  // }
  if (since == 0) {
    Logs.outDev('full update specials odds');
  }
  return pinnacleGet('/v2/odds/special', { sportId: 29, oddsFormat: 'Decimal', leagueIds, since })
  .then(data => {
    const { leagues, last } = data;
    if (!last) {
      return [];
    }
    GLOBAL_DATA.specialsOddsVersion = last;
    return leagues?.flatMap(league => league.specials);
  });
}

/**
 * 更新特殊盘口赔率数据
 * @returns
 */
const updateSpecialsOdds = async () => {
  return getSpecialsOdds()
  .then(specials => {
    if (specials.length) {
      const { gamesMap={} } = GLOBAL_DATA;
      const contestants = Object.values(gamesMap)
      .filter(game => game.specials)
      .map(game => {
        const { specials } = game;
        const contestants = Object.values(specials).map(special => {
          const { contestants } = special;
          return contestants.map(contestant => [contestant.id, contestant]);
        });
        return contestants;
      }).flat(2);
      const contestantsMap = new Map(contestants);
      const lines = specials?.flatMap(special => special.contestantLines) ?? [];
      lines.forEach(line => {
        const { id, handicap, lineId, max, price } = line;
        const contestant = contestantsMap.get(id);
        if (!contestant) {
          return;
        }
        contestant.handicap = handicap;
        contestant.lineId = lineId;
        contestant.max = max;
        contestant.price = price;
      });
    }
  });
}

/**
 * 获取滚球数据
 * @returns {Promise<Array>}
 */
const getInRunning = async () => {
  return pinnacleGet('/v2/inrunning')
  .then(data => {
    const sportId = 29;
    const leagues = data.sports?.find(sport => sport.id == sportId)?.leagues ?? [];
    return leagues.filter(league => {
      const { id } = league;
      const filtedLeaguesSet = new Set(GLOBAL_DATA.filtedLeagues);
      return filtedLeaguesSet.has(id);
    }).flatMap(league => league.events);
  });
}

/**
 * 更新滚球数据
 * @returns {Promise<void>}
 */
const updateInRunning = async () => {
  return getInRunning()
  .then(games => {
    if (!games.length) {
      return;
    }
    const { gamesMap={} } = GLOBAL_DATA;
    games.forEach(game => {
      const { id, state, elapsed } = game;
      const localGame = gamesMap[id];
      if (localGame) {
        Object.assign(localGame, { state, elapsed });
      }
    });
  });
}

/**
 * 获取币种数据
 * @returns {Promise<Array>}
 */
const getCurrencies = async () => {
  return pinnacleGet('/v2/currencies');
}

/**
 * 更新币种数据
 * @returns {Promise<void>}
 */
const updateCurrencies = async () => {
  return getCurrencies()
  .then(data => {
    GLOBAL_DATA.currencies = data;
    GLOBAL_DATA.currenciesUpdatedAt = Date.now();
    Logs.outDev('currencies updated');
  });
}

const getCurrenciesLoopDelay = () => {
  const minDelay = 10 * 60 * 1000;
  const maxDelay = 15 * 60 * 1000;
  return minDelay + Math.floor(Math.random() * (maxDelay - minDelay + 1));
}

/**
 * 同步币种数据循环
 */
const currenciesLoop = () => {
  updateCurrencies()
  .catch(err => {
    Logs.err('failed to update currencies:', err.message, err.source);
  })
  .finally(() => {
    const delay = GLOBAL_DATA.currenciesUpdatedAt ? getCurrenciesLoopDelay() : 60 * 1000;
    setTimeout(currenciesLoop, delay);
  });
}

/**
 * 获取盘口数据
 * @returns {Object}
 */
const getGames = () => {
  const { filtedGames, gamesMap={} } = GLOBAL_DATA;
  const filtedGamesSet = new Set(filtedGames);
  const nowTime = Date.now();
  const gamesData = {};
  Object.values(gamesMap).forEach(game => {
    const { id, liveStatus, parentId, resultingUnit, timestamp } = game;

    if (resultingUnit !== 'Regular') {
      return false;  // 非常规赛事不处理
    }

    const gmtMinus4Date = getDateInTimezone(-4);
    const todayEndTime = new Date(`${gmtMinus4Date} 23:59:59 GMT-4`).getTime();
    const tomorrowEndTime = todayEndTime + 24 * 60 * 60 * 1000;
    if (liveStatus == 1 && timestamp < nowTime) {
      game.marketType = 2;  // 滚球赛事
    }
    else if (liveStatus != 1 && timestamp > nowTime && timestamp <= todayEndTime) {
      game.marketType = 1;  // 今日赛事
    }
    else if (liveStatus != 1 && timestamp > todayEndTime && timestamp <= tomorrowEndTime) {
      game.marketType = 0;  // 明日早盘赛事
    }
    else {
      game.marketType = -1;  // 非近期赛事
    }

    if (game.marketType < 0) {
      return false;  // 非近期赛事不处理
    }

    let actived = false;
    if (liveStatus != 1 && filtedGamesSet.has(id)) {
      actived = true;  // 在赛前列表中
      game.eventId = id;
      game.originId = 0;
    }
    else if (liveStatus == 1 && filtedGamesSet.has(parentId)) {
      actived = true;  // 在滚球列表中
      game.eventId = parentId;
      game.originId = id;
    }

    if (actived) {
      const { filtedGames } = GLOBAL_DATA;
      parseGame(game, filtedGames)?.forEach(gameInfo => {
        const { marketType, ...rest } = gameInfo;
        if (!gamesData[marketType]) {
          gamesData[marketType] = [];
        }
        gamesData[marketType].push(rest);
      });
    }
  });
  return gamesData;
}

/**
 * 同步盘口数据循环
 */
const pinnacleDataLoop = () => {
  updateStraightFixtures()
  .then(() => {
    return Promise.all([
      updateStraightOdds(),
      updateSpecialFixtures(),
      updateInRunning(),
    ]);
  })
  .then(() => {
    return updateSpecialsOdds();
  })
  .then(() => {
    if (!GLOBAL_DATA.loopActive) {
      GLOBAL_DATA.loopActive = true;
      Logs.out('loop active');
      notifyException('Pinnacle API startup.');
    }

    if (GLOBAL_DATA.requestErrorCount > 0) {
      GLOBAL_DATA.requestErrorCount = 0;
      Logs.out('request error count reset');
    }

    const nowTime = Date.now();
    const loopDuration = nowTime - GLOBAL_DATA.loopResultTime;
    GLOBAL_DATA.loopResultTime = nowTime;
    if (loopDuration > 15000) {
      Logs.out('loop duration is too long', loopDuration);
    }
    else {
      Logs.outDev('loop duration', loopDuration);
    }

    const timestamp = Date.now();
    const games = getGames();
    const data = { games, timestamp, tp: TP };

    updateBaseEvents(data);

    if (IS_DEV) {
      setData(gamesMapCacheFile, GLOBAL_DATA.gamesMap);
    }

  })
  .catch(err => {
    Logs.err(err.message, err.source);
    GLOBAL_DATA.requestErrorCount++;
    if (GLOBAL_DATA.loopActive && GLOBAL_DATA.requestErrorCount > 5) {
      const exceptionMessage = 'request errors have reached the limit';
      Logs.out(exceptionMessage);
      GLOBAL_DATA.loopActive = false;

      Logs.out('loop inactive');
      const exceptionList = ['Pinnacle API paused']
      if (exceptionMessage) {
        exceptionList.push(exceptionMessage);
      }
      if (err.source?.data?.message) {
        exceptionList.push(err.source.data.message);
      }
      else if (err.message) {
        exceptionList.push(err.message);
      }
      if (err.source?.username) {
        exceptionList.push(err.source.username);
      }
      notifyException(exceptionList.join('. '));
    }
  })
  .finally(() => {
    const { loopActive } = GLOBAL_DATA;
    let loopDelay = 1000 * 5;
    if (!loopActive) {
      loopDelay = 1000 * 60;
      resetVersionsCount();
    }
    else {
      incrementVersionsCount();
    }
    setTimeout(pinnacleDataLoop, loopDelay);
  });
}

/**
 * 缓存GLOBAL_DATA数据到文件
 * @returns {Promise<void>}
 */
const saveGlobalDataToCache = async () => {
  return setData(globalDataCacheFile, GLOBAL_DATA);
}

/**
 * 从文件加载GLOBAL_DATA数据
 * @returns {Promise<void>}
 */
const loadGlobalDataFromCache = async () => {
  return getData(globalDataCacheFile)
  .then(data => {
    if (!data) {
      return;
    }
    Object.keys(GLOBAL_DATA)?.forEach(key => {
      if (key in data) {
        GLOBAL_DATA[key] = data[key];
      }
    });
  });
}

/**
 * 监听进程退出事件，保存GLOBAL_DATA数据
 * @param {*} code
 */
const saveExit = (code) => {
  saveGlobalDataToCache()
  .then(() => {
    Logs.out('global data saved');
  })
  .catch(err => {
    Logs.err('failed to save global data', err.message);
  })
  .finally(() => {
    process.exit(code);
  });
}
process.on('SIGINT', () => {
  saveExit(0);
});
process.on('SIGTERM', () => {
  saveExit(0);
});
process.on('SIGUSR2', () => {
  saveExit(0);
});

/**
 * 启动同步盘口数据
 */
export const startSyncMarketsData = () => {
  loadGlobalDataFromCache()
  .then(() => {
    Logs.out('global data loaded');
  })
  .catch(err => {
    Logs.err('failed to load global data', err.message);
  })
  .finally(() => {
    GLOBAL_DATA.loopResultTime = Date.now();
    GLOBAL_DATA.loopActive = true;
    return getFiltedGames();
  })
  .then(() => {
    currenciesLoop();
    pinnacleDataLoop();
  });
}