123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372 |
- from rds_model.rds_user_shop_promotion_list import RdsUserShopPromotionList
- import time
- import json
- import sys
- import threading
- import random
- import urllib
- import requests
- from rds_model.db_redis import DbRedis
- from libs.Xg04 import X_Gorgon
- from libs.proxy import Proxy
- from libs.mysql_user_living import *
- start_time = time.time()
- def get_random(i, random_type=1):
- if random_type == 1:
- return str(random.randint(1 * 10 ** (i - 1), 1 * 10 ** i - 1))
- elif random_type == 8:
- seed = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
- sa = []
- for i in range(i):
- sa.append(random.choice(seed))
- salt = ''.join(sa)
- return salt
- else:
- seed = "1234567890abcde"
- sa = []
- for i in range(i):
- sa.append(random.choice(seed))
- salt = ''.join(sa)
- return salt
- def get_random_brand_type():
- brand_type = get_random(3, random_type=8) + '-' + get_random(2, random_type=8) + '00'
- return brand_type
- def get_mc():
- def a():
- seed = "1234567890ABCDEF"
- sa = []
- for i in range(2):
- sa.append(random.choice(seed))
- salt = ''.join(sa)
- return salt
- k = ''
- for i in range(6):
- k += a() + ':'
- return k[:-1]
- def get_trace():
- trace_list = [
- '00-70f99f2209e0b045dd14266ee1da0468-70f99f2209e0b045-01',
- '00-ce7faf4409b7fcc0ae6135fdd4250468-ce7faf4409b7fcc0-01',
- '00-ce7faf3b09b7fcc0ae6042f1d8100468-ce7faf3b09b7fcc0-01',
- '00-cdd79d2309b7fcc0ae6625a4cb190468-cdd79d2309b7fcc0-01',
- '00-cefde9f009b7fcc0ae6750e1349e0468-cefde9f009b7fcc0-01',
- '00-ced2e6ef09b7fcc0ae67dd7bfe000468-ced2e6ef09b7fcc0-01',
- '00-cefbfeb509b7fcc0ae659396a6ea0468-cefbfeb509b7fcc0-01',
- '00-cefaa25409b7fcc0ae657726a3c30468-cefaa25409b7fcc0-01',
- '00-63eb56f009b7fcc0ae600208011d0468-63eb56f009b7fcc0-01'
- ]
- return random.choice(trace_list)
- def get_user_shop_promotion_on_shelf_data(user_id, sec_user_id, result):
- domain = 'api3-normal-c-lq.amemv.com'
- # domain = 'aweme.snssdk.com'
- url = 'https://' + domain + '/aweme/v1/promotion/user/promotion/list/?'
- rticket = str(int(time.time() * 1000))
- mc = get_mc
- udid = '8604' + get_random(11)
- trace_id = get_trace()
- # openudid = '3b22' + str(udid.uuid4())[-12:]
- device_id, iid, udid, openudid, cookie, V1, V2, device_type, device_brand = result[0], result[1], result[3], result[2], result[4], result[8], result[9], result[10], result[11]
-
- query = {
- "user_id" : "2066709087257544",
- #"user_id" : str(user_id),
- "sec_user_id" : "MS4wLjABAAAASJ-Q7Mjt6chYEEju720IdOKArNsreGnDi8ADd3E64SsbAeqoXRQex9mCDG7u5DaY",
- #"sec_user_id" : str(sec_user_id),
- "cursor" : "0",
- "count" : "1000",
- "column_id" : "0",
- "goods_type" : "0",
- "shop_version" : "1",
- "os_api" : "23",
- "device_type" : "HUAWEI+MLA-AL10",
- # "device_type" : device_type,
- "ssmix" : "a",
- "manifest_version_code" : "130001",
- "dpi" : "480",
- "app_name" : "aweme",
- "version_name" : "13.0.0",
- "ts" : "1633763939",
- # "ts" : int(time.time()),
- "cpu_support64" : "true",
- "storage_type" : "0",
- "app_type" : "normal",
- "appTheme" : "dark",
- "ac" : "wifi",
- "host_abi" : "armeabi-v7a",
- "update_version_code" : "13009900",
- "channel" : "tengxun_new",
- "_rticket" : "1633763939688",
- # "_rticket" : rticket,
- "device_platform" : "android",
- "iid" : "1961967166955757",
- # "iid" : str(iid),
- "version_code" : "130000",
- # "mac_address" : mc,
- "mac_address" : "50%3A01%3AD9%3A21%3AED%3AC2",
- "cdid" : "ab363b15-99db-4ef5-bfb1-834a8e564105",
- "openudid" : "291f3ce2efe59345",
- # "openudid" : str(openudid),
- "device_id" : "49388718822",
- # "device_id" : str(device_id),
- "resolution" : "1080*1800",
- "os_version" : "6.0",
- "language" : "zh",
- # "device_brand":device_brand,
- "device_brand" : "HUAWEI",
- "aid" : "1128"
- }
- query_params = urllib.parse.urlencode(query)
- url = url + query_params
- body = ''
- xGorgon = X_Gorgon(query_params, body)
- # userAgent = f'aweme.snssdk.com/130001 (Linux; U; Android 6.0; zh_CN; {device_type}; Build/{device_type}; Cronet/TTNetVersion:414feb46 2020-09-08 QuicVersion:7aee791b 2020-06-05)'
- userAgent = 'com.ss.android.ugc.aweme/130001 (Linux; U; Android 6.0; zh_CN; HUAWEI+MLA-AL10; Build/HUAWEI+MLA-AL10; Cronet/TTNetVersion:414feb46 2020-09-08 QuicVersion:7aee791b 2020-06-05)'
- headers = {
- 'Host': domain,
- 'Connection': 'keep-alive',
- 'passport-sdk-version' : '18',
- 'sdk-version' : '2',
- 'X-SS-DP' : '1128',
- # 'x-tt-trace-id' : trace_id,
- # 'Upgrade-Insecure-Requests': '1',
- 'User-Agent': userAgent,
- # 'accept-encoding': 'gzip, deflate',
- # "x-SS-REQ-TICKET": rticket,
- "x-Gorgon": xGorgon.get('X-Gorgon'),
- "x-Khronos": xGorgon.get('X-Khronos'),
- # "cookie" : cookie,
- "accept": "application/json, text/plain, */*",
- }
- retry = 0
- response_json = None
- while True:
- if retry > 5:
- break
- retry += 1
- proxy = Proxy.dailiyun_get()
- proxies = {
- "http": "http://" + proxy,
- "https": "http://" + proxy
- }
- try:
- response = requests.get(
- url,
- headers=headers,
- proxies=proxies,
- timeout=8
- )
- if (response.status_code == 200) and (response.text is not None) and (response.text != ''):
- response_json = response.json()
- if (response_json.get('promotions') is not None):
- print(
- time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
- + ' 数据获取成功!'
- + '\n'
- + user_id
- + '\n'
- )
- break
- else:
- print(
- time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
- + ' 数据获取失败!'
- + '\n'
- + user_id
- + '\n'
- + response.text
- + Proxy.proxy_info
- + '\n'
- )
-
- else:
- print(
- time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
- + ' 爬取http连接失败!'
- #+ str(response.status_code)
- + str(url)
- + '\n'
- + str(headers)
- + '\n'
- + Proxy.proxy_info
- + '\n'
- + user_id
- + '\n'
- + '爬取结果:' + str(response)
- + '\n'
- )
- time.sleep(1)
- except requests.exceptions.ProxyError as e:
- print(
- time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
- + ' 代理过期!'
- + str(e)
- + '\n'
- + user_id
- + '\n'
- + Proxy.proxy_info
- + '\n'
- )
- Proxy.dailiyun_del_proxy(proxy)
- pass
- except requests.exceptions.ConnectTimeout as e:
- print(
- time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
- + ' ConnectTimeout!'
- + str(e)
- + '\n'
- + user_id
- + '\n'
- + Proxy.proxy_info
- + '\n'
- )
- Proxy.dailiyun_del_proxy(proxy)
- pass
- except Exception as e:
- print(
- time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
- + ' 请求抛出异常!'
- + str(e)
- + '\n'
- + user_id
- + '\n'
- + Proxy.proxy_info
- + '\n'
- )
- pass
- return response_json
- def scrape():
- rds_list = RdsUserShopPromotionList()
- room_info = rds.get_on_shelf_request_param()
- if room_info is None:
- return None
- # room_info = json.loads(room_info)
- room_info = room_info.split('@')
- room_id = room_info[0]
- user_id = room_info[1]
- room_id = str(room_id)
- user_id = str(user_id)
- # 根据用户id获取sec_uid
- user_info = MysqlUserLiving().get_user_info(user_id)
- sec_user_id, = user_info
- print(
- time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
- + ' user_id:' + str(user_id) + ' sec_uid:' + str(sec_user_id)
- )
- key = 'DOUYIN_SCRAPE_DID_IID_TTREQ_1221'
- rdid = DbRedis.connect().rpop(key)
- if rdid:
- result = rdid.split('@@@')
- else:
- return None
- DbRedis.connect().lpush(key, rdid)
- try:
- response_json = get_user_shop_promotion_on_shelf_data(user_id=user_id, sec_user_id=sec_user_id, result=result)
-
- if response_json is None:
- # rds_list.record_score(0)
- # rds_list.push_request_id(room_id)
- print(
- time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
- + ' 数据获取失败!响应数据为空!'
- + '\n'
- + user_id
- + '\n'
- )
- sys.exit(0)
- data = json.dumps({
- "data": response_json.get('promotions'),
- "extra": {
- 'room_id' : room_id,
- 'user_id' : user_id,
- }
- })
- # rds_list.record_score(1)
- rds_list.push_on_shelf_data_list(data)
- except Exception as e:
- # rds_list.record_score(0)
- rds_list.push_request_id_on_shelf(room_id + '@' + user_id)
- print(
- time.strftime("%H:%M:%S", time.localtime())
- + ' '
- + user_id
- + '数据异常:'
- + str(e)
- )
- sys.exit(0)
- if __name__ == "__main__":
- print("主方法开始执行")
- # 并行线程数
- threading_count = int(sys.argv[1])
- num = int(sys.argv[2])
- rds = RdsUserShopPromotionList()
- print(
- time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
- + ' '
- + ' 开始执行,更新直播队列长度:'
- + str(rds.get_on_shelf_len())
- )
- while True:
- sys.stdout.flush()
-
- # 减去主线程
- active_count = threading.active_count() - 1
-
- increment = threading_count - active_count
-
- while increment > 0:
- sys.stdout.flush()
- # scrape()
- task = threading.Thread(target=scrape, args=())
- task.start() # 准备就绪, 等待cpu执行
- increment = increment - 1
-
- current_time = time.time()
- if current_time - start_time > 3600:
- print(
- time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
- + ' 主方法执行终止'
- )
- sys.exit(0)
- time.sleep(1)
|