攻克Data-数据采集与存储-适用于各个平台数据爬虫

import json import requests import random import urllib.request import re requests.adapters.DEFAULT_RETRIES =5 from datetime import datetime, timedelta #------------------------------------------下列实例Start------------------------------------------ class 实例(): def 从数据库账号批量下载实例加去重(): import pymssql #获取日期列表(昨日往前推7天) datelist=Gk.get_date_list(-7,-1) #获取要爬取的店铺id db = pymssql.connect('192.168.1.202','ai','ai123456','Aidata',charset='utf8') cursor=db.cursor() cursor.execute("select 平台店铺id,cookie from [AiData].[dbo].[账号信息] where 平台='京东' and 状态='RUN'and left(平台店铺id,1) like '%[0-9]%'") #cursor.execute("select 平台店铺id,cookie from [AiData].[dbo].[账号信息] where 平台='京东' and 状态='RUN'and 平台店铺id='12745356'") account_sql=cursor.fetchall() #数据已存在 cursor.execute("select distinct date,shopId from [Test].[dbo].[CategorySales_JD]") record_sql=cursor.fetchall() db.commit() cursor.close() db.close() for rec in account_sql: shopid=rec[0] cookie=rec[1] print(shopid) for date in datelist: target=tuple((date,shopid))#tuple为元祖 if target in record_sql: print(f'店铺id{shopid},日期{date}已采集') else: print(f'需采集的店铺id是{shopid},日期是{date}') _headers_dict = { "Sec-Ch-Ua": "\"Not.A/Brand\";v=\"8\", \"Chromium\";v=\"114\", \"Google Chrome\";v=\"114\"", "Sec-Ch-Ua-Mobile": "?0", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36", "Accept": "application/json, text/plain, */*", "User-Mup": "1689141894121", "P-Pin": "%E7%8C%AB%E4%BA%BA%E9%92%88%E7%BA%BA-%E5%85%B3", "Uuid": "98836b6fdb7aa9e34617-18948b3dbe9", "X-Requested-With": "XMLHttpRequest", "User-Mnp": "d9dba01ce39fca777f9950ed12175ec6", "Sec-Ch-Ua-Platform": "\"Windows\"", "Sec-Fetch-Site": "same-origin", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Dest": "empty", "Referer": "https://sz.jd.com/sz/view/dealAnalysis/dealFeatures.html", "Accept-Language": "zh-CN,zh;q=0.9", "Cookie": f"{cookie}", "Host": "sz.jd.com", "Connection": "Close" } # 自动生成的代码，设置的新值会当作变量传入url _url = f'https://sz.jd.com/sz/api/trade/getCategoryFeatureData.ajax?channel=99&date={date}&endDate={date}&startDate={date}&shopId={shopid}'#.format() print(_url) _post_data_string = '''''' response = Gk.get_request(_url, _headers_dict,True) # print(response) Gk.Guid2DB( _url, _headers_dict, _post_data_string, response) #从数据库账号批量下载实例加去重() def 从数据库账号批量下载实例(): import pymssql db = pymssql.connect('192.168.1.202','ai','ai123456','AiData',charset='utf8') cursor = db.cursor() cursor.execute("SELECT [平台店铺ID],[请求头],[Cookie] FROM [账号信息] where 平台='京东' and 状态='RUN' order by 1 desc") result_login = cursor.fetchall() # print(result_login) db.commit() cursor.close( ) db.close() for rec in result_login: cookie=rec[2] # 根据实际代码替换下面代码： _headers_dict = { "Host": "sz.jd.com", "Connection": "keep-alive", "Cookie": f"{cookie}", } # 自动生成的代码，设置的新值会当作变量传入url _url = f'https://sz.jd.com/sz/api/realtime/getRealtimeData.ajax?'#.format() _post_data_string = '''''' response = Gk.get_request(_url, _headers_dict,True) print(response) #Gk.Guid2DB( _url, _headers_dict, _post_data_string, response) #从数据库账号批量下载实例() #--------------------------------------------END------------------------------------------------ def add_months(date, months): month = (date.month + months) % 12 year = date.year + (date.month + months - 1) // 12 if month == 0: month = 12 day = min(date.day, (date.replace(year=year, month=month, day=1) + timedelta(days=-1)).day) return date.replace(year=year, month=month, day=day) #获取月份列表 def get_month_list(start, end): result = [] # 获取当前日期 current_date = datetime.now() # 计算起始日期和结束日期 start_date = add_months(current_date, start) end_date = add_months(current_date, end) while start_date <= end_date: result.append(start_date.strftime("%Y-%m")) start_date = add_months(start_date, 1) return result #获取日期列表 def get_date_list(start,end): import datetime datelist=[] now_time=datetime.datetime.now() for i in range(start,end+1,-1 if start>end else 1): datelist.append((now_time+datetime.timedelta(i)).strftime('%Y-%m-%d')) return datelist #Chrome请头到转字典 def headers_str_to_dict(headers): headers_dict = {} for line in headers.split('\n'): if ':' in line: key, value = re.split(':|=', line.strip(), maxsplit=1) headers_dict[key.strip()] = value.strip() return headers_dict #Get请求 def get_request(url, headers,useFiddlerProxy): if useFiddlerProxy: # 假设 IP 池已经包含了多个代理 IP 地址 proxy_list = ['http://127.0.0.1:8888'] # 随机选择一个代理 IP 地址进行请求 proxies = {"http": random.choice(proxy_list)} proxies = urllib.request.getproxies() if proxies == {'http': 'http://127.0.0.1:8888', 'https': 'https://127.0.0.1:8888'}: proxies = { 'http': 'http://127.0.0.1:8888', 'https': 'http://127.0.0.1:8888' # https -> http } res = requests.get(url, headers=headers, proxies=proxies, verify=False, timeout=5) else : res = requests.get(url, headers=headers, verify=False, timeout=5) content_type = res.headers['Content-Type'] if 'image' in content_type: # 返回图片内容 return res.content else: # 返回文本内容 return res.text #POST请求 def post_request(url, headers, data,useFiddlerProxy): if useFiddlerProxy: # 假设 IP 池已经包含了多个代理 IP 地址 proxy_list = ['http://127.0.0.1:8888'] # 随机选择一个代理 IP 地址进行请求 proxies = {"http": random.choice(proxy_list)} proxies = urllib.request.getproxies() if proxies == {'http': 'http://127.0.0.1:8888', 'https': 'https://127.0.0.1:8888'}: proxies = { 'http': 'http://127.0.0.1:8888', 'https': 'http://127.0.0.1:8888' # https -> http } res = requests.post(url, headers=headers, data=data, proxies=proxies, verify=False, timeout=5) else: res = requests.post(url, headers=headers, data=data, verify=False, timeout=5) content_type = res.headers['Content-Type'] request_header_str = str