Python3：sqlalchemy对mysql数据库操作，非sql语句
1# python3 2# author lizm 3# datetime 2018-02-01 10:00:00 4# -*- coding: utf-8 -*- 5''' 6 数据起始日期：2015-05-08 7 数据库：mysql 8''' 9import requests 10from bs4 import BeautifulSoup 11import json 12import pymysql 13import datetime 14import time 15import sys 16import logging 17from selenium import webdriver 18from sqlalchemy import Column,Integer, String,DateTime,create_engine 19from sqlalchemy.orm import sessionmaker 20from sqlalchemy.ext.declarative import declarative_base 21from sqlalchemy import and_,func 22import configparser 23import math 24 25logger = logging.getLogger() 26#set loghandler 27file = logging.FileHandler(sys.path[0]+"\py_zgjs_log"+time.strftime("%Y%m%d")+".log") 28logger.addHandler(file) 29#set formater 30formatter = logging.Formatter("%(asctime)s %(levelname)s %(message)s") 31file.setFormatter(formatter) 32#set log level 33logger.setLevel(logging.NOTSET) 34 35# 创建对象的基类: 36Base = declarative_base() 37 38class Yztzzqktjb(Base): 39 40 # 表名 41 __tablename__ = 'py_zgjs_yztzzqktjb' 42 43 # 表结构 44 id = Column(Integer,primary_key=True,autoincrement=True) 45 mc = Column(String(200),nullable=False) 46 begindate = Column(String(45),nullable=False) 47 enddate = Column(String(45), nullable=False) 48 sjmc = Column(String(200)) 49 ssjmc = Column(String(200)) 50 sl = Column(String(45)) 51 create_time = Column(DateTime,nullable=False) 52 update_time = Column(DateTime,nullable=False) 53 54 def __init__(self,mc,begindate,enddate,sjmc,ssjmc,sl,create_time,update_time): 55 self.mc = mc 56 self.begindate = begindate 57 self.enddate = enddate 58 self.sjmc = sjmc 59 self.ssjmc = ssjmc 60 self.sl = sl 61 self.create_time = create_time 62 self.update_time = update_time 63 64class ZgjsEntry(object): 65 66 def __init__(self, v1, v2,v3,v4,v5,v6): 67 self.v1 = v1 68 self.v2 = v2 69 self.v3 = v3 70 self.v4 = v4 71 self.v5 = v5 72 self.v6 = v6 73 74 def __get__(self, instance, cls): 75 if instance is None: 76 return self 77 else: 78 return instance.__dict__[self.name] 79 80 def __set__(self, instance, value): 81 instance.__dict__[self.name] = value 82 83 def __delete__(self, instance): 84 del instance.__dict__[self.name] 85 86def dbconfig(): 87 #生成config对象 88 cfg = configparser.ConfigParser() 89 #用config对象读取配置文件 90 path_ = sys.path[0] 91 cfg.read(path_+"\dbconfig.ini") 92 ip = cfg.get("dbserver", "ip") 93 port = cfg.get("dbserver", "port") 94 user = cfg.get("dbserver", "user") 95 password = cfg.get("dbserver", "password") 96 dbname = cfg.get("dbserver", "dbname") 97 endtime = cfg.get("dbtime", "endtime") 98 initdate = cfg.get("dbtime", "mzkbinitdate") 99 interval = cfg.get("dbtime", "interval") 100 return (ip,port,user,password,dbname,endtime,initdate,interval) 101 102def savrData(tableName,zgjsList): 103 msgcode = 0 104 message = '数据保存成功' 105 try: 106 dbcfg = dbconfig() 107 # 初始化数据库连接, 108 # 传入参数：数据库类型+连接库+用户名+密码+主机，字符编码，是否打印建表细节 109 engine = create_engine('mysql+mysqlconnector://'+dbcfg[2]+':'+dbcfg[3]+'@'+dbcfg[0]+':'+dbcfg[1]+'/'+dbcfg[4],encoding='utf-8') 110 # 创建DBSession类型: 111 DBSession = sessionmaker(bind=engine) 112 session = DBSession() 113 try: 114 # 增操作 115 items = [] 116 if tableName == 'Yztzzqktjb': 117 if len(zgjsList)>0: 118 for i in range(0,len(zgjsList)): 119 results = session.query(Yztzzqktjb).filter(and_(Yztzzqktjb.mc == zgjsList[i].v1,Yztzzqktjb.begindate == zgjsList[i].v2,Yztzzqktjb.enddate==zgjsList[i].v3,Yztzzqktjb.sjmc==zgjsList[i].v4,Yztzzqktjb.ssjmc==zgjsList[i].v5)).all() 120 if len(results) > 0: 121 session.query(Yztzzqktjb).filter(and_(Yztzzqktjb.mc == zgjsList[i].v1,Yztzzqktjb.begindate == zgjsList[i].v2,Yztzzqktjb.enddate==zgjsList[i].v3,Yztzzqktjb.sjmc==zgjsList[i].v4,Yztzzqktjb.ssjmc==zgjsList[i].v5)).update({Yztzzqktjb.sl: zgjsList[i].v6,Yztzzqktjb.update_time:time.strftime('%Y-%m-%d %H:%M:%S')}, synchronize_session=False) 122 else: 123 item = Yztzzqktjb(mc=zgjsList[i].v1,begindate=zgjsList[i].v2,enddate=zgjsList[i].v3,sjmc=zgjsList[i].v4,ssjmc=zgjsList[i].v5,sl=zgjsList[i].v6,create_time=time.strftime('%Y-%m-%d %H:%M:%S'),update_time=time.strftime('%Y-%m-%d %H:%M:%S')) 124 items.append(item) 125 else: 126 pass 127 #print("len(items)>>>>>%s" %len(items)) 128 if len(items) > 0: 129 for i in range(0,len(items)): 130 session.add(items[i]) 131 #提交数据 132 session.commit() 133 except Exception as e: 134 msgcode = 1 135 message = '数据保存失败' + str(e) 136 session.rollback() 137 finally: 138 #关闭 139 session.close() 140 except Exception as e: 141 msgcode = 1 142 message = '数据库连接失败'+str(e) 143 logger.info(message) 144 print(message) 145 return msgcode 146 147 148def getData(jsDate, channelIdStr,tableName): 149 zgjsList = [] 150 dateStr = jsDate[0:4]+'.'+jsDate[5:7]+'.'+jsDate[8:10] 151 # 查询按钮跳转url： 152 # http://www.******.cn/cms-search/view.action?action=china 153 url = "http://www.******.cn/cms-search/view.action?action=china" 154 headerDict = {'Host': 'www.*******.cn', 155 'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/61.0.3163.31 Safari/537.36', 156 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8', 157 'Accept-Language': 'zh-CN,zh;q=0.8', 158 'Accept-Encoding': 'gzip, deflate', 159 'Referer': 'http://www.******.cn/cms-search/view.action?action=china', 160 'Connection': 'keep-alive'} 161 data = {'dateType': '', 'dateStr': dateStr, 162 'channelIdStr': channelIdStr} 163 # psot 传递参数 164 res = requests.post(url, data=data, headers=headerDict) 165 # 获取跳转后的页面源码 166 soup = BeautifulSoup(res.content, "html.parser") 167 #获取周报的起始日期 168 SettlementTitle = soup.find('div',class_='SettlementTitle') 169 if SettlementTitle is None: 170 return zgjsList 171 h2 = SettlementTitle.find('h2').text 172 if h2 == '搜索结果': 173 return zgjsList 174 weekdate = h2.strip().split('（')[1].split('）')[0] 175 begindate = weekdate.split('-')[0].replace('.','-') 176 enddate = weekdate.split('-')[1].replace('.','-') 177 178 settlementList = soup.find(id='settlementList') 179 # print(settlementList) 180 if settlementList is None: 181 return zgjsList 182 if settlementList.find('table') is None: 183 return zgjsList 184 185 table_ = settlementList.find('table') 186 tr_list = table_.find('table').find_all('tr') 187 # 上级名称 188 sjmc_1 = '' 189 sjmc_2 = '' 190 sjmc_3 = '' 191 sjmc_4 = '' 192 sjmc_5 = '' 193 sjmc_6 = '' 194 # 上上级名称 195 ssjmc_1 = '' 196 for n in range(1,len(tr_list)): 197 td_list = tr_list[n].find_all('td') 198 if tableName == 'Yztzzqktjb': 199 if n == 1: 200 sjmc_1 = td_list[0].get_text().replace('一、','').strip() 201 if n == 4: 202 sjmc_2 = td_list[0].get_text().replace('二、','').strip() 203 ssjmc_1 = td_list[0].get_text().replace('二、','').strip() 204 if n == 5: 205 sjmc_3 = td_list[0].get_text().replace('1、','').strip() 206 if n == 9: 207 sjmc_4 = td_list[0].get_text().replace('2、','').strip() 208 if n == 13: 209 sjmc_5 = td_list[0].get_text().replace('三、','').strip() 210 if n == 17: 211 sjmc_6 = td_list[0].get_text().replace('四、','').strip() 212 213 if tableName == 'Yztzzqktjb': 214 if n in (6,10,14,18): 215 continue 216 zgjs = ZgjsEntry('','','','','','') 217 zgjs.v2 = begindate 218 zgjs.v3 = enddate 219 220 if tableName == 'Yztzzqktjb': 221 # 上级名称 222 if n in (2,3): 223 zgjs.v4 = sjmc_1 224 if n in (5,9): 225 zgjs.v4 = sjmc_2 226 if n in (7,8): 227 zgjs.v4 = sjmc_3 228 zgjs.v5 = ssjmc_1 229 if n in (11,12): 230 zgjs.v4 = sjmc_4 231 zgjs.v5 = ssjmc_1 232 if n in (15,16): 233 zgjs.v4 = sjmc_5 234 if n in (19,20): 235 zgjs.v4 = sjmc_6 236 for i in range(0,len(td_list)): 237 if i == 0: 238 zgjs.v1 =td_list[i].get_text().replace('一、','').replace('二、','').replace('三、','').replace('四、','').replace('1、','').replace('2、','').strip() 239 if i == 1: 240 zgjs.v6 =td_list[i].get_text().strip().replace(',','') 241 if zgjs is not None: 242 zgjsList.append(zgjs) 243 return zgjsList 244 245# 获取开始日期： 246def getBeginDate(bgdate,tableName): 247 r_date = bgdate 248 try: 249 dbcfg = dbconfig() 250 # 初始化数据库连接, 251 # 传入参数：数据库类型+连接库+用户名+密码+主机，字符编码，是否打印建表细节 252 engine = create_engine('mysql+mysqlconnector://'+dbcfg[2]+':'+dbcfg[3]+'@'+dbcfg[0]+':'+dbcfg[1]+'/'+dbcfg[4],encoding='utf-8') 253 # 创建DBSession类型: 254 DBSession = sessionmaker(bind=engine) 255 session = DBSession() 256 try: 257 if tableName == 'Yztzzqktjb': 258 results = session.query(func.max(Yztzzqktjb.enddate)).all() 259 if len(results) != 0: 260 r_date = results[0] 261 else: 262 pass 263 except Exception as e: 264 print('获取开始日期，查询异常；%s'%str(e)) 265 logger('获取开始日期，查询异常；%s'%str(e)) 266 session.rollback() 267 finally: 268 #关闭 269 session.close() 270 except Exception as e: 271 print('获取开始日期，数据库连接失败；%s'%str(e)) 272 logger('获取开始日期，数据库连接失败；%s'%str(e)) 273 if r_date[0] is None: 274 r_date = bgdate 275 else: 276 begin = time.strptime(r_date[0], "%Y-%m-%d") 277 y,m,d = begin[0:3] 278 r_date = datetime.date(y,m,d) + datetime.timedelta(days=7) 279 r_date = r_date .strftime('%Y-%m-%d') 280 return r_date 281 282def isCheckData(date_): 283 r_code = 0 284 try: 285 dbcfg = dbconfig() 286 # 初始化数据库连接, 287 # 传入参数：数据库类型+连接库+用户名+密码+主机，字符编码，是否打印建表细节 288 engine = create_engine('mysql+mysqlconnector://'+dbcfg[2]+':'+dbcfg[3]+'@'+dbcfg[0]+':'+dbcfg[1]+'/'+dbcfg[4],encoding='utf-8') 289 # 创建DBSession类型: 290 DBSession = sessionmaker(bind=engine) 291 session = DBSession() 292 try: 293 # 294 results = session.query(Yztzzqktjb).filter(and_(func.datediff(Yztzzqktjb.enddate,date_)<6,func.datediff(Yztzzqktjb.enddate,date_)>-2)).all() 295 if len(results) == 0: 296 r_code = 1 297 else: 298 r_code = 0 299 except Exception as e: 300 r_code = 1 301 print('判断是否有数据异常；%s'%str(e)) 302 logger('判断是否有数据异常；%s'%str(e)) 303 session.rollback() 304 finally: 305 #关闭 306 session.close() 307 except Exception as e: 308 r_code = 1 309 print('判断是否有数据，数据库连接异常；%s'%str(e)) 310 logger('判断是否有数据，数据库连接异常；%s'%str(e)) 311 return r_code 312 313# 执行入口 314def main(initdate_): 315 req_list = [ 316 {'report':'6ac54ce22db4474abc234d6edbe53ae7','table':'Yztzzqktjb'} 317 ] 318 for req in req_list: 319 #字符转日期 320 begin = time.strptime(getBeginDate(initdate_,req['table']), "%Y-%m-%d") 321 y,m,d = begin[0:3] 322 #日期格式：2018-01-18 323 begin = datetime.date(y,m,d) 324 #获取当前日期 325 end = datetime.date.today() 326 if (end- begin).days < 0: 327 pass 328 else: 329 for i in range(math.ceil((end - begin).days/7)+1): 330 list_szzj = [] 331 # 日期转字符 332 date_ = (begin+datetime.timedelta(days=i*7)).strftime('%Y-%m-%d') 333 list_mzkb = getData(date_,req['report'],req['table']) 334 if len(list_mzkb): 335 savrData(req['table'],list_mzkb) 336 else: 337 pass 338 time.sleep(0.5) 339 if i % 350 == 0: 340 time.sleep(15) 341 342 343if __name__ == '__main__': 344 vrg_date = '20150509' 345 dbcfg = dbconfig() 346 vrg_endtime = dbcfg[5][0:2]+":"+dbcfg[5][2:4]+":"+dbcfg[5][4:6] 347 var_initdate = dbcfg[6][0:4]+"-"+dbcfg[6][4:6]+"-"+dbcfg[6][6:8] 348 var_interval = int(dbcfg[7]) 349 350 if len(vrg_date) ==8: 351 vrg_date = str(vrg_date[0:4]) + "-" + str(vrg_date[4:6]) + "-" + str(vrg_date[6:8]) 352 end_time = time.strptime(vrg_endtime, "%H:%M:%S") 353 y,m,d = end_time[3:6] 354 end_time = datetime.time(y,m,d) 355 # 循环采集 356 while True: 357 now_time = time.strftime("%H%M%S") 358 main(var_initdate) 359 if isCheckData(vrg_date,) == 0: 360 logger.info("采集数据结束") 361 print("采集数据结束") 362 break 363 # 时间到停止采集 364 if int(end_time.strftime('%H%M%S')) - int(now_time) <= 0: 365 logger.info("采集数据结束") 366 print("采集数据结束") 367 break 368 # 间隔执行时间 369 logger.info("**********************（%s）:没有采集到数据，任务继续执行**********************" %vrg_date) 370 print("********************（%s）:没有采集到数据，任务继续执行**********************" %vrg_date) 371 time.sleep(var_interval) 372 else: 373 logger.info("日期参数格式不正确，请用格式：20180205") 374 print("日期参数格式不正确，请用格式：20180205")