import boto3, sys, datetime, socket, re
def _get_boto_session(profile_name: str) -> boto3.Session:
"""
Get the boto3 session. Works with profile configured in runtime.
:param profile_name: The configured AWS profile name. Pass in None to use default credentials chain.
:returns: The Boto session.
"""
if profile_name == None:
session = boto3.Session(region_name="eu-central-1")
return session
else:
session = boto3.Session(profile_name=profile_name)
return session
def _parse_one_image(line: str) -> str:
if re.match("^ *!\[", line):
percent = re.sub("^ *!\[", "", line.split("](")[0])
return (
''
)
else:
return line
def _parse_one_href(line: str) -> str:
if not re.match("^ *!\[", line) and re.match("^.*\[.*\]\(.*\)", line):
anchor = (
''
+ re.sub("^.*\[", "", line.split("](")[0])
+ ""
)
return re.sub("\[.*\)", anchor, line)
else:
return line
def _parse_one_code_pv1(line: str) -> str:
ret_tab = "
" + (" " * 17)
if re.match("^.*`.*`", line):
code = (
' " + line.replace(" ", " ") + " " + line.replace(" ", " ") + " " + line.replace(" ", " ") + " "
if "None" not in pd_dashboard_url:
nodes_table += (
' "
)
monitor_ip = m_public_ip.strip()
grafana_port = 3000
timeout_seconds = 1
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(timeout_seconds)
result = sock.connect_ex((monitor_ip, int(grafana_port)))
if result == 0:
grafana_dashboard_url = "http://" + monitor_ip + ":3000"
nodes_table += (
'Grafana Dashboard - http://'
+ monitor_ip
+ ":3000"
)
sock.close()
nodes_table += "' + line.split("`")[1].strip() + ""
)
if ">Step" in line:
return re.sub("`.*`", ret_tab + code, line)
else:
return re.sub("`.*`", code, line)
else:
return line
def _parse_one_code(line: str, pos: int) -> str:
if re.match("^.*`", line) and not re.match("^ *```", line):
new_pos = pos + 1
if pos % 2 == 0:
return _parse_one_code(line.replace("`", "", 1), new_pos)
else:
return _parse_one_code(
line.replace("`", '', 1), new_pos
)
else:
return line
def _process_command_begin() -> str:
command_begin = '
"
return command_end
def _process_command(line: str) -> str:
return "
"
return sql_end
def _process_sql(line: str) -> str:
# return "
"
return output_end
def _process_output(line: str) -> str:
return "'
+ parsed_line.replace("# ", "", 1).strip()
+ "
"
)
exercise_title = parsed_line.replace("# ", "", 1).strip()
elif re.match("^### ", parsed_line):
parsed_lines += [
"" + parsed_line.replace("### ", "", 1).strip() + "
"
]
elif re.match("^## ", parsed_line):
parsed_lines += [
"" + parsed_line.replace("## ", "", 1).strip() + "
"
]
elif re.match("^\+ ", parsed_line):
tl = (
'
"
)
ptl = _parse_one_code(_parse_one_href(tl), 1)
parsed_lines += [ptl]
elif re.match("^ *```command", parsed_line):
parsed_lines += [_process_command_begin()]
in_command = True
elif re.match("^ *```output", parsed_line):
parsed_lines += [_process_output_begin()]
in_output = True
elif re.match("^ *```sql", parsed_line):
parsed_lines += [_process_sql_begin()]
in_sql = True
elif re.match("^ *``` *$", parsed_line):
if in_command:
parsed_lines += [_process_command_end()]
in_command = False
if in_output:
parsed_lines += [_process_output_end()]
in_output = False
if in_sql:
parsed_lines += [_process_sql_end()]
in_sql = False
else:
ptl = (
'
"
)
parsed_lines += [ptl]
print('
'
)
single_nodes_table = True
single_create_lab_hint = True
for pl in parsed_lines:
if pl.startswith("Overview") and single_nodes_table:
if len(nodes_table) > 50:
print(nodes_table)
single_nodes_table = False
elif single_create_lab_hint:
print(
"
Click and allow approximately 6 minutes for the countdown timer to finish. Upon completion, you will unlock the detailed exercise tabs above and receive the information of the lab environment.
"
)
single_create_lab_hint = False
elif pl.startswith("Exercise "):
print(
''
)
if len(nodes_table) > 50:
print(
''
)
else:
print(
""
)
if len(nodes_table) > 50:
print(
'
Exercise ", "").replace("
", "").strip()
+ '">
'
)
else:
print(" 50:
print(pl)
else:
None
else:
if "None" not in pd_dashboard_url:
pl = pl.replace(
"TiDB Dashboard",
'TiDB Dashboard',
)
if grafana_dashboard_url != None:
pl = pl.replace(
"Grafana Dashboard",
'Grafana Dashboard',
)
if len(nodes_table) > 50:
print(
pl.replace("monitor1_address", m_public_ip)
.replace("trainer_name", trainer_name)
.replace("us-west-2", "eu-central-1")
.replace("oregon", "frankfurt")
)
else:
print(
pl.replace("us-west-2", "eu-central-1").replace(
"oregon", "frankfurt"
)
)
if len(nodes_table) > 50:
print(
''
)
print(
""
) # For terminate resources
print("
"
}
)
else:
cost = str(cost) + " (covered by RI)"
if step_2_needed != None:
print(""
+ instances["InstanceId"].rjust(19, " ")
+ " "
+ instances["PublicIpAddress"].ljust(16, " ")
+ " "
+ instances["PrivateIpAddress"].ljust(16, " ")
+ " "
+ (name.ljust(13, " ") if name != None else " ")
+ " "
+ (trainer.ljust(10, " ") if trainer != None else " ")
+ " "
+ (student.ljust(7, " ") if student != None else " ")
+ " "
+ (node_role.ljust(9, " ") if node_role != None else " ")
+ " "
+ str(up_time).ljust(14, " ")
+ " " + step_2_needed + "
")
trainer_name_cache = "NONE"
student_name_cache = "NONE"
for key in sorted([*instances_dict]):
trainer_name = key.split(":")[0]
student_name = key.split(":")[1]
if trainer_name != trainer_name_cache or student_name != student_name_cache:
nodes_table += ""
# nodes_table += ("
"
m_public_ip = "monitor1_address"
for key in sorted([*instances_dict]):
trainer_name = key.split(":")[0]
student_name = key.split(":")[1]
trainer_name_cache = trainer_name
student_name_cache = student_name
row = instances_dict[key]
if "monitor1" in row:
m_public_ip = row.split(""
nodes_table += " "
nodes_table += ""
trainer_name_cache = trainer_name
student_name_cache = student_name
nodes_table += instances_dict[key]
nodes_table += ""
nodes_table += "Instance ID Public IP Private IP Suggested Role Instructor Student Name Up Time (mins) "
nodes_table += "")[1].strip()
if len([*instances_dict]) > 0:
None
if m_public_ip != "monitor1_address":
nodes_table += "
TiDB Dashboard - '
+ pd_dashboard_url
+ "
"
else:
None
# print("")
_parse_and_print_md(
normal_font_size,
language,
lab_guide_markdown_file,
m_public_ip,
trainer_name,
pd_dashboard_url,
grafana_dashboard_url,
nodes_table,
)
if __name__ == "__main__":
trainer_name = sys.argv[1]
pd_dashboard_url = sys.argv[2]
lab_name = sys.argv[3]
session = _get_boto_session(None)
_check(
session,
trainer_name,
lab_name + "-lab-guide.md",
"english",
"16",
pd_dashboard_url,
)