Spaces:
Sleeping
Sleeping
| """ | |
| ClickUp Data Source โ ู ุญุงูุงุฉ ClickUp API. | |
| ููุฑุฃ data/clickup_tasks.csv ูููุฑุฌุนู ุจููุณ ุงูุดูู ุงูุฐู ุณููุฑุฌุนู ClickUp API ุงูุญููููุ | |
| ูุจููุณ ุงูู interface ุจุงูุถุจุท. ููู ุงูุฅูุชุงุฌ: ููุชุจ ClickUpAPISource ุจููุณ ุฏุงูุฉ | |
| fetch_tasks() ุงูุชู ุชุณุชุฏุนู GET /api/v2/list/{list_id}/task โ ูุงูู Sync Engine ูู | |
| logic.py ูุง ูุชุบูุฑ ุญุฑููุง ูุงุญุฏูุง. | |
| ุฎุฑูุทุฉ ุงูุญููู (ุงูู ุญุงูุงุฉ โ ClickUp API): | |
| task_id โ task["id"] | |
| task_name โ task["name"] | |
| client โ task["list"]["name"] (ุงูู List = ุงูุนู ูู) | |
| assignee_id โ task["assignees"][0]["id"] | |
| status โ task["status"]["status"] | |
| priority โ task["priority"]["priority"] | |
| task_value โ custom_field "Task Value" (ุงูุญู ุงูุฉ ู ู ุชูุณูู ุงูู ูุงู ) | |
| due_date โ task["due_date"] (ms epoch โ date) | |
| completed_at โ task["date_done"] (ms epoch โ date) | |
| quality_rating โ custom_field "Quality Rating" | |
| client_feedback โ custom_field "Client Feedback" | |
| """ | |
| from __future__ import annotations | |
| import os | |
| from abc import ABC, abstractmethod | |
| import pandas as pd | |
| BASE_DIR = os.path.dirname(os.path.abspath(__file__)) | |
| TASKS_FILE = os.path.join(BASE_DIR, "data", "clickup_tasks.csv") | |
| TASK_COLUMNS = [ | |
| "task_id", "task_name", "client", "assignee_id", "status", "priority", | |
| "task_value", "due_date", "completed_at", "quality_rating", "client_feedback", | |
| ] | |
| class TaskSource(ABC): | |
| """ุงููุงุฌูุฉ ุงูุชู ูุนุชู ุฏ ุนูููุง ุงูู Sync Engine.""" | |
| def fetch_tasks(self) -> pd.DataFrame: | |
| """ููุฑุฌุน ูู ุงูู ูุงู ุจุฃุนู ุฏุฉ TASK_COLUMNS.""" | |
| class CSVClickUpSource(TaskSource): | |
| """ู ุญุงูุงุฉ: ุงูู ุตุฏุฑ ู ูู CSV ุจุฏู ุงุณุชุฏุนุงุก ุงูู API.""" | |
| def __init__(self, path: str = TASKS_FILE): | |
| self.path = path | |
| def fetch_tasks(self) -> pd.DataFrame: | |
| df = pd.read_csv(self.path) | |
| # ููุณ ุงูุชุทุจูุน ุงูุฐู ุณููุฒู ู ุน ุงูู API ุงูุญูููู (ms epoch โ datetime) | |
| df["due_date"] = pd.to_datetime(df["due_date"], errors="coerce") | |
| df["completed_at"] = pd.to_datetime(df["completed_at"], errors="coerce") | |
| df["task_value"] = pd.to_numeric(df["task_value"], errors="coerce").fillna(0).astype(int) | |
| df["quality_rating"] = pd.to_numeric(df["quality_rating"], errors="coerce") | |
| return df | |
| def fetch_tasks_for(self, employee_id: str) -> pd.DataFrame: | |
| df = self.fetch_tasks() | |
| return df[df["assignee_id"] == employee_id] | |
| # --------------------------------------------------------------------------- | |
| # ูุณุฎุฉ ุงูุฅูุชุงุฌ โ ุชููุนููู ููู ุงูุฑุจุท ุงูุญููููุ ุจููุณ ุงูู interface ุชู ุงู ูุง. | |
| # | |
| # class ClickUpAPISource(TaskSource): | |
| # def __init__(self, api_token: str, list_ids: list[str]): | |
| # self.api_token = api_token | |
| # self.list_ids = list_ids | |
| # | |
| # def fetch_tasks(self) -> pd.DataFrame: | |
| # rows = [] | |
| # for list_id in self.list_ids: | |
| # r = requests.get( | |
| # f"https://api.clickup.com/api/v2/list/{list_id}/task", | |
| # headers={"Authorization": self.api_token}, | |
| # params={"include_closed": "true"}, | |
| # ) | |
| # for t in r.json()["tasks"]: | |
| # rows.append(_map_api_task(t)) | |
| # return pd.DataFrame(rows, columns=TASK_COLUMNS) | |
| # --------------------------------------------------------------------------- | |
| def get_task_source() -> TaskSource: | |
| """ููุทุฉ ุงูุชุจุฏูู ุงููุญูุฏุฉ: return ClickUpAPISource(token, lists) ูู ุงูุฅูุชุงุฌ.""" | |
| return CSVClickUpSource() | |