-
-
Save blog2i2j/ca2fad6f837a24a48b37b5d828cb5e03 to your computer and use it in GitHub Desktop.
获取近5年的法定节假日和休息日,使用百度公共API
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import datetime | |
import aiohttp | |
import asyncio | |
import tqdm | |
type_name = ['工作日', '休息日', '节假日'] | |
max_conn = asyncio.Semaphore(15) # 最大并发连接数, 默认15 | |
async def fetch(url, day): | |
await max_conn.acquire() | |
async with aiohttp.ClientSession(headers={'apikey': '这里替换成你的百度apikey'}) as session: | |
async with session.get(url) as response: | |
i = int(await response.text()) | |
max_conn.release() | |
if i > 0: | |
return '{},{}\n'.format(day.strftime('%Y-%m-%d'), type_name[i]) | |
async def main(): | |
begin = datetime.datetime.strptime('20100101', '%Y%m%d') # 起始日期 | |
end = datetime.datetime.strptime('20161231', '%Y%m%d') # 结束日期 | |
tasks = [] | |
while begin <= end: | |
url = 'http://apis.baidu.com/xiaogg/holiday/holiday?d={}'.format(begin.strftime('%Y%m%d')) | |
tasks.append(asyncio.ensure_future(fetch(url, begin))) | |
begin += datetime.timedelta(days=1) | |
responses = [] | |
for f in tqdm.tqdm(asyncio.as_completed(tasks), total=len(tasks)): | |
r = await f | |
if r: | |
responses.append(r) | |
return sorted(responses) | |
rst = asyncio.get_event_loop().run_until_complete(main()) | |
with open('holiday.csv', 'w', encoding='utf8') as csvfile: | |
csvfile.writelines(rst) | |
print('\ndone!') |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment