全国旅游监管平台爬数据比对导游信息
本帖最后由 wangwei0511 于 2023-8-17 15:08 编辑鉴于大家不清楚作用，我说下场景吧。某个景区带团导游免费（购买0元门票），那么就要把这些导游的身份证信息录入到白名单，让导游能够买到票。
怎么证明是不是导游呢？就需要用报备的导游信息去https://mr.mct.gov.cn/（全国旅游监管网站）去进行查询比对，如果真实导游，就录入系统，然后能够买到导游票。
这个程序就是从数据库获取信息至全国旅游监管网站去比对的过程，并把比对的相关信息更新至数据库。
数据库的信息是导游自己信息报备的。
——————————————————————————————————————————————目的：实现从数据库到获取导游信息，和国家旅游局的导游查询信息进行比对，然后将比对结果写入数据库。
场景：某景区带团报备的导游免票，就需要核实报备的导游是否是真实的导游、故通过此代码进行比对。
需要注意的点：旅游局接口有防爬，所以需要有间隔时常，及多次重试机制，目前这个稳定运行半年没问题。
import mysql.connector
import requests
import json
import random
import time
import datetime
def query():
USER_AGENT_LIST = [
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/75.0.3770.90 Safari/537.36',
'Mozilla/5.0 (iPod; U; CPU iPhone OS 4_3_3 like Mac OS X; en-us) AppleWebKit/533.17.9 (KHTML, like Gecko) Version/5.0.2 Mobile/8J2 Safari/6533.18.5'
]
headers = {
'User-Agent': random.choice(USER_AGENT_LIST),
'Content-Type': 'application/json; charset=utf-8'
}
#数据库信息
mydb = mysql.connector.connect(
host="",
user="",
password="",
database=""
)
mycursor = mydb.cursor()
mycursor.execute("SELECT id, guide_cert_no FROM guide_info WHERE query_results IS NULL AND guide_cert_no IS NOT NULL")
current_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) #获取时间
print("导游正常查询：", current_time)#看是否执行查询
for row in mycursor.fetchall():
current_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
print("导游正常查询及执行：", current_time)#看是否执行查询
id = row
guide_cert_no = row
print(f"第 {id}行, 导游证号: {guide_cert_no}")
data = {
'type': '3',
'value': guide_cert_no
}
print(data)
response = requests.post('https://mr.mct.gov.cn/data/guide/verify', headers=headers, data=json.dumps(data))
time.sleep(0.8)
result = json.loads(response.text)
print(result)
# 每遍历19次暂停42秒
if (id + 1) % 19 == 0:
print("暂停42秒...")
time.sleep(42)
if 'error' in result:
max_retry = 2# 最大重试次数为2次
retry_count = 0# 当前重试次数初始化为0
# 循环执行请求直到结果不出错或超过最大重试次数
while retry_count < max_retry:
response = requests.post('https://mr.mct.gov.cn/data/guide/verify', headers=headers, data=json.dumps(data))
time.sleep(0.8)# 每次请求之间间隔0.8秒
result = json.loads(response.text)
print(result)
if 'error' in result:# 如果结果中包含错误信息
retry_count += 1# 当前重试次数加1
time.sleep(20)# 等待10秒后再次执行请求
else:# 如果结果中不包含错误信息
break# 跳出循环
else:# 如果循环执行了max_retry次仍然没有成功，则输出错误信息
result = json.loads(response.text)
reason = result['reason']
print(reason)
#获取时间并显示，写入的文件。
now = datetime.datetime.now()# 获取当前时间和日期
formatted_now = now.strftime("%Y-%m-%d %H:%M:%S")# 将时间格式化为"YYYY-MM-DD HH:MM:SS"格式
update_query = f"UPDATE guide_info SET query_results='{reason}', query_time='{formatted_now}' WHERE id={id}"
mycursor.execute(update_query)
mydb.commit()
else:
data = result['data']
if 'pic_avatar' in data:
pic_avatar = data['pic_avatar']
user_name = data['user_name']
star_level = data['star_level']
guide_card = data['guide_card']
guide_id = data['guide_id']
level = data['level']
leader_card = data['leader_card']
leader_type_name = data['leader_type_name']
languages = data['languages']
comp_name = data['comp_name']
qr = data['qr']
issued_date = data['issued_date']
expired_date = data['expired_date']
pinyin_name = data['pinyin_name']
qr_img_url = data['qr_img_url']
##查询数据库中导游的姓名
select_query = f"SELECT guide_name FROM guide_info WHERE id = {id}"
mycursor.execute(select_query)
result = mycursor.fetchone()
guide_name = result
now = datetime.datetime.now()# 获取当前时间和日期
formatted_now = now.strftime("%Y-%m-%d %H:%M:%S")# 将时间格式化为"YYYY-MM-DD HH:MM:SS"格式
#写入到数据库
update_query = f"UPDATE guide_info SET query_results='查询成功', pic_avatar='{pic_avatar}', user_name='{user_name}', star_level='{star_level}', guide_card='{guide_card}', guide_id='{guide_id}', level='{level}', leader_card='{leader_card}', leader_type_name='{leader_type_name}', languages='{languages}',comp_name='{comp_name}', qr='{qr}', issued_date='{issued_date}', expired_date='{expired_date}', pinyin_name='{pinyin_name}', qr_img_url='{qr_img_url}', query_time='{formatted_now}' WHERE id={id}"
mycursor.execute(update_query)
mydb.commit()
if user_name and user_name == guide_name:
mycursor.execute(f"UPDATE guide_info SET check_name='姓名正确' WHERE id={id}")
#可以使用以下代码来查询当前行的 guide_id_card 值：
select_query = f"SELECT guide_id_card FROM guide_info WHERE id = {id}"
mycursor.execute(select_query)
result = mycursor.fetchone()
guide_id_card = result
# 查询数据库中duplicate_queries列的所有值，排除空值
mycursor.execute("SELECT duplicate_queries FROM guide_info WHERE duplicate_queries IS NOT NULL AND duplicate_queries != ''")
myresult = mycursor.fetchall()
result_str = guide_name + guide_cert_no + guide_id_card
myresult_list = []
for row in myresult:
myresult_list.append(row)
result_str = guide_name + guide_cert_no + guide_id_card
if result_str in myresult_list:
mycursor.execute(f"UPDATE guide_info SET duplicate_queries = CONCAT('重复') WHERE id={id}")
print('重复')
mydb.commit()
else:
mycursor.execute(f"UPDATE guide_info SET duplicate_queries = CONCAT(guide_name, guide_cert_no, guide_id_card) WHERE id={id}")
mydb.commit()
#print('不重复')
# mycursor.execute(f"UPDATE guide_info SET duplicate_queries = CONCAT(guide_name, guide_cert_no, guide_id_card) WHERE id={id}")
#mydb.commit()
else:
tip = data['tip']
now = datetime.datetime.now()# 获取当前时间和日期
formatted_now = now.strftime("%Y-%m-%d %H:%M:%S")# 将时间格式化为"YYYY-MM-DD HH:MM:SS"格式
update_query = f"UPDATE guide_info SET query_results='{tip}', query_time='{formatted_now}' WHERE id={id}"
mycursor.execute(update_query)
mydb.commit()
# 更新last_processed_id和last_updated字段
update_query = "UPDATE guide_status SET last_processed_id = %s, last_updated = %s"
now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
update_params = (id, now)
mycursor.execute(update_query, update_params)
mydb.commit() xyl52p 发表于 2023-9-8 19:44
请教楼主，你们景区针对不带团的导游有免票政策吗？ 有电子导游证的真实导游。
没有哦，只能是带团的导游。但是嘛，如果你想来 可以私聊我。 景区管理方的老板，会来白嫖软件吗。景区管理方的经理，主管，助理，更不会要免费的软件。景区管理的普通工作人员，只要人家拿出导游证，管他真的假的，收到钱又不是自己的，假如是真的被投诉还有风险。 导游信息的数据库是哪里的？ 干嘛用的？ 和大家一样，没看明白作用。 sql执行都来了，这个数据库哪里有 算是杳黑导游的吗？实名认证 这是查导游是否是假的吗 greatzdl 发表于 2023-8-17 12:07
导游信息的数据库是哪里的？
就是国家旅游局，去爬的。 sp0770 发表于 2023-8-17 14:19
算是杳黑导游的吗？实名认证
算是吧，景区需要时真实的导游才能免票，所以需要比对是不是真实的导游。