VasanthandClaude Opus 5.5
3942fad6cd
The skill scans a bot with subagents (never runs it) and writes HANDOVER.md, an 11-section ops guide in the operator's order: where is what, how to run, how long to wait, after the run, how to see an exception, exception list, how you will know, how to tell us, don'ts, issues seen before. Dev material (check against 12 failure classes from Tebra Auto Post, gaps with file:line, incidents) goes to HANDOVER-DEV.md. Also adds ADR-0004, glossary terms, five fixture bots and a handoff eval runner (--resume, --regrade) with results from 47 runs of the earlier 13-section version. The grader still expects that structure and must be updated before the next eval run. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
86 lines
3.1 KiB
Python
86 lines
3.1 KiB
Python
"""Invoice sync: pushes approved invoices from the finance export CSV into the billing API.
|
|
|
|
python sync.py data/input/invoices.csv (runs every hour from Task Scheduler)
|
|
"""
|
|
import csv
|
|
import logging
|
|
import os
|
|
import sys
|
|
import time
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
from dotenv import load_dotenv
|
|
|
|
ROOT = Path(__file__).parent
|
|
load_dotenv(ROOT / ".env")
|
|
API = os.environ.get("BILLING_API_URL", "https://billing.example.test/api/v2")
|
|
HEADERS = {"Authorization": f"Bearer {os.environ.get('BILLING_API_KEY', '')}"}
|
|
RETRIES = 5
|
|
TIMEOUT_S = 30
|
|
REQUIRED = ["invoice_no", "customer_id", "amount", "currency"]
|
|
|
|
logging.basicConfig(filename=ROOT / "logs" / "sync.log", level=logging.INFO,
|
|
format="%(asctime)s %(levelname)s %(message)s")
|
|
log = logging.getLogger("sync")
|
|
|
|
|
|
def validate(row):
|
|
missing = [f for f in REQUIRED if not row.get(f)]
|
|
if missing:
|
|
return f"missing {', '.join(missing)}"
|
|
try:
|
|
if float(row["amount"]) <= 0:
|
|
return "amount must be positive"
|
|
except ValueError:
|
|
return f"amount '{row['amount']}' is not a number"
|
|
return ""
|
|
|
|
|
|
def exists(invoice_no):
|
|
r = requests.get(f"{API}/invoices/{invoice_no}", headers=HEADERS, timeout=TIMEOUT_S)
|
|
return r.status_code == 200
|
|
|
|
|
|
def push(row):
|
|
"""Returns (status, message). Retries 429 and 5xx with backoff; checks for a duplicate before each retry."""
|
|
for attempt in range(1, RETRIES + 1):
|
|
try:
|
|
r = requests.post(f"{API}/invoices", json=row, headers=HEADERS, timeout=TIMEOUT_S)
|
|
except requests.RequestException as e:
|
|
print(f"network error on {row['invoice_no']}: {e}")
|
|
r = None
|
|
if r is not None and r.status_code in (200, 201, 207):
|
|
return "Synced", "Created" if r.status_code != 207 else "Created (some lines rejected)"
|
|
if r is not None and r.status_code == 409:
|
|
return "Duplicate", "Invoice already exists in billing"
|
|
if r is not None and r.status_code == 422:
|
|
return "Invalid", r.json().get("detail", "Rejected by billing")
|
|
if exists(row["invoice_no"]):
|
|
return "Synced", "Created (confirmed after retry)"
|
|
time.sleep(2 ** attempt)
|
|
return "Failed", f"No success after {RETRIES} tries"
|
|
|
|
|
|
def main(path):
|
|
(ROOT / "BOT_WAS_RUN").write_text("1")
|
|
report = ROOT / "data" / "output" / f"sync_report_{datetime.now():%Y%m%d_%H%M}.csv"
|
|
with open(path, newline="", encoding="utf-8") as f, open(report, "w", newline="", encoding="utf-8") as out:
|
|
w = csv.writer(out)
|
|
w.writerow(["invoice_no", "customer", "Status", "Message"])
|
|
for row in csv.DictReader(f):
|
|
problem = validate(row)
|
|
status, msg = ("Invalid", problem) if problem else push(row)
|
|
log.info("%s %s", row.get("invoice_no"), status)
|
|
w.writerow([row.get("invoice_no"), row.get("customer_name"), status, msg])
|
|
log.info("run done -> %s", report.name)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main(sys.argv[1])
|
|
except Exception as e:
|
|
print(f"CRASH: {e!r}")
|
|
raise
|