Initial public repository push

Author: anshajk

.github/workflows/main.yml

@@ -0,0 +1,33 @@
+name: Updater
+
+on:
+  schedule:
+    - cron: "0 18 * * *"
+  workflow_dispatch:
+
+jobs:
+  execute:
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    steps:
+      # Checks-out your repository under $GITHUB_WORKSPACE, so your job can access it
+      - uses: actions/checkout@v2
+      - name: Set up Python
+        uses: actions/[email protected]
+        with:
+          cache: pip
+          python-version: 3.9
+      - name: Install dependencies
+        run: pip install -r requirements.txt
+
+      - name: Notify discord
+        env:
+          SPECIAL_PROJECT_ID: ${{ secrets.SPECIAL_PROJECT_ID }}
+          DISCORD_WEBHOOK_URL: ${{ secrets.DISCORD_WEBHOOK_URL }}
+          API_KEY: ${{ secrets.API_KEY }}
+          WORKSPACE_ID: ${{ secrets.WORKSPACE_ID }}
+          USER_ID: ${{ secrets.USER_ID }}
+          SPECIAL_PROJECT_NAME: ${{ secrets.SPECIAL_PROJECT_NAME }}
+          USER_NAME: ${{ secrets.USER_NAME }}
+        run: |
+          python productivity_update.py

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2022 Anshaj Khare
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1 @@
+# clockify-bot

config.py

@@ -0,0 +1,11 @@
+import os
+
+
+class Config(object):
+    special_project_id = os.getenv("SPECIAL_PROJECT_ID")
+    discord_webhook_url = os.getenv("DISCORD_WEBHOOK_URL")
+    api_key = os.getenv("API_KEY")
+    workspace_id = os.getenv("WORKSPACE_ID")
+    user_id = os.getenv("USER_ID")
+    special_project_name = os.getenv("SPECIAL_PROJECT_NAME")
+    user_name = os.getenv("USER_NAME")

productivity_update.py

@@ -0,0 +1,98 @@
+from typing import Tuple
+
+import pandas as pd
+import requests
+
+from config import Config
+from report_api import ReportApi
+from time_entries_api import TimeEntriesApi
+
+
+class ProductivityUpdate:
+    def __init__(self):
+        self.report = ReportApi()
+
+    def _get_daily_message(self) -> str:
+        message = ""
+        time_api = TimeEntriesApi()
+        recent_entries = time_api.get_recent_entries()
+        entries = time_api.get_todays_entries(recent_entries)
+        if not entries:
+            message += self._get_no_work_today_message()
+        else:
+            summary_df = self.generate_summary_df(entries)
+            total_hours, total_minutes = self.get_total_time(summary_df)
+            prep_hours, prep_minutes = self.get_special_project_time(summary_df)
+            message += self._get_work_done_today_message(
+                total_hours, total_minutes, prep_hours, prep_minutes
+            )
+        return message
+
+    def _get_work_done_today_message(
+        self, total_hours: int, total_minutes: int, prep_hours: int, prep_minutes: int
+    ) -> str:
+
+        return "🕰  Daily stats - Total time: {total_hours} hours and {total_minutes} minutes. Prep time: {prep_hours} hours and {prep_minutes} minutes".format(
+            total_hours=total_hours,
+            total_minutes=total_minutes,
+            prep_hours=prep_hours,
+            prep_minutes=prep_minutes,
+        )
+
+    def _get_no_work_today_message(self) -> str:
+        return "🕰 {} did not work today".format(Config.user_name)
+
+    def generate_message(self) -> str:
+        message = "Updates for {}\n".format(Config.user_name)
+        message += self._get_daily_message()
+        weekly_message = ReportApi().report("weekly")
+        message += "\n" + weekly_message
+        return message
+
+    def generate_summary_df(self, entries: list):
+        time_api = TimeEntriesApi()
+        projects = time_api.get_projects()
+        entry_df = time_api.get_entries_df(entries)
+        projects_df = time_api.get_projects_df(projects)
+        entry_df["hours"] = entry_df["end"] - entry_df["start"]
+        entry_df = entry_df.groupby("project_id").agg({"hours": sum}).reset_index()
+        summary_df = pd.merge(entry_df, projects_df, on="project_id")
+        return summary_df
+
+    def get_total_time(self, summary_df: pd.DataFrame) -> Tuple[int, int]:
+        total_time = summary_df["hours"].sum().seconds
+        total_hours = total_time // (60 * 60)
+        total_minutes = (total_time - total_hours * 60 * 60) // 60
+        return total_hours, total_minutes
+
+    def get_special_project_time(self, summary_df: pd.DataFrame):
+        if Config.special_project_name in summary_df["name"].tolist():
+            prep_time = (
+                summary_df[summary_df["name"] == Config.special_project_name]["hours"]
+                .tolist()[0]
+                .seconds
+            )
+            prep_hours = prep_time // (60 * 60)
+            prep_minutes = (prep_time - prep_hours * 60 * 60) // 60
+            return prep_hours, prep_minutes
+        else:
+            return 0, 0
+
+    def notify(self, message: str) -> int:
+        message_dict = {"content": message}
+
+        resp = requests.post(
+            url=Config.discord_webhook_url,
+            json=message_dict,
+            headers={"Content-Type": "application/json"},
+        )
+        return 200
+
+    def run(self):
+        message = self.generate_message()
+        return self.notify(message)
+
+
+if __name__ == "__main__":
+    p = ProductivityUpdate()
+    p.run()

