Merge pull request #27 from cloudnativedaysjp/feature/cicd-base

CIのベースを作成

Author: tkc66-buzz

.github/workflows/ci.yml

@@ -0,0 +1,39 @@
+name: ci
+
+on:
+  push:
+    branches: "*"
+  pull_request:
+    branches: [master]
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ["3.11"]
+
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v2
+        with:
+          python-version: ${{ matrix.python-version }}
+      - name: Install poetry
+        run: |
+          curl -sSL https://raw.githubusercontent.com/python-poetry/poetry/master/install-poetry.py | python -
+          echo "$HOME/.poetry/bin" >> $GITHUB_PATH
+      - name: Poetry Version
+        run: |
+          ls -la
+          poetry --version
+      - name: Poetry Install Dependencies
+        run: |
+          poetry install --no-interaction
+      - name: Test with pytest
+        run: |
+          poetry run pytest
+      - name: Lint with flake8
+        run: |
+          poetry run flake8 --show-source --exit-zero --statistics --config setup.cfg

cndctl/Dreamkast.py

@@ -1,8 +1,6 @@
-import logging
-
-logger = logging.getLogger(__name__)
 import datetime
 import json
+import logging
 import os
 import sys
 
@@ -11,6 +9,8 @@
 
 from .Cli import Cli
 
+logger = logging.getLogger(__name__)
+
 
 class Dreamkast:
     def __init__(
@@ -88,15 +88,15 @@ def update(self):
     def talks(self):
         logger.debug("dreamkast_update()")
 
-    def put_upload_url(talkid, upload_url, token):
+    def put_upload_url(self, talkid, upload_url, token):
         req_url = "https://event.cloudnativedays.jp/api/v1/talks/{}/video_registration".format(
             talkid
         )
         headers = {"Authorization": "Bearer {}".format(token)}
         data = {"url": ""}
         data["url"] = upload_url
 
-        res = requests.put(req_url, headers=headers, data=json.dumps(data))
+        requests.put(req_url, headers=headers, data=json.dumps(data))
 
     def set_video_registration(self, talkid: int, video_drop_url: str) -> bool:
         """talkごとのアップロードURLをセットします
@@ -133,7 +133,7 @@ def __request_dk_api(
             dict: レスポンスボディを返します
         """
         if not self.__check_dk_env(env_file_path=".dk.env"):
-            logging.error(f"failed check env file. Please type 'dk update'")
+            logging.error("failed check env file. Please type 'dk update'")
             return {}
 
         token = self.__read_token(env_file_path=".dk.env")
@@ -177,12 +177,14 @@ def get_track_talks(self, track_name: str, event_date: str) -> list:
         talks = self.get_talks_in_track_and_event_date(track_name, event_date)
         for talk in talks:
             talk["start_at"] = datetime.datetime.fromisoformat(
-                talk["actualStartTime"]).time()
+                talk["actualStartTime"]
+            ).time()
             talk["end_at"] = datetime.datetime.fromisoformat(
-                talk["actualEndTime"]).time()
+                talk["actualEndTime"]
+            ).time()
             talk["duration"] = datetime.datetime.fromisoformat(
-                talk["actualEndTime"]) - datetime.datetime.fromisoformat(
-                    talk["actualStartTime"])
+                talk["actualEndTime"]
+            ) - datetime.datetime.fromisoformat(talk["actualStartTime"])
 
         return sorted(talks, key=lambda x: x["start_at"])
 
@@ -202,7 +204,7 @@ def get_track_talks_cmd(self, track_name: str, event_date: str) -> None:
 
     def onair_next_cmd(self, track_name: str, event_date: str):
         if event_date == "":
-            event_date = datetime.datetime.now().strftime('%Y-%m-%d')
+            event_date = datetime.datetime.now().strftime("%Y-%m-%d")
 
         if self.onair_next(track_name, event_date):
             sys.exit(0)
@@ -211,11 +213,15 @@ def onair_next_cmd(self, track_name: str, event_date: str):
 
     def onair_next(self, track_name: str, event_date: str):
         if event_date == "":
-            event_date = datetime.datetime.now().strftime('%Y-%m-%d')
+            event_date = datetime.datetime.now().strftime("%Y-%m-%d")
 
         talks = self.get_talks_in_track_and_event_date(track_name, event_date)
         if not talks:
