main
py 178 lines 6.11 KB
Raw
1 #!/usr/bin/env python3
2 """Check whether the Copilot model pricing table is due for manual review."""
3
4 from __future__ import annotations
5
6 import argparse
7 import json
8 from datetime import UTC, date, datetime
9 from pathlib import Path
10
11 from scripts.model_pricing import (
12 MODEL_PRICING,
13 PRICING_FETCHED_DATE,
14 PRICING_REVIEW_INTERVAL_MONTHS,
15 PRICING_SOURCE_URL,
16 TieredModelRate,
17 )
18
19
20 def parse_source_headers(path: Path | None) -> dict[str, str]:
21 if path is None or not path.exists():
22 return {}
23 metadata: dict[str, str] = {}
24 for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
25 if ":" not in line:
26 continue
27 name, value = line.split(":", 1)
28 normalized = name.strip().lower()
29 if normalized in {"etag", "last-modified"}:
30 metadata[normalized] = value.strip()
31 return metadata
32
33
34 def parse_date(value: str) -> date:
35 candidate = value.strip()
36 if candidate.endswith("Z"):
37 candidate = f"{candidate[:-1]}+00:00"
38 if "T" in candidate:
39 return datetime.fromisoformat(candidate).date()
40 return date.fromisoformat(candidate)
41
42
43 def add_months(value: date, months: int) -> date:
44 month_index = value.month - 1 + months
45 year = value.year + month_index // 12
46 month = month_index % 12 + 1
47 month_lengths = [
48 31,
49 29 if year % 4 == 0 and (year % 100 != 0 or year % 400 == 0) else 28,
50 31,
51 30,
52 31,
53 30,
54 31,
55 31,
56 30,
57 31,
58 30,
59 31,
60 ]
61 return date(year, month, min(value.day, month_lengths[month - 1]))
62
63
64 def pricing_status(
65 current_date: date,
66 source_url: str = PRICING_SOURCE_URL,
67 source_headers: dict[str, str] | None = None,
68 ) -> dict[str, object]:
69 fetched_date = parse_date(PRICING_FETCHED_DATE)
70 due_date = add_months(fetched_date, PRICING_REVIEW_INTERVAL_MONTHS)
71 source_url_matches = source_url == PRICING_SOURCE_URL
72 due = current_date >= due_date
73 tiered_models = sorted(
74 model for model, pricing in MODEL_PRICING.items() if isinstance(pricing, TieredModelRate)
75 )
76 return {
77 "needs_review": due or not source_url_matches,
78 "review_due": due,
79 "source_url_matches": source_url_matches,
80 "source_url": PRICING_SOURCE_URL,
81 "requested_source_url": source_url,
82 "fetched_date": PRICING_FETCHED_DATE,
83 "review_interval_months": PRICING_REVIEW_INTERVAL_MONTHS,
84 "due_date": due_date.isoformat(),
85 "current_date": current_date.isoformat(),
86 "model_count": len(MODEL_PRICING),
87 "tiered_models": tiered_models,
88 "source_headers": source_headers or {},
89 }
90
91
92 def render_report(status: dict[str, object]) -> str:
93 result = "required" if status["needs_review"] else "not due"
94 return (
95 "\n".join(
96 [
97 "# Copilot model pricing review",
98 "",
99 f"**Status:** Review {result}.",
100 f"**Source:** {status['source_url']}",
101 f"**Repository pricing fetched:** {status['fetched_date']}",
102 f"**Review interval:** every {status['review_interval_months']} months",
103 f"**Next/due review date:** {status['due_date']}",
104 f"**Workflow check date:** {status['current_date']}",
105 f"**Tracked pricing entries:** {status['model_count']}",
106 f"**Long-context pricing entries:** {', '.join(status['tiered_models'])}",
107 f"**Observed source metadata:** {json.dumps(status['source_headers'], sort_keys=True) if status['source_headers'] else 'not captured'}",
108 "",
109 "This workflow does not change pricing automatically. Please compare the repository pricing table against the GitHub docs, update code/docs/tests if needed, and open a PR.",
110 "",
111 "Checklist:",
112 "- Review `scripts/model_pricing.py` against the source URL.",
113 "- Update cost documentation and tests if rates, model names, or thresholds changed.",
114 "- Keep the source URL and fetched date in sync with the reviewed table.",
115 ]
116 )
117 + "\n"
118 )
119
120
121 def write_github_output(path: Path, status: dict[str, object]) -> None:
122 with path.open("a", encoding="utf-8") as handle:
123 handle.write(f"needs_review={str(status['needs_review']).lower()}\n")
124 handle.write(f"due_date={status['due_date']}\n")
125 handle.write(f"fetched_date={status['fetched_date']}\n")
126
127
128 def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
129 parser = argparse.ArgumentParser(
130 description="Check whether Copilot model pricing needs manual review."
131 )
132 parser.add_argument(
133 "--current-date", default=datetime.now(UTC).date().isoformat(), help="Current UTC date."
134 )
135 parser.add_argument(
136 "--source-url",
137 default=PRICING_SOURCE_URL,
138 help="Expected GitHub Copilot pricing source URL.",
139 )
140 parser.add_argument("--output", type=Path, help="Write a Markdown review report to this path.")
141 parser.add_argument(
142 "--json-output", type=Path, help="Write machine-readable status JSON to this path."
143 )
144 parser.add_argument(
145 "--github-output", type=Path, help="Append step outputs for GitHub Actions."
146 )
147 parser.add_argument(
148 "--source-headers",
149 type=Path,
150 help="Optional HTTP response headers captured from the source URL.",
151 )
152 return parser.parse_args(argv)
153
154
155 def main(argv: list[str] | None = None) -> int:
156 args = parse_args(argv)
157 status = pricing_status(
158 parse_date(args.current_date), args.source_url, parse_source_headers(args.source_headers)
159 )
160 report = render_report(status)
161
162 if args.output:
163 args.output.write_text(report, encoding="utf-8")
164 else:
165 print(report, end="")
166
167 if args.json_output:
168 args.json_output.write_text(
169 json.dumps(status, indent=2, sort_keys=True) + "\n", encoding="utf-8"
170 )
171 if args.github_output:
172 write_github_output(args.github_output, status)
173
174 return 0
175
176
177 if __name__ == "__main__":
178 raise SystemExit(main())