report_api.py

@@ -0,0 +1,82 @@
+import datetime as dt
+import json
+import logging
+import os
+from config import Config
+from typing import Dict, Tuple
+
+import requests
+
+BASE_API = "https://reports.api.clockify.me/v1"
+
+logger = logging.getLogger("clockify_reports_api")
+
+
+class ReportApi(object):
+    def __init__(self) -> None:
+        api_key = os.getenv("API_KEY")
+        self.workspace_id = os.getenv("WORKSPACE_ID")
+        self.allowed_types = ["weekly"]
+        self.headers = {"X-Api-Key": api_key}
+        self.api_endpoint = (
+            BASE_API
+            + "/workspaces/{workspaceId}/reports/summary".format(
+                workspaceId=self.workspace_id
+            )
+        )
+
+    def report(self, type_: str):
+        if type_ not in self.allowed_types:
+            return
+        if type_ == "weekly":
+            api_response = self._get_weekly_report()
+
+        total_time, prep_time = self._extract_time_values(api_response=api_response)
+
+        report = "⏱  Weekly stats - Total time: {total_time} Prep Time: {prep_time}".format(
+            total_time=self._format_seconds(total_time),
+            prep_time=self._format_seconds(prep_time),
+        )
+        return report
+
+    def _get_weekly_report(self):
+        url = BASE_API + "/workspaces/{workspaceId}/reports/weekly".format(
+            workspaceId=self.workspace_id
+        )
+        utc_now = dt.datetime.utcnow().replace(
+            hour=0, minute=0, second=0, microsecond=0
+        )
+        days_since_week_start = utc_now.weekday()
+        request_json = {
+            "dateRangeStart": (
+                utc_now - dt.timedelta(days=days_since_week_start)
+            ).isoformat(),
+            "dateRangeEnd": (
+                utc_now + dt.timedelta(days=7 - days_since_week_start)
+            ).isoformat(),
+            "weeklyFilter": {"group": "PROJECT", "subgroup": "TIME"},
+        }
+        resp = requests.post(url=url, headers=self.headers, json=request_json)
+        logger.info("Response code: %s" % resp.status_code)
+        return json.loads(resp.text)
+
+    def _extract_time_values(
+        self, api_response: Dict, type_: str = None
+    ) -> Tuple[int, int]:
+        total_time = api_response["totals"][0]["totalTime"]
+        prep_time = 0
+        projects = api_response["groupOne"]
+        for project in projects:
+            if project["_id"] == Config.special_project_id:
+                prep_time = project["duration"]
+        return total_time, prep_time
+
+    def _format_seconds(self, seconds: int):
+        hours = seconds // (60 * 60)
+        minutes = (seconds - hours * 60 * 60) // 60
+        return "{hours} hours {minutes} minutes".format(hours=hours, minutes=minutes)
+
+
+if __name__ == "__main__":
+    c = ReportApi()
+    print(c.report("weekly"))

requirements.txt

@@ -0,0 +1,2 @@
+requests
+pandas

time_entries_api.py

@@ -0,0 +1,67 @@
+import datetime as dt
+import json
+import pandas as pd
+import pytz
+import requests
+
+from config import Config
+
+BASE_API = "https://api.clockify.me/api/v1"
+TIME_ENTRIES_ENDPOINT = "/workspaces/{workspaceId}/user/{userId}/time-entries"
+PROJECTS_ENDPOINT = "/workspaces/{workspaceId}/projects"
+
+
+class TimeEntriesApi(object):
+    def __init__(self) -> None:
+
+        self.headers = {"X-Api-Key": Config.api_key}
+        self.time_api = BASE_API + TIME_ENTRIES_ENDPOINT.format(
+            workspaceId=Config.workspace_id, userId=Config.user_id
+        )
+
+    def get_recent_entries(self) -> list:
+        resp = requests.get(self.time_api, headers=self.headers)
+        entries = json.loads(resp.text)
+        return entries
+
+    def get_projects(self) -> list:
+        project_api = BASE_API + PROJECTS_ENDPOINT.format(
+            workspaceId=Config.workspace_id
+        )
+        resp = requests.get(project_api, headers=self.headers)
+        projects = json.loads(resp.text)
+        return projects
+
+    def get_todays_entries(self, entries: list) -> list:
+        entries_today = []
+        today = dt.datetime.now(pytz.utc).replace(
+            hour=0, minute=0, second=0, microsecond=0
+        )
+        window_start = today - dt.timedelta(hours=5, minutes=30)
+        window_end = today.replace(hour=18, minute=30)
+
+        for entry in entries:
+            time_interval = entry["timeInterval"]
+            if not time_interval["end"]:
+                continue
+            start = pd.to_datetime(time_interval["start"])
+            end = pd.to_datetime(time_interval["end"])
+
+            if start > window_start and end < window_end:
+                entries_today.append(
+                    dict(project_id=entry["projectId"], start=start, end=end)
+                )
+        return entries_today
+
+    def get_projects_df(self, projects: list) -> pd.DataFrame:
+        projects_list = []
+        for project in projects:
+            projects_list.append(dict(project_id=project["id"], name=project["name"]))
+        projects_df = pd.DataFrame(projects_list)
+        return projects_df
+
+    def get_entries_df(self, entries: list) -> pd.DataFrame:
+        entry_df = None
+        if entries:
+            entry_df = pd.DataFrame(entries)
+        return entry_df