mr28ekt6u
/
ppre8onyw
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

wb接口

Browse Source
zhouguoxian
ZGX 5 years ago
parent 3dae6c0b89
commit efaca4c00b
5 changed files with 12 additions and 298 deletions
Show Stats Download Patch File Download Diff File

12
We_test.html
Unescape View File

@ -0,0 +1,12 @@
<!doctype html>
<html>
<head>
<script type = "text/javascript" src="{{ url_for('static', filename='hello.js') }}"></script>
<script type = "text/javascript" src="{{ url_for('static', filename='echarts.js') }}"></script>
<script type = "text/javascript" src="{{ url_for('static', filename='jquery-3.5.1.min.js') }}"></script>
</head>
 <body>
<div id="main1" style="width:100%;height: 600px;"></div>
<script src="../static/js/Map_Contop5.js"></script>
 </body>
</html>

138
foreign_datas.py
Unescape View File

@ -1,138 +0,0 @@
import pymysql
import requests
from bs4 import BeautifulSoup
import json
class ForeignCountry:
def __init__(self):
self.died = 0
self.confirmed = 0
self.crued = 0
self.area = ''
self.curConfirm = 0
self.confirmedRelative = 0
self.pub_date = ''
def __str__(self):
return 'area: %s, died: %s, confirmed: %s, crued: %s, curConfirm: %s, confirmedRelative: %s, pub_date: %s' % (self.area, self.died, self.confirmed, self.crued, self.curConfirm, self.confirmedRelative, self.pub_date)
def get_fcountry_info_tuple(self):
return ((self.died, self.confirmed, self.crued, self.area, self.curConfirm, self.confirmedRelative, self.pub_date))
class MyDB:
def __init__(self, host, user, passwd, db):
self.conn = pymysql.connect(host, user, passwd, db)
self.cursor = self.conn.cursor()
# 获取国外国家类参数列表格式
def get_fcountry_args_list(self, all_foreign_countries):
all_args_list = []
for country in all_foreign_countries:
info = country.get_fcountry_info_tuple()
all_args_list.append(info)
return all_args_list
# 保存国外每日疫情数据
def save_outside_daily_datas(self, all_foreign_countries):
curdate=all_foreign_countries[0].pub_date
#先删除当天已有的数据
sql='delete from foreign_daily_datas where pub_date like "%s"'%(curdate[:10]+ '%')
try:
self.cursor.execute(sql)
self.conn.commit()
except Exception as e:
print(e)
sql = 'insert into foreign_daily_datas(died, confirmed, crued, area, curConfirm, confirmedRelative, pub_date) values(%s, %s, %s, %s, %s, %s, %s)'
res = self.get_fcountry_args_list(all_foreign_countries)
print('+++ foreign_daily_datas, data len: %d' % len(res))
try:
self.cursor.executemany(sql, res)
self.conn.commit()
except Exception as e:
print(e)
print('+++ foreign_daily_datas is over.')
def __del__(self):
if self.conn is not None:
self.conn.close()
class DataService:
#解析网页
def __init__(self):
self.url = 'https://voice.baidu.com/act/newpneumonia/newpneumonia'
self.db = MyDB('localhost', 'root', '1999z5g24x','text_data_increasing')
# 抓取网页
def fetch_html_page(self):
res = requests.get(self.url)
res = res.content.decode('utf-8')
return res
# 解析网页
def parse_target_page(self, html):
soup = BeautifulSoup(html)
tag = soup.find('script', attrs={'id':'captain-config'})
tagStr = tag.string
tagDict = json.loads(tagStr)
# 提取数据更新时间\n",
self.pub_date = tagDict['component'][0]['mapLastUpdatedTime']
# 获取国外各国家的数据\n",
self.outsideDatas = tagDict['component'][0]['caseOutsideList']
# 处理字符串
def process_str(self, s):
ret = ''
if s is None or s == '':
ret ='0'
else:
ret = s
return ret
def getOrElse(self,target, key):
ret = ''
if target.get(key) != None:
ret = target[key]
else:
ret = ''
return ret
#解析国外数据
def parse_outside_daily_datas(self):
# self.__outsideDailyDatas为list，一个国家对应一个元素    
all_foreign_countries = []
for item in self.outsideDatas:
country = ForeignCountry()
country.died = self.getOrElse(item, 'died')
country.confirmed = self.getOrElse(item, 'confirmed')
country.crued = self.getOrElse(item, 'crued')
country.area = self.getOrElse(item, 'area')
country.curConfirm = self.getOrElse(item, 'curConfirm')
country.confirmedRelative = self.getOrElse(item, 'confirmedRelative')
country.pub_date = self.pub_date
all_foreign_countries.append(country)
return all_foreign_countries
# 提取内容生成对象
def fetch_page_datas(self):
all_countries = self.parse_outside_daily_datas()
#for item in all_countries:
# print(item)
return all_countries
# 业务函数
def process_data(self):
html = self.fetch_html_page()
self.parse_target_page(html)
all_countries = self.fetch_page_datas()
# 保存数据
self.db.save_outside_daily_datas(all_countries)
# 创建Dataservice对象
ds = DataService()
ds.process_data()

