You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
mipac/compiler/update_api_status.py

164 lines
6.7 KiB

import copy
import hashlib
import json
import sys
from type import OpenAPI
from typing import Any, Literal, TypedDict
sys.path.append("../")
from mipac.utils.util import COLORS # noqa: E402
"""
notSupported: MiPACに実装されていない
supported: MiPACでサポートされている
needToWork: サポートされているが作業が必要
RemovedFromMisskey: Misskeyから削除された
Removed: MiPACから削除されたこれに変更するとremovedから自動で削除されます
"""
STATUS = Literal["notSupported", "supported", "needToWork", 'RemovedFromMisskey', "Removed"]
class IEndpoint(TypedDict):
path: str
status: STATUS
request_body_hash: str | None
response_body_hash: str | None
class ISchema(TypedDict):
name: str
hash: str
status: STATUS
class IData(TypedDict):
endpoints: dict[Literal["support", "removed"], dict[str, IEndpoint]]
schemas: dict[str, ISchema]
def get_sha256_hash(data: dict[str, Any]):
return hashlib.sha256(json.dumps(data, ensure_ascii=False, indent=0).encode('utf-8')).hexdigest()
"""
support: Misskeyに実装されており実装する予定があるまたは実装されている
removed: MiPACでサポートされているがMisskeyから削除された
"""
SECTIONS = Literal["support", "removed"]
with open("./datas/v13_api.json", mode="r", encoding="utf-8") as f:
api: OpenAPI = json.load(f)
with open('./datas/endpoints.json', mode='r', encoding='utf-8') as f:
endpoints: IData= json.load(f)
_endpoints: IData = copy.deepcopy(endpoints)
# パスに関する情報を更新する
for path in api['paths']:
old_data = endpoints["endpoints"]['support'].get(path, None)
current_request_body_hash = get_sha256_hash(api['paths'][path]['post'].get('requestBody', {}))
current_response_body_hash = get_sha256_hash(api['paths'][path]['post'].get('responses', {}))
if old_data is None:
endpoints["endpoints"]['support'][path] = {
'path': path,
'status': "notSupported",
'request_body_hash': get_sha256_hash(api['paths'][path]['post'].get('requestBody', {})),
'response_body_hash': get_sha256_hash(api['paths'][path]['post'].get('responses', {})),
}
else:
# 既存のデータから削除することで残りはremovedにする
del _endpoints["endpoints"]['support'][path]
# ハッシュが変更されているかどうかを確認する
if current_request_body_hash != old_data['request_body_hash']:
print(f"{COLORS.green}[CHANGED: REQUEST] changed request body hash {COLORS.reset} {path} {COLORS.reset}")
endpoints["endpoints"]['support'][path]['request_body_hash'] = current_request_body_hash
if current_response_body_hash != old_data['response_body_hash']:
print(f"{COLORS.green}[CHANGED: RESPONSES] changed responses hash {COLORS.reset} {path} {COLORS.reset}")
endpoints["endpoints"]['support'][path]['response_body_hash'] = current_response_body_hash
# ハッシュが変更されている場合はneedToWorkにする
if current_request_body_hash != old_data['request_body_hash'] or current_response_body_hash != old_data['response_body_hash']:
endpoints["endpoints"]['support'][path]['status'] = "needToWork"
# Misskeyから削除されたエンドポイントをremovedに移動する
for path in _endpoints["endpoints"]['support']:
endpoints["endpoints"]['removed'][path] = _endpoints["endpoints"]['support'][path]
# Misskeyから削除された場合はRemovedFromMisskeyにする
endpoints["endpoints"]['removed'][path]['status'] = "RemovedFromMisskey"
del endpoints["endpoints"]['support'][path]
# MiPACからの削除が完了した場合はremovedから削除する
for path in _endpoints["endpoints"]['removed']:
if endpoints["endpoints"]['removed'][path]['status'] == "Removed":
del endpoints["endpoints"]['removed'][path]
continue
# スキーマに関する情報を更新する
for schema in api['components']['schemas']:
try:
del _endpoints["schemas"][schema]
except KeyError:
pass
old_data = endpoints["schemas"].get(schema, None)
if old_data is None:
endpoints['schemas'][schema] = {
'name': schema,
'hash': get_sha256_hash(api['components']['schemas'][schema]),
'status': "notSupported"
}
else:
current_hash = get_sha256_hash(api['components']['schemas'][schema])
if current_hash != old_data['hash']:
print(f"{COLORS.green}[CHANGED: SCHEMA] changed schema hash {COLORS.reset} {schema} {COLORS.reset}")
endpoints['schemas'][schema]['hash'] = current_hash
endpoints['schemas'][schema]['status'] = "needToWork"
with open('./datas/endpoints.json', mode='w', encoding='utf-8') as f:
json.dump(endpoints, f, ensure_ascii=False, indent=4)
def get_list(data: IData, section: SECTIONS, status: STATUS):
result = ""
for path_name in data['endpoints'][section]:
path = data['endpoints'][section][path_name]
if path['status'] == status:
affix = "(Need to work)" if path['status'] == 'needToWork' else ""
result += f"- [{'x' if path['status'] == 'supported' else ' '}] {path['path']} {affix}\n"
return result
with open('./datas/support_status.md', mode='w', encoding='utf-8') as f:
path_number = len(endpoints['endpoints']['support'])
supported_path_number = len([path for path in endpoints['endpoints']['support'] if endpoints['endpoints']['support'][path]['status'] == 'supported' or endpoints['endpoints']['support'][path]['status'] == 'needToWork'])
supported_endpoints = get_list(endpoints, 'support', 'supported')
not_supported_endpoints = get_list(endpoints, 'support', 'notSupported')
removed_from_misskey_endpoints = get_list(endpoints, 'removed', 'RemovedFromMisskey')
need_to_work_endpoints = get_list(endpoints, 'support', 'needToWork')
support_schemas = ""
for schema_name in endpoints['schemas']:
schema = endpoints['schemas'][schema_name]
affix = "(Need to work)" if schema['status'] == 'needToWork' else ""
support_schemas += f"- [{'x' if schema['status'] == 'supported' else ' '}] {schema['name']} {affix}\n"
f.write(f"""## SUPPORTED ENDPOINTS ({supported_path_number}/{path_number})
{supported_endpoints}
## Not supported endpoints
{"💯" if len(not_supported_endpoints.strip()) == 0 else not_supported_endpoints}
## Changed request body or responses
{"💯" if len(need_to_work_endpoints.strip()) == 0 else need_to_work_endpoints}
## Removed from Misskey
{"💯" if len(removed_from_misskey_endpoints.strip()) == 0 else removed_from_misskey_endpoints}
## SUPPORTED SCHEMAS
{support_schemas}
""")
print("done")