-            logging.error("Could not get Talks. Request config: DATE='%s', TRACK='%s'", event_date, track_name)
+            logging.error(
+                "Could not get Talks. Request config: DATE='%s', TRACK='%s'",
+                event_date,
+                track_name,
+            )
             sys.exit()
 
         for talk in talks:
@@ -234,7 +240,9 @@ def onair_next(self, track_name: str, event_date: str):
         current_onair_talk_id = self.get_current_onair_talk(track_id)["id"]
         # 現在OnAirなTalkが存在しない
         if current_onair_talk_id == 0:
-            logging.error("There is no on-air talk. Please make a talk on-air and then execute it.")
+            logging.error(
+                "There is no on-air talk. Please make a talk on-air and then execute it."
+            )
             return False
 
         sorted_talks = sorted(talks, key=lambda x: x["start_at"])
@@ -246,7 +254,10 @@ def onair_next(self, track_name: str, event_date: str):
                     return True
 
         # OnAirなTalkは存在するが対象の event_date と track に存在しない
-        logging.warning("There is no OnAir Talk for the specified `event_date` and `track` combination. The currently OnAir Talks are: %s", self.get_talk(current_onair_talk_id))
+        logging.warning(
+            "There is no OnAir Talk for the specified `event_date` and `track` combination. The currently OnAir Talks are: %s",
+            self.get_talk(current_onair_talk_id),
+        )
         return False
 
     def onair(self, dk_talk_id):
@@ -274,7 +285,7 @@ def onair(self, dk_talk_id):
         path = f"/talks/{dk_talk_id}"
         data = {"on_air": True}
 
-        print(self.__request_dk_api(path, "put", data)["message"])
+        print(self.__request_dk_api(path, "put", data))
 
     def get_track(self):
         logger.debug("get_track_name()")