11
foreign_table.sql
Unescape View File

@ -1,11 +0,0 @@
create table foreign_daily_datas
(      
Id                INT primary key auto_increment not null comment '编号',      
died              INT comment '累计死亡',      
confirmed         INT comment '累计确诊',      
crued             INT comment '累计治愈',      
area              VARCHAR(4000) comment '国家',      
curConfirm        INT comment '现有确诊',      
confirmedRelative INT comment '较昨日新增确诊',      
pub_date          VARCHAR(4000) comment '发布日期'
) comment '国外各国家每日疫情数据';

92
get_datas.py
Unescape View File

@ -1,92 +0,0 @@
import pymysql
from datetime import datetime, timedelta
import time
class MyDB:
def __init__(self, host, user, passwd, db):
self.conn = pymysql.connect(host, user, passwd, db)
self.cursor = self.conn.cursor()
def __del__(self):
self.conn.close()
# 获取当然日期
def get_cur_date(self):
date = datetime.today()
curdate = date.strftime('%Y.%m.%d')
return curdate
# 获取前N天的日期
def get_pren_date(self, n=1):
predate = datetime.today() + timedelta(-n)
predate = predate.strftime('%Y.%m.%d')
return predate
# 获取现有确诊人数最多的5个国家
def get_country_curconfirm_top5(self):
curdate = self.get_cur_date() # 获取当天的日期
sql = "select area,curConfirm,pub_date from foreign_daily_datas where pub_date like '%s' order by curConfirm desc limit 5" % (curdate + '%')
#print('+++ sql: %s' % sql)
results = []
try:
self.cursor.execute(sql)
results = self.cursor.fetchall()
n = 1
while len(results) <= 0:
predate = self.get_pren_date(n)
sql = "select area,curConfirm,pub_date from foreign_daily_datas where pub_date like '%s' order by curConfirm desc limit 5" % (predate + '%')
#print('+++ presql: %s' % sql)
self.cursor.execute(sql)
results = self.cursor.fetchall()
n += 1
if n >= 30:
break
else:
time.sleep(1)
except Exception as e:
print(e)
return results
#测试
if __name__ == "__main__":
mydb = MyDB('localhost', 'root', '1999z5g24x', 'text_data_increasing')
results = mydb.get_country_curconfirm_top5() # results : []
for i in results:
print(i)
from flask import Flask, render_template
from flask import jsonify
app = Flask(__name__)
@app.route('/hello')
def say_hello():
return '<html><body><h1>Hello World</h1></body></html>'
@app.route('/')
def hello():
return render_template('we_text.html', name='Jerry')
@app.route('/country_topn_curconfirm')
def get_country_curconfirm_topn():
mydb = MyDB('localhost', 'root', '1999z5g24x', 'text_data_increasing')
results = mydb.get_country_curconfirm_top5() # results : []
return jsonify(pub_date=results[0][2], country=[x[0] for x in results], curConfirms=[x[1] for x in results])
app.run()

57
show_in_ wb.js
Unescape View File

@ -1,57 +0,0 @@
var myCharts1 = echarts.init(document.getElementById("main1"));
var option1 = {
title: {
text: '现有确诊人数Top5的国家',
subtext: '',
left: 'center',
textStyle:{
color:'#FF0000',
fontFamily:'微软雅黑'
}
},
tooltip: {
formatter: '<br/>{b}:{c}'
},
xAxis: {
type: 'category',
data: []
},
yAxis: {
type: 'value'
},
series: [
{
data: [],
type: 'bar',
showBackground: true,
backgroundStyle: {
color: 'rgba(220, 220, 220, 0.8)'
}
}
]
};
// 向后台发起请求，获取数据
$.ajax({
cache: false,
type:"GET",
url:"/country_topn_curconfirm",
data: null,
dataType : "json",
async: false,
error: function(request) {
alert("发送请求失败！");
},
success: function(result) {
// pub_date, privinces, curConfirms
option1.title.subtext = "数据更新时间： " + result.pub_date
for(i=0; i<result.country.length; ++i) {
option1.xAxis.data.push(result.country[i])
option1.series[0].data.push(result.curConfirms[i])
}
console.info(result)
}
});
myCharts1.setOption(option1);
Write Preview
Loading…
Cancel
Save