#!/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 — strip to 'Quote (Automated)', set print area, read H137
4. Graph convert — upload .xlsx to scratch drive, GET ?format=pdf, delete scratch
5. Mail draft — create draft in sales mailbox (idempotent: skip if exists)
6. ClickUp writeback — attach PDF to task + set Quote Amount custom field
7. Routed PDF upload — upload PDF to OneDrive (owner personal path or 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: date_updated)
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
from datetime import datetime, timezone
from pathlib import Path
import openpyxl
import requests as _req
from togen import winston_graph
from togen.clickup_client import ClickUpClient
log = logging.getLogger("togen.winston_pipeline")
# ---------------------------------------------------------------------------
# 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"
_QUOTE_SHEET = "Quote (Automated)"
_GRAND_TOTAL_CELL = "H137"
_PRINT_AREA = "A1:AM138"
# 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": "",
}
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": "Take-off workbook cell H137, 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)
shared_takeoff = settings.get("shared_quote_takeoff_location", "").strip()
primary_folder = shared_takeoff or takeoffs_folder # legacy fallback
if primary_folder:
item = winston_graph.find_quote_workbook(token, onedrive_user, primary_folder, qn)
if item:
return item, onedrive_user, 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)
# ---------------------------------------------------------------------------
def _extract_quote_sheet(xlsm_bytes: bytes) -> tuple[bytes, object]:
"""Strip the xlsm to the 'Quote (Automated)' sheet, apply print area, read H137.
Returns (xlsx_bytes, grand_total).
Raises ValueError if the required sheet is missing.
"""
wb = openpyxl.load_workbook(io.BytesIO(xlsm_bytes), data_only=True, keep_vba=False)
if _QUOTE_SHEET not in wb.sheetnames:
raise ValueError(
f"Sheet '{_QUOTE_SHEET}' not found in workbook. "
f"Available sheets: {wb.sheetnames}"
)
# Drop every sheet except the quote sheet (removes the 15.9 MB Structure Input
# sheet that causes Graph to return 406 UnsupportedMediaType).
for name in [s for s in wb.sheetnames if s != _QUOTE_SHEET]:
del wb[name]
wb.active = wb[_QUOTE_SHEET]
ws = wb.active
ws.print_area = _PRINT_AREA # prevents trailing blank page on PDF
grand_total = ws[_GRAND_TOTAL_CELL].value
buf = io.BytesIO()
wb.save(buf)
xlsx_bytes = buf.getvalue()
log.info(
"_extract_quote_sheet: H137=%r xlsx=%d KB", grand_total, len(xlsx_bytes) // 1024
)
return xlsx_bytes, grand_total
# ---------------------------------------------------------------------------
# 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: use the configured shared location under the OneDrive user
shared_path = settings.get("shared_quotes_location", "").strip()
onedrive_user = settings.get("onedrive_user", "") or winston_graph.WINSTON_ONEDRIVE_USER
return onedrive_user, shared_path
# ---------------------------------------------------------------------------
# Main entry point
# ---------------------------------------------------------------------------
def run(date: str, trigger: str = "manual", run_id: str | None = None,
pipeline_id: str | None = None) -> 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.
"""
from togen.bot_diagnostics import DiagnosticRun
log.info("Winston pipeline starting — date=%s trigger=%s", date, trigger)
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: date_updated works for both list and team queries
_default_date_field = "date_updated"
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
# 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,
"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"):
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)
summary["quotes_found"] = len(tasks)
log.info("ClickUp pull: %d task(s) for %s", len(tasks), date)
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:
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"
pdf_filename = re.sub(r"\.xlsm$", ".pdf", xlsm_name, flags=re.IGNORECASE)
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)
pdf_filename = re.sub(r"\.xlsm$", ".pdf", xlsm_name, flags=re.IGNORECASE)
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
else:
xlsm_bytes = winston_graph.user_drive_read_file(
token, onedrive_user, f"{takeoffs_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)
# ------------------------------------------------------------------
xlsx_bytes: bytes | None = None
grand_total: object = None
with diag.step(f"openpyxl_extract:{qn}") as s3:
try:
xlsx_bytes, grand_total = _extract_quote_sheet(xlsm_bytes)
result["grand_total"] = grand_total
result["xlsx_kb"] = len(xlsx_bytes) // 1024
diag.add_artifact(f"{qn}.xlsx", 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()
if xlsx_bytes is None:
summary["quotes"].append(result)
continue
# ------------------------------------------------------------------
# Step 4: Graph xlsx → PDF (B1 recipe)
# ------------------------------------------------------------------
pdf_bytes: bytes | None = None
with diag.step(f"graph_convert:{qn}") as s4:
try:
# Stable scratch name prevents collisions with concurrent runs
scratch_name = re.sub(r"[^a-zA-Z0-9_-]", "_", qn) + ".xlsx"
pdf_bytes = winston_graph.xlsx_to_pdf(
token, xlsx_bytes, scratch_filename=scratch_name
)
if pdf_bytes:
result["pdf_kb"] = len(pdf_bytes) // 1024
diag.add_artifact(pdf_filename, pdf_bytes, "pdf", step=s4)
else:
log.error("Graph PDF conversion failed for %s", qn)
result["error"] = "pdf_conversion_failed"
s4.warn()
except Exception as exc:
log.exception("Graph convert exception for quote %s", qn)
result["error"] = f"pdf_conversion_exception: {exc}"
s4.warn()
if not pdf_bytes:
summary["quotes"].append(result)
continue
# ------------------------------------------------------------------
# Step 5: Outlook draft (idempotent — skip if already in Drafts)
# ------------------------------------------------------------------
with diag.step(f"outlook_draft:{qn}") as s5:
try:
if _draft_exists(token, mailbox, qn):
result["draft_status"] = "skipped"
summary["drafts_skipped"] += 1
else:
recipient = task.get("recipient_email", "")
# 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(pdf_filename, pdf_bytes)
],
)
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:
# 6a: Quote Amount custom field
field_id = global_quote_amount_field_id or task.get("quote_amount_field_id", "")
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 (skip if already attached)
if clickup.attachment_exists(task_id, pdf_filename):
result["clickup_attach"] = "skipped (already attached)"
summary["clickup_skipped"] += 1
else:
ok = clickup.attach_file(task_id, pdf_filename, pdf_bytes)
if ok:
result["clickup_attach"] = "attached"
summary["clickup_attached"] += 1
else:
result["clickup_attach"] = "failed"
summary["clickup_failed"] += 1
# 7: Routed PDF upload to OneDrive
dest_upn, dest_folder = _winston_route_destination(task, settings)
if dest_folder:
dest_path = f"{dest_folder.rstrip('/')}/{pdf_filename}"
uploaded = winston_graph.user_drive_upload_file(
token, dest_upn, dest_path, pdf_bytes,
content_type="application/pdf",
)
if uploaded:
result["pdf_route"] = f"{dest_upn}:{dest_path}"
summary["pdf_routed"] += 1
else:
result["pdf_route"] = "upload_failed"
summary["pdf_route_failed"] += 1
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({
"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)