madmin/Order-Manager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit

Browse Source
This commit is contained in:
madmin
2026-08-26 09:15:35 -05:00
commit ffaaa0c8c8
23 changed files with 1775 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/services/__init__.py
+15
View File
@@ -0,0 +1,15 @@
"""
Registry of available order sources.
Add new services here as they're built (ShipStation, Odoo, ...).
The UI reads this registry rather than importing specific services
directly, so adding a source doesn't require touching main_window.py.
"""
from app.services.jira_service import JiraService
from app.services.shipstation_service import ShipStationService
SERVICE_REGISTRY = {
"jira": JiraService,
"shipstation": ShipStationService,
# "odoo": OdooService, # Odoo push/pull API coming later - file export exists today, see app/services/odoo_export.py
}
app/services/base.py
+44
View File
@@ -0,0 +1,44 @@
"""
Base interface for anything that can fetch orders.
To add a new source later (ShipStation, Odoo, ...):
1. Create app/services/shipstation_service.py
2. Subclass OrderService, implement fetch_orders()
3. Register it in app/services/__init__.py's SERVICE_REGISTRY
4. Add its settings to app/config.py's SETTINGS_SCHEMA
The UI and database layers don't need to change at all.
"""
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import List, TypedDict, Optional
import datetime as dt
class NormalizedOrder(TypedDict):
"""The common shape every service must translate its data into."""
source: str
external_id: str
ticket_number: Optional[str]
company: str
skus: List[str]
summary: str
status: str
source_created_at: Optional[dt.datetime]
raw_data: dict
class OrderService(ABC):
"""Common interface for every order source."""
#: short machine-friendly name, e.g. "jira", "shipstation", "odoo"
name: str = "base"
@abstractmethod
def fetch_orders(self) -> List[NormalizedOrder]:
"""Fetch and return today's orders, normalized to NormalizedOrder."""
raise NotImplementedError
def is_configured(self) -> bool:
"""Override to check that required settings are present."""
return True
app/services/jira_service.py
+228
View File
@@ -0,0 +1,228 @@
"""
JIRA order source.
Pulls tickets matching a JQL query (configured in .env as JIRA_JQL,
editable from the Settings dialog) and normalizes them into orders.
Auth: JIRA Cloud API token (https://id.atlassian.com/manage-profile/security/api-tokens)
using HTTP Basic auth with your account email + the token.
"""
from __future__ import annotations
import datetime as dt
import re
from typing import List, Tuple
import requests
from app import config
from app.companies import parse_mapping, resolve_company_for_skus
from app.services.base import OrderService, NormalizedOrder
SEARCH_PAGE_SIZE = 50
REQUEST_TIMEOUT_SECONDS = 30
# Deliverable values come back like "SH011: Shipping - Return Label iPad -
# Physical in Box" - the part before the colon is the actual SKU code;
# everything after it is a human-readable description.
DELIVERABLE_CODE_PATTERN = re.compile(r"^\s*([A-Za-z]{2,6}\d{1,6})\s*:\s*(.*)$")
class JiraServiceError(Exception):
"""Raised for any JIRA fetch failure, with a message safe to show in the UI."""
class JiraService(OrderService):
name = "jira"
def __init__(self) -> None:
settings = config.load_settings()
self.base_url = settings["JIRA_URL"].rstrip("/")
self.email = settings["JIRA_EMAIL"]
self.api_token = settings["JIRA_API_TOKEN"]
self.jql = settings["JIRA_JQL"]
# Each company has its own set of "deliverable" custom fields in
# JIRA (e.g. "Deliverables" + "Hardware Needed" for Signify;
# "Oak Street Deliverables" + "Oak Street Hardware Needed" for
# Oak Street). We scan all of them and let the SKU code prefix
# (SH.../OK...) decide the company, same as before - this just
# widens where we look for those codes.
self.signify_field_ids = self._split_field_ids(settings["JIRA_SIGNIFY_SKU_FIELDS"])
self.oakstreet_field_ids = self._split_field_ids(settings["JIRA_OAKSTREET_SKU_FIELDS"])
self.all_sku_field_ids = list(
dict.fromkeys(self.signify_field_ids + self.oakstreet_field_ids)
)
self.sku_map = parse_mapping(
settings["COMPANY_SKU_MAP"] or config.DEFAULT_COMPANY_SKU_MAP
)
@staticmethod
def _split_field_ids(raw: str) -> List[str]:
return [f.strip() for f in (raw or "").split(",") if f.strip()]
def is_configured(self) -> bool:
return bool(self.base_url and self.email and self.api_token and self.jql)
def fetch_orders(self) -> List[NormalizedOrder]:
if not self.is_configured():
raise JiraServiceError(
"JIRA is not fully configured yet. Open Settings and fill in "
"JIRA Site URL, Email, API Token, and JQL."
)
issues = self._search_all_issues()
return [self._to_normalized_order(issue) for issue in issues]
# -- internals -----------------------------------------------------
def _search_all_issues(self) -> List[dict]:
url = f"{self.base_url}/rest/api/3/search"
auth = (self.email, self.api_token)
headers = {"Accept": "application/json"}
# Always pull summary/status/created, plus every configured deliverable field.
fields = "summary,status,created"
if self.all_sku_field_ids:
fields += "," + ",".join(self.all_sku_field_ids)
all_issues: List[dict] = []
start_at = 0
while True:
params = {
"jql": self.jql,
"startAt": start_at,
"maxResults": SEARCH_PAGE_SIZE,
"fields": fields,
}
try:
response = requests.get(
url,
params=params,
auth=auth,
headers=headers,
timeout=REQUEST_TIMEOUT_SECONDS,
)
except requests.RequestException as exc:
raise JiraServiceError(f"Could not reach JIRA: {exc}") from exc
if response.status_code == 401:
raise JiraServiceError(
"JIRA rejected the credentials (401). Check your email and API token."
)
if response.status_code == 400:
raise JiraServiceError(
f"JIRA rejected the query (400). Check your JQL or SKU field ID. "
f"Details: {response.text[:300]}"
)
if not response.ok:
raise JiraServiceError(
f"JIRA returned an error ({response.status_code}): {response.text[:300]}"
)
try:
data = response.json()
except ValueError as exc:
raise JiraServiceError("JIRA returned a response that wasn't valid JSON.") from exc
batch = data.get("issues", [])
all_issues.extend(batch)
total = data.get("total", len(all_issues))
start_at += len(batch)
if start_at >= total or not batch:
break
return all_issues
def _raw_values_from_field(self, fields: dict, field_id: str) -> List[str]:
"""
Pull raw string values out of one custom field. JIRA custom fields
can come back several different ways depending on the field type:
- plain string (text fields)
- a single {"value": ...} object (single-select dropdown fields -
this is the shape "Hardware Needed" uses, and it was previously
being silently dropped since it isn't a list)
- a list of strings or {"value": ...} objects (multi-select fields,
e.g. "Deliverables" which can hold several kit items)
Handle all of these.
"""
raw_value = fields.get(field_id)
if raw_value is None:
return []
if isinstance(raw_value, str):
parts = re.split(r"[\n;]+", raw_value)
return [p.strip() for p in parts if p.strip()]
if isinstance(raw_value, dict):
value = raw_value.get("value") or raw_value.get("name") or raw_value.get("id")
return [str(value).strip()] if value else []
if isinstance(raw_value, list):
values = []
for item in raw_value:
if isinstance(item, str):
values.append(item.strip())
elif isinstance(item, dict):
value = item.get("value") or item.get("name") or item.get("id")
if value:
values.append(str(value).strip())
return [v for v in values if v]
return []
def _extract_skus_and_texts(self, fields: dict) -> Tuple[List[str], List[str]]:
"""
Scan every configured deliverable field and return:
- skus: just the code part (e.g. "SH011"), deduped, order preserved
- texts: the full "CODE: description" strings, for use as a
fallback summary when the ticket's actual Summary is blank
(which is the normal case for these order tickets)
"""
skus: List[str] = []
texts: List[str] = []
for field_id in self.all_sku_field_ids:
for raw_value in self._raw_values_from_field(fields, field_id):
texts.append(raw_value)
match = DELIVERABLE_CODE_PATTERN.match(raw_value)
code = match.group(1) if match else raw_value
if code not in skus:
skus.append(code)
return skus, texts
def _to_normalized_order(self, issue: dict) -> NormalizedOrder:
fields = issue.get("fields", {})
created_raw = fields.get("created")
created_at = None
if created_raw:
try:
# JIRA timestamps look like 2024-01-15T09:30:00.000+0000
created_at = dt.datetime.strptime(created_raw[:19], "%Y-%m-%dT%H:%M:%S")
except ValueError:
created_at = None
ticket_number = issue.get("key", "")
skus, deliverable_texts = self._extract_skus_and_texts(fields)
company = resolve_company_for_skus(skus, self.sku_map)
# These order tickets typically leave the JIRA Summary field
# blank - fall back to the deliverables so there's still
# something readable in the dashboard/table.
summary = fields.get("summary") or "; ".join(deliverable_texts)
return NormalizedOrder(
source="jira",
external_id=ticket_number,
ticket_number=ticket_number,
company=company,
skus=skus,
summary=summary,
status=(fields.get("status") or {}).get("name", ""),
source_created_at=created_at,
raw_data=issue,
)
app/services/odoo_export.py
+44
View File
@@ -0,0 +1,44 @@
"""
Odoo file export.
Odoo integration is still in development on your end, so for now we
generate a CSV of orders that fits your import template. This is a
placeholder column layout - once you know the exact columns your Odoo
import template expects, update ODOO_EXPORT_COLUMNS below (or tell me
and I'll adjust it) and the export will follow automatically.
When live Odoo API access is ready, this file is where an
OdooService(OrderService)-style push would go instead/alongside.
"""
from __future__ import annotations
import csv
from pathlib import Path
from typing import List
from app.models import Order
# (header label, function to get the value from an Order)
ODOO_EXPORT_COLUMNS = [
("Order Reference", lambda o: o.ticket_number or o.external_id),
("Company", lambda o: o.company),
("Source", lambda o: o.source),
("SKUs", lambda o: ", ".join(o.skus or [])),
("Status", lambda o: o.status),
("Summary", lambda o: o.summary),
("Source Created At", lambda o: o.source_created_at.isoformat() if o.source_created_at else ""),
]
def export_orders_to_csv(orders: List[Order], filepath: str) -> int:
"""Write orders to a CSV file. Returns the number of rows written."""
path = Path(filepath)
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow([label for label, _ in ODOO_EXPORT_COLUMNS])
for order in orders:
writer.writerow([getter(order) for _, getter in ODOO_EXPORT_COLUMNS])
return len(orders)
app/services/shipstation_service.py
+167
View File
@@ -0,0 +1,167 @@
"""
ShipStation order source (API V2).
Important context (as of ShipStation's current V2 docs): V2 does not
have a dedicated "list orders" endpoint the way the older V1 API did.
The closest equivalent is GET /v2/shipments, where each shipment
carries a store_id and (depending on how the order arrived) an items
array with SKUs. That's what this service pulls from.
Because we're fetching independently of JIRA (no shared internal ID),
we recover the AR-###### ticket number by scanning the whole shipment
payload for the configured pattern, rather than assuming one fixed
field - so this keeps working even before we've confirmed exactly
which field your team puts it in (order notes, a custom field, a tag,
etc). Once that's confirmed, this can be tightened to read that field
directly for speed and reliability.
"""
from __future__ import annotations
import datetime as dt
import json
import re
from typing import List, Optional
import requests
from app import config
from app.companies import parse_mapping, resolve_company_by_store
from app.services.base import OrderService, NormalizedOrder
API_BASE = "https://api.shipstation.com/v2"
PAGE_SIZE = 100
REQUEST_TIMEOUT_SECONDS = 30
class ShipStationServiceError(Exception):
"""Raised for any ShipStation fetch failure, with a message safe to show in the UI."""
class ShipStationService(OrderService):
name = "shipstation"
def __init__(self) -> None:
settings = config.load_settings()
self.api_key = settings["SHIPSTATION_API_KEY"]
self.store_map = parse_mapping(settings["SHIPSTATION_STORE_MAP"])
self.ticket_pattern = re.compile(
settings["TICKET_NUMBER_REGEX"] or config.DEFAULT_TICKET_NUMBER_REGEX
)
def is_configured(self) -> bool:
return bool(self.api_key and self.store_map)
def fetch_orders(self) -> List[NormalizedOrder]:
if not self.is_configured():
raise ShipStationServiceError(
"ShipStation is not fully configured yet. Open Settings and fill in "
"the API Key and the Store ID -> Company mapping."
)
shipments = self._fetch_todays_shipments()
# Only keep shipments belonging to one of our two known stores.
known_store_ids = set(self.store_map.keys())
relevant = [s for s in shipments if str(s.get("store_id")) in known_store_ids]
return [self._to_normalized_order(s) for s in relevant]
# -- internals -----------------------------------------------------
def _fetch_todays_shipments(self) -> List[dict]:
headers = {"API-Key": self.api_key, "Accept": "application/json"}
today_start = dt.datetime.combine(dt.date.today(), dt.time.min)
today_end = today_start + dt.timedelta(days=1)
all_shipments: List[dict] = []
page = 1
while True:
params = {
"created_at_start": today_start.isoformat() + "Z",
"created_at_end": today_end.isoformat() + "Z",
"page": page,
"page_size": PAGE_SIZE,
"sort_by": "created_at",
"sort_dir": "desc",
}
try:
response = requests.get(
f"{API_BASE}/shipments",
params=params,
headers=headers,
timeout=REQUEST_TIMEOUT_SECONDS,
)
except requests.RequestException as exc:
raise ShipStationServiceError(f"Could not reach ShipStation: {exc}") from exc
if response.status_code == 401:
raise ShipStationServiceError(
"ShipStation rejected the API key (401). Check it in Settings."
)
if not response.ok:
raise ShipStationServiceError(
f"ShipStation returned an error ({response.status_code}): {response.text[:300]}"
)
try:
data = response.json()
except ValueError as exc:
raise ShipStationServiceError(
"ShipStation returned a response that wasn't valid JSON."
) from exc
batch = data.get("shipments", [])
all_shipments.extend(batch)
total_pages = data.get("pages", 1)
if page >= total_pages or not batch:
break
page += 1
return all_shipments
def _extract_ticket_number(self, shipment: dict) -> Optional[str]:
try:
blob = json.dumps(shipment)
except (TypeError, ValueError):
return None
match = self.ticket_pattern.search(blob)
return match.group(0) if match else None
@staticmethod
def _extract_skus(shipment: dict) -> List[str]:
items = shipment.get("items") or []
skus = [item.get("sku") for item in items if isinstance(item, dict) and item.get("sku")]
return skus
def _to_normalized_order(self, shipment: dict) -> NormalizedOrder:
created_raw = shipment.get("created_at")
created_at = None
if created_raw:
try:
created_at = dt.datetime.strptime(created_raw[:19], "%Y-%m-%dT%H:%M:%S")
except ValueError:
created_at = None
store_id = str(shipment.get("store_id", ""))
company = resolve_company_by_store(store_id, self.store_map)
skus = self._extract_skus(shipment)
ticket_number = self._extract_ticket_number(shipment)
external_id = shipment.get("shipment_id", "")
summary_bits = [b for b in [ticket_number, ", ".join(skus)] if b]
summary = " - ".join(summary_bits) or external_id
return NormalizedOrder(
source="shipstation",
external_id=external_id,
ticket_number=ticket_number,
company=company,
skus=skus,
summary=summary,
status=shipment.get("shipment_status", ""),
source_created_at=created_at,
raw_data=shipment,
)