| 1 | from __future__ import annotations |
| 2 | |
| 3 | import argparse |
| 4 | import csv |
| 5 | import re |
| 6 | import warnings |
| 7 | from pathlib import Path |
| 8 | |
| 9 | import yaml |
| 10 | |
| 11 | from scripts.sanitize_repo_content import INJECTION_PHRASES |
| 12 | from scripts.topic_paths import analyzed_dir |
| 13 | |
| 14 | FRONTMATTER_PATTERN = re.compile(r"^---\n(.*?)\n---\n(.*)\Z", re.DOTALL) |
| 15 | WEEK_PATTERN = re.compile(r"^(?P<year>\d{4})-W(?P<week>\d{2})$") |
| 16 | SUMMARY_SUFFIX = "-summary.md" |
| 17 | ANALYSIS_SUFFIX = " Analysis" |
| 18 | REQUIRED_ANALYSIS_FIELDS = { |
| 19 | "title", |
| 20 | "date", |
| 21 | "week", |
| 22 | "year", |
| 23 | "tags", |
| 24 | "categories", |
| 25 | "repos_featured", |
| 26 | "stars_tracked", |
| 27 | "top_repo", |
| 28 | "quality_score", |
| 29 | "summary", |
| 30 | } |
| 31 | |
| 32 | # Defense-in-depth: max lengths for frontmatter fields even though upstream |
| 33 | # sanitization should already have applied limits. |
| 34 | _FIELD_MAX_LENGTHS: dict[str, int] = { |
| 35 | "title": 300, |
| 36 | "summary": 1000, |
| 37 | "top_repo": 200, |
| 38 | } |
| 39 | |
| 40 | _INJECTION_PHRASES = INJECTION_PHRASES |
| 41 | |
| 42 | |
| 43 | class GenerationError(ValueError): |
| 44 | pass |
| 45 | |
| 46 | |
| 47 | def _validate_frontmatter_safety(frontmatter: dict[str, object]) -> None: |
| 48 | """Defense-in-depth check: reject frontmatter with injection artifacts.""" |
| 49 | for field, max_len in _FIELD_MAX_LENGTHS.items(): |
| 50 | value = frontmatter.get(field) |
| 51 | if value is None: |
| 52 | continue |
| 53 | # Coerce to string for validation (mirrors transform_summary's str() calls) |
| 54 | if not isinstance(value, str): |
| 55 | value = str(value) |
| 56 | if len(value) > max_len: |
| 57 | raise GenerationError( |
| 58 | f"Frontmatter field '{field}' exceeds safe length " |
| 59 | f"({len(value)} > {max_len}). Possible injection artifact." |
| 60 | ) |
| 61 | lowered = value.lower() |
| 62 | for phrase in _INJECTION_PHRASES: |
| 63 | if phrase in lowered: |
| 64 | raise GenerationError( |
| 65 | f"Frontmatter field '{field}' contains suspicious phrase " |
| 66 | f"'{phrase}'. Possible prompt injection artifact." |
| 67 | ) |
| 68 | |
| 69 | |
| 70 | def parse_args() -> argparse.Namespace: |
| 71 | parser = argparse.ArgumentParser( |
| 72 | description="Generate a Hugo weekly content page from an analyzed summary markdown file." |
| 73 | ) |
| 74 | parser.add_argument( |
| 75 | "summary", |
| 76 | nargs="?", |
| 77 | default=None, |
| 78 | help="Path to data/analyzed/YYYY-WNN-summary.md. Defaults to the newest analyzed summary.", |
| 79 | ) |
| 80 | parser.add_argument( |
| 81 | "--output", |
| 82 | default=None, |
| 83 | help="Optional explicit output path. Defaults to content/weekly/YYYY/WNN.md.", |
| 84 | ) |
| 85 | return parser.parse_args() |
| 86 | |
| 87 | |
| 88 | def parse_week(value: str) -> tuple[int, int]: |
| 89 | match = WEEK_PATTERN.fullmatch(value) |
| 90 | if not match: |
| 91 | raise GenerationError(f"Invalid week value: {value}") |
| 92 | return int(match.group("year")), int(match.group("week")) |
| 93 | |
| 94 | |
| 95 | def week_from_summary_path(path: Path) -> tuple[int, int]: |
| 96 | if not path.name.endswith(SUMMARY_SUFFIX): |
| 97 | raise GenerationError(f"Invalid summary filename: {path.name}") |
| 98 | return parse_week(path.name.removesuffix(SUMMARY_SUFFIX)) |
| 99 | |
| 100 | |
| 101 | def find_latest_summary(root: Path, topic_id: str | None = None) -> Path: |
| 102 | search_dir = analyzed_dir(topic_id) |
| 103 | # When using default (relative) path, resolve via root for backward compat |
| 104 | if topic_id is None: |
| 105 | candidates = list(root.glob(f"data/analyzed/*{SUMMARY_SUFFIX}")) |
| 106 | else: |
| 107 | candidates = list(search_dir.glob(f"*{SUMMARY_SUFFIX}")) |
| 108 | if not candidates: |
| 109 | # Fallback: try the other approach |
| 110 | candidates = ( |
| 111 | list(search_dir.glob(f"*{SUMMARY_SUFFIX}")) |
| 112 | if topic_id is None |
| 113 | else list(root.glob(f"data/analyzed/*{SUMMARY_SUFFIX}")) |
| 114 | ) |
| 115 | if not candidates: |
| 116 | raise GenerationError("No analyzed summaries found under data/analyzed/.") |
| 117 | return max(candidates, key=week_from_summary_path) |
| 118 | |
| 119 | |
| 120 | def parse_scalar(value: str): |
| 121 | value = value.strip() |
| 122 | if not value: |
| 123 | return "" |
| 124 | if value.startswith("[") and value.endswith("]"): |
| 125 | inner = value[1:-1].strip() |
| 126 | if not inner: |
| 127 | return [] |
| 128 | return [ |
| 129 | item.strip().strip('"').strip("'") |
| 130 | for item in csv.reader([inner], skipinitialspace=True).__next__() |
| 131 | ] |
| 132 | if value.startswith(('"', "'")) and value.endswith(('"', "'")): |
| 133 | return value[1:-1] |
| 134 | if re.fullmatch(r"-?\d+", value): |
| 135 | return int(value) |
| 136 | if value.lower() == "true": |
| 137 | return True |
| 138 | if value.lower() == "false": |
| 139 | return False |
| 140 | return value |
| 141 | |
| 142 | |
| 143 | def parse_frontmatter(document: str) -> tuple[dict[str, object], str]: |
| 144 | match = FRONTMATTER_PATTERN.match(document) |
| 145 | if not match: |
| 146 | raise GenerationError("Summary is missing YAML frontmatter.") |
| 147 | |
| 148 | frontmatter_text, body = match.groups() |
| 149 | try: |
| 150 | frontmatter = yaml.safe_load(frontmatter_text) |
| 151 | except yaml.YAMLError as exc: |
| 152 | raise GenerationError(f"Invalid YAML frontmatter: {exc}") from exc |
| 153 | if not isinstance(frontmatter, dict): |
| 154 | raise GenerationError("Frontmatter must be a YAML mapping.") |
| 155 | |
| 156 | missing = REQUIRED_ANALYSIS_FIELDS.difference(frontmatter) |
| 157 | if missing: |
| 158 | raise GenerationError(f"Missing required analysis fields: {', '.join(sorted(missing))}") |
| 159 | |
| 160 | return frontmatter, body.strip() + "\n" |
| 161 | |
| 162 | |
| 163 | def normalize_title(title: str) -> str: |
| 164 | if title.endswith(ANALYSIS_SUFFIX): |
| 165 | return title[: -len(ANALYSIS_SUFFIX)] |
| 166 | return title |
| 167 | |
| 168 | |
| 169 | def ensure_list(value: object, *, field_name: str) -> list[str]: |
| 170 | if not isinstance(value, list) or any(not isinstance(item, str) or not item for item in value): |
| 171 | raise GenerationError(f"{field_name} must be a non-empty list of strings.") |
| 172 | return value |
| 173 | |
| 174 | |
| 175 | def infer_output_path(week: str, root: Path) -> Path: |
| 176 | year, week_number = parse_week(week) |
| 177 | return root / "content" / "weekly" / str(year) / f"W{week_number:02d}.md" |
| 178 | |
| 179 | |
| 180 | def yaml_quote(value: str) -> str: |
| 181 | return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"' |
| 182 | |
| 183 | |
| 184 | def optional_string(value: object) -> str: |
| 185 | return "" if value is None else str(value) |
| 186 | |
| 187 | |
| 188 | def is_local_asset_path(value: object) -> bool: |
| 189 | if not isinstance(value, str) or not value: |
| 190 | return False |
| 191 | if value.startswith(("http://", "https://", "//")): |
| 192 | return False |
| 193 | asset_path = Path(value) |
| 194 | if asset_path.is_absolute(): |
| 195 | return False |
| 196 | if ".." in asset_path.parts: |
| 197 | return False |
| 198 | return True |
| 199 | |
| 200 | |
| 201 | def render_frontmatter(data: dict[str, object]) -> str: |
| 202 | lines = [ |
| 203 | "---", |
| 204 | f"title: {yaml_quote(str(data['title']))}", |
| 205 | f"date: {data['date']}", |
| 206 | f"week: {yaml_quote(str(data['week']))}", |
| 207 | f"tags: [{', '.join(yaml_quote(t) for t in data['tags'])}]", |
| 208 | f"categories: [{', '.join(yaml_quote(c) for c in data['categories'])}]", |
| 209 | f"repos_featured: {data['repos_featured']}", |
| 210 | f"stars_tracked: {data['stars_tracked']}", |
| 211 | f"top_repo: {yaml_quote(str(data['top_repo']))}", |
| 212 | f"summary: {yaml_quote(str(data['summary']))}", |
| 213 | "draft: false", |
| 214 | ] |
| 215 | |
| 216 | # Cover image frontmatter (PaperMod convention) |
| 217 | cover = data.get("cover") |
| 218 | if cover and isinstance(cover, dict): |
| 219 | lines.append("cover:") |
| 220 | if cover.get("image"): |
| 221 | lines.append(f" image: {yaml_quote(str(cover['image']))}") |
| 222 | if cover.get("alt"): |
| 223 | lines.append(f" alt: {yaml_quote(str(cover['alt']))}") |
| 224 | if cover.get("caption"): |
| 225 | lines.append(f" caption: {yaml_quote(str(cover['caption']))}") |
| 226 | if cover.get("attribution"): |
| 227 | lines.append(f" attribution: {yaml_quote(str(cover['attribution']))}") |
| 228 | if cover.get("license"): |
| 229 | lines.append(f" license: {yaml_quote(str(cover['license']))}") |
| 230 | lines.append(" relative: false") |
| 231 | |
| 232 | # Explicit OG image override |
| 233 | if data.get("og_image"): |
| 234 | lines.append(f"og_image: {yaml_quote(str(data['og_image']))}") |
| 235 | |
| 236 | lines.extend(["---", ""]) |
| 237 | return "\n".join(lines) |
| 238 | |
| 239 | |
| 240 | def transform_summary(frontmatter: dict[str, object], body: str) -> str: |
| 241 | _validate_frontmatter_safety(frontmatter) |
| 242 | tags = ensure_list(frontmatter["tags"], field_name="tags") |
| 243 | categories = ensure_list(frontmatter["categories"], field_name="categories") |
| 244 | if "weekly" not in categories: |
| 245 | categories = [*categories, "weekly"] |
| 246 | |
| 247 | page_frontmatter: dict[str, object] = { |
| 248 | "title": normalize_title(str(frontmatter["title"])), |
| 249 | "date": str(frontmatter["date"]), |
| 250 | "week": str(frontmatter["week"]), |
| 251 | "tags": tags, |
| 252 | "categories": categories, |
| 253 | "repos_featured": int(frontmatter["repos_featured"]), |
| 254 | "stars_tracked": int(frontmatter["stars_tracked"]), |
| 255 | "top_repo": str(frontmatter["top_repo"]), |
| 256 | "summary": str(frontmatter["summary"]), |
| 257 | } |
| 258 | |
| 259 | cover = frontmatter.get("cover") if isinstance(frontmatter.get("cover"), dict) else {} |
| 260 | cover_image = frontmatter.get("cover_image") or cover.get("image") |
| 261 | if cover_image: |
| 262 | if is_local_asset_path(cover_image): |
| 263 | cover_alt = frontmatter.get("cover_alt") or cover.get("alt") |
| 264 | cover_attribution = frontmatter.get("cover_attribution") or cover.get("attribution") |
| 265 | cover_license = frontmatter.get("cover_license") or cover.get("license") |
| 266 | page_frontmatter["cover"] = { |
| 267 | "image": str(cover_image), |
| 268 | "alt": optional_string(cover_alt), |
| 269 | "attribution": optional_string(cover_attribution), |
| 270 | "license": optional_string(cover_license), |
| 271 | } |
| 272 | else: |
| 273 | warnings.warn(f"Skipping non-local cover image value: {cover_image}", stacklevel=2) |
| 274 | |
| 275 | # Pass through OG image override — reject URLs to enforce no-hotlinking policy |
| 276 | og_image = frontmatter.get("og_image") |
| 277 | if og_image: |
| 278 | if is_local_asset_path(og_image): |
| 279 | page_frontmatter["og_image"] = str(og_image) |
| 280 | else: |
| 281 | warnings.warn(f"Skipping non-local og_image value: {og_image}", stacklevel=2) |
| 282 | |
| 283 | return render_frontmatter(page_frontmatter) + "\n" + body.lstrip() |
| 284 | |
| 285 | |
| 286 | def generate_content(summary_path: Path, output_path: Path | None = None) -> Path: |
| 287 | root = Path.cwd() |
| 288 | document = summary_path.read_text(encoding="utf-8") |
| 289 | frontmatter, body = parse_frontmatter(document) |
| 290 | target_path = output_path or infer_output_path(str(frontmatter["week"]), root) |
| 291 | target_path.parent.mkdir(parents=True, exist_ok=True) |
| 292 | target_path.write_text(transform_summary(frontmatter, body), encoding="utf-8") |
| 293 | return target_path |
| 294 | |
| 295 | |
| 296 | def main() -> int: |
| 297 | args = parse_args() |
| 298 | root = Path.cwd() |
| 299 | summary_path = Path(args.summary) if args.summary else find_latest_summary(root) |
| 300 | output_path = Path(args.output) if args.output else None |
| 301 | |
| 302 | if not summary_path.exists(): |
| 303 | raise SystemExit(f"Summary file not found: {summary_path}") |
| 304 | |
| 305 | written_path = generate_content(summary_path, output_path) |
| 306 | print(f"Generated {written_path} from {summary_path}") |
| 307 | return 0 |
| 308 | |
| 309 | |
| 310 | if __name__ == "__main__": |
| 311 | raise SystemExit(main()) |