From c6b00c2e85f0eaede3065aadd2ad5c055b2b9b50 Mon Sep 17 00:00:00 2001 From: Kamil Rudnicki Date: Mon, 10 Aug 2026 10:05:48 +0200 Subject: [PATCH 1/2] Netsuite --- .env.example | 13 + README.md | 28 ++ docs/netsuite.md | 132 +++++++ export_time_entries_netsuite.py | 393 ++++++++++++++++++++ fetch_netsuite.py | 402 +++++++++++++++++++++ netsuite_config.example.json | 48 +++ requirements.txt | 1 + src/netsuite_client.py | 334 +++++++++++++++++ sync_projects.py | 57 ++- tests/test_export_time_entries_netsuite.py | 217 +++++++++++ tests/test_fetch_netsuite.py | 130 +++++++ tests/test_netsuite_client.py | 152 ++++++++ tests/test_sync_projects.py | 71 ++++ 13 files changed, 1971 insertions(+), 7 deletions(-) create mode 100644 docs/netsuite.md create mode 100644 export_time_entries_netsuite.py create mode 100644 fetch_netsuite.py create mode 100644 netsuite_config.example.json create mode 100644 src/netsuite_client.py create mode 100644 tests/test_export_time_entries_netsuite.py create mode 100644 tests/test_fetch_netsuite.py create mode 100644 tests/test_netsuite_client.py diff --git a/.env.example b/.env.example index a07cf43..d93e3c5 100644 --- a/.env.example +++ b/.env.example @@ -6,6 +6,8 @@ TIMECAMP_TASK_ID=your_timecamp_task_id_for_redmine_projects_like_170066189 # Optional: skip assigning mandatory tags to a task when more than this many # mandatory tags would need to be added. Leave empty/unset for no limit. # TIMECAMP_MAX_MANDATORY_TAGS_TO_ADD=1 +# Optional: isolate reads/archives to one integration's external_task_id prefix. +# TIMECAMP_SYNC_EXTERNAL_ID_PREFIX=netsuite_ # AZUREDEVOPS integration # AZUREDEVOPS_INSTANCES=Company1:https://dev.azure.com/company1:token1,Company2:https://dev.azure.com/company2:token2 @@ -52,3 +54,14 @@ MONDAY_API_TOKEN=your_monday_api_token # MONDAY_BOARD_IDS=123456789,987654321 # Optional: comma-separated Monday column titles to export as meandatory_tags. # MONDAY_MEANDATORY_TAGS=Client,CoE + +# NetSuite integration (OAuth 2.0 Client Credentials / M2M) +NETSUITE_ACCOUNT_ID=1234567_SB1 +NETSUITE_CLIENT_ID=your_oauth2_integration_client_id +NETSUITE_CERTIFICATE_ID=your_oauth2_m2m_certificate_id +NETSUITE_PRIVATE_KEY_FILE=/absolute/path/to/netsuite-private-key.pem +# NETSUITE_PRIVATE_KEY_PASSPHRASE=optional_private_key_passphrase +# NETSUITE_JWT_ALGORITHM=PS256 +# NETSUITE_CONFIG_FILE=netsuite_config.json +# Temporary alternative for a manually obtained OAuth 2.0 bearer token: +# NETSUITE_ACCESS_TOKEN=your_short_lived_access_token diff --git a/README.md b/README.md index cf68866..f4fa905 100644 --- a/README.md +++ b/README.md @@ -147,6 +147,34 @@ uv run --with-requirements requirements.txt python export_monday_time_logged.py uv run --with-requirements requirements.txt python export_monday_time_logged.py --from 2026-06-01 --to 2026-06-18 --column-title "Time Tracked" --include-main-rows ``` +### NetSuite ↔ TimeCamp POC + +The NetSuite integration synchronizes projects and project tasks into TimeCamp, +then upserts TimeCamp entries as NetSuite `timebill` records. User provisioning +and project-user assignments are intentionally out of scope; employees are read +only to resolve the employee reference required by exported time. + +```bash +cp netsuite_config.example.json netsuite_config.json + +uv run --env-file .env --with-requirements requirements.txt python fetch_netsuite.py +TIMECAMP_SYNC_EXTERNAL_ID_PREFIX=netsuite_ \ + TIMECAMP_SYNC_ACTIONS=tasks,names,estimates,tags,mandatory_tags,archive \ + uv run --env-file .env --with-requirements requirements.txt python sync_projects.py + +# Dry-run by default. +uv run --env-file .env --with-requirements requirements.txt \ + python export_time_entries_netsuite.py --from 2026-08-01 --to 2026-08-03 + +# Write only after the dry-run has no mapping errors. +uv run --env-file .env --with-requirements requirements.txt \ + python export_time_entries_netsuite.py --from 2026-08-01 --to 2026-08-03 --apply +``` + +See [`docs/netsuite.md`](docs/netsuite.md) for the WCG-specific discovery and +mapping contract. The example CAPEX/OPEX field IDs are placeholders and must be +replaced before an applied export. + ### Limiting TimeCamp Sync Actions By default, `sync_projects.py` runs all actions: creating missing tasks, updating changed diff --git a/docs/netsuite.md b/docs/netsuite.md new file mode 100644 index 0000000..20af45b --- /dev/null +++ b/docs/netsuite.md @@ -0,0 +1,132 @@ +# NetSuite ↔ TimeCamp POC + +## Scope + +The integration implements this flow: + +1. SuiteQL reads active NetSuite projects and project tasks. +2. `fetch_netsuite.py` writes the common `tasks.json` contract. +3. `sync_projects.py` creates, renames, estimates, tags, and archives the matching + TimeCamp hierarchy. +4. `export_time_entries_netsuite.py` maps TimeCamp entries back to their NetSuite + employee, project, project task/activity, and CAPEX/OPEX classification. +5. The exporter upserts `timebill` records by `timecamp-{entry_id}` external ID. + Re-running a date range updates the same records instead of duplicating them. + +User provisioning and project-user assignment are not implemented. The exporter +only reads NetSuite employees and TimeCamp users to resolve the employee required +by a `timebill`. Email is the default key. Ambiguous or missing matches block an +applied export; use `time_export.employee_mapping` for explicit TimeCamp user ID +to NetSuite employee ID overrides. + +## Authentication + +Use OAuth 2.0 Client Credentials (M2M). NetSuite requires an integration record, +the `REST Web Services` scope, a role with the necessary record permissions, and +a certificate mapping. Configure: + +```dotenv +NETSUITE_ACCOUNT_ID=1234567_SB1 +NETSUITE_CLIENT_ID=... +NETSUITE_CERTIFICATE_ID=... +NETSUITE_PRIVATE_KEY_FILE=/absolute/path/to/private-key.pem +``` + +`NETSUITE_ACCESS_TOKEN` accepts a short-lived bearer token for local diagnosis, +but it is not a scheduler credential. Token-based OAuth 1.0 authentication is +deliberately not added: Oracle says that from NetSuite 2027.1 new TBA integrations +for REST web services cannot be created. + +## Account discovery before the POC + +NetSuite's REST schema is account-specific. Standard and custom fields must be +confirmed against the WCG Records Catalog; guessing them is unsafe. The client +supports the metadata endpoint, and the relevant record types are `job`, +`projecttask`, `employee`, and `timebill`. + +Copy `netsuite_config.example.json` to `netsuite_config.json`. Both SuiteQL +queries must use a deterministic `ORDER BY`, because the REST endpoint is paged. +The importer expects these aliases: + +| Query | Required aliases | Optional aliases | +| --- | --- | --- | +| `projects` | `id`, `name` | `parent_id`, `capex_opex`, `activity_id`, `is_inactive` | +| `project_tasks` | `id`, `name`, `project_id` | `parent_id`, `capex_opex`, `activity_id`, `estimated_work_hours`, `original_estimate_seconds`, `is_inactive` | +| `employees_query` | `id`, `email` | none | + +The example uses standard field candidates, not a claim about WCG's schema. +Replace them when the Records Catalog or a real SuiteQL call proves otherwise. + +## CAPEX/OPEX + +Alias the WCG project/task classification field to `capex_opex` in SuiteQL. The +importer normalizes it through `classification.value_map`, inherits a missing task +classification from its parent/project, and assigns it under a mandatory TimeCamp +tag list. Unknown non-empty values stop the import instead of corrupting financial +classification. The example sets `classification.required` to `true`, so its +standard query skeleton deliberately cannot be used for synchronization until the +WCG classification field is added to the query. + +For export, make an explicit choice in `time_export.classification`: + +- `mode: "field"` writes the mapped value to a WCG `timebill` field. +- `mode: "project"` writes no classification field because the selected NetSuite + project is the authoritative classification. +- `mode: "omit"` deliberately drops it. This is appropriate only if WCG confirms + CAPEX/OPEX is irrelevant on individual time records. + +The placeholders in the example config intentionally make `field` mode fail until +the real WCG field and list value IDs are entered. + +## Activity and project task mapping + +Time entered on an imported project task exports both the NetSuite project and +project-task IDs. If WCG uses a service item or another activity reference, alias +its ID as `activity_id` and set `time_export.fields.activity` to the corresponding +`timebill` field. `default_activity_id` is the fallback. + +NetSuite custom forms can require extra fields such as approval status, subsidiary, +department, or location. Add invariant values to `time_export.fixed_fields`. Do not +apply an export until the dry-run payload passes against the sandbox metadata and +WCG's approval workflow. + +## Commands + +```bash +uv run --env-file .env --with-requirements requirements.txt \ + python fetch_netsuite.py --config netsuite_config.json --output tasks.json + +TIMECAMP_SYNC_EXTERNAL_ID_PREFIX=netsuite_ \ + TIMECAMP_SYNC_ACTIONS=tasks,names,estimates,tags,mandatory_tags,archive \ + uv run --env-file .env --with-requirements requirements.txt \ + python sync_projects.py --input tasks.json + +uv run --env-file .env --with-requirements requirements.txt \ + python export_time_entries_netsuite.py \ + --config netsuite_config.json --tasks tasks.json \ + --from 2026-08-01 --to 2026-08-03 +``` + +The exporter is dry-run by default. `--apply` is all-or-nothing for local mapping +validation: any unmapped employee, invalid duration, missing project, or unresolved +CAPEX/OPEX value stops the run before the first NetSuite write. Network failure can +still interrupt a batch, but external-ID upserts make the same command safe to retry. + +NetSuite stores `timebill.hours` at minute precision. `duration_rounding` supports +`nearest` (default), `floor`, `ceil`, or `reject`; use `reject` if WCG requires zero +rounding loss. + +## Unresolved WCG decisions + +- Which standard/custom records are the authoritative project and activity sources? +- What field represents project hierarchy, status, project manager, and CAPEX/OPEX? +- Is CAPEX/OPEX derived from the project or stored on each `timebill`? +- Is `caseTaskEvent` the correct project-task field in WCG's REST metadata? +- Is a service `item`, approval status, subsidiary, department, location, or memo required? +- Should an edited/deleted TimeCamp entry update/delete the NetSuite record, and what + happens after approval or posting closes the accounting period? +- What date window, timezone cutoff, and approval state are eligible for export? + +These are specification decisions, not implementation details. The POC should prove +them against a NetSuite sandbox before production credentials or scheduled writes are +allowed. diff --git a/export_time_entries_netsuite.py b/export_time_entries_netsuite.py new file mode 100644 index 0000000..272fd01 --- /dev/null +++ b/export_time_entries_netsuite.py @@ -0,0 +1,393 @@ +import argparse +import json +import os +import re +from collections import Counter +from copy import deepcopy +from dataclasses import dataclass +from datetime import date +from pathlib import Path +from typing import Any, Dict, Iterable, List, Optional, Tuple + +from dotenv import load_dotenv + +from fetch_netsuite import load_config, row_value +from src.netsuite_client import NetSuiteClient +from src.timecamp_client import TimeCampClient + +DEFAULT_CONFIG_FILE = "netsuite_config.json" +DEFAULT_TASKS_FILE = "tasks.json" +DEFAULT_TIMEBILL_FIELDS = { + "employee": "employee", + "date": "tranDate", + "hours": "hours", + "project": "customer", + "project_task": "caseTaskEvent", + "activity": None, + "memo": "memo", +} +VALID_EXTERNAL_ID = re.compile(r"^[A-Za-z0-9_-]+$") +CONFIG_PLACEHOLDER_PREFIX = "REPLACE_WITH_" + + +@dataclass(frozen=True) +class PreparedTimeBill: + timecamp_entry_id: str + external_id: str + payload: Dict[str, Any] + + +def load_tasks(path: str) -> List[Dict[str, Any]]: + try: + tasks = json.loads(Path(path).read_text(encoding="utf-8")) + except FileNotFoundError as exc: + raise ValueError(f"Tasks file not found: {path}") from exc + except json.JSONDecodeError as exc: + raise ValueError(f"Invalid JSON in tasks file {path}: {exc}") from exc + if not isinstance(tasks, list): + raise ValueError(f"{path} must contain a JSON list") + return tasks + + +def parse_positive_seconds(value: Any) -> Optional[int]: + if isinstance(value, bool): + return None + try: + seconds = int(value) + except (TypeError, ValueError): + return None + return seconds if seconds > 0 else 0 + + +def format_netsuite_hours(seconds: int, rounding: str = "nearest") -> str: + if seconds <= 0: + raise ValueError("TimeBill duration must be positive") + normalized_rounding = str(rounding or "nearest").strip().casefold() + if normalized_rounding == "reject": + if seconds % 60: + raise ValueError("Duration is not a whole minute") + minutes = seconds // 60 + elif normalized_rounding == "floor": + minutes = seconds // 60 + elif normalized_rounding == "ceil": + minutes = (seconds + 59) // 60 + elif normalized_rounding == "nearest": + minutes = (seconds + 30) // 60 + else: + raise ValueError( + "time_export.duration_rounding must be nearest, floor, ceil, or reject" + ) + if minutes <= 0: + raise ValueError("Duration rounds to zero minutes") + hours, remaining_minutes = divmod(minutes, 60) + return f"{hours}:{remaining_minutes:02d}" + + +def build_employee_mapping( + timecamp_users: Iterable[Dict[str, Any]], + netsuite_employees: Iterable[Dict[str, Any]], + explicit_mapping: Optional[Dict[str, Any]] = None, +) -> Dict[str, str]: + if explicit_mapping is not None and not isinstance(explicit_mapping, dict): + raise ValueError("time_export.employee_mapping must be an object") + explicit = { + str(timecamp_id): str(netsuite_id) + for timecamp_id, netsuite_id in (explicit_mapping or {}).items() + if netsuite_id not in (None, "") and str(netsuite_id).strip() + } + employees_by_email: Dict[str, List[str]] = {} + for employee in netsuite_employees: + employee_id = row_value(employee, "id") + email = str(row_value(employee, "email") or "").strip().casefold() + if employee_id in (None, "") or not email: + continue + employees_by_email.setdefault(email, []).append(str(employee_id)) + + result = dict(explicit) + for user in timecamp_users: + user_id = user.get("user_id") or user.get("id") + if user_id in (None, "") or str(user_id) in result: + continue + email = str(user.get("email") or "").strip().casefold() + matches = employees_by_email.get(email, []) + if len(matches) == 1: + result[str(user_id)] = matches[0] + return result + + +def reference(record_id: Any) -> Dict[str, str]: + return {"id": str(record_id)} + + +def _configured_fields(export_config: Dict[str, Any]) -> Dict[str, Optional[str]]: + raw_fields = export_config.get("fields") or {} + if not isinstance(raw_fields, dict): + raise ValueError("time_export.fields must be an object") + fields = dict(DEFAULT_TIMEBILL_FIELDS) + for key, value in raw_fields.items(): + fields[str(key)] = str(value).strip() if value else None + for required in ("employee", "date", "hours", "project"): + if not fields.get(required): + raise ValueError(f"time_export.fields.{required} must be configured") + return fields + + +def _apply_classification( + payload: Dict[str, Any], + classification: Optional[str], + export_config: Dict[str, Any], +) -> None: + if not classification: + return + raw_config = export_config.get("classification") + if not isinstance(raw_config, dict): + raise ValueError( + "Time entry has CAPEX/OPEX but time_export.classification is not configured" + ) + mode = str(raw_config.get("mode") or "").strip().casefold() + if mode in {"project", "omit"}: + return + if mode != "field": + raise ValueError( + "time_export.classification.mode must be field, project, or omit" + ) + + field = str(raw_config.get("field") or "").strip() + if not field: + raise ValueError("time_export.classification.field must be configured") + if field.startswith(CONFIG_PLACEHOLDER_PREFIX): + raise ValueError("Replace the placeholder CAPEX/OPEX NetSuite field") + value_map = raw_config.get("value_map") or {} + if not isinstance(value_map, dict): + raise ValueError("time_export.classification.value_map must be an object") + mapped_value = None + for raw_name, raw_value in value_map.items(): + if str(raw_name).strip().casefold() == classification.casefold(): + mapped_value = raw_value + break + if mapped_value in (None, ""): + raise ValueError(f"No NetSuite value mapped for {classification}") + if str(mapped_value).startswith(CONFIG_PLACEHOLDER_PREFIX): + raise ValueError(f"Replace the placeholder NetSuite value for {classification}") + value_format = str(raw_config.get("value_format") or "id").strip().casefold() + if value_format == "id": + payload[field] = reference(mapped_value) + elif value_format == "raw": + payload[field] = mapped_value + else: + raise ValueError("time_export.classification.value_format must be id or raw") + + +def prepare_timebills( + entries: Iterable[Dict[str, Any]], + timecamp_tasks: Iterable[Dict[str, Any]], + source_tasks: Iterable[Dict[str, Any]], + employee_mapping: Dict[str, str], + config: Dict[str, Any], +) -> Tuple[List[PreparedTimeBill], Counter, List[str]]: + export_config = config.get("time_export") or {} + if not isinstance(export_config, dict): + raise ValueError("time_export config must be an object") + fields = _configured_fields(export_config) + fixed_fields = export_config.get("fixed_fields") or {} + if not isinstance(fixed_fields, dict): + raise ValueError("time_export.fixed_fields must be an object") + external_id_prefix = str(export_config.get("external_id_prefix") or "timecamp-") + rounding = str(export_config.get("duration_rounding") or "nearest") + + timecamp_tasks_by_id = { + str(task.get("task_id")): task + for task in timecamp_tasks + if task.get("task_id") not in (None, "") + } + source_tasks_by_id = { + str(task.get("task_id")): task + for task in source_tasks + if task.get("task_id") not in (None, "") + } + + prepared: List[PreparedTimeBill] = [] + skipped: Counter = Counter() + errors: List[str] = [] + for entry in entries: + entry_id = str(entry.get("id") or "").strip() + seconds = parse_positive_seconds(entry.get("duration")) + if not entry_id: + skipped["missing_entry_id"] += 1 + continue + if seconds is None: + errors.append(f"TimeCamp entry {entry_id}: invalid duration") + continue + if seconds == 0: + skipped["zero_duration"] += 1 + continue + + timecamp_task = timecamp_tasks_by_id.get(str(entry.get("task_id"))) + if not timecamp_task: + skipped["missing_timecamp_task"] += 1 + continue + external_task_id = str(timecamp_task.get("external_task_id") or "") + source_task = source_tasks_by_id.get(external_task_id) + netsuite_data = source_task.get("netsuite") if source_task else None + if not isinstance(netsuite_data, dict): + skipped["non_netsuite_task"] += 1 + continue + + user_id = str(entry.get("user_id") or "") + employee_id = employee_mapping.get(user_id) + if not employee_id: + errors.append( + f"TimeCamp entry {entry_id}: no NetSuite employee mapping " + f"for user {user_id}" + ) + continue + project_id = netsuite_data.get("project_id") + if project_id in (None, ""): + errors.append( + f"TimeCamp entry {entry_id}: source task has no NetSuite project_id" + ) + continue + try: + entry_date = date.fromisoformat(str(entry.get("date") or "")) + hours = format_netsuite_hours(seconds, rounding) + except ValueError as exc: + errors.append(f"TimeCamp entry {entry_id}: {exc}") + continue + + external_id = f"{external_id_prefix}{entry_id}" + if not VALID_EXTERNAL_ID.fullmatch(external_id): + errors.append( + f"TimeCamp entry {entry_id}: generated external ID contains " + "unsupported characters" + ) + continue + + payload = deepcopy(fixed_fields) + payload[fields["employee"]] = reference(employee_id) + payload[fields["date"]] = entry_date.isoformat() + payload[fields["hours"]] = hours + payload[fields["project"]] = reference(project_id) + + project_task_id = netsuite_data.get("project_task_id") + if fields.get("project_task") and project_task_id not in (None, ""): + payload[fields["project_task"]] = reference(project_task_id) + + activity_id = netsuite_data.get("activity_id") + if activity_id in (None, ""): + activity_id = export_config.get("default_activity_id") + if fields.get("activity") and activity_id not in (None, ""): + payload[fields["activity"]] = reference(activity_id) + + description = str(entry.get("description") or "").strip() + if fields.get("memo") and description: + payload[fields["memo"]] = description + + try: + _apply_classification( + payload, + netsuite_data.get("capex_opex"), + export_config, + ) + except ValueError as exc: + errors.append(f"TimeCamp entry {entry_id}: {exc}") + continue + + prepared.append( + PreparedTimeBill( + timecamp_entry_id=entry_id, + external_id=external_id, + payload=payload, + ) + ) + + return prepared, skipped, errors + + +def parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Export TimeCamp time entries to NetSuite TimeBill records." + ) + parser.add_argument("--from", dest="start_date", required=True) + parser.add_argument("--to", dest="end_date", required=True) + parser.add_argument( + "--config", + default=os.getenv("NETSUITE_CONFIG_FILE", DEFAULT_CONFIG_FILE), + ) + parser.add_argument("--tasks", default=DEFAULT_TASKS_FILE) + parser.add_argument( + "--apply", + action="store_true", + help="Write to NetSuite. Without this flag the command is a dry-run.", + ) + return parser.parse_args(argv) + + +def main(argv: Optional[List[str]] = None) -> None: + load_dotenv(override=True) + args = parse_args(argv) + start_date = date.fromisoformat(args.start_date) + end_date = date.fromisoformat(args.end_date) + if start_date > end_date: + raise ValueError("--from must be before or equal to --to") + + config = load_config(args.config) + export_config = config.get("time_export") or {} + employees_query = str(export_config.get("employees_query") or "").strip() + explicit_mapping = export_config.get("employee_mapping") or {} + if not employees_query and not explicit_mapping: + raise ValueError( + "Configure time_export.employees_query or time_export.employee_mapping" + ) + + timecamp_token = os.getenv("TIMECAMP_API_TOKEN") + if not timecamp_token: + raise ValueError("TIMECAMP_API_TOKEN must be set in .env") + timecamp = TimeCampClient(timecamp_token) + netsuite = NetSuiteClient.from_env() + source_tasks = load_tasks(args.tasks) + + print(f"Loading TimeCamp entries from {start_date} to {end_date}...") + entries = timecamp.get_time_entries(start_date, end_date) + timecamp_tasks = timecamp.get_tasks() + timecamp_users = timecamp.get_users() + employees = netsuite.suiteql(employees_query) if employees_query else [] + employee_mapping = build_employee_mapping( + timecamp_users, + employees, + explicit_mapping, + ) + prepared, skipped, errors = prepare_timebills( + entries, + timecamp_tasks, + source_tasks, + employee_mapping, + config, + ) + + print(f"Prepared {len(prepared)} NetSuite TimeBill upsert(s)") + for reason, count in sorted(skipped.items()): + print(f" Skipped {reason}: {count}") + if errors: + print(f"Blocked by {len(errors)} mapping/validation error(s):") + for error in errors[:50]: + print(f" - {error}") + if len(errors) > 50: + print(f" - ... and {len(errors) - 50} more") + raise SystemExit(1) + + if not args.apply: + print("Dry-run only; add --apply to write records to NetSuite.") + for item in prepared[:20]: + print( + f" {item.external_id}: {json.dumps(item.payload, ensure_ascii=False)}" + ) + return + + record_type = str(export_config.get("record_type") or "timebill").strip() + for index, item in enumerate(prepared, start=1): + netsuite.upsert_record(record_type, item.external_id, item.payload) + print(f"Upserted {index}/{len(prepared)}: {item.external_id}") + + +if __name__ == "__main__": + main() diff --git a/fetch_netsuite.py b/fetch_netsuite.py new file mode 100644 index 0000000..b4655b3 --- /dev/null +++ b/fetch_netsuite.py @@ -0,0 +1,402 @@ +import argparse +import json +import os +from decimal import ROUND_HALF_UP, Decimal, InvalidOperation +from pathlib import Path +from typing import Any, Dict, Iterable, List, Optional + +from dotenv import load_dotenv + +from src.netsuite_client import NetSuiteClient + +DEFAULT_CONFIG_FILE = "netsuite_config.json" +DEFAULT_OUTPUT_FILE = "tasks.json" +NETSUITE_PROJECT_PREFIX = "netsuite_project_" +NETSUITE_PROJECT_TASK_PREFIX = "netsuite_project_task_" +DEFAULT_CLASSIFICATION_TAG_LIST = "CAPEX / OPEX" + + +def project_external_id(project_id: Any) -> str: + return f"{NETSUITE_PROJECT_PREFIX}{project_id}" + + +def project_task_external_id(project_task_id: Any) -> str: + return f"{NETSUITE_PROJECT_TASK_PREFIX}{project_task_id}" + + +def row_value(row: Dict[str, Any], key: str, default: Any = None) -> Any: + wanted = key.casefold() + for raw_key, value in row.items(): + if str(raw_key).casefold() == wanted: + return value + return default + + +def optional_id(value: Any) -> Optional[str]: + if value in (None, "", 0, "0"): + return None + return str(value).strip() or None + + +def required_id(row: Dict[str, Any], key: str, record_label: str) -> str: + value = optional_id(row_value(row, key)) + if value is None: + raise ValueError(f"{record_label} SuiteQL row has no {key}: {row!r}") + return value + + +def required_name(row: Dict[str, Any], record_label: str) -> str: + value = str(row_value(row, "name") or "").strip() + if not value: + raise ValueError(f"{record_label} SuiteQL row has no name: {row!r}") + return value + + +def is_inactive(row: Dict[str, Any]) -> bool: + value = row_value(row, "is_inactive") + if isinstance(value, str): + return value.strip().casefold() in {"t", "true", "yes", "1"} + return bool(value) + + +def load_config(path: str) -> Dict[str, Any]: + config_path = Path(path) + try: + config = json.loads(config_path.read_text(encoding="utf-8")) + except FileNotFoundError as exc: + raise ValueError( + f"NetSuite config not found: {config_path}. " + "Copy netsuite_config.example.json first." + ) from exc + except json.JSONDecodeError as exc: + raise ValueError( + f"Invalid JSON in NetSuite config {config_path}: {exc}" + ) from exc + + if not isinstance(config, dict): + raise ValueError("NetSuite config must contain a JSON object") + return config + + +def classification_config(config: Dict[str, Any]) -> Dict[str, Any]: + raw = config.get("classification") or {} + if not isinstance(raw, dict): + raise ValueError("classification config must be an object") + return raw + + +def normalize_classification( + value: Any, + config: Dict[str, Any], +) -> Optional[str]: + if value is None or not str(value).strip(): + default = config.get("default") + if default is None or not str(default).strip(): + return None + value = default + + normalized = " ".join(str(value).strip().split()) + value_map = config.get("value_map") or {} + if not isinstance(value_map, dict): + raise ValueError("classification.value_map must be an object") + + mapped_values = { + str(raw_key).strip().casefold(): str(mapped).strip() + for raw_key, mapped in value_map.items() + } + classification = mapped_values.get(normalized.casefold(), normalized.upper()) + allowed_values = config.get("allowed_values", ["CAPEX", "OPEX"]) + allowed_by_key = { + str(allowed).strip().casefold(): str(allowed).strip() + for allowed in allowed_values + } + canonical = allowed_by_key.get(classification.casefold()) + if canonical is None: + raise ValueError( + f"Unknown CAPEX/OPEX value {value!r}; configure classification.value_map" + ) + return canonical + + +def estimate_seconds(row: Dict[str, Any]) -> Optional[int]: + raw_seconds = row_value(row, "original_estimate_seconds") + raw_hours = row_value(row, "estimated_work_hours") + if raw_seconds in (None, "") and raw_hours in (None, ""): + return None + + try: + value = Decimal( + str(raw_seconds if raw_seconds not in (None, "") else raw_hours) + ) + except InvalidOperation as exc: + raise ValueError(f"Invalid NetSuite project task estimate: {row!r}") from exc + if value < 0: + raise ValueError(f"NetSuite project task estimate cannot be negative: {row!r}") + if raw_seconds in (None, ""): + value *= Decimal(3600) + return int(value.quantize(Decimal("1"), rounding=ROUND_HALF_UP)) + + +def build_task_structure( + project_rows: Iterable[Dict[str, Any]], + project_task_rows: Iterable[Dict[str, Any]], + config: Dict[str, Any], +) -> List[Dict[str, Any]]: + class_config = classification_config(config) + tag_list_name = str( + class_config.get("tag_list_name") or DEFAULT_CLASSIFICATION_TAG_LIST + ).strip() + if not tag_list_name: + raise ValueError("classification.tag_list_name must not be empty") + + projects_by_id: Dict[str, Dict[str, Any]] = {} + for row in project_rows: + if is_inactive(row): + continue + project_id = required_id(row, "id", "Project") + if project_id in projects_by_id: + raise ValueError(f"Duplicate NetSuite project id: {project_id}") + projects_by_id[project_id] = row + + project_tasks_by_id: Dict[str, Dict[str, Any]] = {} + task_project_ids: Dict[str, str] = {} + for row in project_task_rows: + if is_inactive(row): + continue + task_id = required_id(row, "id", "Project task") + project_id = required_id(row, "project_id", "Project task") + if task_id in project_tasks_by_id: + raise ValueError(f"Duplicate NetSuite project task id: {task_id}") + if project_id not in projects_by_id: + raise ValueError( + f"NetSuite project task {task_id} references missing active " + f"project {project_id}" + ) + project_tasks_by_id[task_id] = row + task_project_ids[task_id] = project_id + + project_classifications: Dict[str, Optional[str]] = {} + visiting_projects: set[str] = set() + + def get_project_classification(project_id: str) -> Optional[str]: + if project_id in project_classifications: + return project_classifications[project_id] + if project_id in visiting_projects: + raise ValueError(f"Cycle in NetSuite project hierarchy at {project_id}") + visiting_projects.add(project_id) + row = projects_by_id[project_id] + direct_value = row_value(row, "capex_opex") + if direct_value not in (None, ""): + value = normalize_classification(direct_value, class_config) + else: + parent_id = optional_id(row_value(row, "parent_id")) + value = ( + get_project_classification(parent_id) + if parent_id in projects_by_id + else normalize_classification(None, class_config) + ) + visiting_projects.remove(project_id) + project_classifications[project_id] = value + return value + + task_classifications: Dict[str, Optional[str]] = {} + visiting_tasks: set[str] = set() + + def get_task_classification(task_id: str) -> Optional[str]: + if task_id in task_classifications: + return task_classifications[task_id] + if task_id in visiting_tasks: + raise ValueError(f"Cycle in NetSuite project task hierarchy at {task_id}") + visiting_tasks.add(task_id) + row = project_tasks_by_id[task_id] + direct_value = row_value(row, "capex_opex") + parent_id = optional_id(row_value(row, "parent_id")) + if direct_value not in (None, ""): + value = normalize_classification(direct_value, class_config) + elif parent_id in project_tasks_by_id: + if task_project_ids[parent_id] != task_project_ids[task_id]: + raise ValueError( + f"NetSuite project task {task_id} has a parent from another project" + ) + value = get_task_classification(parent_id) + else: + value = get_project_classification(task_project_ids[task_id]) + visiting_tasks.remove(task_id) + task_classifications[task_id] = value + return value + + output: List[Dict[str, Any]] = [] + for project_id in sorted(projects_by_id, key=_id_sort_key): + row = projects_by_id[project_id] + parent_project_id = optional_id(row_value(row, "parent_id")) + parent_id: Any = 0 + if parent_project_id in projects_by_id and parent_project_id != project_id: + parent_id = project_external_id(parent_project_id) + + external_id = project_external_id(project_id) + classification = get_project_classification(project_id) + if class_config.get("required") and not classification: + raise ValueError( + f"NetSuite project {project_id} has no required " + "CAPEX/OPEX classification" + ) + task: Dict[str, Any] = { + "name": required_name(row, "Project"), + "task_id": external_id, + "external_task_id": external_id, + "parent_id": parent_id, + "netsuite": { + "project_id": project_id, + "project_task_id": None, + "activity_id": optional_id(row_value(row, "activity_id")), + "capex_opex": classification, + }, + } + if classification: + task["mandatory_tags"] = {tag_list_name: [classification]} + output.append(task) + + for task_id in sorted(project_tasks_by_id, key=_id_sort_key): + row = project_tasks_by_id[task_id] + project_id = task_project_ids[task_id] + parent_task_id = optional_id(row_value(row, "parent_id")) + if parent_task_id in project_tasks_by_id: + if task_project_ids[parent_task_id] != project_id: + raise ValueError( + f"NetSuite project task {task_id} has a parent from another project" + ) + parent_id = project_task_external_id(parent_task_id) + else: + parent_id = project_external_id(project_id) + + external_id = project_task_external_id(task_id) + classification = get_task_classification(task_id) + if class_config.get("required") and not classification: + raise ValueError( + f"NetSuite project task {task_id} has no required " + "CAPEX/OPEX classification" + ) + task = { + "name": required_name(row, "Project task"), + "task_id": external_id, + "external_task_id": external_id, + "parent_id": parent_id, + "netsuite": { + "project_id": project_id, + "project_task_id": task_id, + "activity_id": optional_id(row_value(row, "activity_id")), + "capex_opex": classification, + }, + } + estimated_seconds = estimate_seconds(row) + if estimated_seconds is not None: + task["original_estimate_seconds"] = estimated_seconds + if classification: + task["mandatory_tags"] = {tag_list_name: [classification]} + output.append(task) + + _validate_output_hierarchy(output) + return output + + +def _id_sort_key(value: str) -> tuple: + try: + return (0, int(value)) + except ValueError: + return (1, value.casefold()) + + +def _validate_output_hierarchy(tasks: List[Dict[str, Any]]) -> None: + tasks_by_id = {str(task["task_id"]): task for task in tasks} + for task in tasks: + seen = set() + current = task + while current.get("parent_id") not in (None, 0, "0"): + current_id = str(current["task_id"]) + if current_id in seen: + raise ValueError(f"Cycle in generated hierarchy at {current_id}") + seen.add(current_id) + parent_id = str(current["parent_id"]) + parent = tasks_by_id.get(parent_id) + if parent is None: + raise ValueError( + f"Generated task {current_id} references missing parent {parent_id}" + ) + current = parent + + +class NetSuiteFetcher: + def __init__( + self, + config: Dict[str, Any], + client: Optional[NetSuiteClient] = None, + ): + self.config = config + self.client = client or NetSuiteClient.from_env() + + def fetch_all_data(self) -> List[Dict[str, Any]]: + suiteql = self.config.get("suiteql") or {} + if not isinstance(suiteql, dict): + raise ValueError("suiteql config must be an object") + projects_query = str(suiteql.get("projects") or "").strip() + project_tasks_query = str(suiteql.get("project_tasks") or "").strip() + if not projects_query or not project_tasks_query: + raise ValueError( + "suiteql.projects and suiteql.project_tasks must be configured" + ) + + print("Fetching projects from NetSuite...") + projects = self.client.suiteql(projects_query) + print(f" Found {len(projects)} project rows") + print("Fetching project tasks from NetSuite...") + project_tasks = self.client.suiteql(project_tasks_query) + print(f" Found {len(project_tasks)} project task rows") + return build_task_structure(projects, project_tasks, self.config) + + def save_to_json(self, data: List[Dict[str, Any]], filename: str) -> str: + Path(filename).write_text( + json.dumps(data, indent=2, ensure_ascii=False) + "\n", + encoding="utf-8", + ) + return filename + + +def parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Fetch NetSuite projects and project tasks for TimeCamp sync." + ) + parser.add_argument( + "--config", + default=os.getenv("NETSUITE_CONFIG_FILE", DEFAULT_CONFIG_FILE), + help=f"NetSuite mapping JSON (default: {DEFAULT_CONFIG_FILE})", + ) + parser.add_argument( + "-o", + "--output", + default=DEFAULT_OUTPUT_FILE, + help=f"Output tasks JSON (default: {DEFAULT_OUTPUT_FILE})", + ) + return parser.parse_args(argv) + + +def main(argv: Optional[List[str]] = None) -> None: + load_dotenv(override=True) + args = parse_args(argv) + config = load_config(args.config) + fetcher = NetSuiteFetcher(config) + data = fetcher.fetch_all_data() + fetcher.save_to_json(data, args.output) + projects = sum( + 1 + for task in data + if str(task["task_id"]).startswith(NETSUITE_PROJECT_PREFIX) + and not str(task["task_id"]).startswith(NETSUITE_PROJECT_TASK_PREFIX) + ) + print(f"Saved {len(data)} items to {args.output}") + print(f" Projects: {projects}") + print(f" Project tasks: {len(data) - projects}") + print(" Users synchronized: 0 (intentionally out of scope)") + + +if __name__ == "__main__": + main() diff --git a/netsuite_config.example.json b/netsuite_config.example.json new file mode 100644 index 0000000..949cdb0 --- /dev/null +++ b/netsuite_config.example.json @@ -0,0 +1,48 @@ +{ + "suiteql": { + "projects": "SELECT id, entityid AS name, parent AS parent_id FROM job WHERE isinactive = 'F' ORDER BY id", + "project_tasks": "SELECT id, title AS name, company AS project_id, parent AS parent_id, estimatedwork AS estimated_work_hours FROM projecttask ORDER BY company, id" + }, + "classification": { + "tag_list_name": "CAPEX / OPEX", + "allowed_values": [ + "CAPEX", + "OPEX" + ], + "required": true, + "value_map": { + "1": "CAPEX", + "2": "OPEX", + "capital": "CAPEX", + "operating": "OPEX" + }, + "default": null + }, + "time_export": { + "record_type": "timebill", + "employees_query": "SELECT id, email FROM employee WHERE isinactive = 'F' AND email IS NOT NULL ORDER BY id", + "employee_mapping": {}, + "external_id_prefix": "timecamp-", + "duration_rounding": "nearest", + "fields": { + "employee": "employee", + "date": "tranDate", + "hours": "hours", + "project": "customer", + "project_task": "caseTaskEvent", + "activity": null, + "memo": "memo" + }, + "default_activity_id": null, + "classification": { + "mode": "field", + "field": "REPLACE_WITH_WCG_CAPEX_OPEX_TIMEBILL_FIELD", + "value_format": "id", + "value_map": { + "CAPEX": "REPLACE_WITH_WCG_CAPEX_VALUE_ID", + "OPEX": "REPLACE_WITH_WCG_OPEX_VALUE_ID" + } + }, + "fixed_fields": {} + } +} diff --git a/requirements.txt b/requirements.txt index af23e3c..0be3ae2 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,3 +3,4 @@ requests==2.33.0 python-dotenv==1.2.2 azure-devops==7.1.0b3 jira==3.8.0 +PyJWT[crypto]==2.10.1 diff --git a/src/netsuite_client.py b/src/netsuite_client.py new file mode 100644 index 0000000..85e27aa --- /dev/null +++ b/src/netsuite_client.py @@ -0,0 +1,334 @@ +import os +import re +import time +import uuid +from pathlib import Path +from typing import Any, Callable, Dict, List, Optional +from urllib.parse import quote + +import requests + +NETSUITE_REQUEST_TIMEOUT = 30 +NETSUITE_SUITEQL_PAGE_SIZE = 1000 +NETSUITE_MAX_SUITEQL_RESULTS = 100_000 +NETSUITE_RECORD_TYPE = re.compile(r"^[A-Za-z0-9_]+$") + + +def normalize_account_subdomain(account_id: str) -> str: + """Return the account-specific subdomain used by SuiteTalk.""" + normalized = str(account_id or "").strip().lower().replace("_", "-") + if not normalized: + raise ValueError("NETSUITE_ACCOUNT_ID must not be empty") + return normalized + + +def netsuite_base_url(account_id: str) -> str: + account = normalize_account_subdomain(account_id) + return f"https://{account}.suitetalk.api.netsuite.com/services/rest" + + +class StaticAccessTokenProvider: + def __init__(self, access_token: str): + self.access_token = str(access_token or "").strip() + if not self.access_token: + raise ValueError("NETSUITE_ACCESS_TOKEN must not be empty") + + def __call__(self) -> str: + return self.access_token + + +class OAuth2ClientCredentialsTokenProvider: + """Get and cache a NetSuite OAuth 2.0 M2M access token.""" + + def __init__( + self, + account_id: str, + client_id: str, + certificate_id: str, + private_key_file: str, + private_key_passphrase: Optional[str] = None, + algorithm: str = "PS256", + base_url: Optional[str] = None, + session: Optional[requests.Session] = None, + now: Callable[[], float] = time.time, + ): + self.account_id = account_id + self.client_id = str(client_id or "").strip() + self.certificate_id = str(certificate_id or "").strip() + self.private_key_file = Path(private_key_file) + self.private_key_passphrase = private_key_passphrase + self.algorithm = str(algorithm or "PS256").strip().upper() + self.base_url = (base_url or netsuite_base_url(account_id)).rstrip("/") + self.session = session or requests.Session() + self.now = now + self._access_token: Optional[str] = None + self._expires_at = 0.0 + + if not self.client_id: + raise ValueError("NETSUITE_CLIENT_ID must not be empty") + if not self.certificate_id: + raise ValueError("NETSUITE_CERTIFICATE_ID must not be empty") + if self.algorithm not in {"PS256", "PS384", "PS512", "ES256", "ES384", "ES512"}: + raise ValueError(f"Unsupported NETSUITE_JWT_ALGORITHM: {self.algorithm}") + + @property + def token_url(self) -> str: + return f"{self.base_url}/auth/oauth2/v1/token" + + def __call__(self) -> str: + now = self.now() + if self._access_token and now < self._expires_at - 60: + return self._access_token + + assertion = self._create_client_assertion(int(now)) + response = self.session.post( + self.token_url, + data={ + "grant_type": "client_credentials", + "client_assertion_type": ( + "urn:ietf:params:oauth:client-assertion-type:jwt-bearer" + ), + "client_assertion": assertion, + }, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + timeout=NETSUITE_REQUEST_TIMEOUT, + ) + response.raise_for_status() + payload = response.json() + access_token = str(payload.get("access_token") or "").strip() + if not access_token: + raise ValueError("NetSuite OAuth token response has no access_token") + + try: + expires_in = int(payload.get("expires_in", 3600)) + except (TypeError, ValueError): + expires_in = 3600 + + self._access_token = access_token + self._expires_at = now + max(1, expires_in) + return access_token + + def _create_client_assertion(self, now: int) -> str: + try: + import jwt + from cryptography.hazmat.primitives import serialization + except ModuleNotFoundError as exc: + raise RuntimeError( + "OAuth 2.0 M2M requires PyJWT[crypto]; install requirements.txt" + ) from exc + + try: + private_key_pem = self.private_key_file.read_bytes() + except OSError as exc: + raise ValueError( + f"Cannot read NETSUITE_PRIVATE_KEY_FILE: {self.private_key_file}" + ) from exc + + password = None + if self.private_key_passphrase: + password = self.private_key_passphrase.encode("utf-8") + + private_key = serialization.load_pem_private_key( + private_key_pem, + password=password, + ) + claims = { + "iss": self.client_id, + "scope": ["rest_webservices"], + "aud": self.token_url, + "iat": now, + "exp": now + 300, + "jti": uuid.uuid4().hex, + } + return jwt.encode( + claims, + private_key, + algorithm=self.algorithm, + headers={"typ": "JWT", "kid": self.certificate_id}, + ) + + +def token_provider_from_env( + account_id: str, + base_url: Optional[str] = None, +) -> Callable[[], str]: + access_token = os.getenv("NETSUITE_ACCESS_TOKEN") + if access_token: + return StaticAccessTokenProvider(access_token) + + required_values = { + "NETSUITE_CLIENT_ID": os.getenv("NETSUITE_CLIENT_ID"), + "NETSUITE_CERTIFICATE_ID": os.getenv("NETSUITE_CERTIFICATE_ID"), + "NETSUITE_PRIVATE_KEY_FILE": os.getenv("NETSUITE_PRIVATE_KEY_FILE"), + } + missing = [name for name, value in required_values.items() if not value] + if missing: + raise ValueError( + "Set NETSUITE_ACCESS_TOKEN or all OAuth 2.0 M2M variables: " + + ", ".join(missing) + ) + + return OAuth2ClientCredentialsTokenProvider( + account_id=account_id, + client_id=required_values["NETSUITE_CLIENT_ID"], + certificate_id=required_values["NETSUITE_CERTIFICATE_ID"], + private_key_file=required_values["NETSUITE_PRIVATE_KEY_FILE"], + private_key_passphrase=os.getenv("NETSUITE_PRIVATE_KEY_PASSPHRASE"), + algorithm=os.getenv("NETSUITE_JWT_ALGORITHM", "PS256"), + base_url=base_url, + ) + + +class NetSuiteClient: + """Small SuiteTalk REST client for SuiteQL and record upserts.""" + + def __init__( + self, + account_id: str, + access_token_provider: Optional[Callable[[], str]] = None, + base_url: Optional[str] = None, + session: Optional[requests.Session] = None, + ): + self.account_id = str(account_id or "").strip() + if not self.account_id: + raise ValueError("NETSUITE_ACCOUNT_ID must be set") + + self.base_url = ( + base_url + or os.getenv("NETSUITE_BASE_URL") + or netsuite_base_url(self.account_id) + ).rstrip("/") + self.access_token_provider = access_token_provider or token_provider_from_env( + self.account_id, + self.base_url, + ) + self.session = session or requests.Session() + self.session.headers.update( + { + "Accept": "application/json", + "Content-Type": "application/json", + "User-Agent": "TimeCamp-NetSuite-Sync", + } + ) + + @classmethod + def from_env(cls) -> "NetSuiteClient": + account_id = os.getenv("NETSUITE_ACCOUNT_ID") + if not account_id: + raise ValueError("NETSUITE_ACCOUNT_ID must be set in .env") + return cls(account_id) + + def _request( + self, + method: str, + path: str, + *, + json: Optional[Dict[str, Any]] = None, + params: Optional[Dict[str, Any]] = None, + headers: Optional[Dict[str, str]] = None, + ) -> Any: + request_headers = { + "Authorization": f"Bearer {self.access_token_provider()}", + } + if headers: + request_headers.update(headers) + + response = self.session.request( + method, + f"{self.base_url}/{path.lstrip('/')}", + json=json, + params=params, + headers=request_headers, + timeout=NETSUITE_REQUEST_TIMEOUT, + ) + response.raise_for_status() + if response.status_code == 204 or not response.content: + return {} + return response.json() + + def suiteql( + self, + query: str, + page_size: int = NETSUITE_SUITEQL_PAGE_SIZE, + ) -> List[Dict[str, Any]]: + normalized_query = str(query or "").strip() + if not normalized_query: + raise ValueError("SuiteQL query must not be empty") + if page_size < 1 or page_size > NETSUITE_SUITEQL_PAGE_SIZE: + raise ValueError( + f"SuiteQL page_size must be between 1 and {NETSUITE_SUITEQL_PAGE_SIZE}" + ) + + rows: List[Dict[str, Any]] = [] + offset = 0 + while True: + payload = self._request( + "POST", + "query/v1/suiteql", + json={"q": normalized_query}, + params={"limit": page_size, "offset": offset}, + headers={"Prefer": "transient"}, + ) + if not isinstance(payload, dict): + raise ValueError("Unexpected NetSuite SuiteQL response") + page = payload.get("items", []) + if not isinstance(page, list): + raise ValueError("Unexpected NetSuite SuiteQL items response") + rows.extend(page) + + if len(rows) > NETSUITE_MAX_SUITEQL_RESULTS: + raise ValueError( + "SuiteQL result exceeds NetSuite's 100,000 row REST limit" + ) + if not payload.get("hasMore"): + break + + try: + next_offset = int(payload.get("offset", offset)) + int( + payload.get("count", len(page)) + ) + except (TypeError, ValueError) as exc: + raise ValueError( + "Invalid NetSuite SuiteQL pagination response" + ) from exc + if next_offset <= offset: + raise ValueError("NetSuite SuiteQL pagination did not advance") + offset = next_offset + + return rows + + def get_metadata(self, record_types: Optional[List[str]] = None) -> Dict[str, Any]: + params = None + if record_types: + params = {"select": ",".join(record_types)} + payload = self._request( + "GET", + "record/v1/metadata-catalog", + params=params, + headers={"Accept": "application/swagger+json"}, + ) + if not isinstance(payload, dict): + raise ValueError("Unexpected NetSuite metadata response") + return payload + + def upsert_record( + self, + record_type: str, + external_id: str, + payload: Dict[str, Any], + ) -> Any: + normalized_record_type = str(record_type or "").strip() + normalized_external_id = str(external_id or "").strip() + if not normalized_record_type: + raise ValueError("NetSuite record_type must not be empty") + if not NETSUITE_RECORD_TYPE.fullmatch(normalized_record_type): + raise ValueError("NetSuite record_type contains unsupported characters") + if not normalized_external_id: + raise ValueError("NetSuite external_id must not be empty") + + encoded_external_id = quote(normalized_external_id, safe="-_") + return self._request( + "PUT", + f"record/v1/{normalized_record_type}/eid:{encoded_external_id}", + json=payload, + ) diff --git a/sync_projects.py b/sync_projects.py index c2d028d..26085c2 100644 --- a/sync_projects.py +++ b/sync_projects.py @@ -34,6 +34,7 @@ TIMECAMP_API_TOKEN = os.getenv('TIMECAMP_API_TOKEN') TIMECAMP_TASK_ID = os.getenv('TIMECAMP_TASK_ID') TIMECAMP_SYNC_ACTIONS = os.getenv('TIMECAMP_SYNC_ACTIONS') +TIMECAMP_SYNC_EXTERNAL_ID_PREFIX = os.getenv('TIMECAMP_SYNC_EXTERNAL_ID_PREFIX') TIMECAMP_STRICT_USER_SYNC = os.getenv('TIMECAMP_STRICT_USER_SYNC') TIMECAMP_MAX_MANDATORY_TAGS_TO_ADD = os.getenv('TIMECAMP_MAX_MANDATORY_TAGS_TO_ADD') TIMECAMP_MANDATORY_TAG_CACHE_FILE = os.getenv( @@ -489,6 +490,42 @@ def get_source_external_task_id(task): return f"sync_{task_id}" + +def is_timecamp_task_in_sync_scope( + external_id, + source_external_ids, + configured_prefix=None, +): + """Return whether a TimeCamp task is owned by this synchronization run.""" + if not external_id: + return False + + external_id = str(external_id) + prefix = str(configured_prefix or "").strip() + if prefix: + return external_id.startswith(prefix) + + return external_id.startswith('sync_') or external_id in source_external_ids + + +def validate_source_external_id_scope(source_external_ids, configured_prefix=None): + prefix = str(configured_prefix or "").strip() + if not prefix: + return + + out_of_scope = sorted( + external_id + for external_id in source_external_ids + if not str(external_id).startswith(prefix) + ) + if out_of_scope: + preview = ", ".join(out_of_scope[:5]) + raise ValueError( + "Source external_task_id values do not match " + f"TIMECAMP_SYNC_EXTERNAL_ID_PREFIX={prefix!r}: {preview}" + ) + + def load_tasks_from_json(filename=DEFAULT_TASKS_FILE): """Load hierarchical tasks from JSON file""" try: @@ -522,6 +559,15 @@ def sync_hierarchical_tasks_to_timecamp( if not azure_tasks: return + source_external_ids = { + get_source_external_task_id(task) + for task in azure_tasks + } + validate_source_external_id_scope( + source_external_ids, + TIMECAMP_SYNC_EXTERNAL_ID_PREFIX, + ) + client = TimeCampClient(TIMECAMP_API_TOKEN) api_metrics_before_setup = get_api_metrics_snapshot(client) @@ -549,17 +595,14 @@ def sync_hierarchical_tasks_to_timecamp( if "users" in enabled_actions: assigned_user_sync = build_assigned_user_sync_result(client, azure_tasks) - source_external_ids = { - get_source_external_task_id(task) - for task in azure_tasks - } - # Create mapping of existing TimeCamp tasks by external_task_id timecamp_tasks_map = {} for entry in timecamp_entries: external_id = entry.get('external_task_id') - if external_id and ( - external_id.startswith('sync_') or external_id in source_external_ids + if is_timecamp_task_in_sync_scope( + external_id, + source_external_ids, + TIMECAMP_SYNC_EXTERNAL_ID_PREFIX, ): timecamp_tasks_map[external_id] = entry diff --git a/tests/test_export_time_entries_netsuite.py b/tests/test_export_time_entries_netsuite.py new file mode 100644 index 0000000..e4ff290 --- /dev/null +++ b/tests/test_export_time_entries_netsuite.py @@ -0,0 +1,217 @@ +import unittest + +from export_time_entries_netsuite import ( + build_employee_mapping, + format_netsuite_hours, + prepare_timebills, +) + + +class ExportTimeEntriesNetSuiteTest(unittest.TestCase): + def setUp(self): + self.config = { + "time_export": { + "duration_rounding": "nearest", + "fields": { + "employee": "employee", + "date": "tranDate", + "hours": "hours", + "project": "customer", + "project_task": "caseTaskEvent", + "activity": "item", + "memo": "memo", + }, + "classification": { + "mode": "field", + "field": "custcol_capex_opex", + "value_format": "id", + "value_map": {"CAPEX": "11", "OPEX": "12"}, + }, + "fixed_fields": {"isBillable": True}, + } + } + self.timecamp_tasks = [ + { + "task_id": "900", + "external_task_id": "netsuite_project_task_100", + } + ] + self.source_tasks = [ + { + "task_id": "netsuite_project_task_100", + "netsuite": { + "project_id": "10", + "project_task_id": "100", + "activity_id": "501", + "capex_opex": "CAPEX", + }, + } + ] + + def test_employee_mapping_uses_unique_email_and_explicit_override(self): + timecamp_users = [ + {"user_id": "1", "email": "One@Example.com"}, + {"user_id": "2", "email": "duplicate@example.com"}, + {"user_id": "3", "email": "explicit@example.com"}, + ] + netsuite_employees = [ + {"ID": "101", "EMAIL": "one@example.com"}, + {"id": "201", "email": "duplicate@example.com"}, + {"id": "202", "email": "duplicate@example.com"}, + ] + + mapping = build_employee_mapping( + timecamp_users, + netsuite_employees, + {"3": "303"}, + ) + + self.assertEqual(mapping, {"1": "101", "3": "303"}) + + def test_employee_mapping_rejects_non_object_override(self): + with self.assertRaisesRegex(ValueError, "must be an object"): + build_employee_mapping([], [], ["invalid"]) + + def test_formats_duration_at_netsuite_minute_precision(self): + self.assertEqual(format_netsuite_hours(5400), "1:30") + self.assertEqual(format_netsuite_hours(3631, "nearest"), "1:01") + self.assertEqual(format_netsuite_hours(3659, "floor"), "1:00") + self.assertEqual(format_netsuite_hours(3601, "ceil"), "1:01") + with self.assertRaisesRegex(ValueError, "not a whole minute"): + format_netsuite_hours(3601, "reject") + + def test_prepares_idempotent_timebill_payload(self): + entries = [ + { + "id": "700", + "task_id": "900", + "user_id": "1", + "date": "2026-08-03", + "duration": "5431", + "description": "Architecture workshop", + } + ] + + prepared, skipped, errors = prepare_timebills( + entries, + self.timecamp_tasks, + self.source_tasks, + {"1": "101"}, + self.config, + ) + + self.assertEqual(skipped, {}) + self.assertEqual(errors, []) + self.assertEqual(len(prepared), 1) + self.assertEqual(prepared[0].external_id, "timecamp-700") + self.assertEqual( + prepared[0].payload, + { + "isBillable": True, + "employee": {"id": "101"}, + "tranDate": "2026-08-03", + "hours": "1:31", + "customer": {"id": "10"}, + "caseTaskEvent": {"id": "100"}, + "item": {"id": "501"}, + "memo": "Architecture workshop", + "custcol_capex_opex": {"id": "11"}, + }, + ) + + def test_non_netsuite_entries_are_skipped_but_unmapped_employee_blocks(self): + entries = [ + { + "id": "700", + "task_id": "900", + "user_id": "missing", + "date": "2026-08-03", + "duration": 3600, + }, + { + "id": "701", + "task_id": "901", + "user_id": "1", + "date": "2026-08-03", + "duration": 3600, + }, + ] + timecamp_tasks = self.timecamp_tasks + [ + {"task_id": "901", "external_task_id": "jira_123"} + ] + + prepared, skipped, errors = prepare_timebills( + entries, + timecamp_tasks, + self.source_tasks, + {}, + self.config, + ) + + self.assertEqual(prepared, []) + self.assertEqual(skipped["non_netsuite_task"], 1) + self.assertEqual(len(errors), 1) + self.assertIn("no NetSuite employee mapping", errors[0]) + + def test_project_classification_mode_does_not_write_custom_field(self): + config = { + "time_export": { + "classification": {"mode": "project"}, + } + } + entries = [ + { + "id": "700", + "task_id": "900", + "user_id": "1", + "date": "2026-08-03", + "duration": 3600, + } + ] + + prepared, _skipped, errors = prepare_timebills( + entries, + self.timecamp_tasks, + self.source_tasks, + {"1": "101"}, + config, + ) + + self.assertEqual(errors, []) + self.assertNotIn("custcol_capex_opex", prepared[0].payload) + + def test_placeholder_classification_config_blocks_export(self): + config = { + "time_export": { + "classification": { + "mode": "field", + "field": "REPLACE_WITH_WCG_FIELD", + "value_map": {"CAPEX": "REPLACE_WITH_WCG_VALUE"}, + } + } + } + entries = [ + { + "id": "700", + "task_id": "900", + "user_id": "1", + "date": "2026-08-03", + "duration": 3600, + } + ] + + prepared, _skipped, errors = prepare_timebills( + entries, + self.timecamp_tasks, + self.source_tasks, + {"1": "101"}, + config, + ) + + self.assertEqual(prepared, []) + self.assertEqual(len(errors), 1) + self.assertIn("placeholder", errors[0]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_fetch_netsuite.py b/tests/test_fetch_netsuite.py new file mode 100644 index 0000000..7bead37 --- /dev/null +++ b/tests/test_fetch_netsuite.py @@ -0,0 +1,130 @@ +import unittest + +from fetch_netsuite import NetSuiteFetcher, build_task_structure + + +class FakeNetSuiteClient: + def __init__(self, responses): + self.responses = list(responses) + self.queries = [] + + def suiteql(self, query): + self.queries.append(query) + return self.responses.pop(0) + + +class FetchNetSuiteTest(unittest.TestCase): + def setUp(self): + self.config = { + "classification": { + "tag_list_name": "CAPEX / OPEX", + "allowed_values": ["CAPEX", "OPEX"], + "value_map": {"1": "CAPEX", "2": "OPEX"}, + } + } + + def test_builds_hierarchy_classification_estimates_and_no_user_sync(self): + projects = [ + {"ID": "10", "NAME": "Parent project", "capex_opex": "1"}, + {"id": "20", "name": "Child project", "parent_id": "10"}, + {"id": "99", "name": "Inactive", "is_inactive": "T"}, + ] + project_tasks = [ + { + "id": "100", + "name": "Design", + "project_id": "20", + "estimated_work_hours": "1.5", + "activity_id": "501", + }, + { + "id": "200", + "name": "Delivery", + "project_id": "20", + "parent_id": "100", + "capex_opex": "2", + }, + ] + + tasks = build_task_structure(projects, project_tasks, self.config) + + self.assertEqual( + [task["task_id"] for task in tasks], + [ + "netsuite_project_10", + "netsuite_project_20", + "netsuite_project_task_100", + "netsuite_project_task_200", + ], + ) + self.assertEqual(tasks[1]["parent_id"], "netsuite_project_10") + self.assertEqual(tasks[2]["parent_id"], "netsuite_project_20") + self.assertEqual(tasks[3]["parent_id"], "netsuite_project_task_100") + self.assertEqual(tasks[2]["original_estimate_seconds"], 5400) + self.assertEqual(tasks[2]["netsuite"]["activity_id"], "501") + self.assertEqual( + tasks[2]["mandatory_tags"], + {"CAPEX / OPEX": ["CAPEX"]}, + ) + self.assertEqual( + tasks[3]["mandatory_tags"], + {"CAPEX / OPEX": ["OPEX"]}, + ) + self.assertTrue( + all("assigned_users" not in task for task in tasks), + "NetSuite import must not synchronize users", + ) + + def test_rejects_unknown_financial_classification(self): + projects = [{"id": "10", "name": "Project", "capex_opex": "MAYBE"}] + + with self.assertRaisesRegex(ValueError, "Unknown CAPEX/OPEX"): + build_task_structure(projects, [], self.config) + + def test_required_classification_rejects_unclassified_project(self): + config = { + "classification": { + "required": True, + "allowed_values": ["CAPEX", "OPEX"], + } + } + + with self.assertRaisesRegex(ValueError, "no required CAPEX/OPEX"): + build_task_structure( + [{"id": "10", "name": "Unclassified"}], + [], + config, + ) + + def test_rejects_project_task_without_active_project(self): + with self.assertRaisesRegex(ValueError, "missing active project 10"): + build_task_structure( + [], + [{"id": "100", "name": "Task", "project_id": "10"}], + self.config, + ) + + def test_fetcher_executes_only_project_and_project_task_queries(self): + config = { + **self.config, + "suiteql": { + "projects": "projects query", + "project_tasks": "tasks query", + }, + } + client = FakeNetSuiteClient( + [ + [{"id": "10", "name": "Project"}], + [{"id": "100", "name": "Task", "project_id": "10"}], + ] + ) + + tasks = NetSuiteFetcher(config, client).fetch_all_data() + + self.assertEqual(client.queries, ["projects query", "tasks query"]) + self.assertEqual(len(tasks), 2) + self.assertTrue(all("assigned_users" not in task for task in tasks)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_netsuite_client.py b/tests/test_netsuite_client.py new file mode 100644 index 0000000..2ac7a70 --- /dev/null +++ b/tests/test_netsuite_client.py @@ -0,0 +1,152 @@ +import tempfile +import unittest +from pathlib import Path + +import jwt +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import rsa + +from src.netsuite_client import ( + NETSUITE_REQUEST_TIMEOUT, + NetSuiteClient, + OAuth2ClientCredentialsTokenProvider, + StaticAccessTokenProvider, + netsuite_base_url, + normalize_account_subdomain, +) + + +class FakeResponse: + def __init__(self, payload=None, status_code=200): + self.payload = payload + self.status_code = status_code + self.content = b"" if payload is None else b"json" + + def raise_for_status(self): + pass + + def json(self): + return self.payload + + +class FakeSession: + def __init__(self, responses): + self.headers = {} + self.responses = list(responses) + self.calls = [] + + def request(self, method, url, **kwargs): + self.calls.append({"method": method, "url": url, **kwargs}) + return self.responses.pop(0) + + +class NetSuiteClientTest(unittest.TestCase): + def test_normalizes_sandbox_account_for_account_specific_url(self): + self.assertEqual(normalize_account_subdomain("1234567_SB1"), "1234567-sb1") + self.assertEqual( + netsuite_base_url("1234567_SB1"), + "https://1234567-sb1.suitetalk.api.netsuite.com/services/rest", + ) + + def test_static_access_token_provider_rejects_empty_token(self): + with self.assertRaisesRegex(ValueError, "must not be empty"): + StaticAccessTokenProvider(" ") + + def test_suiteql_paginates_and_sends_required_prefer_header(self): + session = FakeSession( + [ + FakeResponse( + { + "items": [{"id": "1"}, {"id": "2"}], + "count": 2, + "offset": 0, + "hasMore": True, + } + ), + FakeResponse( + { + "items": [{"id": "3"}], + "count": 1, + "offset": 2, + "hasMore": False, + } + ), + ] + ) + client = NetSuiteClient( + "1234567", + access_token_provider=lambda: "access-token", + session=session, + ) + + rows = client.suiteql("SELECT id FROM job ORDER BY id", page_size=2) + + self.assertEqual(rows, [{"id": "1"}, {"id": "2"}, {"id": "3"}]) + self.assertEqual( + [call["params"]["offset"] for call in session.calls], + [0, 2], + ) + self.assertEqual(session.calls[0]["headers"]["Prefer"], "transient") + self.assertEqual( + session.calls[0]["headers"]["Authorization"], + "Bearer access-token", + ) + self.assertEqual(session.calls[0]["timeout"], NETSUITE_REQUEST_TIMEOUT) + + def test_upsert_uses_put_and_external_id_url(self): + session = FakeSession([FakeResponse(status_code=204)]) + client = NetSuiteClient( + "1234567", + access_token_provider=lambda: "access-token", + session=session, + ) + + response = client.upsert_record( + "timebill", + "timecamp-12:34", + {"hours": "1:30"}, + ) + + self.assertEqual(response, {}) + self.assertEqual(session.calls[0]["method"], "PUT") + self.assertTrue( + session.calls[0]["url"].endswith("/record/v1/timebill/eid:timecamp-12%3A34") + ) + self.assertEqual(session.calls[0]["json"], {"hours": "1:30"}) + + def test_oauth2_m2m_assertion_has_required_netsuite_claims(self): + private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + private_key_pem = private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ) + with tempfile.TemporaryDirectory() as temp_dir: + key_path = Path(temp_dir) / "private-key.pem" + key_path.write_bytes(private_key_pem) + provider = OAuth2ClientCredentialsTokenProvider( + account_id="1234567", + client_id="client-id", + certificate_id="certificate-id", + private_key_file=str(key_path), + now=lambda: 1_800_000_000, + ) + + assertion = provider._create_client_assertion(1_800_000_000) + + header = jwt.get_unverified_header(assertion) + claims = jwt.decode( + assertion, + options={"verify_signature": False, "verify_exp": False}, + ) + self.assertEqual(header["alg"], "PS256") + self.assertEqual(header["kid"], "certificate-id") + self.assertEqual(claims["iss"], "client-id") + self.assertEqual(claims["scope"], ["rest_webservices"]) + self.assertEqual(claims["aud"], provider.token_url) + self.assertEqual(claims["iat"], 1_800_000_000) + self.assertEqual(claims["exp"], 1_800_000_300) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_sync_projects.py b/tests/test_sync_projects.py index aa2b67a..0280e96 100644 --- a/tests/test_sync_projects.py +++ b/tests/test_sync_projects.py @@ -11,6 +11,77 @@ class SyncProjectsCliTest(unittest.TestCase): + def test_external_id_prefix_isolates_integration_archive_scope(self): + source_ids = {"netsuite_project_1"} + + self.assertTrue( + sync_projects.is_timecamp_task_in_sync_scope( + "netsuite_project_1", source_ids, "netsuite_" + ) + ) + self.assertTrue( + sync_projects.is_timecamp_task_in_sync_scope( + "netsuite_project_2", source_ids, "netsuite_" + ), + "stale NetSuite tasks must remain visible so archive can find them", + ) + self.assertFalse( + sync_projects.is_timecamp_task_in_sync_scope( + "sync_jira_1", source_ids, "netsuite_" + ) + ) + + def test_default_external_id_scope_remains_backward_compatible(self): + source_ids = {"monday_1"} + + self.assertTrue( + sync_projects.is_timecamp_task_in_sync_scope( + "sync_jira_1", source_ids + ) + ) + self.assertTrue( + sync_projects.is_timecamp_task_in_sync_scope("monday_1", source_ids) + ) + self.assertFalse( + sync_projects.is_timecamp_task_in_sync_scope("netsuite_old", source_ids) + ) + + def test_rejects_source_ids_outside_configured_archive_scope(self): + with self.assertRaisesRegex(ValueError, "do not match"): + sync_projects.validate_source_external_id_scope( + {"netsuite_project_1", "sync_jira_1"}, + "netsuite_", + ) + + def test_rejects_wrong_scope_before_any_timecamp_api_call(self): + with ( + patch.object( + sync_projects, + "load_tasks_from_json", + return_value=[ + { + "task_id": "jira_1", + "external_task_id": "sync_jira_1", + "parent_id": 0, + "name": "Wrong source", + } + ], + ), + patch.object( + sync_projects, + "TIMECAMP_SYNC_EXTERNAL_ID_PREFIX", + "netsuite_", + ), + patch.object(sync_projects, "TimeCampClient") as timecamp_client, + ): + with self.assertRaisesRegex(ValueError, "do not match"): + sync_projects.sync_hierarchical_tasks_to_timecamp( + {"tasks"}, + "tasks.json", + ) + + timecamp_client.assert_not_called() + def test_main_uses_input_file_for_preview_and_sync(self): with ( patch.object(sync_projects, "TIMECAMP_API_TOKEN", "token"), From eb081d67f22211e605f1f188f42c7b79ac87f2c7 Mon Sep 17 00:00:00 2001 From: Kamil Rudnicki Date: Mon, 10 Aug 2026 18:46:07 +0200 Subject: [PATCH 2/2] Add netsuite_config.json to .gitignore --- .gitignore | 1 + 1 file changed, 1 insertion(+) diff --git a/.gitignore b/.gitignore index f2194f2..a64ee60 100644 --- a/.gitignore +++ b/.gitignore @@ -22,3 +22,4 @@ tasks.json task*.json projects.json +netsuite_config.json