@@ -432,7 +443,7 @@ def get_talks_in_track_and_event_date(
         day_id_this_day = 0
         for day_id in day_ids:
             if day_id["date"] == event_date:
-                day_id_this_day = day_id['id']
+                day_id_this_day = day_id["id"]
 
         talks = self.get_talks(day_id_this_day)
         tracks = self.get_track()
@@ -441,9 +452,7 @@ def get_talks_in_track_and_event_date(
             if track["name"] == track_name:
                 track_id = track["id"]
 
-        talks_for_track = [
-            talk for talk in talks if talk["trackId"] == track_id
-        ]
+        talks_for_track = [talk for talk in talks if talk["trackId"] == track_id]
         return talks_for_track
 
     def get_track_id(self, want_track_name: str, tracks: list) -> int:

cndctl/MediaSource.py

@@ -1,10 +1,10 @@
-import logging
-
-logger = logging.getLogger(__name__)
 import datetime
+import logging
 
 import simpleobsws
 
+logger = logging.getLogger(__name__)
+
 
 class MediaSource:
     def __init__(self, ws) -> None:

cndctl/Nextcloud.py

@@ -1,17 +1,17 @@
-import logging
-
-logger = logging.getLogger(__name__)
-
 import csv
 import json
+import logging
 import os
 import sys
 
 import requests
 import urllib3
+from nextcloud import NextCloud
+
+logger = logging.getLogger(__name__)
+
 
 urllib3.disable_warnings()
-from nextcloud import NextCloud
 
 
 class Nextcloud:
@@ -41,7 +41,7 @@ def put_upload_url(talkid, upload_url, token):
         data = {"url": ""}
         data["url"] = upload_url
 
-        res = requests.put(req_url, headers=headers, data=json.dumps(data))
+        requests.put(req_url, headers=headers, data=json.dumps(data))
 
     def dirsync(self):
         print("uploader dirsync command can not use!")
@@ -67,7 +67,7 @@ def dirsync(self):
             session_kwargs={"verify": False},
         ) as nxc:
             nxc_ping = nxc.get_connection_issues()
-            if not nxc_ping is None:
+            if nxc_ping is not None:
                 logging.error("Nextcloud connection failed! msg: %s", nxc_ping)
                 sys.exit()
 

cndctl/Operator.py

@@ -1,6 +1,7 @@
 import datetime
 import sys
 
+
 class Operator:
     def __init__(self, dreamkast, loop, scene):
         self.dk = dreamkast
@@ -9,7 +10,7 @@ def __init__(self, dreamkast, loop, scene):
 
     def next_cmd(self, track_name: str, event_date: str):
         if event_date == "":
-            event_date = datetime.datetime.now().strftime('%Y-%m-%d')
+            event_date = datetime.datetime.now().strftime("%Y-%m-%d")
 
         print("\n||| CURRENT SWICHER SCENES |||")
         self.loop.run_until_complete(self.scene.get())
@@ -19,20 +20,20 @@ def next_cmd(self, track_name: str, event_date: str):
 
         print("||| CURRENT TRACK TALKS |||")
         self.dk.get_track_talks_cmd(track_name, event_date)
-        
+
         print("\n")
         if not self.dk.onair_next(track_name, event_date):
             sys.exit(1)
 
     def now_cmd(self, track_name: str, event_date: str):
         if event_date == "":
-            event_date = datetime.datetime.now().strftime('%Y-%m-%d')
+            event_date = datetime.datetime.now().strftime("%Y-%m-%d")
 
         tracks = self.dk.get_track()
         for track in tracks:
-            if track['name'] == track_name:
-                track_id = track['id']
+            if track["name"] == track_name:
+                track_id = track["id"]
         current_talk = self.dk.get_current_onair_talk(track_id)
         current_scene = self.loop.run_until_complete(self.scene.current())
         print(f"talk  : {current_talk['id']}_{current_talk['title']}")
-        print(f"scene : {current_scene}")
+        print(f"scene : {current_scene}")

cndctl/Scene.py

@@ -1,12 +1,12 @@
 import logging
-
-logger = logging.getLogger(__name__)
 import sys
 
 import simpleobsws
 
 from .Cli import Cli
 
+logger = logging.getLogger(__name__)
+
 
 class Scene:
     def __init__(self, ws) -> None:
@@ -46,7 +46,7 @@ async def current(self):
             logger.error("Request error. Request:%s Response:%s", request, ret)
             sys.exit()
 
-        return ret.responseData['currentProgramSceneName']
+        return ret.responseData["currentProgramSceneName"]
 
     # cndctl scene next
     async def next(self):

cndctl/Switcher.py

@@ -1,3 +1,4 @@
+import datetime
 import json
 import logging
 
@@ -6,10 +7,7 @@
 from .Dreamkast import Dreamkast
 
 logger = logging.getLogger(__name__)
-import datetime
-import logging
 
-logger = logging.getLogger(__name__)
 logging.basicConfig(
     level=logging.DEBUG, format="%(asctime)s [%(levelname)s] [%(name)s] %(message)s"
 )

cndctl/__init__.py

@@ -1,4 +1,4 @@
-if __name__ == '__main__':
+if __name__ == "__main__":
     from .cndctl import run
 
-    run()
+    run()

cndctl/__main__.py

@@ -1,13 +1,7 @@
-import logging
-
-logger = logging.getLogger(__name__)
-logging.basicConfig(
-    level=logging.WARNING, format="%(asctime)s [%(levelname)s] %(message)s"
-)
-
 import argparse
 import asyncio
 import json
+import logging
 import os
 import sys
 
@@ -16,10 +10,15 @@
 from .Dreamkast import Dreamkast
 from .MediaSource import MediaSource
 from .Nextcloud import Nextcloud
+from .Operator import Operator
 from .Scene import Scene
 from .Source import Source
 from .Switcher import Switcher
-from .Operator import Operator
+
+logger = logging.getLogger(__name__)
+logging.basicConfig(
+    level=logging.WARNING, format="%(asctime)s [%(levelname)s] %(message)s"
+)
 
 parser = argparse.ArgumentParser(description="obs remote controll cli tool")
 parser.add_argument("object")
@@ -47,7 +46,7 @@
 OBS_HOST = ""
 OBS_PORT = ""
 OBS_PASS = ""
-JSON_FILE_PATH =""
+JSON_FILE_PATH = ""
 DK_URL = ""
 DK_CLIENT_ID = ""
 DK_CLIENT_SECRETS = ""
@@ -306,8 +305,7 @@ def run():
                 sys.exit()
             op.now_cmd(EVENT_TRACK, EVENT_DATE)
             sys.exit()
-            
 
     else:
         print(f"undefined command: {args}")
-        sys.exit(1)
+        sys.exit(1)