# -*- coding:utf-8 -*- # @Time : 2023/7/5 10:27 # @Author: luozhipeng # @File : host_update.py import os from python_hosts import Hosts, HostsEntry import sys LOCAL_PATH = os.path.dirname(os.path.abspath(__file__)) BASE_PROJECT_PATH = os.path.abspath(os.path.join(LOCAL_PATH, '../../../{}'.format("UBRD"))) BASIC_PATH = os.path.abspath(os.path.join(LOCAL_PATH, '../../../')) sys.path.append(BASE_PROJECT_PATH) sys.path.append(BASIC_PATH) from base_framework.public_tools.utils import Tools from base_framework.public_tools.apollo import Apollo import requests, time import os import sys import socket import subprocess import requests import urllib.parse obj_apollo = Apollo() obj_tools = Tools() class HostUpdate: def __init__(self): if sys.platform.startswith('win'): self.hosts_location = Hosts(path='C:\Windows\System32\drivers\etc\hosts') elif sys.platform.startswith('darwin'): self.hosts_location = Hosts(path='\etc\hosts') def host_update_by_host_dict(self, host_dict_list: list): entry_list = [] for host in host_dict_list: new_entry = HostsEntry(entry_type='ipv4', address=host['ip'], names=[host['host_name'], '']) entry_list.append(new_entry) self.hosts_location.add(entry_list) self.hosts_location.write() def host_remove_intercept(self,address,name): self.hosts_location.remove_all_matching(address,name) self.hosts_location.write() def get_ip_by_host_domain(self,domain: str): ip=socket.gethostbyname(domain) return ip def wait_dns_flush(self,ip,domain): count=1 while True: cmd1="ipconfig /flushdns" result1=subprocess.run(cmd1,capture_output=True,text=True) output1=result1.stdout print(output1) cmd2="ping {}".format(domain) result2=subprocess.run(cmd2,capture_output=True,text=True) output2=result2.stdout print(output2) cmd="ipconfig /displaydns" result=subprocess.run(cmd,capture_output=True,text=True) output=result.stdout print(output) if output is None: output='' if ip in output: break if count >= 30: break count=count + 1 host_list=[{"ip": ip,"host_name": domain}] self.host_remove_intercept(ip,domain) self.host_update_by_host_dict(host_list) time.sleep(2) class CoreApollo: def __init__(self,show_username=None,password=None,current_evn=None): if not show_username: self.show_username="liuruiquan" if not password: self.password="lrq5823LRQ" self.apollo_url="http://apollo.qa.huohua.cn/signin" if not current_evn: self.current_evn="SIM" self.apollo_host="http://apollo.qa.huohua.cn" self.session=self.core_apollo_login() def core_apollo_login(self): post_data=dict() post_data['login-submit']='登录' post_data['username']=self.show_username post_data['password']=self.password req_session=requests.Session() resp=req_session.post(url=self.apollo_url,data=post_data) return req_session def get_key(self,server,cluster,key,project): req_url="%s/apps/%s/envs/%s/clusters/%s/namespaces" % ( self.apollo_host,server,self.current_evn,cluster) resp=self.session.get(url=req_url).json() for item in resp: base_info=item.get('baseInfo') items=item.get('items') if not base_info['namespaceName'].lower() == project.lower(): continue for key_info in items: item_detail=key_info["item"] if item_detail["key"] == key: return key_info["item"] else: return None def set_key(self,key,value,cluster='default',server="",project=None): if server: self.server=server req_url="%s/apps/%s/envs/%s/clusters/%s/namespaces/%s/item" % ( self.apollo_host,self.server,self.current_evn,cluster,project) items=self.get_key(server,cluster,key,project) if items is None: items=dict() items["value"]=value items["key"]=key items["tableViewOperType"]='create' items["addItemBtnDisabled"]=True else: items["value"]=value items["tableViewOperType"]='update' update_resp=self.session.put(url=req_url,json=items) assert update_resp.status_code == 200 release_body=dict() release_time_stamp=time.localtime() release_time='%s-release' % time.strftime("%Y%m%d%H%M%S",release_time_stamp) release_body["isEmergencyPublish"]=False release_body["releaseComment"]="" release_body["releaseTitle"]=release_time release_url="%s/apps/%s/envs/%s/clusters/%s/namespaces/%s/releases" % ( self.apollo_host,self.server,self.current_evn,cluster,project) resp=self.session.post(url=release_url,json=release_body) assert resp.status_code == 200 time.sleep(5) def set_user_force_true(self,vlaue="true"): self.set_key(value=vlaue,key="user-token.forceFallback",server="peppa-core-api",cluster="check" ,project="application") def set_teacher_force_true(self,vlaue="true"): self.set_key(value=vlaue,key="sso-token.forceFallback",server="peppa-core-api",cluster="check" ,project="application") class EduClassroom: def kw_ubrd_get_online_test_classroom(self,post_data_input=None): post_data = dict() host = "sso.huohua.cn" api_url = "https://{}/authentication/form".format(host) show_username = "liuruiquan" username = "liuruiquan" password = "lrq5823LRQ" post_data['showUsername'] = show_username post_data['username'] = username post_data['password'] = password user_agent = {"User-Agent":"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"} deviceid = {"SSO_DEVICE_ID":"8dfbecf2-064a-4e3f-ac2b-fc09a2401416"} req_session = requests.session() req_session.cookies.update(deviceid) req_session.headers.update(user_agent) resp = req_session.post( url=api_url, data=post_data, allow_redirects=False) # token_header = {"accesstoken": token} # req_session.headers.update(token_header) # print(token_header) client_id = "tic-payment-admin" origin_url="aHR0cHM6Ly9zc28uaHVvaHVhLmNuLw==" # origin_url="aHR0cHM6Ly9zc28uc2ltLmh1b2h1YS5jbi8=" redirect_uri = "https://{}/uim/authorize_proxy?client_id={}&debug_mode=1&origin_url={}".format(host,client_id,origin_url) print(redirect_uri) token_code_url = "https://{}/oauth/authorize?client_id={}&response_type=code&".format(host,client_id,redirect_uri) authorize_data = {"redirect_uri": redirect_uri} authorize_data_uri =urllib.parse.urlencode(authorize_data) resp = req_session.get( url=token_code_url+authorize_data_uri, data=authorize_data, allow_redirects=False) print(resp.url) print(resp.headers) print(resp) # token_url = "https://{}/uim/authorize_proxy?client_id={}&debug_mode=1&origin_url={}&code={}".format(host,client_id,origin_url,code) token_url= resp.headers["Location"] # authorize_code_data = {"client_id":"tic-payment-admin","debug_mode": 1,"origin_url": origin_url,"code":code} print(token_url,"token_url") # print(authorize_code_data) resp = req_session.get( url=token_url, # data=authorize_code_data, allow_redirects=False) print(resp) print(resp.url) for key, value in req_session.cookies.items(): if key == 'peppa_sso_token': token = value token_header = {"accesstoken": token} req_session.headers.update(token_header) print(token_header) print(value) break op_time = obj_tools.get_format_date(r_type=4,add_minutes=10) print(op_time) create_classroom_json = {"courseId":1898,"unionFlag":0,"lessonId":49264,"planStudentCount":4,"timezoneName":"Asia/Shanghai","openTime":op_time,"overseasTag":0,"secondOpenTime":op_time,"teacherId":36573,"type":100} api_url_created_classroom = "https://teach-api.huohua.cn/peppa-teach-api/classroom" create_classroom_resp = req_session.post( url=api_url_created_classroom, json=create_classroom_json) # print(req_session.headers) # print(req_session.cookies) # print(resp.content) # print(resp.url) print(create_classroom_resp.json()) classroom_id = create_classroom_resp.json()['data']["classroomId"] print(create_classroom_resp.json()) # # return req_session add_student_json = {"classroomId":classroom_id,"studentId":274886,"userId":275774,"joinType":0,"ignoreClassHour":1} api_url_add_student = "https://teach-api.huohua.cn/peppa-teach-api/classroom_student/add" add_student_resp = req_session.post( url=api_url_add_student, json=add_student_json) print(add_student_resp.json()) # return req_session add_student_json1 = {"classroomId":classroom_id,"studentId":5635578,"userId":5646962,"joinType":0,"ignoreClassHour":1} add_student_resp1 = req_session.post( url=api_url_add_student, json=add_student_json1) print(add_student_resp1.json()) if __name__ == '__main__': host_list = [{"ip": "127.0.0.1", "host_name": "ts.sim.huohua.cn"}, {"ip": "127.0.0.1", "host_name": "its.sim.huohua.cn"}, {"ip": "127.0.0.1", "host_name": "student-api.sim.huohua.cn"}, {"ip": "127.0.0.1", "host_name": "sentry.sim.huohua.cn"}, {"ip": "127.0.0.1", "host_name": "gray.sim.huohua.cn"}, {"ip": "127.0.0.1", "host_name": "classroom-api.sim.huohua.cn"}, {"ip": "127.0.0.1", "host_name": "logserver.sim.huohua.cn"}, {"ip": "127.0.0.1", "host_name": "zipkin.sim.huohua.cn"}, {"ip": "127.0.0.1", "host_name": "gs.sim.huohua.cn"}] A = EduClassroom() A.kw_ubrd_get_online_test_classroom()