|
| 1 | +#!/usr/bin/env python3 |
| 2 | +from __future__ import annotations |
| 3 | + |
| 4 | +import argparse |
| 5 | +import json |
| 6 | +import os |
| 7 | +import re |
| 8 | +import urllib.error |
| 9 | +import urllib.request |
| 10 | +from pathlib import Path |
| 11 | +from typing import Any |
| 12 | + |
| 13 | +from repo_tools import ROOT, canonical_slug, discover_problem_solutions |
| 14 | +from sync_problem_notes import ( |
| 15 | + get_problem_metadata, |
| 16 | + load_metadata_cache, |
| 17 | + note_path_for_slug, |
| 18 | + save_metadata_cache, |
| 19 | + sync_problem_notes, |
| 20 | +) |
| 21 | +from update_problem_note import replace_complexity, replace_section |
| 22 | + |
| 23 | +OPENAI_RESPONSES_URL = "https://api.openai.com/v1/responses" |
| 24 | +DEFAULT_MODEL = "gpt-5-mini" |
| 25 | +TARGET_SECTIONS = ( |
| 26 | + "Problem Summary", |
| 27 | + "Data Structures Used", |
| 28 | + "Approach", |
| 29 | + "Complexity", |
| 30 | + "Revision Notes", |
| 31 | +) |
| 32 | + |
| 33 | +SYSTEM_INSTRUCTIONS = """ |
| 34 | +You generate concise LeetCode study notes from a problem statement and an accepted solution. |
| 35 | +Return only valid JSON with this exact shape: |
| 36 | +{ |
| 37 | + "summary": "string", |
| 38 | + "data_structures": ["string"], |
| 39 | + "approach": "string", |
| 40 | + "time_complexity": "string", |
| 41 | + "space_complexity": "string", |
| 42 | + "revision_notes": ["string"] |
| 43 | +} |
| 44 | +
|
| 45 | +Requirements: |
| 46 | +- Ground the explanation in the provided solution code, not a generic textbook answer. |
| 47 | +- Keep the summary to 1-2 short sentences. |
| 48 | +- Keep the approach to one short paragraph. |
| 49 | +- `data_structures` should be short labels like "Hash Map" or "Two Pointers". |
| 50 | +- Complexities must be asymptotic Big-O values. |
| 51 | +- `revision_notes` should be 1-3 brief bullets about key invariants, edge cases, or why the approach works. |
| 52 | +- Do not wrap the JSON in markdown fences. |
| 53 | +""".strip() |
| 54 | + |
| 55 | + |
| 56 | +def extract_section_body(content: str, heading: str) -> str: |
| 57 | + pattern = rf"## {re.escape(heading)}\n(.*?)(?=\n## |\Z)" |
| 58 | + match = re.search(pattern, content, flags=re.S) |
| 59 | + if not match: |
| 60 | + raise RuntimeError(f"Section '{heading}' not found in note.") |
| 61 | + return match.group(1).strip() |
| 62 | + |
| 63 | + |
| 64 | +def section_needs_update(content: str, heading: str) -> bool: |
| 65 | + return "TODO" in extract_section_body(content, heading) |
| 66 | + |
| 67 | + |
| 68 | +def complexity_needs_update(content: str) -> bool: |
| 69 | + body = extract_section_body(content, "Complexity") |
| 70 | + return "TODO" in body |
| 71 | + |
| 72 | + |
| 73 | +def format_bullets(values: list[str], fallback: str) -> str: |
| 74 | + cleaned = [value.strip() for value in values if value and value.strip()] |
| 75 | + if not cleaned: |
| 76 | + return fallback |
| 77 | + return "\n".join(f"- {value}" for value in cleaned) |
| 78 | + |
| 79 | + |
| 80 | +def strip_code_fences(text: str) -> str: |
| 81 | + stripped = text.strip() |
| 82 | + if stripped.startswith("```"): |
| 83 | + stripped = re.sub(r"^```[a-zA-Z0-9_-]*\n", "", stripped) |
| 84 | + stripped = re.sub(r"\n```$", "", stripped) |
| 85 | + return stripped.strip() |
| 86 | + |
| 87 | + |
| 88 | +def extract_response_text(payload: dict[str, Any]) -> str: |
| 89 | + output_text = payload.get("output_text") |
| 90 | + if isinstance(output_text, str) and output_text.strip(): |
| 91 | + return output_text.strip() |
| 92 | + |
| 93 | + texts: list[str] = [] |
| 94 | + for item in payload.get("output", []): |
| 95 | + if not isinstance(item, dict): |
| 96 | + continue |
| 97 | + for content in item.get("content", []): |
| 98 | + if not isinstance(content, dict): |
| 99 | + continue |
| 100 | + if content.get("type") not in {"output_text", "text"}: |
| 101 | + continue |
| 102 | + text = content.get("text") |
| 103 | + if isinstance(text, str) and text.strip(): |
| 104 | + texts.append(text.strip()) |
| 105 | + continue |
| 106 | + if isinstance(text, dict): |
| 107 | + value = text.get("value") |
| 108 | + if isinstance(value, str) and value.strip(): |
| 109 | + texts.append(value.strip()) |
| 110 | + return "\n".join(texts).strip() |
| 111 | + |
| 112 | + |
| 113 | +def request_note_draft(prompt: str, api_key: str, model: str) -> dict[str, Any]: |
| 114 | + payload = json.dumps( |
| 115 | + { |
| 116 | + "model": model, |
| 117 | + "reasoning": {"effort": "low"}, |
| 118 | + "instructions": SYSTEM_INSTRUCTIONS, |
| 119 | + "input": prompt, |
| 120 | + } |
| 121 | + ).encode() |
| 122 | + request = urllib.request.Request( |
| 123 | + OPENAI_RESPONSES_URL, |
| 124 | + data=payload, |
| 125 | + headers={ |
| 126 | + "Authorization": f"Bearer {api_key}", |
| 127 | + "Content-Type": "application/json", |
| 128 | + "User-Agent": "LeetCode-Solutions-AI-Notes/1.0", |
| 129 | + }, |
| 130 | + ) |
| 131 | + |
| 132 | + with urllib.request.urlopen(request, timeout=90) as response: |
| 133 | + raw = json.loads(response.read().decode()) |
| 134 | + |
| 135 | + text = extract_response_text(raw) |
| 136 | + if not text: |
| 137 | + raise RuntimeError("OpenAI response did not contain output text.") |
| 138 | + |
| 139 | + parsed = json.loads(strip_code_fences(text)) |
| 140 | + if not isinstance(parsed, dict): |
| 141 | + raise RuntimeError("OpenAI response was not a JSON object.") |
| 142 | + return parsed |
| 143 | + |
| 144 | + |
| 145 | +def read_solution_context(solution_paths: list[str]) -> str: |
| 146 | + snippets: list[str] = [] |
| 147 | + total_chars = 0 |
| 148 | + |
| 149 | + for relative_path in solution_paths[:3]: |
| 150 | + path = ROOT / relative_path |
| 151 | + content = path.read_text() |
| 152 | + snippet = content[:6000] |
| 153 | + snippets.append(f"Path: {relative_path}\n```text\n{snippet}\n```") |
| 154 | + total_chars += len(snippet) |
| 155 | + if total_chars >= 12000: |
| 156 | + break |
| 157 | + |
| 158 | + return "\n\n".join(snippets) |
| 159 | + |
| 160 | + |
| 161 | +def build_prompt(slug: str, metadata: dict[str, Any], solution_paths: list[str]) -> str: |
| 162 | + topic_tags = ", ".join(metadata.get("topic_tags", [])) or "Unknown" |
| 163 | + hints = metadata.get("hints", []) |
| 164 | + hint_lines = "\n".join(f"- {hint}" for hint in hints[:3]) if hints else "- None" |
| 165 | + example_testcases = str(metadata.get("example_testcases") or "").strip() |
| 166 | + if not example_testcases: |
| 167 | + example_testcases = "None" |
| 168 | + |
| 169 | + problem_statement = str(metadata.get("content_markdown") or "").strip() |
| 170 | + if len(problem_statement) > 10000: |
| 171 | + problem_statement = problem_statement[:10000].rstrip() + "\n..." |
| 172 | + |
| 173 | + return f""" |
| 174 | +Problem slug: {slug} |
| 175 | +Title: {metadata["title"]} |
| 176 | +Difficulty: {metadata["difficulty"]} |
| 177 | +Topic tags: {topic_tags} |
| 178 | +LeetCode URL: {metadata["url"]} |
| 179 | +
|
| 180 | +Problem statement: |
| 181 | +{problem_statement} |
| 182 | +
|
| 183 | +Example testcases: |
| 184 | +{example_testcases} |
| 185 | +
|
| 186 | +Hints: |
| 187 | +{hint_lines} |
| 188 | +
|
| 189 | +Accepted solution files: |
| 190 | +{read_solution_context(solution_paths)} |
| 191 | +""".strip() |
| 192 | + |
| 193 | + |
| 194 | +def apply_ai_draft(note_path: Path, draft: dict[str, Any]) -> bool: |
| 195 | + content = note_path.read_text() |
| 196 | + updated = content |
| 197 | + changed = False |
| 198 | + |
| 199 | + if section_needs_update(updated, "Problem Summary"): |
| 200 | + updated = replace_section(updated, "Problem Summary", str(draft.get("summary") or "TODO")) |
| 201 | + changed = True |
| 202 | + |
| 203 | + if section_needs_update(updated, "Data Structures Used"): |
| 204 | + data_structures = draft.get("data_structures") |
| 205 | + body = format_bullets(data_structures if isinstance(data_structures, list) else [], "TODO") |
| 206 | + updated = replace_section(updated, "Data Structures Used", body) |
| 207 | + changed = True |
| 208 | + |
| 209 | + if section_needs_update(updated, "Approach"): |
| 210 | + updated = replace_section(updated, "Approach", str(draft.get("approach") or "TODO")) |
| 211 | + changed = True |
| 212 | + |
| 213 | + if complexity_needs_update(updated): |
| 214 | + updated = replace_complexity( |
| 215 | + updated, |
| 216 | + str(draft.get("time_complexity") or "TODO"), |
| 217 | + str(draft.get("space_complexity") or "TODO"), |
| 218 | + ) |
| 219 | + changed = True |
| 220 | + |
| 221 | + if section_needs_update(updated, "Revision Notes"): |
| 222 | + revision_notes = draft.get("revision_notes") |
| 223 | + body = format_bullets( |
| 224 | + revision_notes if isinstance(revision_notes, list) else [], |
| 225 | + "- TODO", |
| 226 | + ) |
| 227 | + updated = replace_section(updated, "Revision Notes", body) |
| 228 | + changed = True |
| 229 | + |
| 230 | + if changed: |
| 231 | + note_path.write_text(updated) |
| 232 | + return changed |
| 233 | + |
| 234 | + |
| 235 | +def parse_args() -> argparse.Namespace: |
| 236 | + parser = argparse.ArgumentParser(description="Generate AI note drafts for synced LeetCode solutions.") |
| 237 | + parser.add_argument("slugs", nargs="*", help="Optional problem slugs to target.") |
| 238 | + return parser.parse_args() |
| 239 | + |
| 240 | + |
| 241 | +def main() -> int: |
| 242 | + args = parse_args() |
| 243 | + api_key = os.getenv("OPENAI_API_KEY") |
| 244 | + if not api_key: |
| 245 | + print("OPENAI_API_KEY not set; skipping AI note generation.") |
| 246 | + return 0 |
| 247 | + |
| 248 | + model = os.getenv("OPENAI_MODEL", DEFAULT_MODEL) |
| 249 | + discovered = discover_problem_solutions() |
| 250 | + target_slugs = ( |
| 251 | + {canonical_slug(slug) for slug in args.slugs if canonical_slug(slug)} |
| 252 | + if args.slugs |
| 253 | + else set(discovered) |
| 254 | + ) |
| 255 | + |
| 256 | + if not target_slugs: |
| 257 | + print("No synced problems found.") |
| 258 | + return 0 |
| 259 | + |
| 260 | + sync_problem_notes(target_slugs) |
| 261 | + cache = load_metadata_cache() |
| 262 | + |
| 263 | + updated_count = 0 |
| 264 | + for slug in sorted(target_slugs): |
| 265 | + solution_paths = discovered.get(slug) |
| 266 | + if not solution_paths: |
| 267 | + continue |
| 268 | + |
| 269 | + note_path = note_path_for_slug(slug) |
| 270 | + if not note_path.exists(): |
| 271 | + continue |
| 272 | + |
| 273 | + note_content = note_path.read_text() |
| 274 | + if not any(section_needs_update(note_content, heading) for heading in TARGET_SECTIONS if heading != "Complexity") and not complexity_needs_update(note_content): |
| 275 | + continue |
| 276 | + |
| 277 | + metadata = get_problem_metadata(slug, cache) |
| 278 | + prompt = build_prompt(slug, metadata, solution_paths) |
| 279 | + |
| 280 | + try: |
| 281 | + draft = request_note_draft(prompt, api_key, model) |
| 282 | + except (RuntimeError, urllib.error.URLError, json.JSONDecodeError) as exc: |
| 283 | + print(f"Skipping {slug}: {exc}") |
| 284 | + continue |
| 285 | + |
| 286 | + if apply_ai_draft(note_path, draft): |
| 287 | + updated_count += 1 |
| 288 | + print(f"Updated {note_path}") |
| 289 | + |
| 290 | + save_metadata_cache(cache) |
| 291 | + print(f"AI-generated drafts updated: {updated_count}") |
| 292 | + return 0 |
| 293 | + |
| 294 | + |
| 295 | +if __name__ == "__main__": |
| 296 | + raise SystemExit(main()) |
0 commit comments