#!/usr/bin/env python3
"""
winston_pipeline.py — Winston Linux quote pipeline (DVI-815).
Port of the legacy outlook_auto/main.py + clickup_quotes.py to run on
fileshare.icastinc.com (Linux), using app-only Graph (winston_graph.py) and
the B1 convert-to-PDF recipe proven in DVI-807. No Windows/Excel COM required.
Entry point: run(date='YYYY-MM-DD')
Callable from the P2 scheduler and the P4 "Run now" UI button.
Pipeline steps per quote task
------------------------------
1. ClickUp pull — fetch quote tasks matching target date
2. Find + download — locate the .xlsm workbook in OneDrive by quote number
3. openpyxl extract — resolve quote sheet(s) from ClickUp tags via tag_map,
trim empty rows + dynamic print area, read grand total
by label search (DVI-982); build one trimmed single-sheet
.xlsx per matched sheet (DVI-983 — often 2+ per quote)
4. Graph convert — upload each .xlsx to scratch drive, GET ?format=pdf; one
PDF per matched sheet, named {quote}{sheetLabel}{ver}.pdf
5. Mail draft — one draft in sales mailbox with ALL the quote's PDFs
attached (idempotent: skip if exists)
6. ClickUp writeback — attach each PDF to task + set Quote Amount custom field
(single field = summed grand total)
7. Routed PDF upload — upload each PDF to OneDrive (owner personal / shared path)
Configuration (env vars, with fallback to winston_settings.json)
-----------------------------------------------------------------
CLICKUP_API_KEY ClickUp personal API token
CLICKUP_LIST_ID ClickUp list ID to query for quote tasks
CLICKUP_DATE_FIELD Task date field to filter on (default: due_date)
CLICKUP_QUOTE_AMOUNT_FIELD_ID Custom field ID for Quote Amount (auto-detected if absent)
WINSTON_ONEDRIVE_USER UPN of the OneDrive owner holding Takeoffs folder
WINSTON_TAKEOFFS_FOLDER OneDrive path to the .xlsm workbooks (see default below)
WINSTON_SERVICE_USER UPN for the scratch drive used during xlsx→PDF conversion
WINSTON_MAIL_FROM Draft mailbox UPN (default: sales@icastinc.com)
AZURE_CLIENT_ID } App-only Graph credentials (already wired in DVI-806)
AZURE_CLIENT_SECRET }
AZURE_TENANT_ID }
"""
import html as _html
import io
import json
import logging
import os
import re
import threading
from datetime import datetime, timezone
from pathlib import Path
import openpyxl
import requests as _req
from openpyxl.utils import get_column_letter
from togen import winston_graph
from togen.clickup_client import ClickUpClient
log = logging.getLogger("togen.winston_pipeline")
# ---------------------------------------------------------------------------
# Cooperative cancellation (DVI-893)
#
# Python threads can't be forcibly killed, so a long-running Winston run is
# stopped cooperatively: the "Stop" endpoint records the run_id here, and the
# pipeline loop checks is_cancelled() at safe points (between quote tasks) and
# returns early, marking the DiagnosticRun as "cancelled".
# ---------------------------------------------------------------------------
_cancel_lock = threading.Lock()
_cancel_requested: set[str] = set()
def request_cancel(run_id: str) -> None:
"""Flag a run for cooperative cancellation (called from the stop endpoint)."""
with _cancel_lock:
_cancel_requested.add(run_id)
def is_cancelled(run_id: str) -> bool:
with _cancel_lock:
return run_id in _cancel_requested
def _clear_cancel(run_id: str) -> None:
with _cancel_lock:
_cancel_requested.discard(run_id)
# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
_SETTINGS_FILE = Path(__file__).resolve().parent / "winston_settings.json"
_USER_SETTINGS_DIR = Path(__file__).resolve().parent / "user_settings"
_EMAIL_TEMPLATE_FILE = Path(__file__).resolve().parent / "winston_email_template.json"
# DVI-999: detect "Send Quotes 26-XXXX" task names when selecting the primary
# task per quote number during tag-union dedup.
_SEND_QUOTES_RE = re.compile(r"(?i)^send\s+quotes?\s+\d")
# DVI-982: quote sheet(s), grand total, and print area are resolved dynamically
# (mirroring PoC run_macros.py) — no hardcoded sheet name / total cell / print
# area. Row-trim bands + check column and the print-area right edge below are
# the only structural constants (they mirror the fixed template geometry the
# PoC hardcodes in hide_rows_via_com).
_TRIM_ROW_BANDS = ((138, 738), (749, 801)) # line-item bands to trim
_TRIM_CHECK_COL = 15 # column O — <1/blank ⇒ empty row
_PRINT_AREA_LAST_COL = 39 # column AM (legacy right edge)
# OneDrive folder containing the per-quote .xlsm workbooks.
# Mirrors the legacy Windows path "Infrastructure Precast Inc_ - Takeoffs".
_DEFAULT_TAKEOFFS_FOLDER = "Infrastructure Precast Inc_ - Takeoffs"
# ---------------------------------------------------------------------------
# Settings helpers (mirrors _load_winston_settings in app.py; no app.py import
# since app.py starts background loops at import time)
# ---------------------------------------------------------------------------
_WINSTON_DEFAULTS: dict = {
"draft_mailbox": "sales@icastinc.com",
"clickup_api_key": "",
"clickup_list_id": "",
# Team/workspace ID used when no list_id is set (matches PoC team-wide query).
# iCast Inc Workspace = 9014248643
"clickup_team_id": "9014248643",
"clickup_date_field": "",
"onedrive_user": "",
"takeoffs_folder": "",
"shared_quotes_location": "",
"shared_quote_takeoff_location": "",
# DVI-981: tag_map — maps ClickUp tag base keys (version suffix stripped) to
# quote sheet names. Seeded from PoC tag_map.json vocabulary.
# custom: exact tag → exact sheet name (case-insensitive match at resolve time)
# standard: each tag expands to candidates "{TAG} Quote 1/2/3"
# Editable in Winston Admin → Settings → Tag map.
"tag_map": {
"custom": {"X": "QuoteX", "Y": "QuoteY", "Z": "QuoteZ", "LP": "LP"},
"standard": ["SPC", "AC", "BW", "BC", "BR", "ES", "EMH", "RW", "SH", "WQ", "WW"],
},
}
def _load_settings() -> dict:
settings = dict(_WINSTON_DEFAULTS)
if _SETTINGS_FILE.is_file():
try:
saved = json.loads(_SETTINGS_FILE.read_text())
if isinstance(saved, dict):
settings.update(saved)
except (json.JSONDecodeError, OSError):
pass
return settings
def _load_user_docs(email: str) -> dict:
"""Return the 'documents' sub-dict from a user's personal settings JSON."""
safe = re.sub(r"[^a-zA-Z0-9@._-]", "_", email.lower())
fp = _USER_SETTINGS_DIR / f"{safe}.json"
if fp.is_file():
try:
return json.loads(fp.read_text()).get("documents", {})
except (json.JSONDecodeError, OSError):
pass
return {}
# ---------------------------------------------------------------------------
# Email draft template (DVI-857)
#
# The subject/body/recipients of Winston draft emails are driven by an
# admin-editable template stored in winston_email_template.json and managed
# from the Winston → Admin → Email tab. The DEFAULTS below reproduce the exact
# subject and body Winston produced before DVI-857, so behavior is unchanged
# until an admin edits the template.
#
# Templates use {token} placeholders. Available tokens (see EMAIL_TOKENS) are
# fed from the pipeline source data for each quote. Token VALUES are HTML-escaped
# when substituted into the body (values come from ClickUp/OneDrive data), while
# the template body itself is authored by a togen admin and treated as trusted.
# ---------------------------------------------------------------------------
EMAIL_TEMPLATE_DEFAULTS: dict = {
"subject": "Quote {quote_number} — iCast Infrastructure",
"body_html": (
"Please find attached the quote for {quote_number}.
"
"Quote Total: {quote_total}
"
"This quote was prepared by iCast Infrastructure. "
"Please review and let us know if you have any questions.
"
"Thank you,
iCast Infrastructure Sales Team
"
),
"cc": [],
"bcc": [],
"fallback_to": [],
}
# Token catalogue surfaced in the Email tab legend: token -> where it comes from.
EMAIL_TOKENS: list[dict] = [
{"token": "quote_number", "source": "ClickUp task — the quote number (workbook / PDF name)"},
{"token": "quote_total", "source": "Grand total read by label from the resolved quote sheet, currency-formatted (e.g. $1,234.56)"},
{"token": "recipient_email", "source": "ClickUp task — the quote's To recipient"},
{"token": "owner_email", "source": "ClickUp task — the quote owner / salesperson"},
{"token": "date", "source": "Pipeline run target date (YYYY-MM-DD)"},
{"token": "mailbox", "source": "Configured send-as mailbox (Settings → Draft mailbox)"},
]
# ---------------------------------------------------------------------------
# Per-pipeline templates (DVI-859 — Option C)
#
# The on-disk store keeps the single global template's fields at the top level
# (backward compatible with the DVI-857 format) plus two additive keys:
# per_pipeline_enabled : bool — feature flag; when false the single global
# template is always used (current behavior).
# pipelines : dict — { pipeline_id: {sparse field overrides} }
#
# Per-pipeline overrides are SPARSE: only fields the admin actually customized
# are stored. Any field a pipeline doesn't set falls back to the global template
# ("fallback to using the single customization"). The feature ships disabled so
# behavior is unchanged; turning the flag on "simply enables" per-pipeline
# customization without any code change.
# ---------------------------------------------------------------------------
def _load_email_store() -> dict:
"""Return the raw on-disk store (global fields + pipelines + flag)."""
if _EMAIL_TEMPLATE_FILE.is_file():
try:
saved = json.loads(_EMAIL_TEMPLATE_FILE.read_text())
if isinstance(saved, dict):
return saved
except (json.JSONDecodeError, OSError):
pass
return {}
def _global_template(store: dict) -> dict:
"""Resolve the global template = DEFAULTS overlaid with saved top-level fields."""
tpl = dict(EMAIL_TEMPLATE_DEFAULTS)
for k, default in EMAIL_TEMPLATE_DEFAULTS.items():
if k in store and type(store[k]) is type(default):
tpl[k] = store[k]
return tpl
def per_pipeline_enabled() -> bool:
"""Whether per-pipeline email templates are turned on."""
return bool(_load_email_store().get("per_pipeline_enabled"))
def set_per_pipeline_enabled(enabled: bool) -> None:
store = _load_email_store()
store["per_pipeline_enabled"] = bool(enabled)
_EMAIL_TEMPLATE_FILE.write_text(json.dumps(store, indent=2))
def get_pipeline_override(pipeline_id: str) -> dict:
"""Return the sparse override dict saved for a pipeline (empty if none)."""
pipelines = _load_email_store().get("pipelines")
ov = pipelines.get(pipeline_id) if isinstance(pipelines, dict) else None
return dict(ov) if isinstance(ov, dict) else {}
def pipeline_override_ids() -> list:
"""Pipeline ids that currently have a saved per-pipeline override."""
pipelines = _load_email_store().get("pipelines")
return list(pipelines.keys()) if isinstance(pipelines, dict) else []
def load_email_template(pipeline_id: str | None = None) -> dict:
"""Resolve the effective email template for an optional pipeline.
Returns the global template, overlaid per-field with any saved override for
``pipeline_id`` (Option C). The overlay only applies when the per-pipeline
feature is enabled AND a (non-empty) override exists for that field;
otherwise the single global template is used (fallback).
"""
store = _load_email_store()
tpl = _global_template(store)
if pipeline_id and store.get("per_pipeline_enabled"):
overrides = store.get("pipelines")
ov = overrides.get(pipeline_id) if isinstance(overrides, dict) else None
if isinstance(ov, dict):
for k, default in EMAIL_TEMPLATE_DEFAULTS.items():
if k in ov and type(ov[k]) is type(default):
tpl[k] = ov[k]
return tpl
def save_email_template(tpl: dict, pipeline_id: str | None = None) -> None:
"""Persist template fields.
When ``pipeline_id`` is None the fields update the single global template.
When set, the fields are stored as a SPARSE override for that pipeline:
empty strings / empty lists are dropped so those fields keep inheriting the
global template. An override with no non-empty fields is removed entirely.
"""
store = _load_email_store()
if pipeline_id:
pipelines = store.get("pipelines")
if not isinstance(pipelines, dict):
pipelines = {}
override: dict = {}
for k, default in EMAIL_TEMPLATE_DEFAULTS.items():
if k not in tpl or type(tpl[k]) is not type(default):
continue
val = tpl[k]
if (isinstance(val, str) and not val.strip()) or (isinstance(val, list) and not val):
continue # empty → inherit global
override[k] = val
if override:
pipelines[pipeline_id] = override
else:
pipelines.pop(pipeline_id, None)
store["pipelines"] = pipelines
else:
for k, default in EMAIL_TEMPLATE_DEFAULTS.items():
if k in tpl and type(tpl[k]) is type(default):
store[k] = tpl[k]
_EMAIL_TEMPLATE_FILE.write_text(json.dumps(store, indent=2))
def _format_total(grand_total: object) -> str:
if isinstance(grand_total, (int, float)):
return f"${grand_total:,.2f}"
return str(grand_total) if grand_total is not None else "—"
def build_email_context(task: dict, grand_total: object, date: str, mailbox: str) -> dict:
"""Map pipeline source data to the substitution tokens for a quote."""
return {
"quote_number": task.get("quote_number", ""),
"quote_total": _format_total(grand_total),
"recipient_email": task.get("recipient_email", ""),
"owner_email": task.get("owner_email", ""),
"date": date or "",
"mailbox": mailbox or "",
}
def render_template(text: str, context: dict, *, escape: bool = False) -> str:
"""Substitute {token} placeholders from context.
Unknown tokens are left untouched (so stray braces in authored HTML/CSS are
preserved). When escape=True, substituted values are HTML-escaped — used for
the body, where values come from external data. Subject uses escape=False.
"""
def _repl(m: "re.Match") -> str:
key = m.group(1)
if key in context:
val = str(context[key])
return _html.escape(val) if escape else val
return m.group(0)
return re.sub(r"\{(\w+)\}", _repl, text or "")
def render_email(template: dict, context: dict) -> tuple[str, str]:
"""Render (subject, body_html) from a template + context."""
subject = render_template(template.get("subject", ""), context) \
or render_template(EMAIL_TEMPLATE_DEFAULTS["subject"], context)
body_html = render_template(template.get("body_html", ""), context, escape=True) \
or render_template(EMAIL_TEMPLATE_DEFAULTS["body_html"], context, escape=True)
return subject, body_html
def sample_email_context() -> dict:
"""A representative context for the Email tab 'Preview with sample data'."""
return {
"quote_number": "26-2904",
"quote_total": "$48,750.00",
"recipient_email": "customer@example.com",
"owner_email": "jsmith@icastinc.com",
"date": datetime.now(timezone.utc).strftime("%Y-%m-%d"),
"mailbox": "sales@icastinc.com",
}
# ---------------------------------------------------------------------------
# Step 2: Quote Take-offs workbook lookup (isolated — DVI-843)
# ---------------------------------------------------------------------------
def _winston_find_workbook(
token: str, task: dict, settings: dict, onedrive_user: str, takeoffs_folder: str
) -> tuple["dict | None", str, str]:
"""Locate the xlsm workbook using shared-first → personal-fallback rule.
Search order:
1. Shared Quote Take-offs (shared_quote_takeoff_location; falls back to
the legacy takeoffs_folder when the new field is empty).
2. Owner's personal Quote Take-offs (quote_takeoff_location from their
User Settings Documents tab) — only tried when the owner email is
known and the personal path is configured.
Returns (item, user_upn, folder_path) on success, or (None, '', '') if the
workbook was not found in either location. Existing behavior is preserved
when shared_quote_takeoff_location is empty (falls back to takeoffs_folder).
"""
qn = task["quote_number"]
owner_email = task.get("owner_email", "").strip()
# 1. Shared Quote Take-offs (primary) — SharePoint site drive (DVI-947).
# The shared library lives in SharePoint, not the pipeline account's
# OneDrive (togen@ has no mysite), so address the site drive directly.
shared_takeoff = settings.get("shared_quote_takeoff_location", "").strip()
primary_folder = shared_takeoff or takeoffs_folder # legacy fallback
if primary_folder:
item = winston_graph.site_find_quote_workbook(token, primary_folder, qn)
if item:
return item, winston_graph.SITE_DRIVE_UPN, primary_folder
# 2. Personal Quote Take-offs (fallback)
if owner_email:
docs = _load_user_docs(owner_email)
personal_takeoff = docs.get("quote_takeoff_location", "").strip()
if personal_takeoff:
item = winston_graph.find_quote_workbook(token, owner_email, personal_takeoff, qn)
if item:
return item, owner_email, personal_takeoff
return None, "", ""
# ---------------------------------------------------------------------------
# Step 3: openpyxl extract (B1 recipe — DVI-807)
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# DVI-982: dynamic sheet + grand-total resolver (mirrors PoC run_macros.py)
#
# The quote sheet(s) for a task are driven by its ClickUp tags (collected incl.
# subtasks in DVI-981) mapped through the editable ``tag_map`` in
# winston_settings.json — NOT a hardcoded sheet name. The grand total is found
# by label search per sheet (the cell below "grand total"), not a fixed cell.
# ---------------------------------------------------------------------------
def _normalize_sheet_key(name: str) -> str:
"""Case- AND whitespace-insensitive key for sheet matching (PoC parity)."""
return (name or "").replace(" ", "").lower()
def _find_sheet(wb, target: str):
"""Return the worksheet matching `target` case/space-insensitively, or None."""
key = _normalize_sheet_key(target)
for name in wb.sheetnames:
if _normalize_sheet_key(name) == key:
return wb[name]
return None
def _extract_version_suffix(tag: str) -> str:
"""PDF version suffix for a tag: '' for v1/none, 'v2' for v2, … (PoC parity).
(Consumed by the DVI-983 multi-PDF naming; surfaced here so the resolver
carries it alongside each candidate.)
"""
m = re.search(r"\s*v(\d{1,2})$", (tag or "").strip(), flags=re.IGNORECASE)
if not m:
return ""
n = int(m.group(1))
return "" if n <= 1 else f"v{n}"
def get_sheets_for_tag(tag: str, tag_map: dict) -> list[str]:
"""Resolve one ClickUp tag (optional vN suffix) to candidate sheet names.
Mirrors PoC run_macros.get_sheets_for_tag against the winston_settings
tag_map shape ``{"custom": {TAG: sheet}, "standard": [TAG, …]}``:
- a custom tag → its single fixed sheet name
- a standard tag → three candidates ``"{TAG} Quote 1/2/3"``
Lookup is case-insensitive; candidate casing follows tag_map so it matches
real tab names under the space/case-insensitive sheet lookup.
"""
from togen.clickup_client import strip_version_suffix
base_upper = strip_version_suffix(tag).strip().upper()
if not base_upper:
return []
custom = tag_map.get("custom") or {}
for key, val in custom.items():
if str(key).upper() == base_upper:
return [val] if isinstance(val, str) else list(val)
for std_tag in (tag_map.get("standard") or []):
if str(std_tag).upper() == base_upper:
return [f"{std_tag} Quote 1", f"{std_tag} Quote 2", f"{std_tag} Quote 3"]
return []
def candidate_sheet_entries(tags: list, tag_map: dict) -> list[tuple[str, str]]:
"""Ordered, de-duplicated ``(sheet_name, version_suffix)`` candidates for a
task's collected tags (mirrors PoC build_quote_sheet_map inner loop)."""
entries: list[tuple[str, str]] = []
seen: set[str] = set()
for tag in tags or []:
ver = _extract_version_suffix(tag)
for sheet in get_sheets_for_tag(tag, tag_map):
if sheet not in seen:
seen.add(sheet)
entries.append((sheet, ver))
return entries
def _find_grand_total(ws) -> tuple[object, "str | None"]:
"""Find the cell containing 'grand total' (case-insensitive, PARTIAL match)
and return ``(value_of_cell_below, label_cell_coordinate)``.
Mirrors PoC ``Cells.Find("grand total", LookAt=xlPart)`` → the cell below.
Returns ``(None, None)`` when no such label is present.
"""
for row in ws.iter_rows():
for cell in row:
v = cell.value
if isinstance(v, str) and "grand total" in v.strip().lower():
below = ws.cell(row=cell.row + 1, column=cell.column).value
return below, cell.coordinate
return None, None
def resolve_quote_sheets(wb, tags: list, tag_map: dict) -> list[dict]:
"""Resolve a task's tags to the quote sheets that ACTUALLY exist in `wb`.
Returns one dict per matched existing sheet, in candidate order::
{candidate, matched_name, version_suffix, grand_total, grand_total_cell}
Sheet matching is case/space-insensitive; the grand total is read per sheet
by label search. An empty list means no candidate resolved (caller emits a
diagnostic and skips the quote — no crash).
"""
resolved: list[dict] = []
matched_keys: set[str] = set()
for candidate, ver in candidate_sheet_entries(tags, tag_map):
ws = _find_sheet(wb, candidate)
if ws is None:
continue
key = _normalize_sheet_key(ws.title)
if key in matched_keys:
continue
matched_keys.add(key)
total, cell = _find_grand_total(ws)
resolved.append({
"candidate": candidate,
"matched_name": ws.title,
"version_suffix": ver,
"grand_total": total,
"grand_total_cell": cell,
})
return resolved
def _trim_rows_and_print_area(ws) -> dict:
"""Hide empty line-item rows and set a dynamic print area (DVI-982).
(a) Hide rows in the trim bands whose column O (15) is blank or < 1 via
openpyxl ``row_dimensions[r].hidden = True`` (mirrors PoC
hide_rows_via_com over the same bands/column).
(b) Compute a print area whose bottom is the last VISIBLE populated row so
the PDF is still trimmed if the Graph export ignores hidden rows.
Returns a diagnostics dict ``{hidden_rows, last_row, print_area}``.
"""
hidden_rows = 0
hidden_set: set[int] = set()
for begin, end in _TRIM_ROW_BANDS:
for r in range(begin, end + 1):
v = ws.cell(row=r, column=_TRIM_CHECK_COL).value
try:
should_hide = v is None or float(v) < 1
except (TypeError, ValueError):
should_hide = v is None or (isinstance(v, str) and not v.strip())
ws.row_dimensions[r].hidden = should_hide
if should_hide:
hidden_rows += 1
hidden_set.add(r)
# Dynamic bottom = last visible row (<= max_row) holding any content, so a
# synthetic print area trims trailing blanks when the sheet defines none.
last_row = 1
max_row = int(ws.max_row or 1)
max_col = min(int(ws.max_column or 1), _PRINT_AREA_LAST_COL)
for r in range(1, max_row + 1):
if r in hidden_set:
continue
for c in range(1, max_col + 1):
val = ws.cell(row=r, column=c).value
if val is not None and not (isinstance(val, str) and not val.strip()):
last_row = r
break
# DVI-985: a real quote sheet ships with its OWN tuned print area + fit-to-
# width scaling (e.g. A1:AP866, scale 43). Clobbering it with a synthetic
# A1:AM{n} cut off the right-hand columns (AN–AP) and broke the scaling in
# the rendered PDF. Preserve the sheet's print area when it has one; only
# fall back to a synthetic range (over the sheet's real column extent, not a
# hardcoded AM) when the sheet defines none. Graph honors the hidden rows
# regardless (verified on 23-1647Bv2.xlsm), so trailing blanks still collapse.
existing = ws.print_area
if existing:
print_area = existing if isinstance(existing, str) else str(existing)
print_area_source = "sheet"
else:
last_col = min(int(ws.max_column or 1), _PRINT_AREA_LAST_COL)
print_area = f"A1:{get_column_letter(last_col)}{last_row}"
ws.print_area = print_area
print_area_source = "synthetic"
return {"hidden_rows": hidden_rows, "last_row": last_row,
"print_area": print_area, "print_area_source": print_area_source}
def sample_workbook_structure(
wb, *, task: "dict | None" = None, tag_map: "dict | None" = None,
sample_rows: int = 20, sample_cols: int = 12, cell_cap: int = 200,
) -> dict:
"""Summarize a loaded workbook for the run-diagnostics inspector (DVI-948).
For every sheet: name, max_row×max_col dimensions, and a top-left cell grid
(sample_rows × sample_cols of cached values). When ``task`` + ``tag_map`` are
provided (DVI-982), also report the tag-driven sheet resolution: the base
tags, candidate sheet names, and the resolved existing sheets with the grand
total read by label per sheet.
"""
sheets: list[dict] = []
for name in wb.sheetnames:
ws = wb[name]
try:
max_row = int(ws.max_row or 0)
max_col = int(ws.max_column or 0)
except Exception:
max_row = max_col = 0
grid: list[list[str]] = []
for r in range(1, min(sample_rows, max_row) + 1):
row_vals: list[str] = []
for c in range(1, min(sample_cols, max_col) + 1):
v = ws.cell(row=r, column=c).value
s = "" if v is None else str(v)
if len(s) > cell_cap:
s = s[:cell_cap] + "…"
row_vals.append(s)
grid.append(row_vals)
sheets.append({
"name": name,
"max_row": max_row,
"max_col": max_col,
"sample": grid,
})
result: dict = {
"sheets": sheets,
"sheet_names": list(wb.sheetnames),
}
if task is not None and tag_map is not None:
from togen.clickup_client import strip_version_suffix
tags = task.get("tags") or []
candidates = candidate_sheet_entries(tags, tag_map)
resolved = resolve_quote_sheets(wb, tags, tag_map)
# DVI-984: full tag→sheet→total decision chain for the inspector.
# (1) Per tag: base key after version-strip + the candidate sheets it
# maps to through the tag_map.
tag_resolution = [
{
"tag": tag,
"base": strip_version_suffix(tag).strip(),
"version_suffix": _extract_version_suffix(tag),
"candidates": get_sheets_for_tag(tag, tag_map),
}
for tag in tags
]
# (2) Per candidate: did the sheet EXIST (case/space-insensitive), its
# matched tab name, and whether it was the chosen sheet for its tab
# (first-wins de-dupe in resolve_quote_sheets).
existing_keys = {_normalize_sheet_key(n): n for n in wb.sheetnames}
chosen_keys = {_normalize_sheet_key(r["matched_name"]) for r in resolved}
candidate_status = []
for cand, ver in candidates:
k = _normalize_sheet_key(cand)
matched_name = existing_keys.get(k)
candidate_status.append({
"candidate": cand,
"version_suffix": ver,
"exists": matched_name is not None,
"matched_name": matched_name,
"chosen": matched_name is not None and k in chosen_keys,
})
# (3) Per chosen sheet: the grand-total LABEL cell and the VALUE cell
# (one row below) the total was read from.
chosen_sheets = []
for r in resolved:
label_cell = r["grand_total_cell"]
value_cell = None
if label_cell:
m = re.match(r"([A-Za-z]+)(\d+)$", str(label_cell))
if m:
value_cell = f"{m.group(1)}{int(m.group(2)) + 1}"
chosen_sheets.append({
"sheet": r["matched_name"],
"version_suffix": r["version_suffix"],
"grand_total": r["grand_total"],
"grand_total_label_cell": label_cell,
"grand_total_value_cell": value_cell,
})
result["resolution"] = {
"tags": tags,
"tag_resolution": tag_resolution,
"candidates": candidate_status,
"chosen_sheets": chosen_sheets,
"matched": bool(resolved),
}
# Back-compat top-level keys (kept for older diagnostics consumers).
result["tags"] = tags
result["candidate_sheets"] = [s for s, _ in candidates]
result["resolved_sheets"] = [
{
"sheet": r["matched_name"],
"grand_total": r["grand_total"],
"grand_total_cell": r["grand_total_cell"],
}
for r in resolved
]
result["matched"] = bool(resolved)
return result
def _pdf_label(sheet_name: str) -> str:
"""Derive the PoC PDF sheet-label from a matched sheet name.
"BW Quote 1" → "BW" (standard tags: strip trailing " Quote N")
"QuoteX" → "X" (custom tags: strip leading "Quote" before a letter)
Mirrors PoC ``run_macros.process_single_excel_file`` label derivation.
"""
label = re.sub(r"\s+Quote\s+\d+$", "", sheet_name or "", flags=re.IGNORECASE)
label = re.sub(r"^Quote(?=[A-Za-z])", "", label, flags=re.IGNORECASE)
return label
def _pdf_filename(quote_num: str, sheet_name: str, version_suffix: str) -> str:
"""PoC-style per-sheet PDF name ``{quote}{sheetLabel}{versionSuffix}.pdf``.
e.g. quote ``26-2270`` + sheet ``QuoteY`` + tag ``yv2`` → ``26-2270Yv2.pdf``.
(Mirrors PoC exactly: the version suffix carries no leading underscore — see
``_extract_version_suffix`` / PoC ``extract_version_suffix``.)
"""
return f"{quote_num}{_pdf_label(sheet_name)}{version_suffix}.pdf"
def _strip_defined_names(wb, keep: str) -> int:
"""Remove workbook- and sheet-level defined names that make the Graph→Office
converter return 406 UnsupportedMediaType (DVI-1003).
The 2026 template ships Excel-internal ``_xleta.*`` LAMBDA/eta helper names
that the converter rejects. Since the workbook is loaded ``data_only=True``
(formulas baked to static values), no cell references any defined name, so
clearing them is safe. Print area / print titles are worksheet-level in
openpyxl (``ws.print_area``) and are NOT touched here. Returns the count of
names removed. Version-robust across openpyxl defined-name APIs.
"""
removed = 0
try:
dn = wb.defined_names
# openpyxl >= 3.1: DefinedNameDict (dict-like)
if hasattr(dn, "keys"):
removed = len(list(dn.keys()))
try:
dn.clear()
except Exception:
wb.defined_names = type(dn)()
# openpyxl < 3.1: DefinedNameList with .definedName / .delete()
elif hasattr(dn, "definedName"):
removed = len(dn.definedName)
for d in list(dn.definedName):
try:
dn.delete(d.name)
except Exception:
pass
except Exception:
log.exception("_strip_defined_names: workbook-level clear failed")
# sheet-scoped defined names (rare, but _xleta can be scoped)
for name in wb.sheetnames:
ws = wb[name]
wdn = getattr(ws, "defined_names", None)
if wdn is None:
continue
try:
if hasattr(wdn, "keys"):
removed += len(list(wdn.keys()))
wdn.clear()
except Exception:
pass
if removed:
log.info("_strip_defined_names: removed %d defined name(s) for %r", removed, keep)
return removed
def _build_sheet_xlsx(
xlsm_bytes: bytes,
target_name: str,
*,
quote_sheet_names: "set[str] | None" = None,
) -> tuple[bytes, dict]:
"""Produce a single-visible-sheet, trimmed .xlsx for ONE resolved quote sheet.
Loads a fresh workbook from the original bytes (so each per-sheet export is
independent and non-destructive to siblings), then:
- Makes the target sheet VISIBLE and trims its empty rows.
- HIDES other quote sheets (siblings that share the quote pattern).
- DELETES all non-quote helper sheets (Database, Data Trans., input, etc.).
- STRIPS workbook-level defined names (the real 406 trigger, see below).
Why strip defined names (DVI-1003 — the actual 406 root cause):
The Graph→Office converter returns ``406 UnsupportedMediaType`` when the
workbook carries certain Excel-internal defined names — notably the
``_xleta.*`` LAMBDA/eta-reduction helpers that the 2026 template introduces
(e.g. ``_xleta.IF``, ``_xleta.TYPE``). This — NOT file size — is why the 2026
template failed while the smaller 2023 template (DVI-985) converted cleanly.
A live convert test on the exact failing file (26-2819Yv2.xlsx) proved it:
clearing defined names → HTTP 200 clean PDF at any size (1.3 MB and 9.9 MB
both converted); leaving them → 406 even at 0.17 MB. Because the workbook is
loaded ``data_only=True`` (all formulas already baked to static values), NO
remaining cell references a defined name, so clearing them is safe. Per-sheet
print area / print titles live at the worksheet level (``ws.print_area``),
not in ``wb.defined_names``, so layout fidelity (DVI-985) is preserved.
Why also delete helper sheets (DVI-1001): keeps the output small so the Graph
upload stays fast/simple; combined with data_only-baked values there are no
live cross-sheet references, so deletion is non-destructive.
``quote_sheet_names`` should be the full set of resolved quote-sheet titles
for this quote (from :func:`resolve_quote_sheets`), including ``target_name``.
Pass ``None`` to fall back to the old behavior (hide all non-target sheets).
"""
wb = openpyxl.load_workbook(io.BytesIO(xlsm_bytes), data_only=True, keep_vba=False)
ws = _find_sheet(wb, target_name) or wb[target_name]
keep = ws.title
if quote_sheet_names is not None:
to_delete = [n for n in wb.sheetnames if n != keep and n not in quote_sheet_names]
for n in to_delete:
del wb[n]
if to_delete:
log.info("_build_sheet_xlsx: deleted %d helper sheets for %r: %s",
len(to_delete), keep, to_delete)
_strip_defined_names(wb, keep)
for name in wb.sheetnames:
wb[name].sheet_state = "visible" if name == keep else "hidden"
wb.active = wb[keep]
trim = _trim_rows_and_print_area(wb[keep])
buf = io.BytesIO()
wb.save(buf)
return buf.getvalue(), trim
def _sum_grand_totals(resolved: list[dict]) -> "float | None":
"""Quote grand total = sum of the numeric per-sheet grand totals (PoC parity;
one sheet in the common case, so just that sheet's total). Returns ``None``
when no sheet yielded a numeric total.
"""
vals: list[float] = []
for r in resolved:
v = r.get("grand_total")
if isinstance(v, bool):
continue
if isinstance(v, (int, float)):
vals.append(float(v))
elif isinstance(v, str):
try:
vals.append(float(v.replace("$", "").replace(",", "").strip()))
except ValueError:
pass
return round(sum(vals), 2) if vals else None
def _extract_from_workbook(
wb, xlsm_bytes: bytes, task: dict, tag_map: dict, quote_num: str
) -> "dict | None":
"""Resolve the quote sheet(s) from tags and build ONE trimmed xlsx per sheet.
The real workflow emits one PDF per matched tag/sheet (often 2+), so this
returns a list of per-sheet artifacts (DVI-983) rather than a single one.
Returns ``None`` when NO candidate sheet resolves (the caller emits a clear
diagnostic and skips the quote — the old hard ``ValueError`` failure path is
gone). Otherwise returns::
{
"sheets": [ {sheet_name, version_suffix, grand_total, grand_total_cell,
pdf_label, pdf_filename, xlsx_bytes, trim}, … ],
"grand_total": , # ClickUp Quote Amount
"resolved": ,
}
Sheets are deduped by resolved name (in :func:`resolve_quote_sheets`); each
per-sheet xlsx is built from a fresh reload of ``xlsm_bytes`` so building one
never mutates the workbook used for another.
"""
tags = task.get("tags") or []
resolved = resolve_quote_sheets(wb, tags, tag_map)
if not resolved:
return None
all_quote_sheet_names = {r["matched_name"] for r in resolved}
sheets: list[dict] = []
for r in resolved:
sheet_name = r["matched_name"]
version_suffix = r["version_suffix"]
xlsx_bytes, trim = _build_sheet_xlsx(
xlsm_bytes, sheet_name, quote_sheet_names=all_quote_sheet_names
)
pdf_filename = _pdf_filename(quote_num, sheet_name, version_suffix)
sheets.append({
"sheet_name": sheet_name,
"version_suffix": version_suffix,
"grand_total": r["grand_total"],
"grand_total_cell": r["grand_total_cell"],
"pdf_label": _pdf_label(sheet_name),
"pdf_filename": pdf_filename,
"xlsx_bytes": xlsx_bytes,
"trim": trim,
})
log.info(
"_extract_from_workbook: sheet=%r pdf=%s total=%r cell=%s hidden=%d print=%s xlsx=%d KB",
sheet_name, pdf_filename, r["grand_total"], r["grand_total_cell"],
trim["hidden_rows"], trim["print_area"], len(xlsx_bytes) // 1024,
)
grand_total = _sum_grand_totals(resolved)
return {
"sheets": sheets,
"grand_total": grand_total,
"resolved": resolved,
}
# ---------------------------------------------------------------------------
# Step 5: Mail draft dedupe
# ---------------------------------------------------------------------------
def _draft_exists(token: str, mailbox: str, quote_number: str) -> bool:
"""Return True if the sales mailbox Drafts folder already has a message
whose subject contains quote_number (Graph OData contains() filter).
"""
url = f"{winston_graph._GRAPH_BASE}/users/{mailbox}/mailFolders/drafts/messages"
# contains() on subject is supported for mail messages in Graph OData
safe_qn = quote_number.replace("'", "''")
params = {
"$filter": f"contains(subject,'{safe_qn}')",
"$select": "id,subject",
"$top": "1",
}
try:
r = _req.get(
url,
headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
params=params,
timeout=30,
)
if r.status_code == 200:
if r.json().get("value"):
log.info("Draft already exists for %s — skipping (dedupe)", quote_number)
return True
else:
log.warning(
"_draft_exists: Graph returned %s for %s: %s",
r.status_code,
quote_number,
r.text[:200],
)
except Exception:
log.exception("_draft_exists error for quote %s", quote_number)
return False
# ---------------------------------------------------------------------------
# Step 7: Routing (isolated per spec — swap rule here without touching pipeline)
# ---------------------------------------------------------------------------
def _winston_route_destination(quote: dict, settings: dict) -> tuple[str, str]:
"""Return (user_upn, folder_path) for the completed PDF upload (Completed Quotes).
Rule (board decision 2 — DVI-805, updated DVI-843):
If the quote owner has a personal Completed Quotes path (quotes_location)
in their User Settings Documents tab → route there (their OneDrive).
Else → shared Completed Quotes path (shared_quotes_location) in Togen Admin.
The routing *criteria* (owner-based: personal overrides shared) is isolated
here so it can be swapped without touching the rest of the pipeline.
"""
owner_email = quote.get("owner_email", "").strip()
if owner_email:
docs = _load_user_docs(owner_email)
personal_path = docs.get("quotes_location", "").strip()
if personal_path:
return owner_email, personal_path
# Shared fallback: the configured shared Completed Quotes location lives in
# the SharePoint site library, not a personal OneDrive (DVI-947). Signal the
# site drive with the SITE_DRIVE_UPN sentinel.
shared_path = settings.get("shared_quotes_location", "").strip()
return winston_graph.SITE_DRIVE_UPN, shared_path
# ---------------------------------------------------------------------------
# Per-run summary report (DVI-976)
#
# After every run (scheduled or manual, including dry runs and early-exit
# paths) email the configured Report Recipients an aggregate summary. On a
# dry run the report is clearly banner-flagged so recipients know no external
# writes were performed.
# ---------------------------------------------------------------------------
_REPORT_METRICS = [
("quotes_found", "Quotes found"),
("drafts_created", "Drafts created"),
("drafts_skipped", "Drafts skipped (already existed)"),
("drafts_failed", "Drafts failed"),
("clickup_amounts_set", "ClickUp amounts set"),
("clickup_amounts_failed", "ClickUp amounts failed"),
("clickup_attached", "ClickUp PDFs attached"),
("clickup_failed", "ClickUp attach failed"),
("pdf_routed", "PDFs routed to OneDrive"),
("pdf_route_failed", "PDF routing failed"),
("writes_skipped_dry_run", "Writes skipped (dry run)"),
]
def _build_report_html(summary: dict, date: str, trigger: str) -> str:
"""Render the per-run summary as branded HTML for the report email."""
dry_run = bool(summary.get("dry_run"))
esc = _html.escape
banner = ""
if dry_run:
banner = (
''
'DRY RUN — no actual actions were taken. No Outlook drafts, ClickUp '
'writebacks, or file uploads were performed. The counts below reflect '
'what would have happened.
'
)
notes = []
if summary.get("config_errors"):
notes.append("Configuration errors: " + esc("; ".join(summary["config_errors"])))
if summary.get("error"):
notes.append("Run error: " + esc(str(summary["error"])))
if summary.get("cancelled"):
notes.append("Run was cancelled before completing.")
notes_html = ""
if notes:
notes_html = (
''
+ "
".join(notes) + "
"
)
metric_rows = "".join(
f'| {esc(label)} | '
f'{int(summary.get(key, 0))} |
'
for key, label in _REPORT_METRICS
)
quotes = summary.get("quotes") or []
quote_rows = ""
for q in quotes:
qn = esc(str(q.get("quote_number", q.get("quote", "—"))))
recipient = esc(str(q.get("recipient", "") or "—"))
draft = esc(str(q.get("draft_status", "—")))
route = esc(str(q.get("pdf_route", "—")))
err = q.get("error")
err_html = (f'{esc(str(err))}
'
if err else "")
quote_rows += (
f'| {qn}{err_html} | '
f'{recipient} | '
f'{draft} | '
f'{route} |
'
)
if not quote_rows:
quote_rows = ('| No quotes processed. |
')
return f"""\
Winston Pipeline Run Report
Date: {esc(date)} |
Trigger: {esc(str(trigger))} |
Run ID: {esc(str(summary.get('run_id', '')))}
{banner}
{notes_html}
Summary
Quotes
| Quote |
Recipient |
Draft |
PDF route |
{quote_rows}
Automated report from Winston. Recipients are configured under
Bots → Winston → Admin → Settings → Report Recipients.
"""
def _send_run_report(summary: dict, date: str, trigger: str) -> bool:
"""Email the run summary to the configured Report Recipients (DVI-976).
Sends regardless of dry-run mode. Returns True if a message was accepted,
False if there were no recipients or the send failed.
"""
settings = _load_settings()
to_list = [e for e in (settings.get("report_to") or []) if e]
cc_list = [e for e in (settings.get("report_cc") or []) if e]
if not to_list and not cc_list:
log.info("Winston report: no Report Recipients configured — skipping send")
return False
token = winston_graph.acquire_graph_token()
if not token:
log.error("Winston report: could not acquire Graph token — report not sent")
return False
dry_run = bool(summary.get("dry_run"))
subject = (
f"Winston pipeline report — {date}"
+ (" [DRY RUN]" if dry_run else "")
)
html_body = _build_report_html(summary, date, trigger)
# Reports are sent from togen@icastinc.com, NOT the sales draft mailbox
# (DVI-976 board decision): sales@ is reserved for the quote drafts/sends
# that land in its inbox, so run reports must come from the service account.
mailbox = (os.environ.get("AZURE_MAIL_FROM", "") or "togen@icastinc.com").strip()
sent = winston_graph.mail_send(
token, subject, html_body, to_list,
mailbox=mailbox, cc_addresses=cc_list,
)
if sent:
log.info("Winston report sent to=%d cc=%d dry_run=%s run_id=%s",
len(to_list), len(cc_list), dry_run, summary.get("run_id"))
else:
log.error("Winston report send failed run_id=%s", summary.get("run_id"))
return sent
# ---------------------------------------------------------------------------
# Main entry point
# ---------------------------------------------------------------------------
def run(date: str, trigger: str = "manual", run_id: str | None = None,
pipeline_id: str | None = None, dry_run: bool = False) -> dict:
"""Run the pipeline and email the per-run summary report (DVI-976).
Thin wrapper around :func:`_run_impl` that always attempts to send the
summary report to the configured Report Recipients afterwards — including
on dry runs and early-exit paths (config errors, no quotes, token failure).
Report delivery never affects the returned summary or raises.
"""
summary = _run_impl(date, trigger, run_id=run_id,
pipeline_id=pipeline_id, dry_run=dry_run)
try:
_send_run_report(summary, date, trigger)
except Exception:
log.exception("Winston summary report send failed (run_id=%s)",
summary.get("run_id"))
return summary
def _run_impl(date: str, trigger: str = "manual", run_id: str | None = None,
pipeline_id: str | None = None, dry_run: bool = False) -> dict:
"""Process all quote tasks for the given date (YYYY-MM-DD).
Returns a summary dict with aggregate counts and per-quote results.
Designed to be called by the scheduler and the "Run now" UI endpoint.
trigger: source of the run ("manual", "user", "scheduler:").
run_id: pre-generated UUID to use for the DiagnosticRun (so callers can
return it to the UI before the background thread starts).
pipeline_id: Date Logic pipeline id driving this run. Selects the
per-pipeline email template override when one exists (Option C);
falls back to the global template otherwise.
dry_run: when True (DVI-908) the pipeline validates the full flow —
ClickUp pull, workbook download, xlsx extract, xlsx→PDF convert,
and diagnostics artifacts — but performs NO external writes: no
Outlook draft, no ClickUp writeback, no Completed-Quotes/OneDrive
upload. The generated PDF/xlsx are still saved as diagnostics
artifacts (a temporary location) so the result is reviewable in
Diagnostics without touching production systems.
"""
from togen.bot_diagnostics import DiagnosticRun
log.info("Winston pipeline starting — date=%s trigger=%s dry_run=%s",
date, trigger, dry_run)
settings = _load_settings()
# Resolve configuration (env vars take precedence over settings file)
clickup_api_key = (
os.environ.get("CLICKUP_API_KEY") or settings.get("clickup_api_key", "")
).strip()
clickup_list_id = (
settings.get("clickup_list_id", "") or os.environ.get("CLICKUP_LIST_ID", "")
).strip()
clickup_team_id = (
settings.get("clickup_team_id", "") or os.environ.get("CLICKUP_TEAM_ID", "")
).strip()
# date_field default: due_date (PoC parity — DVI-999); Send Quotes tasks
# carry a due_date but may not have a recent date_updated, so using
# date_updated silently excludes them from the query window.
_default_date_field = "due_date"
date_field = (
settings.get("clickup_date_field", "")
or os.environ.get("CLICKUP_DATE_FIELD", _default_date_field)
)
global_quote_amount_field_id = os.environ.get("CLICKUP_QUOTE_AMOUNT_FIELD_ID", "").strip()
takeoffs_folder = (
settings.get("takeoffs_folder", "")
or os.environ.get("WINSTON_TAKEOFFS_FOLDER", _DEFAULT_TAKEOFFS_FOLDER)
)
mailbox = (settings.get("draft_mailbox") or winston_graph.WINSTON_MAIL_FROM).strip()
onedrive_user = settings.get("onedrive_user", "") or winston_graph.WINSTON_ONEDRIVE_USER
# DVI-982: tag→sheet map that drives dynamic quote-sheet resolution. Editable
# in Winston Admin → Settings; falls back to the seeded PoC vocabulary.
tag_map = settings.get("tag_map") or _WINSTON_DEFAULTS["tag_map"]
# Email draft template (DVI-857/859): admin-editable subject/body/recipients.
# Resolves the per-pipeline override when set (Option C), else the global one.
email_template = load_email_template(pipeline_id)
diag = DiagnosticRun(bot="winston", trigger=trigger, target_date=date, run_id=run_id)
summary: dict = {
"run_id": diag.id,
"date": date,
"dry_run": bool(dry_run),
"writes_skipped_dry_run": 0,
"quotes_found": 0,
"drafts_created": 0,
"drafts_skipped": 0,
"drafts_failed": 0,
"clickup_amounts_set": 0,
"clickup_amounts_skipped": 0,
"clickup_amounts_failed": 0,
"clickup_attached": 0,
"clickup_skipped": 0,
"clickup_failed": 0,
"pdf_routed": 0,
"pdf_route_failed": 0,
"quotes": [],
}
with diag:
# Guard: required config
errors = []
if not clickup_api_key:
errors.append("missing_clickup_api_key (set CLICKUP_API_KEY or clickup_api_key in winston_settings.json)")
if not clickup_list_id and not clickup_team_id:
errors.append("missing_clickup_source: set clickup_list_id or clickup_team_id in winston_settings.json")
if not onedrive_user:
errors.append("missing_onedrive_user (set WINSTON_ONEDRIVE_USER env var)")
if errors:
for e in errors:
log.error("Winston pipeline config error: %s", e)
summary["config_errors"] = errors
diag.fail("; ".join(errors))
return summary
# ------------------------------------------------------------------
# Step 1: ClickUp pull
# ------------------------------------------------------------------
tasks: list = []
with diag.step("clickup_pull") as s1:
clickup = ClickUpClient(clickup_api_key)
if clickup_list_id:
tasks = clickup.get_quote_tasks(clickup_list_id, date, date_field=date_field)
else:
# PoC-compatible: team-wide query by due_date (no list scope)
tasks = clickup.get_quote_tasks_by_team(clickup_team_id, date, date_field=date_field)
# DVI-999: union tags per quote number + deduplicate to one task per
# quote (PoC parity — build_quote_sheet_map). Tags live on email-line
# subtasks under "Send Quotes 26-XXXX" tasks, not on plain "Quote
# 26-XXXX" tasks; merging across all tasks sharing a quote number
# ensures those subtask tags drive sheet resolution.
_quote_groups: dict[str, list] = {}
for _qt in tasks:
_quote_groups.setdefault(_qt["quote_number"], []).append(_qt)
_deduped: list[dict] = []
for _qn, _group in _quote_groups.items():
_seen_tags: set[str] = set()
_all_tags: list[str] = []
for _gt in _group:
for _tag in (_gt.get("tags") or []):
if _tag not in _seen_tags:
_seen_tags.add(_tag)
_all_tags.append(_tag)
# Primary task: prefer non-Send-Quotes tasks (they carry the
# workbook path + ClickUp writeback fields); fall back to first.
_primary = next(
(_gt for _gt in _group
if not _SEND_QUOTES_RE.match(_gt.get("task_name", ""))),
_group[0],
)
_primary["tags"] = _all_tags
_deduped.append(_primary)
tasks = _deduped
summary["quotes_found"] = len(tasks)
log.info("ClickUp pull: %d task(s) for %s", len(tasks), date)
# DVI-972: surface the retrieved ClickUp data in the Diagnostics
# pane — a query header + per-task table (rendered client-side) plus
# a downloadable clickup_tasks.json artifact for deep inspection.
# DVI-981: log raw + base tags per task for dry-run diagnostics.
from togen.clickup_client import strip_version_suffix
for _t in tasks:
_raw_tags = _t.get("tags") or []
_base_tags = [strip_version_suffix(tg) for tg in _raw_tags]
log.info(
"ClickUp task %s tags_raw=%s tags_base=%s",
_t["quote_number"], _raw_tags, _base_tags,
)
s1.set_detail({
"clickup_query": {
"scope": "list" if clickup_list_id else "team",
"list_id": clickup_list_id or None,
"team_id": None if clickup_list_id else clickup_team_id,
"target_date": date,
"date_field": date_field,
"count": len(tasks),
},
"tasks": tasks,
})
try:
diag.add_artifact(
"clickup_tasks.json",
json.dumps(tasks, default=str, ensure_ascii=False).encode("utf-8"),
"json", step=s1,
)
except Exception:
log.exception("Failed to persist clickup_tasks.json artifact")
if not tasks:
log.info("No quote tasks found for %s — done", date)
diag.set_summary(summary)
return summary
# Acquire Graph token once for the whole batch
token = winston_graph.acquire_graph_token()
if not token:
log.error("Failed to acquire Graph token — aborting")
summary["error"] = "graph_token_failed"
diag.fail("graph_token_failed")
return summary
for task in tasks:
if is_cancelled(diag.id):
log.warning("Winston run %s cancelled by user — stopping", diag.id)
summary["cancelled"] = True
diag.cancel()
break
qn = task["quote_number"]
task_id = task["clickup_task_id"]
result: dict = {"quote_number": qn, "task_id": task_id}
# ------------------------------------------------------------------
# Step 2: Find + download xlsm from OneDrive
# ------------------------------------------------------------------
xlsm_bytes: bytes | None = None
xlsm_name = f"{qn}.xlsm"
with diag.step(f"find_download:{qn}") as s2:
try:
item, _src_upn, _src_folder = _winston_find_workbook(
token, task, settings, onedrive_user, takeoffs_folder
)
if not item:
log.warning(
"Workbook not found for quote %s in any configured "
"Quote Take-offs location", qn
)
result["error"] = "workbook_not_found"
s2.warn()
else:
xlsm_name = item.get("name", xlsm_name)
# DVI-948: record the found file's metadata so the run
# diagnostics dialog can show a File card (name/path/size/
# modified) + an Open-online link.
_parent = item.get("parentReference") or {}
_ppath = _parent.get("path") or ""
_full_path = (f"{_ppath}/{xlsm_name}" if _ppath
else f"{_src_folder}/{xlsm_name}")
s2.set_detail({
"file_name": xlsm_name,
"source": _src_upn,
"folder": _src_folder,
"full_path": _full_path,
"size_kb": (round((item.get("size") or 0) / 1024, 1)
if item.get("size") is not None else None),
"modified": item.get("lastModifiedDateTime"),
"web_url": item.get("webUrl"),
})
dl_url = item.get("@microsoft.graph.downloadUrl")
if dl_url:
dl_resp = _req.get(dl_url, timeout=120)
xlsm_bytes = dl_resp.content if dl_resp.status_code == 200 else None
elif _src_upn == winston_graph.SITE_DRIVE_UPN:
xlsm_bytes = winston_graph.site_drive_read_file(
token, f"{_src_folder}/{xlsm_name}"
)
else:
xlsm_bytes = winston_graph.user_drive_read_file(
token, _src_upn, f"{_src_folder}/{xlsm_name}"
)
if xlsm_bytes:
log.info("Downloaded %s (%d KB)", xlsm_name, len(xlsm_bytes) // 1024)
else:
log.error("Download failed for %s", xlsm_name)
result["error"] = "download_failed"
s2.warn()
except Exception as exc:
log.exception("Download exception for quote %s", qn)
result["error"] = f"download_exception: {exc}"
s2.warn()
if not xlsm_bytes:
summary["quotes"].append(result)
continue
# ------------------------------------------------------------------
# Step 3: openpyxl extract (B1 recipe) — one xlsx per matched sheet
# ------------------------------------------------------------------
# DVI-983: a quote can resolve to several tag/sheet variants, so the
# extract yields a LIST of per-sheet artifacts (each its own trimmed
# single-sheet xlsx + PoC-style PDF name). ``grand_total`` is the
# summed numeric total used for the ClickUp Quote Amount field.
sheets: list[dict] = []
grand_total: object = None
with diag.step(f"openpyxl_extract:{qn}") as s3:
try:
# DVI-948/982: load once, capture a workbook-structure snapshot
# (with the tag→sheet resolution) for the diagnostics inspector,
# then resolve the quote sheet(s) + grand total dynamically.
wb = openpyxl.load_workbook(
io.BytesIO(xlsm_bytes), data_only=True, keep_vba=False
)
structure = sample_workbook_structure(wb, task=task, tag_map=tag_map)
try:
diag.add_artifact(
"workbook_structure.json",
json.dumps(structure, default=str, ensure_ascii=False).encode("utf-8"),
"json", step=s3,
)
except Exception:
log.warning("could not store workbook_structure.json for %s", qn)
extract = _extract_from_workbook(wb, xlsm_bytes, task, tag_map, qn)
if extract is None:
# DVI-982: no candidate sheet resolved from tags — emit a
# clear diagnostic and skip the quote (no crash / no
# "'Quote (Automated)' not found" failure path).
candidates = structure.get("candidate_sheets", [])
log.warning(
"Quote %s: no quote sheet resolved — tags=%s candidates=%s available=%s",
qn, task.get("tags") or [], candidates, wb.sheetnames,
)
result["error"] = "no_quote_sheet_resolved"
result["sheet_resolution"] = {
"tags": task.get("tags") or [],
"candidate_sheets": candidates,
"available_sheets": list(structure.get("sheet_names", [])),
}
s3.set_detail({
"matched": False,
"tags": task.get("tags") or [],
"candidate_sheets": candidates,
"workbook_structure": structure,
})
s3.warn()
try:
diag.add_artifact(xlsm_name, xlsm_bytes, "xlsm", step=s3)
except Exception:
log.warning("could not store source xlsm artifact for %s", qn)
else:
sheets = extract["sheets"]
grand_total = extract["grand_total"]
result["grand_total"] = grand_total
result["quote_sheets"] = [s["sheet_name"] for s in sheets]
result["pdf_filenames"] = [s["pdf_filename"] for s in sheets]
result["xlsx_kb"] = sum(len(s["xlsx_bytes"]) for s in sheets) // 1024
s3.set_detail({
"matched": True,
"grand_total": grand_total,
"sheets": [
{
"sheet": s["sheet_name"],
"pdf": s["pdf_filename"],
"grand_total": s["grand_total"],
"grand_total_cell": s["grand_total_cell"],
"trim": s["trim"],
}
for s in sheets
],
"resolved_sheets": [s["sheet_name"] for s in sheets],
"workbook_structure": structure,
})
for s in sheets:
xlsx_artifact = re.sub(r"\.pdf$", ".xlsx", s["pdf_filename"], flags=re.IGNORECASE)
diag.add_artifact(xlsx_artifact, s["xlsx_bytes"], "xlsx", step=s3)
except Exception as exc:
log.error("openpyxl extract failed for %s: %s", qn, exc)
result["error"] = f"openpyxl_failed: {exc}"
s3.warn()
# DVI-948: on failure keep the source .xlsm so it can be
# downloaded and inspected from the diagnostics dialog.
try:
diag.add_artifact(xlsm_name, xlsm_bytes, "xlsm", step=s3)
except Exception:
log.warning("could not store source xlsm artifact for %s", qn)
if not sheets:
summary["quotes"].append(result)
continue
# ------------------------------------------------------------------
# Step 4: Graph xlsx → PDF (B1 recipe) — one PDF per matched sheet
# ------------------------------------------------------------------
# DVI-983: convert every per-sheet xlsx to its own PDF. A quote
# succeeds as long as at least one sheet converts; sheets that fail
# are recorded but don't abort the rest.
pdfs: list[dict] = [] # {filename, bytes, sheet_name}
with diag.step(f"graph_convert:{qn}") as s4:
for s in sheets:
try:
# Stable scratch name prevents collisions with concurrent runs
scratch_name = (
re.sub(r"[^a-zA-Z0-9_-]", "_",
re.sub(r"\.pdf$", "", s["pdf_filename"], flags=re.IGNORECASE))
+ ".xlsx"
)
pdf_bytes = winston_graph.xlsx_to_pdf(
token, s["xlsx_bytes"], scratch_filename=scratch_name
)
if pdf_bytes:
diag.add_artifact(s["pdf_filename"], pdf_bytes, "pdf", step=s4)
pdfs.append({
"filename": s["pdf_filename"],
"bytes": pdf_bytes,
"sheet_name": s["sheet_name"],
})
else:
log.error("Graph PDF conversion failed for %s [%s]",
qn, s["sheet_name"])
s4.warn()
except Exception as exc:
log.exception("Graph convert exception for quote %s [%s]",
qn, s["sheet_name"])
result.setdefault("error", f"pdf_conversion_exception: {exc}")
s4.warn()
if not pdfs:
result.setdefault("error", "pdf_conversion_failed")
summary["quotes"].append(result)
continue
result["pdf_kb"] = sum(len(p["bytes"]) for p in pdfs) // 1024
if len(pdfs) < len(sheets):
result["pdf_partial"] = f"{len(pdfs)}/{len(sheets)} sheets converted"
# ------------------------------------------------------------------
# Step 5: Outlook draft (idempotent — skip if already in Drafts)
# ------------------------------------------------------------------
with diag.step(f"outlook_draft:{qn}") as s5:
try:
recipient = task.get("recipient_email", "")
# DVI-983: one draft per quote, with ALL its per-sheet PDFs
# attached (mirrors PoC main.py step 3).
attach_names = ", ".join(p["filename"] for p in pdfs)
if dry_run:
# Render what would be sent (for Diagnostics visibility)
# but create no draft.
ctx = build_email_context(task, grand_total, date, mailbox)
subject, _body = render_email(email_template, ctx)
result["draft_status"] = "dry_run (no draft created)"
result["recipient"] = recipient or "(no recipient)"
summary["writes_skipped_dry_run"] += 1
log.info(
"[DRY RUN] Would create draft quote=%s to=%s mailbox=%s subject=%r attach=%s",
qn, recipient or "(none)", mailbox, subject, attach_names,
)
elif _draft_exists(token, mailbox, qn):
result["draft_status"] = "skipped"
summary["drafts_skipped"] += 1
else:
# Recipients: prefer the quote's ClickUp recipient; if none,
# fall back to the template's configured fallback list.
if recipient:
to_list = [recipient]
else:
to_list = [a for a in email_template.get("fallback_to", []) if a]
ctx = build_email_context(task, grand_total, date, mailbox)
subject, body_html = render_email(email_template, ctx)
draft = winston_graph.mail_create_draft(
token,
subject=subject,
body_html=body_html,
to_addresses=to_list,
mailbox=mailbox,
cc_addresses=[a for a in email_template.get("cc", []) if a],
bcc_addresses=[a for a in email_template.get("bcc", []) if a],
attachments=[
winston_graph.encode_attachment(p["filename"], p["bytes"])
for p in pdfs
],
)
if draft:
result["draft_status"] = "created"
result["draft_mailbox"] = mailbox
result["recipient"] = recipient or "(no recipient)"
summary["drafts_created"] += 1
log.info(
"Draft created quote=%s to=%s mailbox=%s",
qn, recipient or "(none)", mailbox,
)
else:
result["draft_status"] = "failed"
summary["drafts_failed"] += 1
s5.warn()
except Exception as exc:
log.exception("Mail draft exception for quote %s", qn)
result["draft_status"] = "failed"
summary["drafts_failed"] += 1
s5.warn()
# ------------------------------------------------------------------
# Step 6: ClickUp write-back + Step 7: Routed PDF upload
# ------------------------------------------------------------------
with diag.step(f"clickup_writeback_route:{qn}") as s67:
try:
field_id = global_quote_amount_field_id or task.get("quote_amount_field_id", "")
# DVI-983: the Quote Amount is a single field (set once from the
# summed grand total), while the PDF attach + OneDrive route run
# once per generated PDF.
pdf_names = [p["filename"] for p in pdfs]
dest_upn, dest_folder = _winston_route_destination(task, settings)
if dry_run:
# Validate routing/targets and log intent, but write nothing.
result["clickup_amount"] = (
f"dry_run (would set {_format_total(grand_total)})"
if field_id and grand_total is not None
else "dry_run (skipped)"
)
result["clickup_attach"] = (
f"dry_run (would attach {', '.join(pdf_names)})"
)
if dest_folder:
dest_paths = [f"{dest_folder.rstrip('/')}/{n}" for n in pdf_names]
result["pdf_route"] = (
f"dry_run (would upload {len(dest_paths)} to "
f"{dest_upn}: {', '.join(dest_paths)})"
)
else:
result["pdf_route"] = "dry_run (no destination configured)"
summary["writes_skipped_dry_run"] += 1
log.info(
"[DRY RUN] Would write back quote=%s: clickup_amount=%r attach=%r route=%r",
qn, result["clickup_amount"], result["clickup_attach"], result["pdf_route"],
)
summary["quotes"].append(result)
continue
# 6a: Quote Amount custom field (single field, summed total)
if field_id and grand_total is not None:
ok = clickup.set_custom_field(task_id, field_id, grand_total)
if ok:
result["clickup_amount"] = "set"
summary["clickup_amounts_set"] += 1
else:
result["clickup_amount"] = "failed"
summary["clickup_amounts_failed"] += 1
else:
reason = "no_field_id" if not field_id else "no_grand_total"
result["clickup_amount"] = f"skipped ({reason})"
summary["clickup_amounts_skipped"] += 1
# 6b: PDF attachment — attach each PDF (skip if already attached)
attach_states: list[str] = []
for p in pdfs:
if clickup.attachment_exists(task_id, p["filename"]):
attach_states.append(f"{p['filename']}: skipped")
summary["clickup_skipped"] += 1
elif clickup.attach_file(task_id, p["filename"], p["bytes"]):
attach_states.append(f"{p['filename']}: attached")
summary["clickup_attached"] += 1
else:
attach_states.append(f"{p['filename']}: failed")
summary["clickup_failed"] += 1
result["clickup_attach"] = "; ".join(attach_states)
# 7: Routed PDF upload to OneDrive — one upload per PDF
if dest_folder:
route_states: list[str] = []
for p in pdfs:
dest_path = f"{dest_folder.rstrip('/')}/{p['filename']}"
if dest_upn == winston_graph.SITE_DRIVE_UPN:
uploaded = winston_graph.site_drive_upload_file(
token, dest_path, p["bytes"],
content_type="application/pdf",
)
else:
uploaded = winston_graph.user_drive_upload_file(
token, dest_upn, dest_path, p["bytes"],
content_type="application/pdf",
)
if uploaded:
route_states.append(f"{dest_upn}:{dest_path}")
summary["pdf_routed"] += 1
else:
route_states.append(f"{p['filename']}: upload_failed")
summary["pdf_route_failed"] += 1
result["pdf_route"] = "; ".join(route_states)
else:
result["pdf_route"] = "skipped (no destination configured)"
log.warning(
"No PDF route destination for %s — "
"set shared_quotes_location in Togen Admin → Documents → Quotes",
qn,
)
except Exception as exc:
log.exception("ClickUp/route exception for quote %s", qn)
result.setdefault("error", f"writeback_exception: {exc}")
s67.warn()
summary["quotes"].append(result)
# Mark overall run status based on soft failures
if summary.get("drafts_failed") or summary.get("pdf_route_failed"):
diag.warn()
diag.set_summary({
"dry_run": summary["dry_run"],
"writes_skipped_dry_run": summary["writes_skipped_dry_run"],
"quotes_found": summary["quotes_found"],
"drafts_created": summary["drafts_created"],
"drafts_skipped": summary["drafts_skipped"],
"drafts_failed": summary["drafts_failed"],
"clickup_amounts_set": summary["clickup_amounts_set"],
"clickup_amounts_failed": summary["clickup_amounts_failed"],
"clickup_attached": summary["clickup_attached"],
"clickup_failed": summary["clickup_failed"],
"pdf_routed": summary["pdf_routed"],
"pdf_route_failed": summary["pdf_route_failed"],
"quotes": summary["quotes"],
})
log.info(
"Winston pipeline complete — date=%s found=%d drafts=+%d/~%d/!%d"
" attached=+%d/~%d/!%d routed=%d",
date,
summary["quotes_found"],
summary["drafts_created"],
summary["drafts_skipped"],
summary["drafts_failed"],
summary["clickup_attached"],
summary["clickup_skipped"],
summary["clickup_failed"],
summary["pdf_routed"],
)
return summary
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
if __name__ == "__main__":
import argparse
import json as _json
import sys
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s %(message)s")
ap = argparse.ArgumentParser(description="Winston quote pipeline (DVI-815)")
ap.add_argument("--date", required=True, help="Target date YYYY-MM-DD")
args = ap.parse_args()
result = run(args.date)
print(_json.dumps(result, indent=2, default=str))
has_error = bool(result.get("error") or result.get("config_errors"))
sys.exit(1 if has_error else 0)