Files
2025-09-29 00:52:08 +02:00

185 lines
7.6 KiB
Python
Executable File

"""
Commonly accessed databases
"""
import base64
import pymongo
from RS import Config
from RS.Core import DatabaseConnection
# Uncomment this once we get newer version of Python that supports ssl.create_default_context
# from RS.Core.Database import ElasticSearch
DEFAULT_USERNAME = "tools"
DEFAULT_PASSWORD = base64.b64decode('cm9ja3N0YXIxQQ==')
class MONGO_NYC(object):
_server = "rsgadcmgo1"
_port = 27017
@classmethod
def Server(cls):
return cls._server
@classmethod
def Port(cls):
return cls._port
@classmethod
def TestServer(cls):
client = pymongo.MongoClient(host=cls._server, port=cls._port)
db = client["rdr-release-dev-1-0-0-0-Test"]
return db['Test']
@classmethod
def ClientsTracking(cls):
client = pymongo.MongoClient(host=cls._server, port=cls._port)
db = client["rdr-release-dev-1-0-0-0-ToolsTracking"]
return db['MobuClientInformation']
@classmethod
def ToolsTracking(cls):
client = pymongo.MongoClient(host=cls._server, port=cls._port)
db = client["rdr-release-dev-1-0-0-0-ToolsTracking"]
return db['MobuToolsTracking']
class NYC(object):
_server = r"animtracking-sql"
@classmethod
def Server(cls):
return cls._server
@classmethod
def AnimationStats(cls, cache=True):
return DatabaseConnection.DatabaseConnection(
driver='{SQL Server}',
server=cls._server,
database="AnimationStats_{}".format(Config.Project.Name),
username=DEFAULT_USERNAME,
password=DEFAULT_PASSWORD,
cache=cache
)
@classmethod
def CutsceneStats(cls, cache=True):
return DatabaseConnection.DatabaseConnection(
driver="{SQL Server}",
server=cls._server,
database="CutsceneStats_{}".format(Config.Project.Name),
username=DEFAULT_USERNAME,
password=DEFAULT_PASSWORD,
cache=cache
)
@classmethod
def LevelStats(cls, cache=True):
return DatabaseConnection.DatabaseConnection(
driver="{SQL Server}",
server=cls._server,
database="LevelStats_{}".format(Config.Project.Name),
cache=cache)
@classmethod
def GradingStats(cls, cache=True):
return DatabaseConnection.DatabaseConnection(
driver="{SQL Server}",
server=cls._server,
database="gradingJobs",
cache=cache)
@classmethod
def CutsceneLogging(cls, cache=True):
return DatabaseConnection.DatabaseConnection(
driver="{SQL Server}",
server=cls._server,
database="CutsceneDatabase_Logging",
cache=cache
)
@classmethod
def AutomationJobs(cls, cache=True):
return DatabaseConnection.DatabaseConnection(
driver="{SQL Server}",
server=cls._server,
database="AutomationJobs",
cache=cache
)
@classmethod
def CaptureRenders(cls, cache=True):
return DatabaseConnection.DatabaseConnection(
driver="{SQL Server}",
server=cls._server,
database="CaptureRenders",
username=DEFAULT_USERNAME,
password=DEFAULT_PASSWORD,
cache=cache
)
@classmethod
def FaceAudioMocap(cls, cache=True):
return DatabaseConnection.DatabaseConnection(
driver="{SQL Server}",
server=cls._server,
database="FaceAudioMocap",
cache=cache
)
@classmethod
def RockstarTools(cls, cache=True):
return DatabaseConnection.DatabaseConnection(
driver="{SQL Server}",
server=cls._server,
database="RockstarTools",
cache=cache
)
class NYCDEV(object):
_server = r"RSGNYCSQL2\RSGNYCSQL2"
@classmethod
def Server(cls):
return cls._server
@classmethod
def FaceAudioMocap(cls, cache=True):
return DatabaseConnection.DatabaseConnection(
driver="{SQL Server}",
server=cls._server,
database="FaceAudioMocap",
cache=cache
)
class SD(object):
_server = r"RSGSANSQL1\PRODUCTION"
@classmethod
def Server(cls):
return cls._server
@classmethod
def RockstarTools(cls, cache=True):
return DatabaseConnection.DatabaseConnection(
driver="{SQL Server}",
server=cls._server,
database="RockstarTools",
username=DEFAULT_USERNAME,
password=DEFAULT_PASSWORD,
cache=cache
)
# Uncomment this once we get newer version of Python that supports ssl.create_default_context
# class Elastic(object):
# @classmethod
# def ToolsTracker(cls, dev=False):
# INDEX = "telemetry_motionbuilder_tools_tracking"
# DOC_TYPE = "motionbuilder_ta_tool"
# return ElasticSearch.ElasticSearch(INDEX, DOC_TYPE, server=ElasticSearch.Server.DEV_SERVER if dev else ElasticSearch.Server.PROD_SERVER)