Created
May 28, 2021 07:55
-
-
Save hp0404/5691bc42cfcba439f0b91c896446ac08 to your computer and use it in GitHub Desktop.
kmplus-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
| # -*- coding: utf-8 -*- | |
| import os | |
| import logging | |
| import requests | |
| import pandas as pd | |
| from time import sleep | |
| KEY = os.environ.get("kmplus", "") | |
| KMP = "https://kmplus.ukravtodor.gov.ua/api/v1/kmp" | |
| ROUTE = "https://kmplus.ukravtodor.gov.ua/api/v1/route/id" | |
| def remote_call(endpoint): | |
| """Виконує запит до АРІ.""" | |
| try: | |
| data = requests.get( | |
| url=endpoint, | |
| headers={"Accept-Language": "uk", "Authorization": KEY}, | |
| data={}, | |
| timeout=5, | |
| #verify=False # if SSLCertVerificationError occurs | |
| ) | |
| if data.status_code == 200: | |
| return data.json() | |
| logging.info(f"{'/'.join(endpoint.split('/')[-2:])} resulting in {data.status_code} status code") | |
| return | |
| except KeyboardInterrupt: | |
| raise | |
| except: | |
| logging.exception("Timed out") | |
| return | |
| def process(data): | |
| """Опрацьовує вміст відповіді.""" | |
| yield { | |
| "id": data["detail"]["id"], | |
| "name": data["detail"]["name"], | |
| "sc": data["detail"]["sc"], | |
| "ut": data["detail"]["ut"], | |
| "owner": data["detail"]["owner"], | |
| "kmp": data["detail"]["kmp"], | |
| "latitude": data["detail"]["coord"][0], | |
| "longitude": data["detail"]["coord"][1] | |
| } | |
| def fetch(sc, min_km=0, max_km=224, m_freq=2): | |
| """Завантажує дані з АРІ v1/kmp/ | |
| Parameters | |
| ---------- | |
| sc: str | |
| Індекс дороги | |
| min_km: int | |
| Початок відрізку дороги | |
| max_km: int | |
| Кінець відрізку дороги | |
| m_freq: int | |
| Кількість відрізків в межах одного кілометру потрібно завантажити. | |
| Наприклад, якщо m_freq = 2, то 1000/2 = 500, або [0, 500]; | |
| якщо m_freq = 5, то 1000/5 = 200, або [0, 200, 400, 600, 800] | |
| """ | |
| for km in range(min_km, max_km+1): | |
| for m in range(0, 1000, 1000 // m_freq): | |
| kmp = f"{km}+{str(m).zfill(3)}" | |
| if m == 0: | |
| logging.info(f"{sc} - {kmp}") | |
| response = remote_call(f"{KMP}/{sc}/{kmp}") | |
| if response: | |
| yield from process(response) | |
| if __name__ == "__main__": | |
| logging.basicConfig( | |
| level=logging.INFO, | |
| format="%(asctime)s [%(levelname)s]: %(message)s", | |
| handlers=[ | |
| logging.FileHandler("log.txt"), | |
| logging.StreamHandler() | |
| ] | |
| ) | |
| for road in ( | |
| "T-04-06", | |
| "T-04-28", | |
| "T-05-09", | |
| "T-05-12", | |
| "T-05-13", | |
| "T-05-14", | |
| "T-05-15", | |
| "T-05-16", | |
| "T-05-18", | |
| "T-05-21", | |
| "T-05-23", | |
| "T-05-24", | |
| "T-05-25", | |
| "T-13-02", | |
| "T-13-06", | |
| "T-13-09" | |
| ): | |
| meta = remote_call(f"{ROUTE}/{road}") | |
| if not meta: | |
| continue | |
| start = int(meta["data"]["kmp_begin"].split(".")[0]) | |
| end = int(meta["data"]["kmp_end"].split(".")[0]) | |
| logging.info(f"{road=}, {start=}, {end=}") | |
| data = fetch(sc=road, min_km=start, max_km=end, m_freq=100) | |
| result = pd.DataFrame(data) | |
| result.drop_duplicates("kmp").to_csv(f"{road}.csv", index=False) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment