Created
January 30, 2024 02:44
-
-
Save taikomatsu/b2ddcf3eab78da88ceb04d623be44ee9 to your computer and use it in GitHub Desktop.
Nuke license usage is reflected in Notion's Database.
This file contains 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 requests | |
import json | |
import os | |
from socket import gethostname | |
import datetime | |
import psutil | |
NOTION_API_KEY = 'your notion api key' | |
DATABASE_ID = 'your database id' | |
NUM_LICENSES = 5 | |
STAT_ACTIVE = '使用中' | |
STAT_INACTIVE = '不使用' | |
headers = { | |
'Accept': 'application/json', | |
'Notion-Version': '2022-06-28', | |
'Authorization': 'Bearer '+ NOTION_API_KEY | |
} | |
db_patch_url = f'https://api.notion.com/v1/databases/{DATABASE_ID}' | |
db_query_url = f'{db_patch_url}/query' | |
def get_active_users_data(): | |
data = { | |
'filter': { | |
'property': 'Status', | |
'status': { | |
'equals': STAT_ACTIVE | |
} | |
} | |
} | |
response = requests.post(db_query_url, headers=headers, json=data) | |
data = response.json() | |
return data | |
def get_active_user_name(data): | |
names = [] | |
for o in data['results']: | |
names.append(o['properties']['Name']['title'][0]['text']['content']) | |
return names | |
def _get_all_process_name(): | |
pnames = [p.info["name"] for p in psutil.process_iter(attrs=["pid", "name"])] | |
return sorted(pnames) | |
def _get_nuke_process_name(): | |
import nuke | |
ver = nuke.NUKE_VERSION_STRING.split('v')[0] | |
return f'Nuke{ver}.exe' | |
def _check_still_nuke_run(): | |
pnames = get_all_process_name() | |
pname = get_nuke_process_name() | |
nprocesses = pnames.count(pname) | |
return nprocesses>1 | |
def get_num_active_license(data): | |
return len(data['results']) | |
def check_available_license(): | |
data = get_active_users_data() | |
active_users = get_active_user_name(data) | |
active = get_username() in active_users | |
nactives = get_num_active_license(data) | |
if active: | |
nactives -= 1 | |
if nactives>=NUM_LICENSES: | |
raise Exception('No licenses available.') | |
def get_self_data(): | |
data = { | |
'filter': { | |
'property': 'Name', | |
'rich_text': { | |
'equals': get_username() | |
} | |
} | |
} | |
response = requests.post(db_query_url, headers=headers, json=data) | |
data = response.json() | |
return data | |
def get_username(): | |
return os.environ['USERNAME'] | |
def get_current_time(): | |
return datetime.datetime.now().strftime('%Y-%m-%dT%H:%M:%S.000+09:00') | |
def get_page_id(data, index=0): | |
return data['results'][index]['id'] | |
def get_page_url(page_id): | |
return f'https://api.notion.com/v1/pages/{page_id}' | |
def modify_status(status, _data={}): | |
data = { | |
'Status': { | |
'status': { | |
'name': status | |
} | |
} | |
} | |
return {**_data, **data} | |
def modify_hostname(hostname, _data={}): | |
data = { | |
'Host Name': { | |
'type': 'rich_text', | |
'rich_text': [ | |
{ | |
'type': 'text', | |
'text': { | |
'content': hostname | |
} | |
} | |
] | |
} | |
} | |
return {**_data, **data} | |
def modify_date(current_time, _data={}): | |
data = { | |
'Status Updated': { | |
'date': { | |
'start': current_time | |
} | |
} | |
} | |
return {**_data, **data} | |
def check_update_error(data): | |
if data['object']=='error': | |
msg = data['code'] + ': ' + data['message'] | |
raise Exception(msg) | |
def update_page(url, data): | |
response = requests.patch(url, headers=headers, json=data) | |
check_update_error(response.json()) | |
def start_nuke(): | |
check_available_license() | |
data = get_self_data() | |
page_id = get_page_id(data) | |
page_url = get_page_url(page_id) | |
data = modify_status(STAT_ACTIVE) | |
data = modify_hostname(gethostname(), data) | |
data = modify_date(get_current_time(), data) | |
update_page(page_url, {'properties': data}) | |
def close_nuke(): | |
data = get_self_data() | |
page_id = get_page_id(data) | |
page_url = get_page_url(page_id) | |
stat = STAT_INACTIVE | |
data = modify_status(stat) | |
data = modify_hostname(gethostname(), data) | |
data = modify_date(get_current_time(), data) | |
update_page(page_url, {'properties': data}) | |
def main(): | |
check_available_license() | |
data = get_self_data() | |
page_id = get_page_id(data) | |
page_url = get_page_url(page_id) | |
data = modify_status(STAT_ACTIVE) | |
data = modify_hostname(gethostname(), data) | |
data = modify_date(get_current_time(), data) | |
update_page(page_url, {'properties': data}) | |
print('# Done') | |
if __name__ == '__main__': | |
main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment