Skip to content

Instantly share code, notes, and snippets.

@timercrack
Created August 16, 2016 09:41
Show Gist options
  • Save timercrack/92ee16aa3be53be190d082dab42dfb1c to your computer and use it in GitHub Desktop.
Save timercrack/92ee16aa3be53be190d082dab42dfb1c to your computer and use it in GitHub Desktop.
获取近5年的法定节假日和休息日,使用百度公共API
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!')
@timercrack
Copy link
Author

python3.5测试通过。

@marvinxu-free
Copy link

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment