Pythonリアルタイムデータ収集 - 新型コロナウイルス-Python チュートリアル-php.cn

Pythonリアルタイムデータ収集 - 新型コロナウイルス

angryTom

リリース： 2020-02-03 17:51:00

転載

4008 人が閲覧しました

Pythonリアルタイムデータ収集 - 新型コロナウイルス

#Python リアルタイムデータ収集 - 新型コロナウイルス

ソースコード出典: https://github.com/Programming-With-Love/2019-nCoV

流行データの時間: 2020.2.1

##プロジェクトの関連スクリーンショット:

National Data Display

国内データの表示

##外国データの表示

##詳細を表示指定した領域のデータ

ソースコード、必要なモジュール(pip installモジュール名など)のインストールに注意してください

import requests
import re
from bs4 import BeautifulSoup
from time import sleep
import json
from prettytable import ALL
from prettytable import PrettyTable

hubei = {}
guangdong = {}
zhejiang = {}
beijing = {}
shanghai = {}
hunan = {}
anhui = {}
chongqing = {}
sichuan = {}
shandong = {}
guangxi = {}
fujian = {}
jiangsu = {}
henan = {}
hainan = {}
tianjin = {}
jiangxi = {}
shanxi1 = {} # 陕西
guizhou = {}
liaoning = {}
xianggang = {}
heilongjiang = {}
aomen = {}
xinjiang = {}
gansu = {}
yunnan = {}
taiwan = {}
shanxi2 = {} # 山西
jilin = {}
hebei = {}
ningxia = {}
neimenggu = {}
qinghai = {} # none
xizang = {} # none
provinces_idx = [hubei, guangdong, zhejiang, chongqing, hunan, anhui, beijing,
                 shanghai, henan, guangxi, shandong, jiangxi, jiangsu, sichuan,
                 liaoning, fujian, heilongjiang, hainan, tianjin, hebei, shanxi2,
                 yunnan, xianggang, shanxi1, guizhou, jilin, gansu, taiwan,
                 xinjiang, ningxia, aomen, neimenggu, qinghai, xizang]
map = {
    &#39;湖北&#39;:0, &#39;广东&#39;:1, &#39;浙江&#39;:2, &#39;北京&#39;:3, &#39;上海&#39;:4, &#39;湖南&#39;:5, &#39;安徽&#39;:6, &#39;重庆&#39;:7,
    &#39;四川&#39;:8, &#39;山东&#39;:9, &#39;广西&#39;:10, &#39;福建&#39;:11, &#39;江苏&#39;:12, &#39;河南&#39;:13, &#39;海南&#39;:14,
    &#39;天津&#39;:15, &#39;江西&#39;:16, &#39;陕西&#39;:17, &#39;贵州&#39;:18, &#39;辽宁&#39;:19, &#39;香港&#39;:20, &#39;黑龙江&#39;:21,
    &#39;澳门&#39;:22, &#39;新疆&#39;:23, &#39;甘肃&#39;:24, &#39;云南&#39;:25, &#39;台湾&#39;:26, &#39;山西&#39;:27, &#39;吉林&#39;:28,
    &#39;河北&#39;:29, &#39;宁夏&#39;:30, &#39;内蒙古&#39;:31, &#39;青海&#39;:32, &#39;西藏&#39;:33
}


