| 1 | # CI Data Source Integration Pattern |
| 2 | |
| 3 | confidence: high |
| 4 | discovered_by: Farnsworth (TechCrunch integration), Bender (implementation) |
| 5 | date: 2026-05-19 |
| 6 | |
| 7 | ## Pattern |
| 8 | |
| 9 | Scripts often exist but aren't wired into the CI pipeline. Prevent script-orphaning by following this pattern: |
| 10 | |
| 11 | 1. **Define DataSource adapter** with standardized interface: |
| 12 | - `get_name()` → source name (e.g., "techcrunch", "github") |
| 13 | - `get_rate_limits()` → rate limit policy |
| 14 | - `crawl(since, until)` → structured output (list of dicts) |
| 15 | |
| 16 | 2. **Wire script into workflow** immediately after creation: |
| 17 | - Add explicit step in CI that calls the script |
| 18 | - Set input parameters (dates, topics, output paths) |
| 19 | - Capture exit codes and log output |
| 20 | - Integrate output into next pipeline stage |
| 21 | |
| 22 | 3. **Document integration point** in PRD: |
| 23 | - Which workflow file calls it |
| 24 | - Input parameters and environment variables |
| 25 | - Output format and schema |
| 26 | - Rate limit behavior and retry policy |
| 27 | |
| 28 | 4. **Test the wire** before PR merge: |
| 29 | - Run the workflow end-to-end |
| 30 | - Verify script actually executes (not skipped by conditions) |
| 31 | - Check output format matches downstream consumer expectations |
| 32 | |
| 33 | ## When to Use |
| 34 | |
| 35 | - Creating new data crawlers (RSS, APIs, GitHub) |
| 36 | - Adding new analysis stages (preprocessing, enrichment) |
| 37 | - Integrating external tools or scripts into CI/CD |
| 38 | - Multi-stage pipelines where data flows from stage to stage |
| 39 | |
| 40 | ## Implementation |
| 41 | |
| 42 | ### DataSource Adapter Pattern |
| 43 | |
| 44 | ```python |
| 45 | class TechCrunchSource: |
| 46 | """TechCrunch RSS data source following the DataSource protocol.""" |
| 47 | |
| 48 | def get_name(self) -> str: |
| 49 | return "techcrunch" |
| 50 | |
| 51 | def get_rate_limits(self) -> dict: |
| 52 | return {"requests_per_minute": 10} |
| 53 | |
| 54 | def crawl( |
| 55 | self, |
| 56 | since: datetime, |
| 57 | until: datetime, |
| 58 | feed_url: str = FEED_URL, |
| 59 | ) -> list[dict[str, Any]]: |
| 60 | """Crawl TechCrunch RSS feed and return structured articles.""" |
| 61 | feed = fetch_feed(feed_url) |
| 62 | articles: list[dict[str, Any]] = [] |
| 63 | |
| 64 | for entry in feed.entries: |
| 65 | pub_date = parse_published_date(entry) |
| 66 | if pub_date is None or pub_date < since or pub_date >= until: |
| 67 | continue |
| 68 | |
| 69 | article = { |
| 70 | "title": getattr(entry, "title", ""), |
| 71 | "url": getattr(entry, "link", ""), |
| 72 | "published_at": iso_timestamp(pub_date), |
| 73 | "categories": extract_categories(entry), |
| 74 | "summary": extract_summary(entry), |
| 75 | "github_links": extract_github_urls(entry), |
| 76 | "entities": extract_entities(entry.title), |
| 77 | } |
| 78 | article["relevance_score"] = compute_relevance_score(article) |
| 79 | articles.append(article) |
| 80 | |
| 81 | return articles |
| 82 | ``` |
| 83 | |
| 84 | ### Workflow Integration |
| 85 | |
| 86 | ```yaml |
| 87 | crawl-techcrunch: |
| 88 | runs-on: ubuntu-latest |
| 89 | steps: |
| 90 | - uses: actions/checkout@v4 |
| 91 | |
| 92 | - name: Set up Python |
| 93 | uses: actions/setup-python@v4 |
| 94 | with: |
| 95 | python-version: "3.11" |
| 96 | |
| 97 | - name: Install dependencies |
| 98 | run: pip install -r requirements.txt |
| 99 | |
| 100 | - name: Crawl TechCrunch RSS |
| 101 | env: |
| 102 | TOPIC: ai-ml |
| 103 | OUTPUT: data/raw/ai-ml/${{ needs.weekly.outputs.week }}-techcrunch.json |
| 104 | run: python scripts/techcrunch_crawler.py \ |
| 105 | --topic "$TOPIC" \ |
| 106 | --output "$OUTPUT" \ |
| 107 | --since "${{ needs.weekly.outputs.since }}" \ |
| 108 | --until "${{ needs.weekly.outputs.until }}" |
| 109 | |
| 110 | - name: Upload crawl results |
| 111 | uses: actions/upload-artifact@v3 |
| 112 | with: |
| 113 | name: techcrunch-crawl |
| 114 | path: data/raw/ |
| 115 | retention-days: 7 |
| 116 | ``` |
| 117 | |
| 118 | ### Output Schema Documentation |
| 119 | |
| 120 | ```markdown |
| 121 | ## TechCrunch Crawler Output |
| 122 | |
| 123 | **File:** `data/raw/{topic}/{week}-techcrunch.json` |
| 124 | |
| 125 | **Schema:** |
| 126 | ```json |
| 127 | { |
| 128 | "week": "2026-W21", |
| 129 | "source": "techcrunch", |
| 130 | "crawled_at": "2026-05-19T19:31:31Z", |
| 131 | "articles": [ |
| 132 | { |
| 133 | "title": "...", |
| 134 | "url": "https://techcrunch.com/...", |
| 135 | "published_at": "2026-05-19T12:00:00Z", |
| 136 | "categories": ["ai", "ml"], |
| 137 | "summary": "...", |
| 138 | "github_links": ["https://github.com/owner/repo"], |
| 139 | "entities": ["OpenAI", "Anthropic"], |
| 140 | "relevance_score": 0.85 |
| 141 | } |
| 142 | ], |
| 143 | "metadata": { |
| 144 | "total_articles": 250, |
| 145 | "relevant_articles": 45, |
| 146 | "github_links_found": 12 |
| 147 | } |
| 148 | } |
| 149 | ``` |
| 150 | ``` |
| 151 | |
| 152 | ## Examples |
| 153 | |
| 154 | From `scripts/techcrunch_crawler.py`: |
| 155 | |
| 156 | ```python |
| 157 | def main(argv: list[str] | None = None) -> int: |
| 158 | parser = argparse.ArgumentParser( |
| 159 | description="Crawl TechCrunch RSS feed for SquadScope" |
| 160 | ) |
| 161 | parser.add_argument("--topic", default="general") |
| 162 | parser.add_argument("--output", default=None) |
| 163 | parser.add_argument("--since", default=None) |
| 164 | parser.add_argument("--until", default=None) |
| 165 | args = parser.parse_args(argv) |
| 166 | |
| 167 | now = datetime.now(UTC) |
| 168 | since = ( |
| 169 | datetime.strptime(args.since, "%Y-%m-%d").replace(tzinfo=UTC) |
| 170 | if args.since |
| 171 | else now - timedelta(days=7) |
| 172 | ) |
| 173 | until = ( |
| 174 | datetime.strptime(args.until, "%Y-%m-%d").replace(tzinfo=UTC) |
| 175 | if args.until |
| 176 | else now |
| 177 | ) |
| 178 | |
| 179 | source = TechCrunchSource() |
| 180 | articles = source.crawl(since=since, until=until) |
| 181 | output = build_output(articles, crawled_at=now) |
| 182 | |
| 183 | if args.output: |
| 184 | out_path = Path(args.output) |
| 185 | else: |
| 186 | out_dir = raw_dir(args.topic) |
| 187 | out_dir.mkdir(parents=True, exist_ok=True) |
| 188 | out_path = out_dir / f"{week_slug(now)}-techcrunch.json" |
| 189 | |
| 190 | out_path.parent.mkdir(parents=True, exist_ok=True) |
| 191 | with open(out_path, "w", encoding="utf-8") as f: |
| 192 | json.dump(output, f, indent=2, ensure_ascii=False) |
| 193 | |
| 194 | print(f"Crawled {output['metadata']['total_articles']} articles → {out_path}") |
| 195 | return 0 |
| 196 | ``` |
| 197 | |
| 198 | ### Config-Driven Parallel RSS Sources |
| 199 | |
| 200 | For small sets of external RSS feeds in the weekly Actions pipeline, prefer one config file plus bounded in-process parallel fetches over one job per feed. This avoids repeated checkout/setup/artifact overhead, keeps a single enrichment artifact contract, and lets maintainers add or remove sources without editing workflow topology. |
| 201 | |
| 202 | ## Notes |
| 203 | |
| 204 | - Standardize output schemas across all data sources for seamless pipeline integration |
| 205 | - Test scripts locally before adding to workflow to catch parameter/path issues |
| 206 | - Document rate limit behavior so workflow can be tuned for cost/speed tradeoffs |
| 207 | - Use artifact uploads to pass data between workflow jobs (cleaner than file system) |