Added TimeTracking generator based on jira activity
This commit is contained in:
Kilian Schuettler
@@ -1,3 +1,5 @@
|
||||
/.env
|
||||
__pycache__/
|
||||
*.py[cod]
|
||||
autostamps.json
|
||||
timestamp_history.json
|
||||
|
||||
@@ -20,17 +20,26 @@ scheduled time in the background.
|
||||
- **Daily timeline** – two-column "Kommen / Gehen" view of every stamp of the day.
|
||||
- **Local persistence** – stamps and scheduled auto-stamps are stored as JSON files
|
||||
in the project directory (`timestamp_history.json`, `autostamps.json`).
|
||||
- **Weekly Jira review** – a separate Streamlit page shows one likely task per day,
|
||||
its Jira activity, and editable module/project/use-case/phase recommendations.
|
||||
- **Excel export** – fills the provided workbook template with every day's complete
|
||||
net stamped time after legal break deductions. Optional Azure OpenAI classification
|
||||
is used when configured; a local deterministic fallback remains available.
|
||||
|
||||
## Project layout
|
||||
|
||||
```
|
||||
stempelbot/
|
||||
app.py # Streamlit UI
|
||||
cli.py # `stempelbot` entry point → runs `streamlit run app.py`
|
||||
Stempelbot.py # Streamlit UI
|
||||
cli.py # `stempelbot` entry point → runs `streamlit run Stempelbot.py`
|
||||
client.py # CoeoClient: login + stamp HTTP calls
|
||||
autostamp.py # AutoStampService: background scheduler
|
||||
stamp_history.py # Local JSON stamp history
|
||||
time_calculator.py # Work/break time calculations
|
||||
jira_client.py # Jira Cloud activity retrieval and normalization
|
||||
weekly_report.py # Weekly aggregation and task recommendations
|
||||
excel_export.py # Template-preserving XLSX export
|
||||
pages/ # Weekly Jira & Excel Streamlit page
|
||||
settings.py # Pydantic settings loaded from .env
|
||||
```
|
||||
|
||||
@@ -73,6 +82,29 @@ STAMPHISTORY_FILE=timestamp_history.json
|
||||
|
||||
# Python logging level, e.g. DEBUG / INFO / WARNING / ERROR
|
||||
LOG_LEVEL=INFO
|
||||
|
||||
# Optional: weekly Jira reporting (Jira Cloud API token authentication)
|
||||
# Use the tenant origin only; browser paths are normalized automatically.
|
||||
JIRA_URL=https://your-company.atlassian.net
|
||||
JIRA_EMAIL=you@example.com
|
||||
JIRA_API_TOKEN=your-jira-api-token
|
||||
# Optional for scoped API tokens; normally discovered automatically after a 401.
|
||||
# JIRA_CLOUD_ID=00000000-0000-0000-0000-000000000000
|
||||
# Optional custom JQL. {start} and {end} are replaced with ISO dates; end is exclusive.
|
||||
# JIRA_JQL=project in (ABC, XYZ) AND updated >= "{start}" AND updated < "{end}"
|
||||
# JIRA_TIMEOUT_SEC=30
|
||||
# JIRA_VERIFY_TLS=true
|
||||
|
||||
# Optional: improve the daily classification with an Azure OpenAI deployment
|
||||
AZURE_OPENAI_ENDPOINT=https://your-resource.openai.azure.com
|
||||
AZURE_OPENAI_API_KEY=your-azure-openai-key
|
||||
AZURE_OPENAI_DEPLOYMENT=gpt-5.6-luna
|
||||
# AZURE_OPENAI_API_VERSION=2025-04-01-preview
|
||||
|
||||
# Optional workbook defaults
|
||||
TIMETRACKING_NAME=Firstname Lastname
|
||||
# TIMETRACKING_TEMPLATE_FILE=C:\path\to\template.xlsx
|
||||
TIMETRACKING_EXPORT_LOCATION=C:\path\to\weekly\exports
|
||||
```
|
||||
|
||||
> ⚠️ Credentials are sent to the coeo portal from your local machine. Keep the
|
||||
@@ -86,7 +118,7 @@ Start the Streamlit app via the Poetry script:
|
||||
poetry run stempelbot
|
||||
```
|
||||
|
||||
This is equivalent to `streamlit run stempelbot/app.py`. Any extra CLI arguments
|
||||
This is equivalent to `streamlit run stempelbot/Stempelbot.py`. Any extra CLI arguments
|
||||
are forwarded to Streamlit, e.g.:
|
||||
|
||||
```powershell
|
||||
@@ -113,6 +145,35 @@ running** for auto-stamps to fire.
|
||||
The **"Heutiger Verlauf"** section lists all of today's stamps split into
|
||||
*Kommen* (even index) and *Gehen* (odd index) columns.
|
||||
|
||||
### Weekly Jira & Excel export
|
||||
|
||||
Open **Weekly Jira & Excel** in Streamlit's page navigation, choose any date in the
|
||||
desired ISO week, and click **Load Jira activity & recommendations**. The page:
|
||||
|
||||
1. loads issues created, assigned, reported, or worklogged by the configured Jira user;
|
||||
2. retains authored comments, changelog entries, worklogs, and relevant assigned updates;
|
||||
3. calculates net time from complete local stamp pairs for each day;
|
||||
4. recommends one workbook classification and one short task summary per day;
|
||||
5. allows every classification and summary to be edited before saving the workbook to
|
||||
`TIMETRACKING_EXPORT_LOCATION`.
|
||||
|
||||
The full net duration for a day is placed on exactly one row, so no stamped minute is
|
||||
split or omitted. Incomplete historical stamp pairs are **not guessed**: the open pair
|
||||
is ignored and visibly flagged for review. Jira and Azure credentials are optional for
|
||||
normal stamping; without Jira, the weekly page still loads local time with editable
|
||||
fallback rows. Azure receives only the selected week's normalized Jira metadata and net
|
||||
minutes, and is skipped entirely when its settings are absent.
|
||||
|
||||
The output is a new in-memory `.xlsx`; the source template is never overwritten.
|
||||
Dropdowns are restored as portable Excel validations after export, and formulas/styles,
|
||||
merged cells, sheet names, and print layout are retained. Formula recalculation is
|
||||
requested when the generated workbook is opened in Excel.
|
||||
|
||||
Both regular and scoped Atlassian API tokens are supported. Regular tokens use the
|
||||
tenant REST URL. If that URL returns HTTP 401, the client discovers the tenant Cloud ID
|
||||
and retries through `https://api.atlassian.com/ex/jira/{cloudId}` as required for scoped
|
||||
tokens. Set `JIRA_CLOUD_ID` only if automatic discovery is unavailable.
|
||||
|
||||
## How stamping works
|
||||
|
||||
`CoeoClient` in `client.py` performs two HTTP POSTs against
|
||||
|
||||
Binary file not shown.
Generated
+36
-9
@@ -1,4 +1,4 @@
|
||||
# This file is automatically @generated by Poetry 2.3.4 and should not be changed by hand.
|
||||
# This file is automatically @generated by Poetry 2.4.1 and should not be changed by hand.
|
||||
|
||||
[[package]]
|
||||
name = "altair"
|
||||
@@ -236,6 +236,18 @@ files = [
|
||||
{file = "colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44"},
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "et-xmlfile"
|
||||
version = "2.0.0"
|
||||
description = "An implementation of lxml.xmlfile for the standard library"
|
||||
optional = false
|
||||
python-versions = ">=3.8"
|
||||
groups = ["main"]
|
||||
files = [
|
||||
{file = "et_xmlfile-2.0.0-py3-none-any.whl", hash = "sha256:7a91720bc756843502c3b7504c77b8fe44217c85c537d85037f0f536151b2caa"},
|
||||
{file = "et_xmlfile-2.0.0.tar.gz", hash = "sha256:dab3f4764309081ce75662649be815c4c9081e88f0837825f90fd28317d4da54"},
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "gitdb"
|
||||
version = "4.0.12"
|
||||
@@ -639,6 +651,21 @@ files = [
|
||||
{file = "numpy-2.4.1.tar.gz", hash = "sha256:a1ceafc5042451a858231588a104093474c6a5c57dcc724841f5c888d237d690"},
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "openpyxl"
|
||||
version = "3.1.5"
|
||||
description = "A Python library to read/write Excel 2010 xlsx/xlsm files"
|
||||
optional = false
|
||||
python-versions = ">=3.8"
|
||||
groups = ["main"]
|
||||
files = [
|
||||
{file = "openpyxl-3.1.5-py2.py3-none-any.whl", hash = "sha256:5282c12b107bffeef825f4617dc029afaf41d0ea60823bbb665ef3079dc79de2"},
|
||||
{file = "openpyxl-3.1.5.tar.gz", hash = "sha256:cf0e3cf56142039133628b5acffe8ef0c12bc902d2aadd3e0fe5878dc08d1050"},
|
||||
]
|
||||
|
||||
[package.dependencies]
|
||||
et-xmlfile = "*"
|
||||
|
||||
[[package]]
|
||||
name = "packaging"
|
||||
version = "26.0"
|
||||
@@ -1264,25 +1291,25 @@ typing-extensions = {version = ">=4.4.0", markers = "python_version < \"3.13\""}
|
||||
|
||||
[[package]]
|
||||
name = "requests"
|
||||
version = "2.32.5"
|
||||
version = "2.34.2"
|
||||
description = "Python HTTP for Humans."
|
||||
optional = false
|
||||
python-versions = ">=3.9"
|
||||
python-versions = ">=3.10"
|
||||
groups = ["main"]
|
||||
files = [
|
||||
{file = "requests-2.32.5-py3-none-any.whl", hash = "sha256:2462f94637a34fd532264295e186976db0f5d453d1cdd31473c85a6a161affb6"},
|
||||
{file = "requests-2.32.5.tar.gz", hash = "sha256:dbba0bac56e100853db0ea71b82b4dfd5fe2bf6d3754a8893c3af500cec7d7cf"},
|
||||
{file = "requests-2.34.2-py3-none-any.whl", hash = "sha256:2a0d60c172f83ac6ab31e4554906c0f3b3588d37b5cb939b1c061f4907e278e0"},
|
||||
{file = "requests-2.34.2.tar.gz", hash = "sha256:f288924cae4e29463698d6d60bc6a4da69c89185ad1e0bcc4104f584e960b9ed"},
|
||||
]
|
||||
|
||||
[package.dependencies]
|
||||
certifi = ">=2017.4.17"
|
||||
certifi = ">=2023.5.7"
|
||||
charset_normalizer = ">=2,<4"
|
||||
idna = ">=2.5,<4"
|
||||
urllib3 = ">=1.21.1,<3"
|
||||
urllib3 = ">=1.26,<3"
|
||||
|
||||
[package.extras]
|
||||
socks = ["PySocks (>=1.5.6,!=1.5.7)"]
|
||||
use-chardet-on-py3 = ["chardet (>=3.0.2,<6)"]
|
||||
use-chardet-on-py3 = ["chardet (>=3.0.2,<8)"]
|
||||
|
||||
[[package]]
|
||||
name = "rpds-py"
|
||||
@@ -1629,4 +1656,4 @@ watchmedo = ["PyYAML (>=3.10)"]
|
||||
[metadata]
|
||||
lock-version = "2.1"
|
||||
python-versions = "^3.12"
|
||||
content-hash = "113ddcce235f83c6b490f723f8337b78145b5fdaf2755ceafa35ad7d54122928"
|
||||
content-hash = "be1a0fad20b040e9145ca6d36f2a918fa1038efcf81d7657614301a0e7326ad2"
|
||||
|
||||
@@ -11,6 +11,8 @@ streamlit = "^1.46"
|
||||
pip-system-certs = "^5.3"
|
||||
pydantic-settings = "^2.12.0"
|
||||
playwright = "^1.61.0"
|
||||
openpyxl = "^3.1.5"
|
||||
requests = "^2.34.2"
|
||||
|
||||
[tool.poetry.scripts]
|
||||
stempelbot = "stempelbot.cli:start"
|
||||
|
||||
@@ -11,7 +11,7 @@ from smarttime_client import get_smarttime_client
|
||||
|
||||
stamps = StampHistory.get_today_stamps()
|
||||
|
||||
# Initialize Logic Class
|
||||
# Initialize logic class
|
||||
logic = TimeLogic(stamps)
|
||||
stats = logic.calculate()
|
||||
autostamp = get_autostamp()
|
||||
@@ -309,3 +309,4 @@ with st.container(border=True):
|
||||
st.session_state.pop("st_snap", None)
|
||||
st.rerun()
|
||||
st.caption(f"Abgefragt: {snap.query_time.strftime('%H:%M:%S')}")
|
||||
|
||||
+2
-2
@@ -5,9 +5,9 @@ import sys
|
||||
|
||||
def start():
|
||||
"""Run the Streamlit application."""
|
||||
# Resolve the absolute path to app.py relative to this file
|
||||
# Resolve the absolute path to Stempelbot.py relative to this file
|
||||
current_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
app_path = os.path.join(current_dir, "app.py")
|
||||
app_path = os.path.join(current_dir, "Stempelbot.py")
|
||||
|
||||
# Run streamlit, passing along any extra command line arguments
|
||||
subprocess.run(["streamlit", "run", app_path] + sys.argv[1:])
|
||||
|
||||
@@ -0,0 +1,153 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import warnings
|
||||
from io import BytesIO
|
||||
from pathlib import Path
|
||||
|
||||
from openpyxl import load_workbook
|
||||
from openpyxl.workbook.defined_name import DefinedName
|
||||
from openpyxl.worksheet.datavalidation import DataValidation
|
||||
|
||||
from weekly_report import DayRecommendation, TemplateOptions, iso_week_bounds
|
||||
|
||||
SHEET_NAME = "weekly time tracking"
|
||||
DROPDOWN_SHEET = "Dropdowns"
|
||||
FIRST_DATA_ROW = 6
|
||||
LAST_DATA_ROW = 14
|
||||
|
||||
|
||||
class ExcelExportError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
def load_template_options(template_file: str | Path) -> TemplateOptions:
|
||||
path = Path(template_file)
|
||||
if not path.is_file():
|
||||
raise ExcelExportError(f"Excel template not found: {path}")
|
||||
with warnings.catch_warnings():
|
||||
warnings.filterwarnings("ignore", message="Data Validation extension is not supported")
|
||||
workbook = load_workbook(path, read_only=True, data_only=False)
|
||||
try:
|
||||
sheet = workbook[DROPDOWN_SHEET]
|
||||
return TemplateOptions(
|
||||
modules=_values(sheet, "F", 3, 14),
|
||||
projects=_values(sheet, "G", 3, 20),
|
||||
use_cases=_values(sheet, "H", 3, 48),
|
||||
phases=_values(sheet, "I", 3, 14),
|
||||
)
|
||||
except KeyError as exc:
|
||||
raise ExcelExportError("The template is missing its Dropdowns sheet.") from exc
|
||||
finally:
|
||||
workbook.close()
|
||||
|
||||
|
||||
def export_weekly_workbook(
|
||||
template_file: str | Path,
|
||||
employee_name: str,
|
||||
year: int,
|
||||
week: int,
|
||||
rows: list[DayRecommendation],
|
||||
) -> bytes:
|
||||
if len(rows) > LAST_DATA_ROW - FIRST_DATA_ROW + 1:
|
||||
raise ExcelExportError("The template has room for at most nine daily rows.")
|
||||
path = Path(template_file)
|
||||
if not path.is_file():
|
||||
raise ExcelExportError(f"Excel template not found: {path}")
|
||||
|
||||
with warnings.catch_warnings():
|
||||
warnings.filterwarnings("ignore", message="Data Validation extension is not supported")
|
||||
workbook = load_workbook(path, data_only=False)
|
||||
try:
|
||||
if SHEET_NAME not in workbook.sheetnames or DROPDOWN_SHEET not in workbook.sheetnames:
|
||||
raise ExcelExportError("The template has an unexpected sheet layout.")
|
||||
sheet = workbook[SHEET_NAME]
|
||||
start, end = iso_week_bounds(year, week)
|
||||
sheet["B3"] = employee_name.strip() or "NAME"
|
||||
sheet["B6"] = f"KW {week}"
|
||||
sheet["C6"] = f"{start:%d.%m.} bis {end:%d.%m.}"
|
||||
|
||||
for row_number in range(FIRST_DATA_ROW, LAST_DATA_ROW + 1):
|
||||
for column in "DEFGHI":
|
||||
sheet[f"{column}{row_number}"] = None
|
||||
|
||||
for row_number, item in enumerate(rows, start=FIRST_DATA_ROW):
|
||||
sheet[f"D{row_number}"] = item.module
|
||||
sheet[f"E{row_number}"] = item.project
|
||||
sheet[f"F{row_number}"] = item.use_case
|
||||
sheet[f"G{row_number}"] = item.phase
|
||||
# Decimal hours are exact to the source minute; formatting controls display only.
|
||||
sheet[f"H{row_number}"] = item.minutes / 60
|
||||
sheet[f"H{row_number}"].number_format = "0.00"
|
||||
note = f"{item.day:%a %d.%m.}: {item.task_summary}"
|
||||
if item.issue_keys:
|
||||
note += f" [{item.issue_keys}]"
|
||||
if item.warning:
|
||||
note += f" — {item.warning}"
|
||||
sheet[f"I{row_number}"] = note
|
||||
|
||||
_restore_validations(workbook)
|
||||
workbook.calculation.fullCalcOnLoad = True
|
||||
workbook.calculation.forceFullCalc = True
|
||||
output = BytesIO()
|
||||
workbook.save(output)
|
||||
return output.getvalue()
|
||||
finally:
|
||||
workbook.close()
|
||||
|
||||
|
||||
def export_filename(employee_name: str, year: int, week: int) -> str:
|
||||
safe_name = re.sub(r"[^\w.-]+", "_", employee_name.strip(), flags=re.UNICODE).strip("_")
|
||||
suffix = f"_{safe_name}" if safe_name else ""
|
||||
return f"TimeTracking{suffix}_{year}_KW{week:02d}.xlsx"
|
||||
|
||||
|
||||
def save_workbook(workbook: bytes, export_location: str | Path, filename: str) -> Path:
|
||||
"""Write workbook bytes to the configured directory and return the saved path."""
|
||||
directory = Path(export_location).expanduser()
|
||||
if Path(filename).name != filename:
|
||||
raise ExcelExportError("The Excel export filename is invalid.")
|
||||
try:
|
||||
directory.mkdir(parents=True, exist_ok=True)
|
||||
if not directory.is_dir():
|
||||
raise ExcelExportError(f"Excel export location is not a directory: {directory}")
|
||||
destination = directory / filename
|
||||
destination.write_bytes(workbook)
|
||||
return destination.resolve()
|
||||
except ExcelExportError:
|
||||
raise
|
||||
except OSError as exc:
|
||||
raise ExcelExportError(f"Could not save Excel export to {directory}: {exc}") from exc
|
||||
|
||||
|
||||
def _values(sheet, column: str, first: int, last: int) -> tuple[str, ...]:
|
||||
return tuple(
|
||||
str(value).strip()
|
||||
for row in range(first, last + 1)
|
||||
if (value := sheet[f"{column}{row}"].value) is not None and str(value).strip()
|
||||
)
|
||||
|
||||
|
||||
def _restore_validations(workbook) -> None:
|
||||
"""Replace unsupported x14 dropdowns with portable named-range validations."""
|
||||
sheet = workbook[SHEET_NAME]
|
||||
sheet.data_validations.dataValidation = []
|
||||
ranges = {
|
||||
"_tt_weeks": ("Dropdowns!$B$3:$B$55", "B6"),
|
||||
"_tt_modules": ("Dropdowns!$F$3:$F$14", "D6:D14"),
|
||||
"_tt_projects": ("Dropdowns!$G$3:$G$20", "E6:E14"),
|
||||
"_tt_use_cases": ("Dropdowns!$H$3:$H$48", "F6:F14"),
|
||||
"_tt_phases": ("Dropdowns!$I$3:$I$14", "G6:G14"),
|
||||
}
|
||||
existing = {name.name for name in workbook.defined_names.values()}
|
||||
for name, (reference, cells) in ranges.items():
|
||||
if name not in existing:
|
||||
workbook.defined_names.add(DefinedName(name, attr_text=reference))
|
||||
validation = DataValidation(type="list", formula1=f"={name}", allow_blank=True)
|
||||
validation.error = "Select a value from the template list."
|
||||
validation.errorTitle = "Invalid time-tracking value"
|
||||
validation.showErrorMessage = True
|
||||
sheet.add_data_validation(validation)
|
||||
validation.add(cells)
|
||||
|
||||
|
||||
@@ -0,0 +1,413 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from datetime import date, datetime, timedelta, timezone
|
||||
from typing import Any
|
||||
from urllib.parse import quote, urlsplit
|
||||
|
||||
import requests
|
||||
from requests.auth import HTTPBasicAuth
|
||||
|
||||
from settings import settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_SENSITIVE_HEADERS = {
|
||||
"authorization",
|
||||
"cookie",
|
||||
"proxy-authorization",
|
||||
"set-cookie",
|
||||
"x-api-key",
|
||||
}
|
||||
_LOG_BODY_LIMIT = 6000
|
||||
|
||||
|
||||
def _safe_headers(headers: Any) -> dict[str, str]:
|
||||
if not headers:
|
||||
return {}
|
||||
return {
|
||||
str(name): "<redacted>" if str(name).casefold() in _SENSITIVE_HEADERS else str(value)
|
||||
for name, value in headers.items()
|
||||
}
|
||||
|
||||
|
||||
def _body_preview(body: Any) -> str:
|
||||
if body is None or body == "":
|
||||
return "<empty response body>"
|
||||
text = str(body)
|
||||
if len(text) <= _LOG_BODY_LIMIT:
|
||||
return text
|
||||
head_size = _LOG_BODY_LIMIT - 1000
|
||||
omitted = len(text) - _LOG_BODY_LIMIT
|
||||
return (
|
||||
f"{text[:head_size]}\n"
|
||||
f"... <{omitted} characters omitted; total response size {len(text)}> ...\n"
|
||||
f"{text[-1000:]}"
|
||||
)
|
||||
|
||||
|
||||
def _normalize_site_url(base_url: str) -> str:
|
||||
value = base_url.strip().rstrip("/")
|
||||
parsed = urlsplit(value)
|
||||
if not parsed.scheme or not parsed.netloc:
|
||||
raise JiraError("JIRA_URL must be an absolute URL, for example https://company.atlassian.net")
|
||||
# Browser URLs under /jira/... serve the SPA HTML, not REST responses. Jira Cloud's
|
||||
# REST API always starts at the tenant origin.
|
||||
if parsed.hostname and parsed.hostname.casefold().endswith(".atlassian.net"):
|
||||
return f"{parsed.scheme}://{parsed.netloc}"
|
||||
return value
|
||||
|
||||
|
||||
def _is_atlassian_cloud(url: str) -> bool:
|
||||
hostname = urlsplit(url).hostname
|
||||
return bool(hostname and hostname.casefold().endswith(".atlassian.net"))
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class JiraActivity:
|
||||
day: date
|
||||
issue_key: str
|
||||
summary: str
|
||||
project: str
|
||||
activity_type: str
|
||||
detail: str
|
||||
url: str
|
||||
|
||||
|
||||
class JiraError(RuntimeError):
|
||||
"""A safe, user-displayable Jira integration error."""
|
||||
|
||||
|
||||
class JiraClient:
|
||||
def __init__(
|
||||
self,
|
||||
base_url: str,
|
||||
email: str,
|
||||
api_token: str,
|
||||
*,
|
||||
timeout: int = 30,
|
||||
verify_tls: bool = True,
|
||||
cloud_id: str | None = None,
|
||||
session: requests.Session | None = None,
|
||||
):
|
||||
self.site_url = _normalize_site_url(base_url)
|
||||
self.cloud_id = cloud_id.strip() if cloud_id else None
|
||||
self.base_url = self._api_base_url()
|
||||
self.timeout = timeout
|
||||
self.verify_tls = verify_tls
|
||||
self.session = session or requests.Session()
|
||||
self.session.auth = HTTPBasicAuth(email, api_token)
|
||||
self.session.headers.update({"Accept": "application/json"})
|
||||
|
||||
def _api_base_url(self) -> str:
|
||||
if self.cloud_id:
|
||||
return f"https://api.atlassian.com/ex/jira/{quote(self.cloud_id, safe='')}"
|
||||
return self.site_url
|
||||
|
||||
def _discover_cloud_id(self) -> str | None:
|
||||
"""Discover Jira Cloud's tenant ID for scoped API-token requests."""
|
||||
if not _is_atlassian_cloud(self.site_url):
|
||||
return None
|
||||
url = f"{self.site_url}/_edge/tenant_info"
|
||||
try:
|
||||
response = self.session.get(
|
||||
url,
|
||||
timeout=self.timeout,
|
||||
verify=self.verify_tls,
|
||||
)
|
||||
response.raise_for_status()
|
||||
payload = response.json()
|
||||
cloud_id = payload.get("cloudId") if isinstance(payload, dict) else None
|
||||
if cloud_id:
|
||||
return str(cloud_id)
|
||||
logger.error(
|
||||
"Atlassian tenant discovery returned no cloudId from %s: %s",
|
||||
url,
|
||||
_body_preview(getattr(response, "text", None)),
|
||||
)
|
||||
except (requests.RequestException, ValueError):
|
||||
logger.exception(
|
||||
"Could not discover the Atlassian Cloud ID from %s", url
|
||||
)
|
||||
return None
|
||||
|
||||
def _get(self, path: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
|
||||
response: requests.Response | None = None
|
||||
try:
|
||||
response = self.session.get(
|
||||
f"{self.base_url}{path}",
|
||||
params=params,
|
||||
timeout=self.timeout,
|
||||
verify=self.verify_tls,
|
||||
)
|
||||
if (
|
||||
response.status_code == 401
|
||||
and not self.cloud_id
|
||||
and (cloud_id := self._discover_cloud_id())
|
||||
):
|
||||
self.cloud_id = cloud_id
|
||||
self.base_url = self._api_base_url()
|
||||
logger.debug(
|
||||
"Using the Atlassian scoped-token API gateway for cloud %s.",
|
||||
cloud_id,
|
||||
)
|
||||
response = self.session.get(
|
||||
f"{self.base_url}{path}",
|
||||
params=params,
|
||||
timeout=self.timeout,
|
||||
verify=self.verify_tls,
|
||||
)
|
||||
response.raise_for_status()
|
||||
payload = response.json()
|
||||
if not isinstance(payload, dict):
|
||||
logger.error(
|
||||
"Jira returned an unexpected JSON payload for GET %s%s:\n%r",
|
||||
self.base_url,
|
||||
path,
|
||||
_body_preview(repr(payload)),
|
||||
)
|
||||
raise JiraError("Jira returned an unexpected response.")
|
||||
return payload
|
||||
except requests.RequestException as exc:
|
||||
failed_response = exc.response if exc.response is not None else response
|
||||
failed_request = getattr(failed_response, "request", None)
|
||||
status = getattr(failed_response, "status_code", None)
|
||||
method = getattr(failed_request, "method", "GET")
|
||||
url = getattr(failed_request, "url", None)
|
||||
if not url:
|
||||
url = f"{self.base_url}{path}"
|
||||
if params:
|
||||
url += f" params={params!r}"
|
||||
reason = getattr(failed_response, "reason", None)
|
||||
body = getattr(failed_response, "text", None)
|
||||
logger.exception(
|
||||
"Jira HTTP request failed\n"
|
||||
" Request: %s %s\n"
|
||||
" Request headers: %r\n"
|
||||
" Status: %s%s\n"
|
||||
" Response headers: %r\n"
|
||||
" Response body:\n%s",
|
||||
method,
|
||||
url,
|
||||
_safe_headers(getattr(failed_request, "headers", None)),
|
||||
status if status is not None else "no response",
|
||||
f" {reason}" if reason else "",
|
||||
_safe_headers(getattr(failed_response, "headers", None)),
|
||||
_body_preview(body),
|
||||
)
|
||||
suffix = f" (HTTP {status})" if status else ""
|
||||
raise JiraError(f"Jira request failed{suffix}.") from exc
|
||||
except ValueError as exc:
|
||||
logger.exception(
|
||||
"Jira returned invalid JSON for GET %s%s\nResponse body:\n%s",
|
||||
self.base_url,
|
||||
path,
|
||||
_body_preview(getattr(response, "text", None)),
|
||||
)
|
||||
raise JiraError("Jira returned invalid JSON.") from exc
|
||||
|
||||
def get_week_activity(
|
||||
self, start: date, end: date, custom_jql: str | None = None
|
||||
) -> list[JiraActivity]:
|
||||
"""Collect relevant issue activity for the inclusive date range."""
|
||||
if end < start:
|
||||
raise ValueError("end must not be before start")
|
||||
|
||||
me = self._get("/rest/api/3/myself")
|
||||
identities = {
|
||||
str(value).casefold()
|
||||
for value in (me.get("accountId"), me.get("emailAddress"), me.get("displayName"))
|
||||
if value
|
||||
}
|
||||
exclusive_end = end + timedelta(days=1)
|
||||
account_id = str(me.get("accountId") or "").replace('"', '\\"')
|
||||
updated_by_clause = (
|
||||
f' OR issuekey in updatedBy("{account_id}", "{start.isoformat()}", '
|
||||
f'"{exclusive_end.isoformat()}")'
|
||||
if account_id
|
||||
else ""
|
||||
)
|
||||
default_jql = (
|
||||
f'updated >= "{start.isoformat()}" AND updated < "{exclusive_end.isoformat()}" '
|
||||
f"AND (assignee = currentUser() OR assignee WAS currentUser() DURING "
|
||||
f'(\"{start.isoformat()}\", \"{exclusive_end.isoformat()}\") '
|
||||
f"OR reporter = currentUser() OR creator = currentUser() "
|
||||
f"OR worklogAuthor = currentUser(){updated_by_clause}) ORDER BY updated ASC"
|
||||
)
|
||||
jql = (custom_jql or default_jql).replace(
|
||||
"{start}", start.isoformat()
|
||||
).replace("{end}", exclusive_end.isoformat())
|
||||
|
||||
issues: list[dict[str, Any]] = []
|
||||
token: str | None = None
|
||||
seen_tokens: set[str] = set()
|
||||
while True:
|
||||
params: dict[str, Any] = {
|
||||
"jql": jql,
|
||||
"fields": "summary,project,created,updated,creator,reporter,assignee,comment,worklog",
|
||||
"expand": "changelog",
|
||||
"maxResults": 100,
|
||||
}
|
||||
if token:
|
||||
params["nextPageToken"] = token
|
||||
page = self._get("/rest/api/3/search/jql", params)
|
||||
page_issues = page.get("issues", [])
|
||||
if not isinstance(page_issues, list):
|
||||
raise JiraError("Jira returned an invalid issue list.")
|
||||
issues.extend(item for item in page_issues if isinstance(item, dict))
|
||||
token = page.get("nextPageToken")
|
||||
if not token or token in seen_tokens or page.get("isLast") is True:
|
||||
break
|
||||
seen_tokens.add(token)
|
||||
|
||||
activities: list[JiraActivity] = []
|
||||
for issue in issues:
|
||||
self._hydrate_truncated_activity(issue)
|
||||
activities.extend(
|
||||
self._normalize_issue(issue, identities, start, exclusive_end)
|
||||
)
|
||||
return sorted(
|
||||
set(activities), key=lambda item: (item.day, item.issue_key, item.activity_type, item.detail)
|
||||
)
|
||||
|
||||
def _hydrate_truncated_activity(self, issue: dict[str, Any]) -> None:
|
||||
"""Fetch activity collections that Jira only partially embeds in search."""
|
||||
key = quote(str(issue.get("key") or ""), safe="")
|
||||
if not key:
|
||||
return
|
||||
fields = issue.setdefault("fields", {})
|
||||
changelog = issue.get("changelog") or {}
|
||||
if _is_truncated(changelog, "histories"):
|
||||
issue["changelog"] = {
|
||||
"histories": self._paginated_values(
|
||||
f"/rest/api/3/issue/{key}/changelog", "values"
|
||||
)
|
||||
}
|
||||
for field, path, collection in (
|
||||
("comment", "comment", "comments"),
|
||||
("worklog", "worklog", "worklogs"),
|
||||
):
|
||||
embedded = fields.get(field) or {}
|
||||
if _is_truncated(embedded, collection):
|
||||
fields[field] = {
|
||||
collection: self._paginated_values(
|
||||
f"/rest/api/3/issue/{key}/{path}", collection
|
||||
)
|
||||
}
|
||||
|
||||
def _paginated_values(self, path: str, collection: str) -> list[dict[str, Any]]:
|
||||
values: list[dict[str, Any]] = []
|
||||
start_at = 0
|
||||
while True:
|
||||
page = self._get(path, {"startAt": start_at, "maxResults": 100})
|
||||
items = page.get(collection, [])
|
||||
if not isinstance(items, list):
|
||||
raise JiraError("Jira returned an invalid activity list.")
|
||||
values.extend(item for item in items if isinstance(item, dict))
|
||||
start_at += len(items)
|
||||
total = int(page.get("total", start_at))
|
||||
if not items or start_at >= total:
|
||||
return values
|
||||
|
||||
def _normalize_issue(
|
||||
self,
|
||||
issue: dict[str, Any],
|
||||
identities: set[str],
|
||||
start: date,
|
||||
exclusive_end: date,
|
||||
) -> list[JiraActivity]:
|
||||
fields = issue.get("fields") or {}
|
||||
key = str(issue.get("key") or "?")
|
||||
summary = str(fields.get("summary") or "(no summary)")
|
||||
project_data = fields.get("project") or {}
|
||||
project = str(project_data.get("name") or project_data.get("key") or "")
|
||||
url = f"{self.site_url}/browse/{key}"
|
||||
result: list[JiraActivity] = []
|
||||
|
||||
def add(timestamp: Any, kind: str, detail: str) -> None:
|
||||
day = _jira_date(timestamp)
|
||||
if day is not None and start <= day < exclusive_end:
|
||||
result.append(JiraActivity(day, key, summary, project, kind, detail, url))
|
||||
|
||||
creator = fields.get("creator") or {}
|
||||
if _is_me(creator, identities):
|
||||
add(fields.get("created"), "created", "Issue created")
|
||||
|
||||
changelog = issue.get("changelog") or {}
|
||||
for history in changelog.get("histories", []) or []:
|
||||
if _is_me(history.get("author") or {}, identities):
|
||||
changes = []
|
||||
for item in history.get("items", []) or []:
|
||||
field = str(item.get("field") or "field")
|
||||
target = item.get("toString")
|
||||
changes.append(f"{field} → {target}" if target else f"updated {field}")
|
||||
add(history.get("created"), "change", ", ".join(changes) or "Issue updated")
|
||||
|
||||
comments = (fields.get("comment") or {}).get("comments", []) or []
|
||||
for comment in comments:
|
||||
if _is_me(comment.get("author") or {}, identities):
|
||||
add(comment.get("created"), "comment", "Comment posted")
|
||||
|
||||
worklogs = (fields.get("worklog") or {}).get("worklogs", []) or []
|
||||
for worklog in worklogs:
|
||||
if _is_me(worklog.get("author") or {}, identities):
|
||||
seconds = int(worklog.get("timeSpentSeconds") or 0)
|
||||
detail = f"Work logged ({seconds // 3600:g}h)" if seconds else "Work logged"
|
||||
add(worklog.get("started") or worklog.get("created"), "worklog", detail)
|
||||
|
||||
if not result and _is_me(fields.get("assignee") or {}, identities):
|
||||
add(fields.get("updated"), "assigned", "Assigned issue updated")
|
||||
return result
|
||||
|
||||
|
||||
def _is_me(user: dict[str, Any], identities: set[str]) -> bool:
|
||||
values = (user.get("accountId"), user.get("emailAddress"), user.get("displayName"))
|
||||
return any(str(value).casefold() in identities for value in values if value)
|
||||
|
||||
|
||||
def _is_truncated(container: dict[str, Any], collection: str) -> bool:
|
||||
values = container.get(collection, []) or []
|
||||
try:
|
||||
return int(container.get("total", len(values))) > len(values)
|
||||
except (TypeError, ValueError):
|
||||
return False
|
||||
|
||||
|
||||
def _jira_date(value: Any) -> date | None:
|
||||
if not value:
|
||||
return None
|
||||
text = str(value).strip().replace("Z", "+00:00")
|
||||
try:
|
||||
parsed = datetime.fromisoformat(text)
|
||||
if parsed.tzinfo is not None:
|
||||
parsed = parsed.astimezone()
|
||||
return parsed.date()
|
||||
except ValueError:
|
||||
try:
|
||||
return datetime.strptime(text[:10], "%Y-%m-%d").replace(tzinfo=timezone.utc).date()
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def get_jira_client() -> JiraClient:
|
||||
missing = [
|
||||
name
|
||||
for name, value in (
|
||||
("JIRA_URL", settings.JIRA_URL),
|
||||
("JIRA_EMAIL", settings.JIRA_EMAIL),
|
||||
("JIRA_API_TOKEN", settings.JIRA_API_TOKEN),
|
||||
)
|
||||
if not value
|
||||
]
|
||||
if missing:
|
||||
raise JiraError(f"Missing Jira configuration: {', '.join(missing)}")
|
||||
return JiraClient(
|
||||
settings.JIRA_URL,
|
||||
settings.JIRA_EMAIL,
|
||||
settings.JIRA_API_TOKEN,
|
||||
timeout=settings.JIRA_TIMEOUT_SEC,
|
||||
verify_tls=settings.JIRA_VERIFY_TLS,
|
||||
cloud_id=settings.JIRA_CLOUD_ID,
|
||||
)
|
||||
|
||||
@@ -0,0 +1,193 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import replace
|
||||
from datetime import date
|
||||
|
||||
import streamlit as st
|
||||
|
||||
from excel_export import (
|
||||
ExcelExportError,
|
||||
export_filename,
|
||||
export_weekly_workbook,
|
||||
load_template_options,
|
||||
save_workbook,
|
||||
)
|
||||
from jira_client import JiraError, get_jira_client
|
||||
from settings import settings
|
||||
from stamp_history import StampHistory
|
||||
from weekly_report import (
|
||||
AzureRecommendationError,
|
||||
DayRecommendation,
|
||||
calculate_week,
|
||||
improve_with_azure,
|
||||
iso_week_bounds,
|
||||
)
|
||||
|
||||
st.set_page_config(page_title="Weekly Jira & Excel", page_icon="📊", layout="wide")
|
||||
st.title("📊 Weekly Jira & Excel")
|
||||
st.caption(
|
||||
"One likely task per day. Net stamped time (after legal break deductions) is assigned "
|
||||
"in full and can be reviewed before export."
|
||||
)
|
||||
|
||||
try:
|
||||
template_options = load_template_options(settings.TIMETRACKING_TEMPLATE_FILE)
|
||||
except ExcelExportError as exc:
|
||||
st.error(str(exc), icon="🚨")
|
||||
st.stop()
|
||||
|
||||
selected_date = st.date_input("Select any day in the week", value=date.today())
|
||||
iso_year, iso_week, _ = selected_date.isocalendar()
|
||||
week_start, week_end = iso_week_bounds(iso_year, iso_week)
|
||||
employee_name = st.text_input("Name for the workbook", value=settings.TIMETRACKING_NAME)
|
||||
st.caption(f"ISO week {iso_week}, {week_start:%d.%m.%Y} – {week_end:%d.%m.%Y}")
|
||||
|
||||
if st.button("Load Jira activity & recommendations", type="primary", width="stretch"):
|
||||
activities = []
|
||||
jira_error = None
|
||||
with st.spinner("Loading Jira and calculating the week..."):
|
||||
try:
|
||||
activities = get_jira_client().get_week_activity(
|
||||
week_start, week_end, settings.JIRA_JQL
|
||||
)
|
||||
except JiraError as exc:
|
||||
jira_error = str(exc)
|
||||
|
||||
stamps_by_day = StampHistory.get_stamps_between(week_start, week_end)
|
||||
rows = calculate_week(
|
||||
iso_year, iso_week, stamps_by_day, activities, template_options
|
||||
)
|
||||
azure_error = None
|
||||
try:
|
||||
rows = improve_with_azure(rows, activities, template_options)
|
||||
except AzureRecommendationError as exc:
|
||||
azure_error = str(exc)
|
||||
|
||||
st.session_state["weekly_report"] = {
|
||||
"key": (iso_year, iso_week),
|
||||
"rows": rows,
|
||||
"activities": activities,
|
||||
"jira_error": jira_error,
|
||||
"azure_error": azure_error,
|
||||
}
|
||||
|
||||
report = st.session_state.get("weekly_report")
|
||||
if report and report.get("key") == (iso_year, iso_week):
|
||||
rows: list[DayRecommendation] = report["rows"]
|
||||
activities = report["activities"]
|
||||
|
||||
if report.get("jira_error"):
|
||||
st.warning(
|
||||
f"{report['jira_error']} Showing stamped time with editable fallback values.",
|
||||
icon="⚠️",
|
||||
)
|
||||
if report.get("azure_error"):
|
||||
st.info(
|
||||
f"{report['azure_error']} Deterministic recommendations are shown instead.",
|
||||
icon="ℹ️",
|
||||
)
|
||||
|
||||
st.subheader("Recommended Excel entries")
|
||||
total_minutes = sum(item.minutes for item in rows)
|
||||
c1, c2, c3 = st.columns(3)
|
||||
c1.metric("Net time", f"{total_minutes // 60:02d}:{total_minutes % 60:02d}")
|
||||
c2.metric("Days", sum(item.minutes > 0 for item in rows))
|
||||
c3.metric("Jira activities", len(activities))
|
||||
|
||||
editor_data = [
|
||||
{
|
||||
"Date": item.day.isoformat(),
|
||||
"Module": item.module,
|
||||
"Project": item.project,
|
||||
"Use Case": item.use_case,
|
||||
"Implementation Phase": item.phase,
|
||||
"Hours": item.hours,
|
||||
"Most likely task": item.task_summary,
|
||||
"Issues": item.issue_keys,
|
||||
"Warning": item.warning,
|
||||
}
|
||||
for item in rows
|
||||
]
|
||||
edited = st.data_editor(
|
||||
editor_data,
|
||||
hide_index=True,
|
||||
width="stretch",
|
||||
disabled=["Date", "Hours", "Issues", "Warning"],
|
||||
column_config={
|
||||
"Module": st.column_config.SelectboxColumn(options=list(template_options.modules)),
|
||||
"Project": st.column_config.SelectboxColumn(options=list(template_options.projects)),
|
||||
"Use Case": st.column_config.SelectboxColumn(options=list(template_options.use_cases)),
|
||||
"Implementation Phase": st.column_config.SelectboxColumn(options=list(template_options.phases)),
|
||||
"Hours": st.column_config.NumberColumn(format="%.2f"),
|
||||
"Most likely task": st.column_config.TextColumn(width="large"),
|
||||
"Warning": st.column_config.TextColumn(width="medium"),
|
||||
},
|
||||
key=f"weekly_editor_{iso_year}_{iso_week}",
|
||||
)
|
||||
|
||||
records = edited.to_dict("records") if hasattr(edited, "to_dict") else edited
|
||||
edited_rows = []
|
||||
source_by_date = {item.day.isoformat(): item for item in rows}
|
||||
for record in records:
|
||||
source = source_by_date[str(record["Date"])]
|
||||
edited_rows.append(
|
||||
replace(
|
||||
source,
|
||||
module=str(record.get("Module") or ""),
|
||||
project=str(record.get("Project") or ""),
|
||||
use_case=str(record.get("Use Case") or ""),
|
||||
phase=str(record.get("Implementation Phase") or ""),
|
||||
task_summary=str(record.get("Most likely task") or ""),
|
||||
)
|
||||
)
|
||||
|
||||
warnings_found = [item for item in edited_rows if item.warning]
|
||||
if warnings_found:
|
||||
st.warning(
|
||||
"Some stamp data is incomplete. Open intervals are not guessed; review these "
|
||||
"days before submitting the workbook.",
|
||||
icon="⚠️",
|
||||
)
|
||||
|
||||
if st.button("Export Excel for selected week", type="primary", width="stretch"):
|
||||
if not settings.TIMETRACKING_EXPORT_LOCATION:
|
||||
st.error("TIMETRACKING_EXPORT_LOCATION is not configured.", icon="🚨")
|
||||
else:
|
||||
try:
|
||||
workbook = export_weekly_workbook(
|
||||
settings.TIMETRACKING_TEMPLATE_FILE,
|
||||
employee_name,
|
||||
iso_year,
|
||||
iso_week,
|
||||
edited_rows,
|
||||
)
|
||||
destination = save_workbook(
|
||||
workbook,
|
||||
settings.TIMETRACKING_EXPORT_LOCATION,
|
||||
export_filename(employee_name, iso_year, iso_week),
|
||||
)
|
||||
st.success(f"Excel exported to: {destination}", icon="✅")
|
||||
except ExcelExportError as exc:
|
||||
st.error(str(exc), icon="🚨")
|
||||
|
||||
st.subheader("Jira activity rundown")
|
||||
if not activities:
|
||||
st.info("No Jira activity was found for this week.")
|
||||
else:
|
||||
for day_offset in range(7):
|
||||
day = week_start.fromordinal(week_start.toordinal() + day_offset)
|
||||
day_items = [item for item in activities if item.day == day]
|
||||
if not day_items:
|
||||
continue
|
||||
with st.expander(f"{day:%A, %d.%m.%Y} — {len(day_items)} activities", expanded=True):
|
||||
for item in day_items:
|
||||
st.markdown(
|
||||
f"- **[{item.issue_key}]({item.url}) — {item.summary}** "
|
||||
f"\n {item.project or 'No project'} · {item.activity_type} · {item.detail}"
|
||||
)
|
||||
elif report:
|
||||
st.info("The selected week changed. Load it to refresh Jira activity and stamped time.")
|
||||
else:
|
||||
st.info("Choose a week and load its activity to begin.")
|
||||
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import logging
|
||||
from pathlib import Path
|
||||
|
||||
from pydantic_settings import BaseSettings, SettingsConfigDict
|
||||
|
||||
@@ -22,5 +23,26 @@ class StempelSettings(BaseSettings):
|
||||
# "Letzte Buchung" observed on SmartTime Plus.
|
||||
SMART_TIME_VERIFY_TOLERANCE_SEC: int = 90
|
||||
|
||||
# Optional weekly Jira / Excel reporting integration.
|
||||
JIRA_URL: str | None = None
|
||||
JIRA_EMAIL: str | None = None
|
||||
JIRA_API_TOKEN: str | None = None
|
||||
JIRA_CLOUD_ID: str | None = None
|
||||
JIRA_JQL: str | None = None
|
||||
JIRA_TIMEOUT_SEC: int = 30
|
||||
JIRA_VERIFY_TLS: bool = True
|
||||
|
||||
AZURE_OPENAI_ENDPOINT: str | None = None
|
||||
AZURE_OPENAI_API_KEY: str | None = None
|
||||
AZURE_OPENAI_DEPLOYMENT: str = "gpt-5.6-luna"
|
||||
AZURE_OPENAI_API_VERSION: str = "2025-04-01-preview"
|
||||
|
||||
TIMETRACKING_NAME: str = ""
|
||||
TIMETRACKING_TEMPLATE_FILE: str = str(
|
||||
Path(__file__).resolve().parent.parent
|
||||
/ "TimeTracking Activation_Firstname_Lastname_CW_Version 15.xlsx"
|
||||
)
|
||||
TIMETRACKING_EXPORT_LOCATION: str | None = None
|
||||
|
||||
settings = StempelSettings()
|
||||
logging.basicConfig(level=settings.LOG_LEVEL)
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import json
|
||||
import os
|
||||
from datetime import datetime
|
||||
from datetime import date, datetime, timedelta
|
||||
|
||||
from settings import settings
|
||||
|
||||
@@ -23,6 +23,23 @@ class StampHistory:
|
||||
today_key = datetime.now().strftime("%Y-%m-%d")
|
||||
return data.get(today_key, [])
|
||||
|
||||
@classmethod
|
||||
def get_stamps(cls, day: date):
|
||||
return cls._load_data().get(day.isoformat(), [])
|
||||
|
||||
@classmethod
|
||||
def get_stamps_between(cls, start: date, end: date):
|
||||
"""Return stamps keyed by date for an inclusive range."""
|
||||
if end < start:
|
||||
raise ValueError("end must not be before start")
|
||||
data = cls._load_data()
|
||||
result = {}
|
||||
day = start
|
||||
while day <= end:
|
||||
result[day] = list(data.get(day.isoformat(), []))
|
||||
day += timedelta(days=1)
|
||||
return result
|
||||
|
||||
@classmethod
|
||||
def save_stamp(cls, timestamp_str):
|
||||
data = cls._load_data()
|
||||
|
||||
@@ -1,6 +1,5 @@
|
||||
import time
|
||||
from datetime import datetime, timedelta
|
||||
from dataclasses import dataclass
|
||||
from datetime import date, datetime, timedelta
|
||||
|
||||
@dataclass
|
||||
class TimeStats:
|
||||
@@ -16,20 +15,30 @@ class TimeStats:
|
||||
|
||||
|
||||
class TimeLogic:
|
||||
def __init__(self, stamps_str_list):
|
||||
# Convert ["08:00", "12:00"] to datetime objects for today
|
||||
self.today_str = datetime.now().strftime("%Y-%m-%d")
|
||||
def __init__(
|
||||
self,
|
||||
stamps_str_list,
|
||||
work_date: date | None = None,
|
||||
now: datetime | None = None,
|
||||
include_open_interval: bool = True,
|
||||
):
|
||||
# Defaults preserve the live dashboard behaviour; reports provide a date
|
||||
# and disable incomplete open intervals.
|
||||
self.now = now or datetime.now()
|
||||
self.work_date = work_date or self.now.date()
|
||||
self.include_open_interval = include_open_interval
|
||||
self.stamps = []
|
||||
for s in stamps_str_list:
|
||||
try:
|
||||
dt = datetime.strptime(f"{self.today_str} {s}", "%Y-%m-%d %H:%M")
|
||||
parsed = datetime.strptime(s, "%H:%M").time()
|
||||
dt = datetime.combine(self.work_date, parsed)
|
||||
self.stamps.append(dt)
|
||||
except ValueError:
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
self.stamps.sort()
|
||||
|
||||
def calculate(self) -> TimeStats:
|
||||
now = datetime.now()
|
||||
now = self.now
|
||||
start_time = now
|
||||
# 1. Calculate Gross Work and Taken Breaks
|
||||
gross_work = timedelta(0)
|
||||
@@ -55,7 +64,7 @@ class TimeLogic:
|
||||
taken_breaks += current_stamp - prev_out
|
||||
|
||||
# If this is the LAST stamp, calculate work until NOW
|
||||
if i == count - 1:
|
||||
if i == count - 1 and self.include_open_interval:
|
||||
gross_work += now - current_stamp
|
||||
else:
|
||||
# Odd index (1, 3, 5) => "OUT"
|
||||
|
||||
@@ -0,0 +1,253 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
from collections import Counter
|
||||
from dataclasses import asdict, dataclass
|
||||
from datetime import date, timedelta
|
||||
from typing import Any, Iterable
|
||||
|
||||
import requests
|
||||
|
||||
from jira_client import JiraActivity
|
||||
from settings import settings
|
||||
from time_calculator import TimeLogic
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class TemplateOptions:
|
||||
modules: tuple[str, ...]
|
||||
projects: tuple[str, ...]
|
||||
use_cases: tuple[str, ...]
|
||||
phases: tuple[str, ...]
|
||||
|
||||
|
||||
@dataclass
|
||||
class DayRecommendation:
|
||||
day: date
|
||||
minutes: int
|
||||
module: str = ""
|
||||
project: str = ""
|
||||
use_case: str = ""
|
||||
phase: str = ""
|
||||
task_summary: str = "No Jira activity found"
|
||||
issue_keys: str = ""
|
||||
warning: str = ""
|
||||
|
||||
@property
|
||||
def hours(self) -> float:
|
||||
return round(self.minutes / 60, 2)
|
||||
|
||||
|
||||
def iso_week_bounds(year: int, week: int) -> tuple[date, date]:
|
||||
start = date.fromisocalendar(year, week, 1)
|
||||
return start, start + timedelta(days=6)
|
||||
|
||||
|
||||
def calculate_week(
|
||||
year: int,
|
||||
week: int,
|
||||
stamps_by_day: dict[date, list[str]],
|
||||
activities: Iterable[JiraActivity],
|
||||
options: TemplateOptions,
|
||||
) -> list[DayRecommendation]:
|
||||
start, end = iso_week_bounds(year, week)
|
||||
by_day: dict[date, list[JiraActivity]] = {}
|
||||
for activity in activities:
|
||||
if start <= activity.day <= end:
|
||||
by_day.setdefault(activity.day, []).append(activity)
|
||||
|
||||
rows: list[DayRecommendation] = []
|
||||
day = start
|
||||
while day <= end:
|
||||
stamps = stamps_by_day.get(day, [])
|
||||
stats = TimeLogic(
|
||||
stamps, work_date=day, include_open_interval=False
|
||||
).calculate()
|
||||
if stamps or by_day.get(day):
|
||||
row = _fallback_recommendation(day, stats.net_work_minutes, by_day.get(day, []), options)
|
||||
warnings = []
|
||||
if len(stamps) % 2:
|
||||
warnings.append("Incomplete stamp pair ignored")
|
||||
if stamps and stats.net_work_minutes == 0:
|
||||
warnings.append("No complete positive work interval")
|
||||
row.warning = "; ".join(warnings)
|
||||
rows.append(row)
|
||||
day += timedelta(days=1)
|
||||
return rows
|
||||
|
||||
|
||||
def _fallback_recommendation(
|
||||
day: date,
|
||||
minutes: int,
|
||||
activities: list[JiraActivity],
|
||||
options: TemplateOptions,
|
||||
) -> DayRecommendation:
|
||||
text = " ".join(
|
||||
f"{activity.project} {activity.issue_key} {activity.summary} {activity.detail}"
|
||||
for activity in activities
|
||||
)
|
||||
project = _best_option(text, options.projects)
|
||||
if not project and activities and "Other" in options.projects:
|
||||
project = "Other"
|
||||
use_case = _best_option(text, options.use_cases)
|
||||
if not use_case:
|
||||
use_case = next(
|
||||
(value for value in options.use_cases if "not related" in value.casefold()), ""
|
||||
)
|
||||
module = _best_option(text, options.modules)
|
||||
phase = _phase_for(text, options.phases)
|
||||
keys = list(dict.fromkeys(activity.issue_key for activity in activities))
|
||||
if activities:
|
||||
counts = Counter(activity.issue_key for activity in activities)
|
||||
primary_key = counts.most_common(1)[0][0]
|
||||
primary = next(item for item in activities if item.issue_key == primary_key)
|
||||
summary = f"{primary_key}: {primary.summary}"
|
||||
if len(keys) > 1:
|
||||
summary += f" (+{len(keys) - 1} more)"
|
||||
else:
|
||||
summary = "No Jira activity found"
|
||||
return DayRecommendation(
|
||||
day=day,
|
||||
minutes=minutes,
|
||||
module=module,
|
||||
project=project,
|
||||
use_case=use_case,
|
||||
phase=phase,
|
||||
task_summary=summary,
|
||||
issue_keys=", ".join(keys),
|
||||
)
|
||||
|
||||
|
||||
def _best_option(text: str, options: tuple[str, ...]) -> str:
|
||||
normalized = _normalize(text)
|
||||
candidates = []
|
||||
for option in options:
|
||||
tokens = [token for token in _normalize(option).split() if len(token) > 2]
|
||||
score = sum(1 for token in set(tokens) if token in normalized)
|
||||
if score:
|
||||
candidates.append((score, len(option), option))
|
||||
return max(candidates, default=(0, 0, ""))[2]
|
||||
|
||||
|
||||
def _normalize(value: str) -> str:
|
||||
return re.sub(r"[^a-z0-9]+", " ", value.casefold()).strip()
|
||||
|
||||
|
||||
def _phase_for(text: str, phases: tuple[str, ...]) -> str:
|
||||
lowered = text.casefold()
|
||||
hints = (
|
||||
(("test", "qa", "quality"), "Testing"),
|
||||
(("deploy", "release"), "Deployment"),
|
||||
(("architect", "design"), "Architecture"),
|
||||
(("requirement", "story", "spec"), "Requirementmanagement"),
|
||||
(("meeting", "workshop", "sync"), "Meetings"),
|
||||
(("handover",), "Handover"),
|
||||
(("hypercare",), "Hypercare"),
|
||||
(("develop", "implement", "code", "fix", "bug"), "Development"),
|
||||
)
|
||||
for words, expected in hints:
|
||||
if any(word in lowered for word in words):
|
||||
match = next((phase for phase in phases if expected.casefold() in phase.casefold()), None)
|
||||
if match:
|
||||
return match
|
||||
return next(
|
||||
(phase for phase in phases if "development" in phase.casefold()),
|
||||
phases[0] if phases else "",
|
||||
)
|
||||
|
||||
|
||||
class AzureRecommendationError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
def improve_with_azure(
|
||||
rows: list[DayRecommendation],
|
||||
activities: Iterable[JiraActivity],
|
||||
options: TemplateOptions,
|
||||
) -> list[DayRecommendation]:
|
||||
"""Improve day-level recommendations in one Azure OpenAI request."""
|
||||
if not rows:
|
||||
return rows
|
||||
if not (
|
||||
settings.AZURE_OPENAI_ENDPOINT
|
||||
and settings.AZURE_OPENAI_API_KEY
|
||||
and settings.AZURE_OPENAI_DEPLOYMENT
|
||||
):
|
||||
return rows
|
||||
|
||||
activities_by_day: dict[str, list[dict[str, str]]] = {}
|
||||
for item in activities:
|
||||
activities_by_day.setdefault(item.day.isoformat(), []).append(
|
||||
{
|
||||
"key": item.issue_key,
|
||||
"summary": item.summary,
|
||||
"project": item.project,
|
||||
"activity": item.activity_type,
|
||||
"detail": item.detail,
|
||||
}
|
||||
)
|
||||
input_rows = [
|
||||
{
|
||||
"date": row.day.isoformat(),
|
||||
"net_minutes": row.minutes,
|
||||
"jira_activity": activities_by_day.get(row.day.isoformat(), []),
|
||||
"fallback": asdict(row),
|
||||
}
|
||||
for row in rows
|
||||
]
|
||||
prompt = {
|
||||
"instruction": (
|
||||
"Choose the single most likely task worked on for each day. Return one JSON "
|
||||
"object with a 'days' array. Every day needs date, module, project, use_case, "
|
||||
"phase, and task_summary. Choose classification values exactly from the supplied "
|
||||
"lists or use an empty string. Keep task_summary to one short sentence. Do not "
|
||||
"change dates or durations; all of a day's time stays on this one row. Prefer "
|
||||
"concrete authored changes/comments/worklogs over merely assigned issues."
|
||||
),
|
||||
"allowed_values": asdict(options),
|
||||
"days": input_rows,
|
||||
}
|
||||
url = (
|
||||
f"{settings.AZURE_OPENAI_ENDPOINT.rstrip('/')}"
|
||||
f"/openai/deployments/{settings.AZURE_OPENAI_DEPLOYMENT}/chat/completions"
|
||||
)
|
||||
try:
|
||||
response = requests.post(
|
||||
url,
|
||||
params={"api-version": settings.AZURE_OPENAI_API_VERSION},
|
||||
headers={"api-key": settings.AZURE_OPENAI_API_KEY, "Content-Type": "application/json"},
|
||||
json={
|
||||
"messages": [
|
||||
{"role": "system", "content": "You classify weekly Jira activity for a time sheet."},
|
||||
{"role": "user", "content": json.dumps(prompt, ensure_ascii=False, default=str)},
|
||||
],
|
||||
"response_format": {"type": "json_object"},
|
||||
"max_completion_tokens": 2500,
|
||||
},
|
||||
timeout=60,
|
||||
)
|
||||
response.raise_for_status()
|
||||
content = response.json()["choices"][0]["message"]["content"]
|
||||
generated = json.loads(content).get("days", [])
|
||||
except (requests.RequestException, ValueError, KeyError, IndexError, TypeError) as exc:
|
||||
raise AzureRecommendationError("Azure OpenAI recommendation failed.") from exc
|
||||
|
||||
by_date = {str(item.get("date")): item for item in generated if isinstance(item, dict)}
|
||||
for row in rows:
|
||||
item = by_date.get(row.day.isoformat())
|
||||
if not item:
|
||||
continue
|
||||
row.module = _allowed(item.get("module"), options.modules, row.module)
|
||||
row.project = _allowed(item.get("project"), options.projects, row.project)
|
||||
row.use_case = _allowed(item.get("use_case"), options.use_cases, row.use_case)
|
||||
row.phase = _allowed(item.get("phase"), options.phases, row.phase)
|
||||
summary = str(item.get("task_summary") or "").strip()
|
||||
if summary:
|
||||
row.task_summary = summary[:500]
|
||||
return rows
|
||||
|
||||
|
||||
def _allowed(value: Any, choices: tuple[str, ...], fallback: str) -> str:
|
||||
return value if isinstance(value, str) and value in choices else fallback
|
||||
|
||||
Reference in New Issue
Block a user