def getTime(text):
    TitleTime = str(text)
    TitleTime = re.findall(&#39;<span>(.*?)</span>&#39;, TitleTime)
    return TitleTime[0]

def getAllCountry(text):
    AllCountry = str(text)
    AllCountry = AllCountry.replace("[<p class=\"confirmedNumber___3WrF5\"><span class=\"content___2hIPS\">", "")
    AllCountry = AllCountry.replace("<span style=\"color: #4169e2\">", "")
    AllCountry = re.sub("</span>", "", AllCountry)
    AllCountry = AllCountry.replace("</p>]", "")
    
    AllCountry = AllCountry.replace("<span style=\"color: rgb(65, 105, 226);\">", "")
    AllCountry = re.sub("<span>", "", AllCountry)
    AllCountry = re.sub("<p>", "", AllCountry)
    AllCountry = re.sub("</p>", "", AllCountry)
    return AllCountry 

def query(province):
    table = PrettyTable([&#39;地区&#39;, &#39;确诊&#39;, &#39;死亡&#39;, &#39;治愈&#39;])

    for (k, v) in province.items():
        name = k
        table.add_row([name, v[0] if v[0] != 0 else &#39;-&#39;, v[1] if v[1] != 0 else &#39;-&#39;, v[2] if v[2] != 0 else &#39;-&#39;])
    if len(province.keys()) != 0:
        print(table)
    else:
        print("暂无")

def getInfo(text):
    text = str(text)
    text = re.sub("<p class=\"descText___Ui3tV\">", "", text)
    text = re.sub("</p>", "", text)
    return text

def is_json(json_str):
    try:
        json.loads(json_str)
    except ValueError:
        return False
    return True

def ff(str, num):
    return str[:num] + str[num+1:]
        

def main():
    url = "https://3g.dxy.cn/newh5/view/pneumonia"

    try:
        headers = {}
        headers[&#39;user-agent&#39;] = &#39;Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/70.0.3538.77 Safari/537.36&#39; #http头大小写不敏感
        headers[&#39;accept&#39;] = &#39;text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8&#39;
        headers[&#39;Connection&#39;] = &#39;keep-alive&#39;
        headers[&#39;Upgrade-Insecure-Requests&#39;] = &#39;1&#39;

        r = requests.get(url, headers=headers)
        r.raise_for_status()
        r.encoding = r.apparent_encoding
        soup = BeautifulSoup(r.text,&#39;lxml&#39;)
        table = PrettyTable([&#39;地区&#39;, &#39;确诊&#39;, &#39;死亡&#39;, &#39;治愈&#39;])
        table.hrules = ALL

        #### 截至时间
        # TitleTime = getTime(soup.select(&#39;.title___2d1_B&#39;))
        
        print()
        # print("              ",TitleTime + "\n")

        while True:
            r = requests.get("https://service-f9fjwngp-1252021671.bj.apigw.tencentcs.com/release/pneumonia")
            json_str = json.loads(r.text)
            if json_str[&#39;error&#39;] == 0:
                break

        print("==================================全国数据==================================")
        print()
        
        print("     确诊 " + str(json_str[&#39;data&#39;][&#39;statistics&#39;][&#39;confirmedCount&#39;]) + " 例"
            + "       " + "疑似 " + str(json_str[&#39;data&#39;][&#39;statistics&#39;][&#39;suspectedCount&#39;]) + " 例"
            + "       " + "死亡" + str(json_str[&#39;data&#39;][&#39;statistics&#39;][&#39;deadCount&#39;]) + " 例"
            + "       " + "治愈" + str(json_str[&#39;data&#39;][&#39;statistics&#39;][&#39;curedCount&#39;]) + " 例\n")

        print("==================================相关情况==================================")
        print()

        print("传染源：" + json_str[&#39;data&#39;][&#39;statistics&#39;][&#39;infectSource&#39;])
        print("病毒：" + json_str[&#39;data&#39;][&#39;statistics&#39;][&#39;virus&#39;])
        print("传播途径：" + json_str[&#39;data&#39;][&#39;statistics&#39;][&#39;passWay&#39;])
        print(json_str[&#39;data&#39;][&#39;statistics&#39;][&#39;remark1&#39;])
        print(json_str[&#39;data&#39;][&#39;statistics&#39;][&#39;remark2&#39;] + "\n")
            
        print("==================================国内情况==================================")
        print()
        
        json_provinces = re.findall("{\"provinceName\":(.*?)]}", str(soup))

        idx = 0
        for province in json_provinces:
            if is_json(province):
                pass

            else:
                province = "{\"provinceName\":" + province + "]}"
                province = json.loads(province)
                
            province_name = province[&#39;provinceShortName&#39;] if province[&#39;provinceShortName&#39;] != 0 else &#39;-&#39;
            confirmed = province[&#39;confirmedCount&#39;] if province[&#39;confirmedCount&#39;] != 0 else &#39;-&#39;
            suspected = province[&#39;suspectedCount&#39;] if province[&#39;suspectedCount&#39;] != 0 else &#39;-&#39;
            cured = province[&#39;curedCount&#39;] if province[&#39;curedCount&#39;] != 0 else &#39;-&#39;
            dead = province[&#39;deadCount&#39;] if province[&#39;deadCount&#39;] != 0 else &#39;-&#39;
            table.add_row([province_name, confirmed, dead, cured])
            map[province_name] = idx
            idx = idx + 1
            for city in province[&#39;cities&#39;]:
                provinces_idx[map[province_name]][city[&#39;cityName&#39;]] = [city[&#39;confirmedCount&#39;], city[&#39;deadCount&#39;], city[&#39;curedCount&#39;]]

        print(table)
        
        
        print()
        print("==================================国外情况==================================")
        print()

        json_provinces = str(re.findall("\"id\":949(.*?)]}", str(soup)))
        json_provinces = json_provinces[:1] + "{\"id\":949" + json_provinces[2:]
        json_provinces = json_provinces[:len(json_provinces) - 2] + json_provinces[len(json_provinces) - 1:]
        provinces = json.loads(json_provinces)

        table = PrettyTable([&#39;地区&#39;, &#39;确诊&#39;, &#39;死亡&#39;, &#39;治愈&#39;])
        for province in provinces:
            confirmed = province[&#39;confirmedCount&#39;] if province[&#39;confirmedCount&#39;] != 0 else &#39;-&#39;
            dead = province[&#39;deadCount&#39;] if province[&#39;deadCount&#39;] != 0 else &#39;-&#39;
            cured = province[&#39;curedCount&#39;] if province[&#39;curedCount&#39;] != 0 else &#39;-&#39;
            table.add_row([province[&#39;provinceName&#39;], confirmed, dead, cured])
        
        print(table)
        print()
        
        print("==================================最新消息==================================")
        print()
        
            
        idx = 0
        for news in json_str[&#39;data&#39;][&#39;timeline&#39;]:
            if idx == 5:
                break
            print(news[&#39;pubDateStr&#39;] + "  " + news[&#39;title&#39;])
            idx = idx + 1
        

        print()
        key = input("请输入您想查询详细信息的省份，例如 湖北\n")
        print()
        if key in map.keys():
            query(provinces_idx[map[key]])
        else:
            print("暂无相关信息")
            
        print("\n欢迎提出各种意见")
    except:
        print("连接失败")

if __name__ == &#39;__main__&#39;:
    main()
    sleep(30)

ログイン後にコピー

最後に、皆さんがあらゆる種類の毒に免疫があることを願っています。さあ、中国！！私たちは必ずこれを乗り越えます！！

以上がPythonリアルタイムデータ収集 - 新型コロナウイルスの詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。